Compare commits
11
Commits
6f76b3f15a
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a1d2b846fd | ||
|
|
30dc340a10 | ||
|
|
e6f3f370da | ||
|
|
cffc26d98e | ||
|
|
ff127a8c21 | ||
|
|
97cbf38d97 | ||
|
|
76b5f827be | ||
|
|
acd63f570d | ||
|
|
8c34a0f0c2 | ||
|
|
0a8f9f45bd | ||
|
|
eb7c1aa500 |
@@ -2,3 +2,4 @@ __pycache__/
|
||||
*.pyc
|
||||
.venv/
|
||||
test_media/seek_validation/
|
||||
test_media/*_fixed*
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
# WeAreSyncing
|
||||
Fix audio synchronization by detecting and marking natural 'claps' in the video, or automatically determine the offset using the difference between the audio and video stream ends.
|
||||
|
||||
This utility was vibe coded with Claude Code AI, proceed at your own discretion.
|
||||
@@ -8,3 +8,4 @@ Terminal=false
|
||||
Categories=AudioVideo;Video;
|
||||
MimeType=video/mp4;video/quicktime;video/x-matroska;video/x-msvideo;video/webm;video/mpeg;
|
||||
StartupNotify=true
|
||||
Name[en_US]=WeAreSyncing
|
||||
+62
-10
@@ -5,6 +5,7 @@ paths use one implementation instead of drifting apart.
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import warnings
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional
|
||||
|
||||
@@ -18,16 +19,22 @@ class ClapDetectionUnavailable(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
# Shared with the GUI's detection-settings dialog (main.py) so both the CLI
|
||||
# and GUI default to, and can be reset to, the exact same values.
|
||||
DEFAULT_THRESHOLD = 0.7
|
||||
DEFAULT_MIN_INTERVAL = 0.3
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ClapEvent:
|
||||
time_sec: float
|
||||
strength: float
|
||||
|
||||
|
||||
def _probe_audio_start_time(input_path: str) -> float:
|
||||
def _probe_audio_start_time(input_path: str, stream_specifier: str) -> Optional[float]:
|
||||
result = subprocess.run(
|
||||
[
|
||||
"ffprobe", "-v", "error", "-select_streams", "a:0",
|
||||
"ffprobe", "-v", "error", "-select_streams", stream_specifier,
|
||||
"-show_entries", "stream=start_time",
|
||||
"-of", "default=noprint_wrappers=1:nokey=1",
|
||||
input_path,
|
||||
@@ -37,10 +44,20 @@ def _probe_audio_start_time(input_path: str) -> float:
|
||||
try:
|
||||
return float(result.stdout.strip())
|
||||
except ValueError:
|
||||
return 0.0
|
||||
# Some containers/codecs legitimately report "N/A" (genuinely no
|
||||
# start offset to speak of), but this could also mean ffprobe just
|
||||
# couldn't express a real nonzero offset for this container -- we
|
||||
# can't tell those apart here, so surface it instead of silently
|
||||
# assuming 0, which would leave onset times subtly, silently wrong.
|
||||
warnings.warn(
|
||||
f"could not determine audio start_time for {input_path!r} "
|
||||
f"(ffprobe returned {result.stdout.strip()!r}); assuming 0.0, "
|
||||
"which may be wrong for this file's container/codec."
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _extract_audio_wav(input_path: str) -> str:
|
||||
def _extract_audio_wav(input_path: str, audio_stream_index: Optional[int] = None) -> str:
|
||||
# The audio stream's own start_time is exactly how a container encodes
|
||||
# "this track starts playing N seconds later than the others" -- i.e.
|
||||
# it's the very thing this whole tool exists to fix. ffmpeg's default
|
||||
@@ -48,14 +65,44 @@ def _extract_audio_wav(input_path: str) -> str:
|
||||
# which would silently make onset detection blind to that offset.
|
||||
# Compensate by inserting the missing lead-in as real silence, so times
|
||||
# measured from this WAV's t=0 still match the file's true timeline.
|
||||
audio_start = _probe_audio_start_time(input_path)
|
||||
#
|
||||
# audio_stream_index (ffprobe/ffmpeg global stream index, e.g. mpv's
|
||||
# own `ff-index` for the track it's currently playing) makes stream
|
||||
# selection explicit and -- critically -- lets the caller guarantee
|
||||
# this matches whichever audio track is actually audible in the
|
||||
# player. Without it, ffmpeg's own default-stream heuristic picks
|
||||
# *a* audio track, which isn't guaranteed to be the same one on
|
||||
# multi-audio-track files (extra language dubs, a separate mic/system
|
||||
# audio track from a screen recorder, etc.) -- something our own
|
||||
# single-track synthetic test fixtures never exercise, but real-world
|
||||
# files often have.
|
||||
stream_specifier = f"{audio_stream_index}" if audio_stream_index is not None else "a:0"
|
||||
audio_start = _probe_audio_start_time(input_path, stream_specifier) or 0.0
|
||||
|
||||
fd, wav_path = tempfile.mkstemp(suffix=".wav")
|
||||
os.close(fd)
|
||||
cmd = ["ffmpeg", "-y", "-i", input_path, "-vn"]
|
||||
cmd += ["-map", f"0:{audio_stream_index}" if audio_stream_index is not None else "a:0"]
|
||||
# aresample=async=1 keeps the decoded audio conformed to its own
|
||||
# internal PTS, filling any gap with silence instead of silently
|
||||
# closing it up. Long recordings (stream captures, OBS-style long
|
||||
# sessions) can have a genuine multi-second discontinuity mid-file --
|
||||
# e.g. a brief audio-device dropout -- from a real capture glitch, not
|
||||
# anything to do with sync between tracks. Plain -vn extraction just
|
||||
# concatenates packets and ignores the gap, silently shortening the
|
||||
# output and shifting every onset after that point earlier by however
|
||||
# long the gap was. Confirmed on a real ~110min capture with exactly
|
||||
# one such gap (+1.621s at ~15.5min in): without this filter the
|
||||
# extracted WAV came out 1.6s short of the source's true duration;
|
||||
# with it, the duration matches exactly. This is a different failure
|
||||
# mode from the leading start_time offset below (which aresample alone
|
||||
# does NOT fix -- it re-zeros to its own first frame, same as plain
|
||||
# extraction), so both are needed together, not either/or.
|
||||
af_chain = ["aresample=async=1"]
|
||||
if audio_start > 0.001:
|
||||
delay_ms = int(round(audio_start * 1000))
|
||||
cmd += ["-af", f"adelay={delay_ms}|{delay_ms}"]
|
||||
af_chain.append(f"adelay={delay_ms}|{delay_ms}")
|
||||
cmd += ["-af", ",".join(af_chain)]
|
||||
cmd += ["-acodec", "pcm_s16le", "-ar", "22050", "-ac", "1", wav_path]
|
||||
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
@@ -68,14 +115,19 @@ def _extract_audio_wav(input_path: str) -> str:
|
||||
def detect_claps(
|
||||
input_path: str,
|
||||
start_offset: float = 0.0,
|
||||
threshold: float = 0.7,
|
||||
min_interval: float = 0.3,
|
||||
threshold: float = DEFAULT_THRESHOLD,
|
||||
min_interval: float = DEFAULT_MIN_INTERVAL,
|
||||
max_results: Optional[int] = None,
|
||||
audio_stream_index: Optional[int] = None,
|
||||
) -> List[ClapEvent]:
|
||||
"""Detect sharp audio onsets in input_path. `start_offset`: only onsets
|
||||
at or after this absolute timestamp are returned (for a from-cursor
|
||||
search). `max_results` caps how many are returned, for a "find the next
|
||||
one" search instead of a full scan.
|
||||
one" search instead of a full scan. `audio_stream_index` (an ffmpeg
|
||||
global stream index) pins detection to a specific audio track on
|
||||
multi-audio-track files -- pass the ff-index of whichever track mpv is
|
||||
actually playing (see main.py) so detection can't silently analyze a
|
||||
different track than the one being listened to.
|
||||
|
||||
The whole file is always decoded and analyzed regardless of
|
||||
start_offset -- an earlier version tried to skip ahead with ffmpeg's
|
||||
@@ -92,7 +144,7 @@ def detect_claps(
|
||||
"librosa is not installed. Install with: pip install librosa numpy soundfile"
|
||||
)
|
||||
|
||||
wav_path = _extract_audio_wav(input_path)
|
||||
wav_path = _extract_audio_wav(input_path, audio_stream_index=audio_stream_index)
|
||||
try:
|
||||
y, sr = librosa.load(wav_path, sr=22050)
|
||||
onset_env = librosa.onset.onset_strength(y=y, sr=sr)
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
@@ -6,53 +8,87 @@ from PyQt6.QtCore import Qt, QTimer, QThread, pyqtSignal
|
||||
from PyQt6.QtGui import QShortcut, QKeySequence, QAction
|
||||
from PyQt6.QtWidgets import (
|
||||
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
|
||||
QPushButton, QLabel, QSlider, QCheckBox, QDoubleSpinBox, QFileDialog,
|
||||
QMessageBox, QPlainTextEdit, QGroupBox,
|
||||
QPushButton, QLabel, QDoubleSpinBox, QFileDialog,
|
||||
QMessageBox, QPlainTextEdit, QGroupBox, QTabWidget, QDialog, QDialogButtonBox, QFormLayout,
|
||||
)
|
||||
|
||||
from player_widget import PlayerWidget
|
||||
from player_panel import PlayerPanel, format_timestamp
|
||||
from sync_logic import compute_offset, SyncOffset
|
||||
from ffmpeg_export import export_synced, FfmpegExportError, ExportResult
|
||||
from clap_detection import detect_claps, ClapDetectionUnavailable, ClapEvent
|
||||
from marker_strip import MarkerStrip
|
||||
from clap_detection import (
|
||||
detect_claps, ClapDetectionUnavailable, ClapEvent,
|
||||
DEFAULT_THRESHOLD, DEFAULT_MIN_INTERVAL,
|
||||
)
|
||||
from stream_edges import probe_stream_edges, StreamEdges, StreamEdgesError
|
||||
|
||||
|
||||
def format_timestamp(seconds: float) -> str:
|
||||
m, s = divmod(max(seconds, 0.0), 60)
|
||||
return f"{int(m):02d}:{s:06.3f}"
|
||||
class ClapSettingsDialog(QDialog):
|
||||
"""Lets the threshold/min-interval knobs from clap_detection.py be
|
||||
tuned from the GUI instead of only via the CLI's -t/-m flags. Keeps its
|
||||
own pending values so Cancel doesn't touch the caller's settings."""
|
||||
|
||||
def __init__(self, threshold: float, min_interval: float, parent=None):
|
||||
super().__init__(parent)
|
||||
self.setWindowTitle("Clap Detection Settings")
|
||||
|
||||
class ClickableSlider(QSlider):
|
||||
"""QSlider normally only jumps to the clicked position if you click
|
||||
exactly on the handle; clicking elsewhere in the groove pages by a
|
||||
fixed step. Override so any click along the track jumps straight
|
||||
there, then let the base class take over so dragging still works
|
||||
(the handle is now under the cursor, so its own drag tracking picks
|
||||
up seamlessly).
|
||||
"""
|
||||
self.threshold_spin = QDoubleSpinBox()
|
||||
self.threshold_spin.setRange(0.1, 2.0)
|
||||
self.threshold_spin.setSingleStep(0.05)
|
||||
self.threshold_spin.setDecimals(2)
|
||||
self.threshold_spin.setValue(threshold)
|
||||
|
||||
def mousePressEvent(self, event):
|
||||
if event.button() == Qt.MouseButton.LeftButton and self.maximum() > self.minimum():
|
||||
ratio = min(max(event.position().x() / max(self.width(), 1), 0.0), 1.0)
|
||||
value = round(self.minimum() + ratio * (self.maximum() - self.minimum()))
|
||||
self.setValue(value)
|
||||
self.sliderMoved.emit(value)
|
||||
super().mousePressEvent(event)
|
||||
self.min_interval_spin = QDoubleSpinBox()
|
||||
self.min_interval_spin.setRange(0.05, 5.0)
|
||||
self.min_interval_spin.setSingleStep(0.05)
|
||||
self.min_interval_spin.setDecimals(2)
|
||||
self.min_interval_spin.setValue(min_interval)
|
||||
|
||||
form = QFormLayout()
|
||||
form.addRow("Sensitivity threshold (higher = less sensitive):", self.threshold_spin)
|
||||
form.addRow("Minimum seconds between claps:", self.min_interval_spin)
|
||||
|
||||
reset_button = QPushButton("Reset to Defaults")
|
||||
reset_button.clicked.connect(self._reset_to_defaults)
|
||||
|
||||
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel)
|
||||
buttons.accepted.connect(self.accept)
|
||||
buttons.rejected.connect(self.reject)
|
||||
|
||||
button_row = QHBoxLayout()
|
||||
button_row.addWidget(reset_button)
|
||||
button_row.addStretch(1)
|
||||
button_row.addWidget(buttons)
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.addLayout(form)
|
||||
layout.addLayout(button_row)
|
||||
|
||||
def _reset_to_defaults(self):
|
||||
self.threshold_spin.setValue(DEFAULT_THRESHOLD)
|
||||
self.min_interval_spin.setValue(DEFAULT_MIN_INTERVAL)
|
||||
|
||||
def values(self):
|
||||
return self.threshold_spin.value(), self.min_interval_spin.value()
|
||||
|
||||
|
||||
class ExportWorker(QThread):
|
||||
finished_ok = pyqtSignal(object) # ExportResult
|
||||
finished_err = pyqtSignal(str, str) # message, stderr
|
||||
|
||||
def __init__(self, input_path, delay_ms, output_path):
|
||||
def __init__(self, input_path, delay_ms, output_path, clip_start=None, clip_duration=None):
|
||||
super().__init__()
|
||||
self.input_path = input_path
|
||||
self.delay_ms = delay_ms
|
||||
self.output_path = output_path
|
||||
self.clip_start = clip_start
|
||||
self.clip_duration = clip_duration
|
||||
|
||||
def run(self):
|
||||
try:
|
||||
result: ExportResult = export_synced(self.input_path, self.delay_ms, self.output_path)
|
||||
result: ExportResult = export_synced(
|
||||
self.input_path, self.delay_ms, self.output_path,
|
||||
clip_start=self.clip_start, clip_duration=self.clip_duration,
|
||||
)
|
||||
self.finished_ok.emit(result)
|
||||
except FfmpegExportError as e:
|
||||
self.finished_err.emit(str(e), e.stderr)
|
||||
@@ -62,11 +98,15 @@ class DetectWorker(QThread):
|
||||
finished_ok = pyqtSignal(list) # List[ClapEvent]
|
||||
finished_err = pyqtSignal(str)
|
||||
|
||||
def __init__(self, input_path, start_offset=0.0, max_results=None):
|
||||
def __init__(self, input_path, start_offset=0.0, max_results=None, audio_stream_index=None,
|
||||
threshold=DEFAULT_THRESHOLD, min_interval=DEFAULT_MIN_INTERVAL):
|
||||
super().__init__()
|
||||
self.input_path = input_path
|
||||
self.start_offset = start_offset
|
||||
self.max_results = max_results
|
||||
self.audio_stream_index = audio_stream_index
|
||||
self.threshold = threshold
|
||||
self.min_interval = min_interval
|
||||
|
||||
def run(self):
|
||||
try:
|
||||
@@ -74,30 +114,66 @@ class DetectWorker(QThread):
|
||||
self.input_path,
|
||||
start_offset=self.start_offset,
|
||||
max_results=self.max_results,
|
||||
audio_stream_index=self.audio_stream_index,
|
||||
threshold=self.threshold,
|
||||
min_interval=self.min_interval,
|
||||
)
|
||||
self.finished_ok.emit(events)
|
||||
except (ClapDetectionUnavailable, RuntimeError) as e:
|
||||
self.finished_err.emit(str(e))
|
||||
|
||||
|
||||
class EstimateWorker(QThread):
|
||||
finished_ok = pyqtSignal(object) # StreamEdges
|
||||
finished_err = pyqtSignal(str)
|
||||
|
||||
def __init__(self, input_path, video_stream_index=None, audio_stream_index=None):
|
||||
super().__init__()
|
||||
self.input_path = input_path
|
||||
self.video_stream_index = video_stream_index
|
||||
self.audio_stream_index = audio_stream_index
|
||||
|
||||
def run(self):
|
||||
try:
|
||||
edges = probe_stream_edges(
|
||||
self.input_path,
|
||||
video_stream_index=self.video_stream_index,
|
||||
audio_stream_index=self.audio_stream_index,
|
||||
)
|
||||
self.finished_ok.emit(edges)
|
||||
except StreamEdgesError as e:
|
||||
self.finished_err.emit(str(e))
|
||||
|
||||
|
||||
class MainWindow(QMainWindow):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.setWindowTitle("A/V Sync Marker")
|
||||
self.setWindowTitle("WeAreSyncing")
|
||||
self.resize(960, 720)
|
||||
|
||||
self.input_path: Optional[str] = None
|
||||
self.video_mark: Optional[float] = None
|
||||
self.audio_mark: Optional[float] = None
|
||||
# The offset currently feeding Preview/Export -- derived either from
|
||||
# the two markers or from the stream-ends estimate, whichever was
|
||||
# applied last.
|
||||
self.offset: Optional[SyncOffset] = None
|
||||
self._export_worker: Optional[ExportWorker] = None
|
||||
self._detect_worker: Optional[DetectWorker] = None
|
||||
self._slider_being_dragged = False
|
||||
self._estimate_worker: Optional[EstimateWorker] = None
|
||||
self._preview_worker: Optional[ExportWorker] = None
|
||||
self._preview_temp_path: Optional[str] = None # currently loaded in preview_panel
|
||||
self._pending_preview_path: Optional[str] = None # being built by _preview_worker
|
||||
self.clap_threshold: float = DEFAULT_THRESHOLD
|
||||
self.clap_min_interval: float = DEFAULT_MIN_INTERVAL
|
||||
|
||||
self.player = PlayerWidget()
|
||||
self.player.setMinimumHeight(400)
|
||||
self.player.fileDropped.connect(self._load)
|
||||
self.player.rightClicked.connect(self._toggle_play)
|
||||
# Two independent monitors (Kdenlive-style project/clip monitor
|
||||
# tabs): each is a full player with its own timeline and transport,
|
||||
# not just a shared video frame with one set of controls.
|
||||
self.source_panel = PlayerPanel(show_marker_strip=True)
|
||||
self.source_panel.fileDropped.connect(self._load)
|
||||
|
||||
self.preview_panel = PlayerPanel(show_marker_strip=False)
|
||||
|
||||
self._build_ui()
|
||||
self._build_menu()
|
||||
@@ -110,86 +186,67 @@ class MainWindow(QMainWindow):
|
||||
|
||||
# -- UI construction ---------------------------------------------------
|
||||
|
||||
def _active_panel(self) -> PlayerPanel:
|
||||
return self.source_panel if self.video_tabs.currentIndex() == 0 else self.preview_panel
|
||||
|
||||
def _build_ui(self):
|
||||
central = QWidget()
|
||||
layout = QVBoxLayout(central)
|
||||
|
||||
layout.addWidget(self.player, stretch=1)
|
||||
|
||||
# Timeline directly under the video (LosslessCut-style), transport
|
||||
# controls below that.
|
||||
seek_row = QHBoxLayout()
|
||||
self.position_label = QLabel("00:00.000")
|
||||
seek_row.addWidget(self.position_label)
|
||||
|
||||
slider_stack = QVBoxLayout()
|
||||
slider_stack.setSpacing(0)
|
||||
self.marker_strip = MarkerStrip()
|
||||
self.marker_strip.markerClicked.connect(self._on_marker_clicked)
|
||||
slider_stack.addWidget(self.marker_strip)
|
||||
|
||||
self.seek_slider = ClickableSlider(Qt.Orientation.Horizontal)
|
||||
self.seek_slider.setRange(0, 0)
|
||||
self.seek_slider.sliderPressed.connect(lambda: setattr(self, "_slider_being_dragged", True))
|
||||
self.seek_slider.sliderMoved.connect(self._on_slider_moved)
|
||||
self.seek_slider.sliderReleased.connect(self._on_slider_released)
|
||||
slider_stack.addWidget(self.seek_slider)
|
||||
seek_row.addLayout(slider_stack, stretch=1)
|
||||
|
||||
self.duration_label = QLabel("00:00.000")
|
||||
seek_row.addWidget(self.duration_label)
|
||||
layout.addLayout(seek_row)
|
||||
|
||||
# Transport row: frame-step/play controls centered as a group, mute
|
||||
# pinned to the right. Load File now lives in the File menu.
|
||||
transport = QHBoxLayout()
|
||||
transport.addStretch(1)
|
||||
|
||||
self.prev_frame_button = QPushButton("⏮ Frame (,)")
|
||||
self.prev_frame_button.clicked.connect(lambda: self._step_frame(-1))
|
||||
transport.addWidget(self.prev_frame_button)
|
||||
|
||||
self.play_button = QPushButton("Play")
|
||||
self.play_button.clicked.connect(self._toggle_play)
|
||||
transport.addWidget(self.play_button)
|
||||
|
||||
self.next_frame_button = QPushButton("Frame ⏭ (.)")
|
||||
self.next_frame_button.clicked.connect(lambda: self._step_frame(1))
|
||||
transport.addWidget(self.next_frame_button)
|
||||
|
||||
transport.addStretch(1)
|
||||
|
||||
self.mute_checkbox = QCheckBox("Mute audio")
|
||||
self.mute_checkbox.toggled.connect(self._on_mute_toggled)
|
||||
transport.addWidget(self.mute_checkbox)
|
||||
|
||||
layout.addLayout(transport)
|
||||
self.video_tabs = QTabWidget()
|
||||
self.video_tabs.addTab(self.source_panel, "Source")
|
||||
self.video_tabs.addTab(self.preview_panel, "Preview")
|
||||
layout.addWidget(self.video_tabs, stretch=1)
|
||||
|
||||
# Marker row
|
||||
marker_group = QGroupBox("Sync markers")
|
||||
marker_layout = QHBoxLayout(marker_group)
|
||||
|
||||
self.mark_video_button = QPushButton("Mark Video Event")
|
||||
self.mark_video_button.clicked.connect(self._mark_video)
|
||||
marker_layout.addWidget(self.mark_video_button)
|
||||
self.video_mark_label = QLabel("video: —")
|
||||
marker_layout.addWidget(self.video_mark_label)
|
||||
|
||||
self.mark_audio_button = QPushButton("Mark Audio Event")
|
||||
self.mark_audio_button.clicked.connect(self._mark_audio)
|
||||
marker_layout.addWidget(self.mark_audio_button)
|
||||
self.audio_mark_label = QLabel("audio: —")
|
||||
marker_layout.addWidget(self.audio_mark_label)
|
||||
|
||||
self.mark_video_button = QPushButton("Mark Video Event")
|
||||
self.mark_video_button.clicked.connect(self._mark_video)
|
||||
marker_layout.addWidget(self.mark_video_button)
|
||||
self.video_mark_label = QLabel("video: —")
|
||||
marker_layout.addWidget(self.video_mark_label)
|
||||
|
||||
# Re-applies the marker-derived offset after the stream-ends
|
||||
# estimate replaced it, without having to re-mark an event.
|
||||
self.use_marker_offset_button = QPushButton("Use Marker Offset")
|
||||
self.use_marker_offset_button.setEnabled(False)
|
||||
self.use_marker_offset_button.clicked.connect(self._recompute_offset)
|
||||
marker_layout.addWidget(self.use_marker_offset_button)
|
||||
|
||||
marker_layout.addStretch(1)
|
||||
|
||||
self.detect_all_button = QPushButton("Detect All Claps")
|
||||
self.detect_all_button.setEnabled(False)
|
||||
self.detect_all_button.clicked.connect(self._on_detect_all_clicked)
|
||||
marker_layout.addWidget(self.detect_all_button)
|
||||
|
||||
self.detect_next_button = QPushButton("Detect Next Clap")
|
||||
self.detect_next_button.setEnabled(False)
|
||||
self.detect_next_button.clicked.connect(self._on_detect_next_clicked)
|
||||
marker_layout.addWidget(self.detect_next_button)
|
||||
|
||||
layout.addWidget(marker_group)
|
||||
|
||||
# Offset + export row
|
||||
export_group = QGroupBox("Offset and export")
|
||||
# Offset + preview + export row
|
||||
export_group = QGroupBox("Offset, preview and export")
|
||||
export_layout = QHBoxLayout(export_group)
|
||||
|
||||
self.offset_label = QLabel("Offset: — (mark both events)")
|
||||
export_layout.addWidget(self.offset_label, stretch=1)
|
||||
|
||||
self.estimate_button = QPushButton("Estimate from Stream Ends")
|
||||
self.estimate_button.setEnabled(False)
|
||||
self.estimate_button.clicked.connect(self._on_estimate_clicked)
|
||||
export_layout.addWidget(self.estimate_button)
|
||||
|
||||
export_layout.addWidget(QLabel("Override (ms):"))
|
||||
self.override_spinbox = QDoubleSpinBox()
|
||||
self.override_spinbox.setRange(-600000, 600000)
|
||||
@@ -197,6 +254,11 @@ class MainWindow(QMainWindow):
|
||||
self.override_spinbox.setEnabled(False)
|
||||
export_layout.addWidget(self.override_spinbox)
|
||||
|
||||
self.preview_button = QPushButton("Preview Sync…")
|
||||
self.preview_button.setEnabled(False)
|
||||
self.preview_button.clicked.connect(self._on_preview_clicked)
|
||||
export_layout.addWidget(self.preview_button)
|
||||
|
||||
self.export_button = QPushButton("Export…")
|
||||
self.export_button.setEnabled(False)
|
||||
self.export_button.clicked.connect(self._on_export_clicked)
|
||||
@@ -238,14 +300,40 @@ class MainWindow(QMainWindow):
|
||||
self.detect_next_action.triggered.connect(self._on_detect_next_clicked)
|
||||
tools_menu.addAction(self.detect_next_action)
|
||||
|
||||
tools_menu.addSeparator()
|
||||
|
||||
self.estimate_action = QAction("Estimate Offset from Stream Ends", self)
|
||||
self.estimate_action.setEnabled(False)
|
||||
self.estimate_action.triggered.connect(self._on_estimate_clicked)
|
||||
tools_menu.addAction(self.estimate_action)
|
||||
|
||||
tools_menu.addSeparator()
|
||||
|
||||
settings_action = QAction("Clap Detection Settings…", self)
|
||||
settings_action.triggered.connect(self._on_clap_settings_clicked)
|
||||
tools_menu.addAction(settings_action)
|
||||
|
||||
def _on_clap_settings_clicked(self):
|
||||
dialog = ClapSettingsDialog(self.clap_threshold, self.clap_min_interval, parent=self)
|
||||
if dialog.exec() == QDialog.DialogCode.Accepted:
|
||||
self.clap_threshold, self.clap_min_interval = dialog.values()
|
||||
self.statusBar().showMessage(
|
||||
f"Clap detection settings updated: threshold={self.clap_threshold:.2f}, "
|
||||
f"min_interval={self.clap_min_interval:.2f}s", 4000,
|
||||
)
|
||||
|
||||
def _wire_shortcuts(self):
|
||||
QShortcut(QKeySequence(","), self, activated=lambda: self._step_frame(-1))
|
||||
QShortcut(QKeySequence("."), self, activated=lambda: self._step_frame(1))
|
||||
QShortcut(QKeySequence(Qt.Key.Key_Space), self, activated=self._toggle_play)
|
||||
# These act on whichever tab (Source/Preview) is currently active --
|
||||
# each is a fully independent monitor with its own transport, so the
|
||||
# keyboard shortcuts follow the same "whichever one you're looking
|
||||
# at" rule as the on-screen buttons.
|
||||
QShortcut(QKeySequence(","), self, activated=lambda: self._active_panel().step_frame(-1))
|
||||
QShortcut(QKeySequence("."), self, activated=lambda: self._active_panel().step_frame(1))
|
||||
QShortcut(QKeySequence(Qt.Key.Key_Space), self, activated=lambda: self._active_panel().toggle_play())
|
||||
# QShortcut auto-repeats by default while the key is held, so this
|
||||
# keeps skipping for as long as the arrow key is down.
|
||||
QShortcut(QKeySequence(Qt.Key.Key_Left), self, activated=lambda: self._seek_relative(-5.0))
|
||||
QShortcut(QKeySequence(Qt.Key.Key_Right), self, activated=lambda: self._seek_relative(5.0))
|
||||
QShortcut(QKeySequence(Qt.Key.Key_Left), self, activated=lambda: self._active_panel().seek_relative(-5.0))
|
||||
QShortcut(QKeySequence(Qt.Key.Key_Right), self, activated=lambda: self._active_panel().seek_relative(5.0))
|
||||
|
||||
# -- transport -----------------------------------------------------
|
||||
|
||||
@@ -255,15 +343,15 @@ class MainWindow(QMainWindow):
|
||||
self._load(path)
|
||||
|
||||
def _load(self, path: str):
|
||||
# player.load() blocks the GUI thread until mpv reports the file's
|
||||
# duration, so show the status message and force one repaint before
|
||||
# starting it -- otherwise Qt wouldn't paint the message until after
|
||||
# the blocking call already returned.
|
||||
# source_panel.load() blocks the GUI thread until mpv reports the
|
||||
# file's duration, so show the status message and force one repaint
|
||||
# before starting it -- otherwise Qt wouldn't paint the message
|
||||
# until after the blocking call already returned.
|
||||
self.statusBar().showMessage(f"Loading {Path(path).name}…")
|
||||
QApplication.processEvents()
|
||||
|
||||
self.input_path = path
|
||||
self.player.load(path)
|
||||
self.source_panel.load(path)
|
||||
self.video_mark = None
|
||||
self.audio_mark = None
|
||||
self.offset = None
|
||||
@@ -271,77 +359,39 @@ class MainWindow(QMainWindow):
|
||||
self.audio_mark_label.setText("audio: —")
|
||||
self.offset_label.setText("Offset: — (mark both events)")
|
||||
self.export_button.setEnabled(False)
|
||||
self.preview_button.setEnabled(False)
|
||||
self.override_spinbox.setEnabled(False)
|
||||
self.use_marker_offset_button.setEnabled(False)
|
||||
self.log_view.clear()
|
||||
|
||||
duration = self.player.mpv.duration or 0.0
|
||||
self.seek_slider.setRange(0, int(duration * 1000))
|
||||
self.duration_label.setText(format_timestamp(duration))
|
||||
self.setWindowTitle(f"A/V Sync Marker — {Path(path).name}")
|
||||
|
||||
self.marker_strip.set_duration(duration)
|
||||
self.marker_strip.clear()
|
||||
self.setWindowTitle(f"WeAreSyncing — {Path(path).name}")
|
||||
self.statusBar().showMessage(f"Loaded {Path(path).name}", 4000)
|
||||
self.detect_all_action.setEnabled(True)
|
||||
self.detect_all_button.setEnabled(True)
|
||||
self.detect_next_action.setEnabled(True)
|
||||
self.detect_next_button.setEnabled(True)
|
||||
self.estimate_action.setEnabled(True)
|
||||
self.estimate_button.setEnabled(True)
|
||||
|
||||
def _toggle_play(self):
|
||||
if self.player.mpv.pause:
|
||||
self.player.play()
|
||||
self.play_button.setText("Pause")
|
||||
else:
|
||||
self.player.pause()
|
||||
self.play_button.setText("Play")
|
||||
|
||||
def _step_frame(self, n: int):
|
||||
self.player.pause()
|
||||
self.play_button.setText("Play")
|
||||
self.player.step_frame(n)
|
||||
self._update_position_ui()
|
||||
|
||||
def _on_mute_toggled(self, checked: bool):
|
||||
self.player.mpv.mute = checked
|
||||
|
||||
def _seek_relative(self, delta_seconds: float):
|
||||
# Unlike frame-stepping (which always pauses -- you're precisely
|
||||
# inspecting a frame), a 5s skip is a coarse nudge while scrubbing
|
||||
# around; seeking doesn't change mpv's pause state on its own, so
|
||||
# just not touching play/pause here is enough to preserve whichever
|
||||
# state the user was already in.
|
||||
if self.input_path is None:
|
||||
return
|
||||
duration = self.player.mpv.duration or 0.0
|
||||
target = min(max(self.player.get_position() + delta_seconds, 0.0), duration)
|
||||
self.player.seek(target)
|
||||
self._update_position_ui()
|
||||
|
||||
def _on_slider_moved(self, value: int):
|
||||
self.player.seek(value / 1000.0)
|
||||
self._update_position_ui()
|
||||
|
||||
def _on_slider_released(self):
|
||||
self._slider_being_dragged = False
|
||||
self._update_position_ui()
|
||||
# A loaded preview belongs to the *previous* source/offset -- drop it.
|
||||
if self._preview_temp_path and os.path.exists(self._preview_temp_path):
|
||||
os.unlink(self._preview_temp_path)
|
||||
self._preview_temp_path = None
|
||||
self.video_tabs.setCurrentIndex(0)
|
||||
|
||||
def _update_position_ui(self):
|
||||
if self.input_path is None:
|
||||
return
|
||||
pos = self.player.get_position()
|
||||
self.position_label.setText(format_timestamp(pos))
|
||||
if not self._slider_being_dragged:
|
||||
self.seek_slider.blockSignals(True)
|
||||
self.seek_slider.setValue(int(pos * 1000))
|
||||
self.seek_slider.blockSignals(False)
|
||||
self.source_panel.update_position()
|
||||
self.preview_panel.update_position()
|
||||
|
||||
# -- markers / offset -----------------------------------------------
|
||||
|
||||
def _mark_video(self):
|
||||
self.video_mark = self.player.get_position()
|
||||
self.video_mark = self.source_panel.player.get_position()
|
||||
self.video_mark_label.setText(f"video: {format_timestamp(self.video_mark)}")
|
||||
self._recompute_offset()
|
||||
|
||||
def _mark_audio(self):
|
||||
self.audio_mark = self.player.get_position()
|
||||
self.audio_mark = self.source_panel.player.get_position()
|
||||
self.audio_mark_label.setText(f"audio: {format_timestamp(self.audio_mark)}")
|
||||
self._recompute_offset()
|
||||
|
||||
@@ -349,43 +399,196 @@ class MainWindow(QMainWindow):
|
||||
if self.video_mark is None or self.audio_mark is None:
|
||||
return
|
||||
self.offset = compute_offset(self.video_mark, self.audio_mark)
|
||||
self.offset_label.setText(self.offset.describe())
|
||||
self.override_spinbox.setValue(self.offset.delay_ms)
|
||||
self._apply_offset(self.offset, self.offset.describe())
|
||||
self.use_marker_offset_button.setEnabled(True)
|
||||
|
||||
def _apply_offset(self, offset: SyncOffset, label: str):
|
||||
"""Shared by both offset sources (markers, stream-ends estimate):
|
||||
whichever is applied last is what Preview/Export use."""
|
||||
self.offset = offset
|
||||
self.offset_label.setText(label)
|
||||
self.override_spinbox.setValue(offset.delay_ms)
|
||||
self.override_spinbox.setEnabled(True)
|
||||
self.export_button.setEnabled(True)
|
||||
self.preview_button.setEnabled(True)
|
||||
|
||||
# -- stream-ends estimate ----------------------------------------------
|
||||
|
||||
def _current_video_stream_index(self) -> Optional[int]:
|
||||
for track in self.source_panel.player.mpv.track_list:
|
||||
if track.get("type") == "video" and track.get("selected"):
|
||||
return track.get("ff-index")
|
||||
return None
|
||||
|
||||
def _on_estimate_clicked(self):
|
||||
if self.input_path is None:
|
||||
return
|
||||
self.estimate_action.setEnabled(False)
|
||||
self.estimate_button.setEnabled(False)
|
||||
self.statusBar().showMessage("Estimating offset from stream ends…")
|
||||
|
||||
self._estimate_worker = EstimateWorker(
|
||||
self.input_path,
|
||||
video_stream_index=self._current_video_stream_index(),
|
||||
audio_stream_index=self._current_audio_stream_index(),
|
||||
)
|
||||
self._estimate_worker.finished_ok.connect(self._on_estimate_done)
|
||||
self._estimate_worker.finished_err.connect(self._on_estimate_failed)
|
||||
self._estimate_worker.start()
|
||||
|
||||
def _on_estimate_done(self, edges: StreamEdges):
|
||||
self.estimate_action.setEnabled(True)
|
||||
self.estimate_button.setEnabled(True)
|
||||
|
||||
offset = SyncOffset(delay_ms=round(edges.end_delay_ms))
|
||||
self._apply_offset(offset, f"{offset.describe()} — estimated from stream ends")
|
||||
|
||||
self.log_view.appendPlainText(
|
||||
f"Stream-ends estimate: video {edges.video_start:.3f}–{edges.video_end:.3f}s, "
|
||||
f"audio {edges.audio_start:.3f}–{edges.audio_end:.3f}s → "
|
||||
f"end-based {edges.end_delay_ms:+.0f} ms, start-based {edges.start_delay_ms:+.0f} ms"
|
||||
)
|
||||
if edges.estimates_disagree:
|
||||
message = (
|
||||
f"Estimated {offset.delay_ms:+.0f} ms from stream ends, but stream starts suggest "
|
||||
f"{edges.start_delay_ms:+.0f} ms — verify with Preview."
|
||||
)
|
||||
self.log_view.appendPlainText("WARNING: " + message)
|
||||
else:
|
||||
message = f"Estimated {offset.delay_ms:+.0f} ms from stream ends (assumes both tracks stopped recording together) — verify with Preview."
|
||||
self.statusBar().showMessage(message, 10000)
|
||||
|
||||
def _on_estimate_failed(self, message: str):
|
||||
self.estimate_action.setEnabled(True)
|
||||
self.estimate_button.setEnabled(True)
|
||||
self.statusBar().showMessage("Offset estimate failed (see log).", 8000)
|
||||
self.log_view.appendPlainText(f"Stream-ends estimate FAILED: {message}")
|
||||
|
||||
# -- preview -----------------------------------------------------
|
||||
|
||||
PREVIEW_MARGIN_SEC = 3.0
|
||||
PREVIEW_DURATION_SEC = 8.0
|
||||
|
||||
def _on_preview_clicked(self):
|
||||
if self.input_path is None or self.offset is None:
|
||||
return
|
||||
delay_ms = self.override_spinbox.value()
|
||||
|
||||
# Build just a short clip around the marked events rather than the
|
||||
# whole file -- lossless stream-copy of a few seconds is fast
|
||||
# regardless of how long the source file is, unlike previewing the
|
||||
# full result. Centering on the midpoint of the two original marks
|
||||
# keeps both the video and audio events (and wherever the offset
|
||||
# moves them to) inside the window for typical human-scale marking
|
||||
# errors.
|
||||
#
|
||||
# An estimated offset may have no markers behind it; center on the
|
||||
# Source playhead instead, so the user can park it on a sync event
|
||||
# (or any stretch of speech) before previewing.
|
||||
if self.video_mark is not None and self.audio_mark is not None:
|
||||
center = (self.video_mark + self.audio_mark) / 2.0
|
||||
else:
|
||||
center = self.source_panel.player.get_position()
|
||||
clip_start = max(center - self.PREVIEW_MARGIN_SEC, 0.0)
|
||||
|
||||
input_p = Path(self.input_path)
|
||||
fd, temp_path = tempfile.mkstemp(suffix=input_p.suffix, prefix="avsync_preview_")
|
||||
os.close(fd)
|
||||
self._pending_preview_path = temp_path
|
||||
|
||||
self.preview_button.setEnabled(False)
|
||||
self.preview_button.setText("Building preview…")
|
||||
self.statusBar().showMessage("Building preview clip…")
|
||||
|
||||
self._preview_worker = ExportWorker(
|
||||
self.input_path, delay_ms, temp_path,
|
||||
clip_start=f"{clip_start:.3f}", clip_duration=self.PREVIEW_DURATION_SEC,
|
||||
)
|
||||
self._preview_worker.finished_ok.connect(self._on_preview_ready)
|
||||
self._preview_worker.finished_err.connect(self._on_preview_failed)
|
||||
self._preview_worker.start()
|
||||
|
||||
def _on_preview_ready(self, result: ExportResult):
|
||||
self.preview_button.setEnabled(True)
|
||||
self.preview_button.setText("Preview Sync…")
|
||||
self.statusBar().showMessage("Preview ready.", 4000)
|
||||
if result.stderr.strip():
|
||||
self.log_view.appendPlainText("ffmpeg preview log:\n" + result.stderr)
|
||||
|
||||
# The preview panel is long-lived (one mpv instance for the whole
|
||||
# session), so swap files instead of spinning up a new player each
|
||||
# time: load the new clip first, then delete the old temp file --
|
||||
# only safe to unlink it once mpv is no longer holding it open.
|
||||
old_path = self._preview_temp_path
|
||||
new_path = self._pending_preview_path
|
||||
self.preview_panel.load(new_path)
|
||||
self.preview_panel.force_play()
|
||||
self._preview_temp_path = new_path
|
||||
self._pending_preview_path = None
|
||||
self.video_tabs.setCurrentIndex(1)
|
||||
|
||||
if old_path and old_path != new_path and os.path.exists(old_path):
|
||||
os.unlink(old_path)
|
||||
|
||||
def _on_preview_failed(self, message: str, stderr: str):
|
||||
self.preview_button.setEnabled(True)
|
||||
self.preview_button.setText("Preview Sync…")
|
||||
self.statusBar().showMessage("Preview failed (see log).", 8000)
|
||||
self.log_view.appendPlainText(f"Preview FAILED: {message}\n{stderr}")
|
||||
if self._pending_preview_path and os.path.exists(self._pending_preview_path):
|
||||
os.unlink(self._pending_preview_path)
|
||||
self._pending_preview_path = None
|
||||
|
||||
# -- clap detection -----------------------------------------------------
|
||||
|
||||
def _on_marker_clicked(self, time_sec: float):
|
||||
self.player.seek(time_sec)
|
||||
self._update_position_ui()
|
||||
|
||||
def _on_detect_all_clicked(self):
|
||||
self._start_detection(start_offset=0.0, max_results=None, label="Scanning whole file for claps…")
|
||||
|
||||
def _on_detect_next_clicked(self):
|
||||
cursor = self.player.get_position()
|
||||
cursor = self.source_panel.player.get_position()
|
||||
self._start_detection(start_offset=cursor, max_results=1, label=f"Scanning from {format_timestamp(cursor)} for next clap…")
|
||||
|
||||
def _current_audio_stream_index(self) -> Optional[int]:
|
||||
"""The ffmpeg/ffprobe global stream index of whichever audio track
|
||||
mpv is currently playing in the Source panel. Passed through to
|
||||
clap detection so it analyzes the exact track being listened to --
|
||||
without this, ffmpeg's own default-stream selection could silently
|
||||
pick a *different* audio track on multi-audio-track files (extra
|
||||
language dubs, a separate mic/system-audio track from a screen
|
||||
recorder), producing markers for the wrong track's timing entirely.
|
||||
"""
|
||||
for track in self.source_panel.player.mpv.track_list:
|
||||
if track.get("type") == "audio" and track.get("selected"):
|
||||
return track.get("ff-index")
|
||||
return None
|
||||
|
||||
def _start_detection(self, start_offset: float, max_results, label: str):
|
||||
if self.input_path is None:
|
||||
return
|
||||
self.detect_all_action.setEnabled(False)
|
||||
self.detect_all_button.setEnabled(False)
|
||||
self.detect_next_action.setEnabled(False)
|
||||
self.detect_next_button.setEnabled(False)
|
||||
self.statusBar().showMessage(label)
|
||||
|
||||
self._detect_worker = DetectWorker(self.input_path, start_offset=start_offset, max_results=max_results)
|
||||
self._detect_worker = DetectWorker(
|
||||
self.input_path, start_offset=start_offset, max_results=max_results,
|
||||
audio_stream_index=self._current_audio_stream_index(),
|
||||
threshold=self.clap_threshold, min_interval=self.clap_min_interval,
|
||||
)
|
||||
self._detect_worker.finished_ok.connect(self._on_detect_done)
|
||||
self._detect_worker.finished_err.connect(self._on_detect_failed)
|
||||
self._detect_worker.start()
|
||||
|
||||
def _on_detect_done(self, events: list):
|
||||
self.detect_all_action.setEnabled(True)
|
||||
self.detect_all_button.setEnabled(True)
|
||||
self.detect_next_action.setEnabled(True)
|
||||
self.detect_next_button.setEnabled(True)
|
||||
|
||||
existing = set(self.marker_strip.markers())
|
||||
existing = set(self.source_panel.marker_strip.markers())
|
||||
existing.update(e.time_sec for e in events)
|
||||
self.marker_strip.set_markers(list(existing))
|
||||
self.source_panel.marker_strip.set_markers(list(existing))
|
||||
|
||||
if not events:
|
||||
self.statusBar().showMessage("No claps found.", 5000)
|
||||
@@ -398,7 +601,9 @@ class MainWindow(QMainWindow):
|
||||
|
||||
def _on_detect_failed(self, message: str):
|
||||
self.detect_all_action.setEnabled(True)
|
||||
self.detect_all_button.setEnabled(True)
|
||||
self.detect_next_action.setEnabled(True)
|
||||
self.detect_next_button.setEnabled(True)
|
||||
self.statusBar().showMessage("Detection failed (see log).", 8000)
|
||||
self.log_view.appendPlainText(f"Clap detection FAILED: {message}")
|
||||
|
||||
@@ -410,7 +615,11 @@ class MainWindow(QMainWindow):
|
||||
delay_ms = self.override_spinbox.value()
|
||||
|
||||
input_p = Path(self.input_path)
|
||||
suggested = str(input_p.with_name(f"{input_p.stem}_synced{input_p.suffix}"))
|
||||
# Matches the original ffdelay script's naming convention
|
||||
# (e.g. "_delayed500ms" / "_delayed-500ms") rather than inventing
|
||||
# a new one.
|
||||
delayed_str = f"{int(round(delay_ms))}"
|
||||
suggested = str(input_p.with_name(f"{input_p.stem}_delayed{delayed_str}ms{input_p.suffix}"))
|
||||
output_path, _ = QFileDialog.getSaveFileName(self, "Export synced file", suggested)
|
||||
if not output_path:
|
||||
return
|
||||
@@ -442,7 +651,15 @@ class MainWindow(QMainWindow):
|
||||
QMessageBox.critical(self, "Export failed", f"{message}\n\n{stderr[-2000:]}")
|
||||
|
||||
def closeEvent(self, event):
|
||||
self.player.shutdown()
|
||||
# Stop the timer before tearing down mpv -- otherwise a tick that
|
||||
# fires between shutdown() and the process actually exiting calls
|
||||
# get_position() on an already-terminated core and crashes
|
||||
# (mpv.ShutdownError, observed as a hard crash in testing).
|
||||
self._position_timer.stop()
|
||||
self.source_panel.shutdown()
|
||||
self.preview_panel.shutdown()
|
||||
if self._preview_temp_path and os.path.exists(self._preview_temp_path):
|
||||
os.unlink(self._preview_temp_path)
|
||||
super().closeEvent(event)
|
||||
|
||||
|
||||
|
||||
+176
@@ -0,0 +1,176 @@
|
||||
from typing import Optional
|
||||
|
||||
from PyQt6.QtCore import Qt, pyqtSignal
|
||||
from PyQt6.QtWidgets import QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QCheckBox, QSlider
|
||||
|
||||
from player_widget import PlayerWidget
|
||||
from marker_strip import MarkerStrip
|
||||
|
||||
|
||||
def format_timestamp(seconds: float) -> str:
|
||||
m, s = divmod(max(seconds, 0.0), 60)
|
||||
return f"{int(m):02d}:{s:06.3f}"
|
||||
|
||||
|
||||
class ClickableSlider(QSlider):
|
||||
"""QSlider normally only jumps to the clicked position if you click
|
||||
exactly on the handle; clicking elsewhere in the groove pages by a
|
||||
fixed step. Override so any click along the track jumps straight
|
||||
there, then let the base class take over so dragging still works
|
||||
(the handle is now under the cursor, so its own drag tracking picks
|
||||
up seamlessly).
|
||||
"""
|
||||
|
||||
def mousePressEvent(self, event):
|
||||
if event.button() == Qt.MouseButton.LeftButton and self.maximum() > self.minimum():
|
||||
ratio = min(max(event.position().x() / max(self.width(), 1), 0.0), 1.0)
|
||||
value = round(self.minimum() + ratio * (self.maximum() - self.minimum()))
|
||||
self.setValue(value)
|
||||
self.sliderMoved.emit(value)
|
||||
super().mousePressEvent(event)
|
||||
|
||||
|
||||
class PlayerPanel(QWidget):
|
||||
"""A self-contained video monitor: player + its own timeline + its own
|
||||
transport row (Kdenlive-style project/clip monitor). Two of these are
|
||||
used side by side as tabs (Source, Preview) in main.py, each with an
|
||||
independent PlayerWidget/mpv instance and independent playback state --
|
||||
switching tabs switches which monitor's timeline and transport you're
|
||||
looking at, not just the video frame.
|
||||
"""
|
||||
|
||||
fileDropped = pyqtSignal(str)
|
||||
|
||||
def __init__(self, show_marker_strip: bool = False, parent=None):
|
||||
super().__init__(parent)
|
||||
|
||||
self.player = PlayerWidget()
|
||||
self.player.setMinimumHeight(320)
|
||||
self.player.fileDropped.connect(self.fileDropped)
|
||||
self.player.rightClicked.connect(self.toggle_play)
|
||||
|
||||
self._loaded = False
|
||||
self._slider_being_dragged = False
|
||||
|
||||
layout = QVBoxLayout(self)
|
||||
layout.setContentsMargins(0, 0, 0, 8)
|
||||
layout.addWidget(self.player, stretch=1)
|
||||
|
||||
seek_row = QHBoxLayout()
|
||||
self.position_label = QLabel("00:00.000")
|
||||
seek_row.addWidget(self.position_label)
|
||||
|
||||
slider_stack = QVBoxLayout()
|
||||
slider_stack.setSpacing(0)
|
||||
|
||||
self.marker_strip: Optional[MarkerStrip] = None
|
||||
if show_marker_strip:
|
||||
self.marker_strip = MarkerStrip()
|
||||
self.marker_strip.markerClicked.connect(self._on_marker_clicked)
|
||||
slider_stack.addWidget(self.marker_strip)
|
||||
|
||||
self.seek_slider = ClickableSlider(Qt.Orientation.Horizontal)
|
||||
self.seek_slider.setRange(0, 0)
|
||||
self.seek_slider.sliderPressed.connect(lambda: setattr(self, "_slider_being_dragged", True))
|
||||
self.seek_slider.sliderMoved.connect(self._on_slider_moved)
|
||||
self.seek_slider.sliderReleased.connect(self._on_slider_released)
|
||||
slider_stack.addWidget(self.seek_slider)
|
||||
seek_row.addLayout(slider_stack, stretch=1)
|
||||
|
||||
self.duration_label = QLabel("00:00.000")
|
||||
seek_row.addWidget(self.duration_label)
|
||||
layout.addLayout(seek_row)
|
||||
|
||||
transport = QHBoxLayout()
|
||||
transport.addStretch(1)
|
||||
|
||||
self.prev_frame_button = QPushButton("⏮ Frame")
|
||||
self.prev_frame_button.clicked.connect(lambda: self.step_frame(-1))
|
||||
transport.addWidget(self.prev_frame_button)
|
||||
|
||||
self.play_button = QPushButton("Play")
|
||||
self.play_button.clicked.connect(self.toggle_play)
|
||||
transport.addWidget(self.play_button)
|
||||
|
||||
self.next_frame_button = QPushButton("Frame ⏭")
|
||||
self.next_frame_button.clicked.connect(lambda: self.step_frame(1))
|
||||
transport.addWidget(self.next_frame_button)
|
||||
|
||||
transport.addStretch(1)
|
||||
|
||||
self.mute_checkbox = QCheckBox("Mute audio")
|
||||
self.mute_checkbox.toggled.connect(lambda checked: setattr(self.player.mpv, "mute", checked))
|
||||
transport.addWidget(self.mute_checkbox)
|
||||
|
||||
layout.addLayout(transport)
|
||||
|
||||
# -- playback -----------------------------------------------------
|
||||
|
||||
def load(self, path: str):
|
||||
self.player.load(path)
|
||||
self._loaded = True
|
||||
duration = self.player.mpv.duration or 0.0
|
||||
self.seek_slider.setRange(0, int(duration * 1000))
|
||||
self.duration_label.setText(format_timestamp(duration))
|
||||
if self.marker_strip is not None:
|
||||
self.marker_strip.set_duration(duration)
|
||||
self.marker_strip.clear()
|
||||
self.play_button.setText("Play")
|
||||
self.update_position()
|
||||
|
||||
def toggle_play(self):
|
||||
if self.player.mpv.pause:
|
||||
self.player.play()
|
||||
self.play_button.setText("Pause")
|
||||
else:
|
||||
self.player.pause()
|
||||
self.play_button.setText("Play")
|
||||
|
||||
def force_play(self):
|
||||
self.player.play()
|
||||
self.play_button.setText("Pause")
|
||||
|
||||
def step_frame(self, n: int):
|
||||
if not self._loaded:
|
||||
return
|
||||
self.player.pause()
|
||||
self.play_button.setText("Play")
|
||||
self.player.step_frame(n)
|
||||
self.update_position()
|
||||
|
||||
def seek_relative(self, delta_seconds: float):
|
||||
# Unlike frame-stepping (which always pauses -- you're precisely
|
||||
# inspecting a frame), a coarse skip is meant to preserve whichever
|
||||
# play/pause state the user was already in; seeking doesn't change
|
||||
# mpv's pause state on its own, so not touching it here is enough.
|
||||
if not self._loaded:
|
||||
return
|
||||
duration = self.player.mpv.duration or 0.0
|
||||
target = min(max(self.player.get_position() + delta_seconds, 0.0), duration)
|
||||
self.player.seek(target)
|
||||
self.update_position()
|
||||
|
||||
def _on_slider_moved(self, value: int):
|
||||
self.player.seek(value / 1000.0)
|
||||
self.update_position()
|
||||
|
||||
def _on_slider_released(self):
|
||||
self._slider_being_dragged = False
|
||||
self.update_position()
|
||||
|
||||
def _on_marker_clicked(self, time_sec: float):
|
||||
self.player.seek(time_sec)
|
||||
self.update_position()
|
||||
|
||||
def update_position(self):
|
||||
if not self._loaded:
|
||||
return
|
||||
pos = self.player.get_position()
|
||||
self.position_label.setText(format_timestamp(pos))
|
||||
if not self._slider_being_dragged:
|
||||
self.seek_slider.blockSignals(True)
|
||||
self.seek_slider.setValue(int(pos * 1000))
|
||||
self.seek_slider.blockSignals(False)
|
||||
|
||||
def shutdown(self):
|
||||
self.player.shutdown()
|
||||
@@ -9,5 +9,5 @@
|
||||
# /etc/ld.so.conf.d/ would fix this system-wide instead, if preferred.
|
||||
set -e
|
||||
cd "$(dirname "$0")"
|
||||
export LD_LIBRARY_PATH="/usr/local/lib:${LD_LIBRARY_PATH}"
|
||||
# export LD_LIBRARY_PATH="/usr/local/lib:${LD_LIBRARY_PATH}"
|
||||
exec .venv/bin/python3 "$@"
|
||||
|
||||
@@ -11,7 +11,7 @@ from datetime import timedelta
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from clap_detection import detect_claps, ClapDetectionUnavailable
|
||||
from clap_detection import detect_claps, ClapDetectionUnavailable, DEFAULT_THRESHOLD, DEFAULT_MIN_INTERVAL
|
||||
|
||||
|
||||
def format_timestamp(seconds):
|
||||
@@ -43,10 +43,10 @@ Threshold guide:
|
||||
)
|
||||
|
||||
parser.add_argument('input_file', help='Audio or video file')
|
||||
parser.add_argument('-t', '--threshold', type=float, default=0.7,
|
||||
help='Detection threshold (0-1, default: 0.7, higher = less sensitive)')
|
||||
parser.add_argument('-m', '--min-interval', type=float, default=0.3,
|
||||
help='Minimum seconds between detections (default: 0.3)')
|
||||
parser.add_argument('-t', '--threshold', type=float, default=DEFAULT_THRESHOLD,
|
||||
help=f'Detection threshold (0-1, default: {DEFAULT_THRESHOLD}, higher = less sensitive)')
|
||||
parser.add_argument('-m', '--min-interval', type=float, default=DEFAULT_MIN_INTERVAL,
|
||||
help=f'Minimum seconds between detections (default: {DEFAULT_MIN_INTERVAL})')
|
||||
parser.add_argument('-s', '--show-strength', action='store_true',
|
||||
help='Show onset strength values')
|
||||
|
||||
|
||||
@@ -51,39 +51,40 @@ def main():
|
||||
assert not win.toggle_log_action.isChecked()
|
||||
print("PASS: log panel hidden by default")
|
||||
|
||||
win.player.seek(5.0)
|
||||
win.source_panel.player.seek(5.0)
|
||||
qapp.processEvents()
|
||||
QTest.keyClick(win, Qt.Key.Key_Right)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
pos = win.player.get_position()
|
||||
pos = win.source_panel.player.get_position()
|
||||
print(f"after Right arrow from 5.0s: pos={pos:.3f}s")
|
||||
assert abs(pos - 10.0) < 0.5, f"expected ~10.0s, got {pos:.3f}s" # 10s clip, clamps at end
|
||||
|
||||
win.player.seek(5.0)
|
||||
win.source_panel.player.seek(5.0)
|
||||
qapp.processEvents()
|
||||
QTest.keyClick(win, Qt.Key.Key_Left)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
pos = win.player.get_position()
|
||||
pos = win.source_panel.player.get_position()
|
||||
print(f"after Left arrow from 5.0s: pos={pos:.3f}s")
|
||||
assert abs(pos - 0.0) < 0.5, f"expected ~0.0s, got {pos:.3f}s"
|
||||
|
||||
win.player.seek(2.0)
|
||||
win.source_panel.player.seek(2.0)
|
||||
qapp.processEvents()
|
||||
QTest.keyClick(win, Qt.Key.Key_Left)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
pos = win.player.get_position()
|
||||
pos = win.source_panel.player.get_position()
|
||||
print(f"after Left arrow from 2.0s (should clamp to 0, not go negative): pos={pos:.3f}s")
|
||||
assert 0.0 <= pos < 0.5, f"expected clamped to ~0.0s, got {pos:.3f}s"
|
||||
|
||||
print("PASS: arrow keys skip 5s and clamp to file bounds")
|
||||
|
||||
win.player.shutdown()
|
||||
win.source_panel.shutdown()
|
||||
win.preview_panel.shutdown()
|
||||
print("\nARROW SEEK SMOKE TEST PASSED")
|
||||
|
||||
|
||||
|
||||
@@ -47,7 +47,7 @@ def main():
|
||||
ok = wait_for(lambda: win.detect_all_action.isEnabled(), qapp)
|
||||
assert ok, "detection did not finish in time"
|
||||
|
||||
markers = win.marker_strip.markers()
|
||||
markers = win.source_panel.marker_strip.markers()
|
||||
print(f"detected markers: {markers}")
|
||||
assert len(markers) >= 1, "expected at least one clap marker"
|
||||
# ground truth beep is at t=5.350s in this fixture; onset detection has
|
||||
@@ -59,12 +59,12 @@ def main():
|
||||
# Simulate a real mouse click on the marker to prove click-to-seek works
|
||||
# end to end, not just that the signal is wired.
|
||||
target_time = min(markers, key=lambda m: abs(m - 5.350))
|
||||
strip = win.marker_strip
|
||||
strip.resize(win.seek_slider.width(), strip.height())
|
||||
strip = win.source_panel.marker_strip
|
||||
strip.resize(win.source_panel.seek_slider.width(), strip.height())
|
||||
x = strip._x_for_time(target_time)
|
||||
click_pos = QPointF(x, 5)
|
||||
|
||||
before_pos = win.player.get_position()
|
||||
before_pos = win.source_panel.player.get_position()
|
||||
event = QMouseEvent(
|
||||
QMouseEvent.Type.MouseButtonPress, click_pos, click_pos,
|
||||
Qt.MouseButton.LeftButton, Qt.MouseButton.LeftButton, Qt.KeyboardModifier.NoModifier,
|
||||
@@ -74,11 +74,12 @@ def main():
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
|
||||
after_pos = win.player.get_position()
|
||||
after_pos = win.source_panel.player.get_position()
|
||||
print(f"clicked marker at x={x:.1f} (t={target_time:.3f}s); player position before={before_pos:.3f}s after={after_pos:.3f}s")
|
||||
assert abs(after_pos - target_time) < 0.1, "clicking the marker should seek the player to that timestamp"
|
||||
|
||||
win.player.shutdown()
|
||||
win.source_panel.shutdown()
|
||||
win.preview_panel.shutdown()
|
||||
print("\nCLAP DETECTION SMOKE TEST PASSED")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
"""Verifies the Clap Detection Settings dialog: defaults match
|
||||
clap_detection.py's DEFAULT_THRESHOLD/DEFAULT_MIN_INTERVAL, changing values
|
||||
and accepting actually updates MainWindow state and gets used by the next
|
||||
detection run, Cancel leaves state untouched, and Reset to Defaults in the
|
||||
dialog restores the shared constants (not just whatever MainWindow happened
|
||||
to start with).
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from PyQt6.QtWidgets import QApplication, QDialog
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
import main as app_main
|
||||
from clap_detection import DEFAULT_THRESHOLD, DEFAULT_MIN_INTERVAL
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
CLIP = str(ROOT / "test_media" / "sync_test_INSYNC.mp4")
|
||||
|
||||
|
||||
def wait_for(predicate, qapp, timeout=20):
|
||||
deadline = time.time() + timeout
|
||||
while not predicate() and time.time() < deadline:
|
||||
qapp.processEvents()
|
||||
time.sleep(0.05)
|
||||
return predicate()
|
||||
|
||||
|
||||
def main():
|
||||
qapp = QApplication(sys.argv)
|
||||
win = app_main.MainWindow()
|
||||
win.show()
|
||||
qapp.processEvents()
|
||||
|
||||
assert win.clap_threshold == DEFAULT_THRESHOLD
|
||||
assert win.clap_min_interval == DEFAULT_MIN_INTERVAL
|
||||
print(f"PASS: MainWindow starts with defaults ({DEFAULT_THRESHOLD}, {DEFAULT_MIN_INTERVAL})")
|
||||
|
||||
# -- Dialog opens pre-filled with current settings
|
||||
dialog = app_main.ClapSettingsDialog(win.clap_threshold, win.clap_min_interval)
|
||||
assert dialog.threshold_spin.value() == DEFAULT_THRESHOLD
|
||||
assert dialog.min_interval_spin.value() == DEFAULT_MIN_INTERVAL
|
||||
|
||||
# -- Changing values and cancelling shouldn't touch MainWindow state
|
||||
dialog.threshold_spin.setValue(0.3)
|
||||
dialog.min_interval_spin.setValue(1.5)
|
||||
dialog.reject()
|
||||
assert win.clap_threshold == DEFAULT_THRESHOLD, "cancel should not have changed MainWindow state"
|
||||
print("PASS: Cancel leaves settings untouched")
|
||||
|
||||
# -- Changing values and accepting (simulated via the real handler) does
|
||||
# update MainWindow state and gets used by the actual worker.
|
||||
dialog2 = app_main.ClapSettingsDialog(win.clap_threshold, win.clap_min_interval)
|
||||
dialog2.threshold_spin.setValue(0.3) # very sensitive
|
||||
dialog2.min_interval_spin.setValue(0.1)
|
||||
win.clap_threshold, win.clap_min_interval = dialog2.values()
|
||||
assert win.clap_threshold == 0.3
|
||||
assert win.clap_min_interval == 0.1
|
||||
print("PASS: Accepting new values updates MainWindow state")
|
||||
|
||||
# -- Reset to Defaults button restores the shared constants regardless
|
||||
# of what was passed in (even if MainWindow itself had non-default
|
||||
# settings already).
|
||||
dialog3 = app_main.ClapSettingsDialog(0.3, 0.1)
|
||||
assert dialog3.threshold_spin.value() == 0.3
|
||||
dialog3._reset_to_defaults()
|
||||
assert dialog3.threshold_spin.value() == DEFAULT_THRESHOLD
|
||||
assert dialog3.min_interval_spin.value() == DEFAULT_MIN_INTERVAL
|
||||
print("PASS: Reset to Defaults restores DEFAULT_THRESHOLD/DEFAULT_MIN_INTERVAL")
|
||||
|
||||
# -- End-to-end: a changed (very sensitive) threshold actually reaches
|
||||
# the real DetectWorker/detect_claps() call for a real detection run.
|
||||
win._load(CLIP)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
|
||||
win.clap_threshold = 0.3
|
||||
win.clap_min_interval = 0.1
|
||||
win.detect_all_button.click()
|
||||
qapp.processEvents()
|
||||
assert win._detect_worker.threshold == 0.3
|
||||
assert win._detect_worker.min_interval == 0.1
|
||||
ok = wait_for(lambda: win.detect_all_button.isEnabled(), qapp)
|
||||
assert ok, "detection did not finish in time"
|
||||
print("PASS: changed settings actually reach the DetectWorker for a real run")
|
||||
|
||||
win.close()
|
||||
qapp.processEvents()
|
||||
print("\nCLAP SETTINGS SMOKE TEST PASSED")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -30,13 +30,19 @@ def main():
|
||||
win.show()
|
||||
qapp.processEvents()
|
||||
|
||||
# -- Tools menu has the clap detection actions, no buttons in the layout
|
||||
# -- Tools menu has the clap detection actions, AND both Detect All Claps
|
||||
# and Detect Next Clap are also buttons in the Sync markers panel (each
|
||||
# kept in sync with its menu action's enabled state, since they're two
|
||||
# separate widgets).
|
||||
menu_titles = [a.text() for a in win.menuBar().actions()]
|
||||
assert "&Tools" in menu_titles, f"expected a Tools menu, got {menu_titles}"
|
||||
assert not hasattr(win, "detect_all_button"), "old button attribute should be gone"
|
||||
assert win.detect_all_action.text() == "Detect All Claps"
|
||||
assert win.detect_next_action.text() == "Detect Next Clap"
|
||||
print("PASS: Tools menu has plain-labeled clap detection actions")
|
||||
assert win.detect_all_button.text() == "Detect All Claps"
|
||||
assert win.detect_next_button.text() == "Detect Next Clap"
|
||||
assert win.detect_all_action.isEnabled() == win.detect_all_button.isEnabled()
|
||||
assert win.detect_next_action.isEnabled() == win.detect_next_button.isEnabled()
|
||||
print("PASS: Tools menu has plain-labeled clap detection actions, both buttons back in sync markers panel")
|
||||
|
||||
# -- Load via drag-and-drop simulation on the player widget
|
||||
mime = QMimeData()
|
||||
@@ -46,7 +52,7 @@ def main():
|
||||
Qt.MouseButton.LeftButton, Qt.KeyboardModifier.NoModifier,
|
||||
QDropEvent.Type.Drop,
|
||||
)
|
||||
win.player.dropEvent(drop_event)
|
||||
win.source_panel.player.dropEvent(drop_event)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.3)
|
||||
qapp.processEvents()
|
||||
@@ -54,42 +60,43 @@ def main():
|
||||
print(f"PASS: drag-and-drop loaded {Path(win.input_path).name}")
|
||||
|
||||
# -- Clicking directly on the slider track (not the handle) jumps there
|
||||
duration_ms = win.seek_slider.maximum()
|
||||
duration_ms = win.source_panel.seek_slider.maximum()
|
||||
assert duration_ms > 0
|
||||
target_x = win.seek_slider.width() * 0.5
|
||||
click_pos = QPointF(target_x, win.seek_slider.height() / 2)
|
||||
target_x = win.source_panel.seek_slider.width() * 0.5
|
||||
click_pos = QPointF(target_x, win.source_panel.seek_slider.height() / 2)
|
||||
press_event = QMouseEvent(
|
||||
QMouseEvent.Type.MouseButtonPress, click_pos, click_pos,
|
||||
Qt.MouseButton.LeftButton, Qt.MouseButton.LeftButton, Qt.KeyboardModifier.NoModifier,
|
||||
)
|
||||
before = win.player.get_position()
|
||||
win.seek_slider.mousePressEvent(press_event)
|
||||
before = win.source_panel.player.get_position()
|
||||
win.source_panel.seek_slider.mousePressEvent(press_event)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
after = win.player.get_position()
|
||||
after = win.source_panel.player.get_position()
|
||||
expected = 0.5 * (duration_ms / 1000.0)
|
||||
print(f"clicked slider track at 50%: before={before:.3f}s after={after:.3f}s expected~{expected:.3f}s")
|
||||
assert abs(after - expected) < 0.5, "clicking the track should jump the player near that position"
|
||||
print("PASS: click-anywhere-on-track seeking works")
|
||||
|
||||
# -- Dragging (sliderMoved) seeks live, without waiting for release
|
||||
win.seek_slider.sliderPressed.emit()
|
||||
win.source_panel.seek_slider.sliderPressed.emit()
|
||||
qapp.processEvents()
|
||||
drag_value = int(0.2 * duration_ms)
|
||||
win.seek_slider.setValue(drag_value)
|
||||
win.seek_slider.sliderMoved.emit(drag_value)
|
||||
win.source_panel.seek_slider.setValue(drag_value)
|
||||
win.source_panel.seek_slider.sliderMoved.emit(drag_value)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
live_pos = win.player.get_position()
|
||||
live_pos = win.source_panel.player.get_position()
|
||||
expected_drag = drag_value / 1000.0
|
||||
print(f"dragged to 20% (still 'pressed', not released): live player position={live_pos:.3f}s expected~{expected_drag:.3f}s")
|
||||
assert abs(live_pos - expected_drag) < 0.5, "dragging should seek immediately, before release"
|
||||
win.seek_slider.sliderReleased.emit()
|
||||
win.source_panel.seek_slider.sliderReleased.emit()
|
||||
print("PASS: live-seek during drag works")
|
||||
|
||||
win.player.shutdown()
|
||||
win.source_panel.shutdown()
|
||||
win.preview_panel.shutdown()
|
||||
print("\nALL INTERACTION SMOKE TESTS PASSED")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Verifies clap detection analyzes whichever audio track mpv is actually
|
||||
playing, not whichever one ffmpeg's own default-stream heuristic would
|
||||
pick. Uses a synthetic fixture with two audio tracks with claps at very
|
||||
different times (3s vs 7s) -- if detection ignored mpv's active track
|
||||
selection, switching mpv to the second track would still incorrectly
|
||||
report the first track's 3s clap instead of the second track's 7s one.
|
||||
|
||||
This directly tests the bug hypothesis from user report: 'sometimes too
|
||||
early, sometimes too late, by different amounts' -- exactly what you'd
|
||||
see if detection sometimes analyzed the wrong audio track on
|
||||
multi-audio-track files, since a wrong-track clap position is essentially
|
||||
unrelated to the true one (not a small timestamp-math error).
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from PyQt6.QtWidgets import QApplication
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
import main as app_main
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
CLIP = str(ROOT / "test_media" / "sync_test_MULTITRACK.mp4")
|
||||
|
||||
|
||||
def wait_for(predicate, qapp, timeout=20):
|
||||
deadline = time.time() + timeout
|
||||
while not predicate() and time.time() < deadline:
|
||||
qapp.processEvents()
|
||||
time.sleep(0.05)
|
||||
return predicate()
|
||||
|
||||
|
||||
def main():
|
||||
qapp = QApplication(sys.argv)
|
||||
win = app_main.MainWindow()
|
||||
win.show()
|
||||
qapp.processEvents()
|
||||
|
||||
win._load(CLIP)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.3)
|
||||
qapp.processEvents()
|
||||
|
||||
mpv = win.source_panel.player.mpv
|
||||
print(f"default aid: {mpv.aid}")
|
||||
default_index = win._current_audio_stream_index()
|
||||
print(f"resolved ff-index for default track: {default_index}")
|
||||
assert default_index == 1, f"expected default track's ff-index to be 1, got {default_index}"
|
||||
|
||||
win.detect_all_button.click()
|
||||
qapp.processEvents()
|
||||
ok = wait_for(lambda: win.detect_all_button.isEnabled(), qapp)
|
||||
assert ok, "detection did not finish in time"
|
||||
markers = win.source_panel.marker_strip.markers()
|
||||
print(f"markers with default track selected: {markers}")
|
||||
assert any(abs(m - 3.0) < 0.5 for m in markers), f"expected a marker near 3.0s (default track's clap): {markers}"
|
||||
assert not any(abs(m - 7.0) < 0.5 for m in markers), f"should not see the OTHER track's 7.0s clap yet: {markers}"
|
||||
|
||||
# Now switch mpv to the second audio track (aid=2), simulating a file
|
||||
# where the user is listening to a non-default track, and rerun
|
||||
# detection -- it must follow the switch, not keep reporting the first
|
||||
# track's clap.
|
||||
mpv.aid = 2
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
switched_index = win._current_audio_stream_index()
|
||||
print(f"resolved ff-index after switching to aid=2: {switched_index}")
|
||||
assert switched_index == 2, f"expected ff-index 2 after switching tracks, got {switched_index}"
|
||||
|
||||
win.source_panel.marker_strip.clear()
|
||||
win.detect_all_button.click()
|
||||
qapp.processEvents()
|
||||
ok = wait_for(lambda: win.detect_all_button.isEnabled(), qapp)
|
||||
assert ok, "detection did not finish in time"
|
||||
markers = win.source_panel.marker_strip.markers()
|
||||
print(f"markers with second track selected: {markers}")
|
||||
assert any(abs(m - 7.0) < 0.5 for m in markers), f"expected a marker near 7.0s (second track's clap) after switching: {markers}"
|
||||
assert not any(abs(m - 3.0) < 0.5 for m in markers), f"should not see the FIRST track's 3.0s clap after switching: {markers}"
|
||||
|
||||
win.close()
|
||||
qapp.processEvents()
|
||||
print("\nPASS: clap detection follows mpv's actual selected audio track, not ffmpeg's default")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,14 +1,14 @@
|
||||
"""Verifies arrow-key skips preserve whichever play/pause state the user
|
||||
was already in, instead of always forcing a pause.
|
||||
|
||||
Calls the real MainWindow._seek_relative() method directly rather than
|
||||
Calls the real PlayerPanel.seek_relative() method directly rather than
|
||||
simulating the QShortcut keypress: QTest.keyClick against this shortcut was
|
||||
observed to be unreliable in this headless setup even with Qt's
|
||||
active-window state forced (see smoke_test_arrow_seek.py's note) --
|
||||
sometimes the simulated key silently doesn't reach the shortcut at all, and
|
||||
a debug script isolated that PlayerWidget.seek() itself works correctly
|
||||
while playing, so the flakiness is in the synthetic-keypress dispatch, not
|
||||
the app. The QShortcut -> _seek_relative wiring itself (a single line) was
|
||||
the app. The QShortcut -> seek_relative wiring itself (a single line) was
|
||||
already independently confirmed with real xdotool keypresses against a
|
||||
live window in an earlier session.
|
||||
"""
|
||||
@@ -37,37 +37,38 @@ def main():
|
||||
qapp.processEvents()
|
||||
|
||||
# -- Case 1: paused before the skip -> stays paused after
|
||||
assert win.player.mpv.pause is True
|
||||
before_pos = win.player.get_position()
|
||||
win._seek_relative(5.0)
|
||||
assert win.source_panel.player.mpv.pause is True
|
||||
before_pos = win.source_panel.player.get_position()
|
||||
win.source_panel.seek_relative(5.0)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
after_pos = win.player.get_position()
|
||||
after_pos = win.source_panel.player.get_position()
|
||||
print(f"paused case: before={before_pos:.3f}s after={after_pos:.3f}s")
|
||||
assert win.player.mpv.pause is True, "should still be paused after skip"
|
||||
assert win.play_button.text() == "Play"
|
||||
assert win.source_panel.player.mpv.pause is True, "should still be paused after skip"
|
||||
assert win.source_panel.play_button.text() == "Play"
|
||||
assert after_pos > before_pos + 4.0, "should have actually skipped forward"
|
||||
print("PASS: stays paused across arrow-key skip when it started paused")
|
||||
|
||||
# -- Case 2: playing before the skip -> stays playing after
|
||||
win.player.seek(0.0)
|
||||
win.player.play()
|
||||
win.play_button.setText("Pause")
|
||||
win.source_panel.player.seek(0.0)
|
||||
win.source_panel.player.play()
|
||||
win.source_panel.play_button.setText("Pause")
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
before_pos = win.player.get_position()
|
||||
win._seek_relative(5.0)
|
||||
before_pos = win.source_panel.player.get_position()
|
||||
win.source_panel.seek_relative(5.0)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
assert win.player.mpv.pause is False, "should still be playing after skip"
|
||||
assert win.play_button.text() == "Pause"
|
||||
after_pos = win.player.get_position()
|
||||
assert win.source_panel.player.mpv.pause is False, "should still be playing after skip"
|
||||
assert win.source_panel.play_button.text() == "Pause"
|
||||
after_pos = win.source_panel.player.get_position()
|
||||
print(f"playing case: before={before_pos:.3f}s after={after_pos:.3f}s (should have jumped ~5s and kept playing)")
|
||||
assert after_pos > before_pos + 4.0, "should have actually skipped forward"
|
||||
|
||||
win.player.shutdown()
|
||||
win.source_panel.shutdown()
|
||||
win.preview_panel.shutdown()
|
||||
print("\nPRESERVE PLAY STATE SMOKE TEST PASSED")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Verifies the sync preview feature: building a short offset-corrected
|
||||
clip and playing it in the Preview tab (Kdenlive-style project/clip
|
||||
monitor tabs) instead of a popup dialog, plus the Sync markers panel
|
||||
showing Mark Audio Event before Mark Video Event with a Detect All Claps
|
||||
button alongside them.
|
||||
"""
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from PyQt6.QtWidgets import QApplication
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
import main as app_main
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
CLIP = str(ROOT / "test_media" / "sync_test_OUTOFSYNC_audio+350ms.mp4")
|
||||
|
||||
|
||||
def main():
|
||||
qapp = QApplication(sys.argv)
|
||||
win = app_main.MainWindow()
|
||||
win.show()
|
||||
qapp.processEvents()
|
||||
|
||||
# -- Marker panel order: Mark Audio Event before Mark Video Event, with
|
||||
# Detect All Claps in the same panel.
|
||||
marker_layout = win.mark_audio_button.parentWidget().layout()
|
||||
widgets = [marker_layout.itemAt(i).widget() for i in range(marker_layout.count())]
|
||||
widgets = [w for w in widgets if w is not None]
|
||||
audio_idx = widgets.index(win.mark_audio_button)
|
||||
video_idx = widgets.index(win.mark_video_button)
|
||||
detect_idx = widgets.index(win.detect_all_button)
|
||||
assert audio_idx < video_idx, "Mark Audio Event should come before Mark Video Event"
|
||||
print(f"PASS: marker panel order is audio({audio_idx}) < video({video_idx}), detect button present at {detect_idx}")
|
||||
|
||||
# -- Two tabs: Source (main player) and Preview
|
||||
assert win.video_tabs.count() == 2
|
||||
assert win.video_tabs.tabText(0) == "Source"
|
||||
assert win.video_tabs.tabText(1) == "Preview"
|
||||
assert win.video_tabs.widget(0) is win.source_panel
|
||||
assert win.video_tabs.widget(1) is win.preview_panel
|
||||
print("PASS: video area has Source and Preview tabs")
|
||||
|
||||
win._load(CLIP)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
assert win.video_tabs.currentIndex() == 0, "loading a new source should land on the Source tab"
|
||||
|
||||
win.source_panel.player.seek(5.000)
|
||||
qapp.processEvents()
|
||||
win._mark_video()
|
||||
|
||||
win.source_panel.player.seek(5.350)
|
||||
qapp.processEvents()
|
||||
win._mark_audio()
|
||||
|
||||
assert win.preview_button.isEnabled()
|
||||
win.preview_button.click()
|
||||
qapp.processEvents()
|
||||
assert not win.preview_button.isEnabled(), "preview button should disable while building"
|
||||
pending_path = win._pending_preview_path
|
||||
print(f"building preview clip at: {pending_path}")
|
||||
|
||||
deadline = time.time() + 20
|
||||
while win.video_tabs.currentIndex() != 1 and time.time() < deadline:
|
||||
qapp.processEvents()
|
||||
time.sleep(0.05)
|
||||
|
||||
assert win.video_tabs.currentIndex() == 1, "finishing a preview should switch to the Preview tab"
|
||||
assert win.preview_button.isEnabled()
|
||||
assert win._preview_temp_path == pending_path
|
||||
assert Path(win._preview_temp_path).exists()
|
||||
print("PASS: preview build switched to the Preview tab automatically")
|
||||
|
||||
qapp.processEvents()
|
||||
time.sleep(0.3)
|
||||
qapp.processEvents()
|
||||
|
||||
duration = win.preview_panel.player.mpv.duration
|
||||
print(f"preview clip duration: {duration:.3f}s (expected ~{win.PREVIEW_DURATION_SEC}s, clamped to file bounds)")
|
||||
assert duration is not None and 0 < duration <= win.PREVIEW_DURATION_SEC + 0.5
|
||||
assert win.preview_panel.player.mpv.pause is False, "preview should auto-play"
|
||||
assert win.preview_panel.play_button.text() == "Pause"
|
||||
|
||||
# -- The Preview tab has its own functional timeline (not a static video
|
||||
# + a single play button): its own seek slider, sized to the preview
|
||||
# clip's own duration, that seeks the preview player independently of
|
||||
# the Source tab's slider/player.
|
||||
assert win.preview_panel.seek_slider is not win.source_panel.seek_slider
|
||||
preview_duration_ms = win.preview_panel.seek_slider.maximum()
|
||||
assert preview_duration_ms > 0
|
||||
print(f"Preview tab's own seek slider range: 0..{preview_duration_ms}ms")
|
||||
target_ms = int(preview_duration_ms * 0.7)
|
||||
win.preview_panel.seek_slider.setValue(target_ms)
|
||||
win.preview_panel.seek_slider.sliderMoved.emit(target_ms)
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
preview_pos = win.preview_panel.player.get_position()
|
||||
source_pos = win.source_panel.player.get_position()
|
||||
print(f"dragged Preview tab's own slider to 70%: preview player now at {preview_pos:.3f}s, source player still at {source_pos:.3f}s")
|
||||
assert abs(preview_pos - target_ms / 1000.0) < 0.5, "Preview tab's slider should seek the preview player"
|
||||
print("PASS: Preview tab has its own independent, functional timeline")
|
||||
|
||||
# -- Switching back to Source shouldn't disturb the main player, and the
|
||||
# preview player keeps running independently (two live mpv instances).
|
||||
main_pos_before_switch = win.source_panel.player.get_position()
|
||||
win.video_tabs.setCurrentIndex(0)
|
||||
qapp.processEvents()
|
||||
main_pos_after_switch = win.source_panel.player.get_position()
|
||||
print(f"main player position: before switch={main_pos_before_switch:.3f}s after={main_pos_after_switch:.3f}s")
|
||||
assert abs(main_pos_after_switch - main_pos_before_switch) < 0.1, "switching tabs shouldn't move the main player"
|
||||
assert win.source_panel.player.mpv.pause is True, "main player should still be paused (untouched by preview building)"
|
||||
print("PASS: switching tabs doesn't disturb the main player")
|
||||
|
||||
# -- Building a second preview swaps the file without leaking the first
|
||||
# temp file.
|
||||
first_preview_path = win._preview_temp_path
|
||||
win.preview_button.click()
|
||||
qapp.processEvents()
|
||||
deadline = time.time() + 20
|
||||
while win._pending_preview_path is not None and time.time() < deadline:
|
||||
qapp.processEvents()
|
||||
time.sleep(0.05)
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
assert not Path(first_preview_path).exists(), "first preview temp file should be cleaned up after being replaced"
|
||||
assert Path(win._preview_temp_path).exists()
|
||||
print("PASS: rebuilding the preview cleans up the previous temp file")
|
||||
|
||||
win.close()
|
||||
qapp.processEvents()
|
||||
assert not Path(win._preview_temp_path).exists() if win._preview_temp_path else True
|
||||
|
||||
print("\nPREVIEW TAB SMOKE TEST PASSED")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -42,24 +42,25 @@ def main():
|
||||
print("PASS: Load File lives in the File menu")
|
||||
|
||||
# -- Right-click on the player toggles play/pause
|
||||
assert win.player.mpv.pause is True, "should start paused"
|
||||
win.player.mpv.command("keypress", "MBTN_RIGHT")
|
||||
assert win.source_panel.player.mpv.pause is True, "should start paused"
|
||||
win.source_panel.player.mpv.command("keypress", "MBTN_RIGHT")
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
assert win.player.mpv.pause is False, "right-click should have started playback"
|
||||
assert win.play_button.text() == "Pause"
|
||||
assert win.source_panel.player.mpv.pause is False, "right-click should have started playback"
|
||||
assert win.source_panel.play_button.text() == "Pause"
|
||||
print("PASS: right-click on player started playback")
|
||||
|
||||
win.player.mpv.command("keypress", "MBTN_RIGHT")
|
||||
win.source_panel.player.mpv.command("keypress", "MBTN_RIGHT")
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
assert win.player.mpv.pause is True, "second right-click should pause again"
|
||||
assert win.play_button.text() == "Play"
|
||||
assert win.source_panel.player.mpv.pause is True, "second right-click should pause again"
|
||||
assert win.source_panel.play_button.text() == "Play"
|
||||
print("PASS: right-click on player paused again")
|
||||
|
||||
win.player.shutdown()
|
||||
win.source_panel.shutdown()
|
||||
win.preview_panel.shutdown()
|
||||
print("\nRIGHT-CLICK / FILE MENU SMOKE TEST PASSED")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,157 @@
|
||||
"""Verifies the stream-ends offset estimate: it lands on the known offset of
|
||||
each synthetic fixture, it feeds Preview/Export exactly like a
|
||||
marker-derived offset does (including with no markers set at all), and
|
||||
Use Marker Offset brings the marker-derived value back afterwards without
|
||||
re-marking -- while the manual marker workflow itself behaves as before.
|
||||
|
||||
Also covers the real-world shape this feature exists for, which none of
|
||||
the fixtures in test_media have: both tracks start at t=0 but the audio
|
||||
ends early (it started capturing late, and the recorder stopped both
|
||||
together). Generated on the fly as a 60s clip, long enough that the probe
|
||||
only reads the tail window instead of the whole file.
|
||||
"""
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
from PyQt6.QtWidgets import QApplication
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
import main as app_main
|
||||
from stream_edges import probe_stream_edges, TAIL_WINDOW_SEC
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
MEDIA = ROOT / "test_media"
|
||||
|
||||
|
||||
def wait_for(predicate, qapp, timeout=20):
|
||||
deadline = time.time() + timeout
|
||||
while not predicate() and time.time() < deadline:
|
||||
qapp.processEvents()
|
||||
time.sleep(0.05)
|
||||
return predicate()
|
||||
|
||||
|
||||
def load(win, qapp, path):
|
||||
win._load(str(path))
|
||||
qapp.processEvents()
|
||||
time.sleep(0.2)
|
||||
qapp.processEvents()
|
||||
|
||||
|
||||
def estimate(win, qapp):
|
||||
win.estimate_button.click()
|
||||
qapp.processEvents()
|
||||
assert wait_for(lambda: win.estimate_button.isEnabled(), qapp), "estimate did not finish in time"
|
||||
|
||||
|
||||
def make_audio_ends_early_clip(path: Path):
|
||||
subprocess.run([
|
||||
"ffmpeg", "-y", "-v", "error",
|
||||
"-f", "lavfi", "-i", "testsrc=size=320x240:rate=30:duration=60",
|
||||
"-f", "lavfi", "-i", "sine=frequency=1000:duration=59",
|
||||
"-c:v", "libx264", "-preset", "ultrafast", "-c:a", "aac",
|
||||
str(path),
|
||||
], check=True)
|
||||
|
||||
|
||||
def main():
|
||||
# -- Probe alone, against every fixture's known offset.
|
||||
expected = {
|
||||
"sync_test_INSYNC.mp4": 0,
|
||||
"sync_test_OUTOFSYNC_audio+350ms.mp4": -350,
|
||||
"sync_test_OUTOFSYNC_video+350ms.mp4": 350,
|
||||
}
|
||||
for name, want in expected.items():
|
||||
edges = probe_stream_edges(str(MEDIA / name))
|
||||
print(f"{name}: end-based {edges.end_delay_ms:+.1f} ms, start-based {edges.start_delay_ms:+.1f} ms")
|
||||
assert abs(edges.end_delay_ms - want) < 5, f"{name}: expected ~{want} ms"
|
||||
print("PASS: end-based estimate matches every fixture's known offset")
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
clip = Path(tmp) / "audio_ends_early.mp4"
|
||||
make_audio_ends_early_clip(clip)
|
||||
assert 60 > TAIL_WINDOW_SEC, "clip should be long enough to exercise the tail-window read"
|
||||
edges = probe_stream_edges(str(clip))
|
||||
print(f"audio-ends-early clip: {edges}, end-based {edges.end_delay_ms:+.1f} ms")
|
||||
assert abs(edges.end_delay_ms - 1000) < 30, "expected ~+1000 ms (delay audio)"
|
||||
assert edges.estimates_disagree, "starts both at 0 but ends 1s apart -- should flag disagreement"
|
||||
print("PASS: audio-ends-early clip estimates ~+1000 ms and flags the start/end disagreement")
|
||||
|
||||
qapp = QApplication(sys.argv)
|
||||
win = app_main.MainWindow()
|
||||
win.show()
|
||||
qapp.processEvents()
|
||||
|
||||
assert not win.estimate_button.isEnabled(), "estimate needs a loaded file"
|
||||
assert not win.use_marker_offset_button.isEnabled()
|
||||
|
||||
# -- Estimate with no markers: enables Preview/Export on its own.
|
||||
load(win, qapp, MEDIA / "sync_test_OUTOFSYNC_video+350ms.mp4")
|
||||
assert win.estimate_button.isEnabled() and win.estimate_action.isEnabled()
|
||||
assert not win.export_button.isEnabled() and not win.preview_button.isEnabled()
|
||||
estimate(win, qapp)
|
||||
print(f"offset label after estimate: {win.offset_label.text()!r}")
|
||||
assert win.offset.delay_ms == 350
|
||||
assert win.override_spinbox.value() == 350 and win.override_spinbox.isEnabled()
|
||||
assert win.export_button.isEnabled() and win.preview_button.isEnabled()
|
||||
assert "estimated from stream ends" in win.offset_label.text()
|
||||
assert not win.use_marker_offset_button.isEnabled(), "no markers yet, nothing to go back to"
|
||||
assert win.video_mark is None and win.audio_mark is None, "estimate must not touch the markers"
|
||||
print("PASS: estimate with no markers sets the offset and enables Preview/Export")
|
||||
|
||||
# -- Preview works from an estimate with no markers (centered on the playhead).
|
||||
win.source_panel.player.seek(5.0)
|
||||
qapp.processEvents()
|
||||
win.preview_button.click()
|
||||
qapp.processEvents()
|
||||
assert wait_for(lambda: win.video_tabs.currentIndex() == 1, qapp), "preview did not build"
|
||||
print("PASS: Preview builds from an estimated offset with no markers set")
|
||||
win.video_tabs.setCurrentIndex(0)
|
||||
|
||||
# -- Manual markers behave as before and override the estimate.
|
||||
load(win, qapp, MEDIA / "sync_test_OUTOFSYNC_audio+350ms.mp4")
|
||||
assert win.offset_label.text() == "Offset: — (mark both events)"
|
||||
win.source_panel.player.seek(5.000)
|
||||
qapp.processEvents()
|
||||
win._mark_video()
|
||||
win.source_panel.player.seek(5.350)
|
||||
qapp.processEvents()
|
||||
win._mark_audio()
|
||||
marker_label = win.offset_label.text()
|
||||
print(f"offset label from markers: {marker_label!r}")
|
||||
assert abs(win.offset.delay_ms - (-350)) < 40
|
||||
assert marker_label == win.offset.describe()
|
||||
assert win.use_marker_offset_button.isEnabled()
|
||||
marker_delay = win.offset.delay_ms
|
||||
print("PASS: manual markers set the offset exactly as before")
|
||||
|
||||
# -- Estimate replaces it; Use Marker Offset restores it without re-marking.
|
||||
estimate(win, qapp)
|
||||
assert win.offset.delay_ms == -349, f"got {win.offset.delay_ms}"
|
||||
assert win.override_spinbox.value() == -349
|
||||
assert win.video_mark is not None and win.audio_mark is not None, "estimate must not clear markers"
|
||||
win.use_marker_offset_button.click()
|
||||
qapp.processEvents()
|
||||
assert win.offset.delay_ms == marker_delay
|
||||
assert win.override_spinbox.value() == round(marker_delay)
|
||||
assert win.offset_label.text() == marker_label
|
||||
print("PASS: Use Marker Offset restores the marker-derived offset after an estimate")
|
||||
|
||||
# -- Loading a new file resets everything, including Use Marker Offset.
|
||||
load(win, qapp, MEDIA / "sync_test_INSYNC.mp4")
|
||||
assert not win.use_marker_offset_button.isEnabled()
|
||||
assert not win.export_button.isEnabled()
|
||||
estimate(win, qapp)
|
||||
assert win.offset.delay_ms == 0
|
||||
print("PASS: new file resets marker state; in-sync file estimates 0 ms")
|
||||
|
||||
win.close()
|
||||
qapp.processEvents()
|
||||
print("\nSTREAM ENDS ESTIMATE SMOKE TEST PASSED")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -30,14 +30,14 @@ def main():
|
||||
qapp.processEvents()
|
||||
time.sleep(0.3)
|
||||
qapp.processEvents()
|
||||
print(f"loaded: duration slider range = {win.seek_slider.maximum()}ms")
|
||||
print(f"loaded: duration slider range = {win.source_panel.seek_slider.maximum()}ms")
|
||||
|
||||
win.player.seek(5.000)
|
||||
win.source_panel.player.seek(5.000)
|
||||
qapp.processEvents()
|
||||
win._mark_video()
|
||||
print(f"video mark: {win.video_mark:.3f}s -> label: {win.video_mark_label.text()}")
|
||||
|
||||
win.player.seek(5.350)
|
||||
win.source_panel.player.seek(5.350)
|
||||
qapp.processEvents()
|
||||
win._mark_audio()
|
||||
print(f"audio mark: {win.audio_mark:.3f}s -> label: {win.audio_mark_label.text()}")
|
||||
@@ -85,7 +85,8 @@ def main():
|
||||
assert Path(EXPORT_OUT).exists()
|
||||
print(f"exported file exists: {EXPORT_OUT}, size={Path(EXPORT_OUT).stat().st_size} bytes")
|
||||
|
||||
win.player.shutdown()
|
||||
win.source_panel.shutdown()
|
||||
win.preview_panel.shutdown()
|
||||
print("\nUI SMOKE TEST PASSED")
|
||||
|
||||
|
||||
|
||||
@@ -31,7 +31,8 @@ def main():
|
||||
qapp.processEvents()
|
||||
assert not win.log_view.isVisible()
|
||||
|
||||
win.player.shutdown()
|
||||
win.source_panel.shutdown()
|
||||
win.preview_panel.shutdown()
|
||||
print("PASS: View > Show ffmpeg Log toggles the log panel correctly")
|
||||
|
||||
|
||||
|
||||
@@ -54,11 +54,11 @@ def find_audio_event_time(path: Path) -> float:
|
||||
return times[0]
|
||||
|
||||
|
||||
def run_case(fixture_name: str, expected_target: str, output_name: str):
|
||||
def run_case(fixture_name: str, expected_target: str, output_name: str, clip_start=None, clip_duration=None):
|
||||
input_path = ROOT / "test_media" / fixture_name
|
||||
output_path = OUT_DIR / output_name
|
||||
|
||||
print(f"Input fixture: {fixture_name}")
|
||||
print(f"Input fixture: {fixture_name}" + (f" (clip {clip_start}+{clip_duration}s)" if clip_start else ""))
|
||||
video_mark = find_video_event_time(input_path)
|
||||
audio_mark = find_audio_event_time(input_path)
|
||||
print(f" detected flash at t={video_mark:.3f}s, beep at t={audio_mark:.3f}s")
|
||||
@@ -67,7 +67,7 @@ def run_case(fixture_name: str, expected_target: str, output_name: str):
|
||||
print(f" {offset.describe()}")
|
||||
assert offset.target == expected_target, f"expected {expected_target} branch, got {offset.target}"
|
||||
|
||||
result = export_synced(str(input_path), offset.delay_ms, str(output_path))
|
||||
result = export_synced(str(input_path), offset.delay_ms, str(output_path), clip_start=clip_start, clip_duration=clip_duration)
|
||||
print(f" ran: {' '.join(result.command)}")
|
||||
|
||||
after_video_t = find_video_event_time(output_path)
|
||||
@@ -84,7 +84,17 @@ def main():
|
||||
OUT_DIR.mkdir(exist_ok=True)
|
||||
run_case("sync_test_OUTOFSYNC_audio+350ms.mp4", "video", "_export_roundtrip_negative.mp4")
|
||||
run_case("sync_test_OUTOFSYNC_video+350ms.mp4", "audio", "_export_roundtrip_positive.mp4")
|
||||
print("PASS: both ffmpeg command branches (positive and negative delay) round-trip correctly.")
|
||||
# Clip mode (used by the GUI's preview feature) is a genuinely different
|
||||
# code path -- ffdelay's own documented quirk is that -ss/-t must be
|
||||
# applied to BOTH inputs, in the right position relative to -itsoffset,
|
||||
# or the clip covers different real-world windows per input and the
|
||||
# "preview" would silently show wrong sync. Exercise it explicitly
|
||||
# rather than assuming the full-file case generalizes.
|
||||
run_case(
|
||||
"sync_test_OUTOFSYNC_audio+350ms.mp4", "video", "_export_roundtrip_clip_mode.mp4",
|
||||
clip_start="2.000", clip_duration=8,
|
||||
)
|
||||
print("PASS: both ffmpeg command branches (positive and negative delay) round-trip correctly, including clip mode.")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+145
@@ -0,0 +1,145 @@
|
||||
"""Offset estimation from where the video and audio tracks start and end,
|
||||
as an optional alternative to marking a sync event by hand.
|
||||
|
||||
Premise: a recorder that stops both tracks at the same real-world moment,
|
||||
but whose audio started capturing late (or early) while still being
|
||||
written as starting at t=0, produces a file where the audio runs out
|
||||
before (or after) the video by exactly the sync error. So
|
||||
|
||||
delay_ms = (video_end - audio_end) * 1000
|
||||
|
||||
using the same sign convention as sync_logic.compute_offset (positive:
|
||||
audio is ahead and gets delayed; negative: video gets delayed).
|
||||
|
||||
This is only an *estimate*: a file whose audio genuinely stopped early
|
||||
(a mic cutting out before the camera did) looks identical but has nothing
|
||||
to do with sync. The start-based figure is reported alongside it as a
|
||||
cross-check -- it's what carries the offset when the container encodes it
|
||||
as a stream start_time instead, and a disagreement between the two is
|
||||
worth surfacing.
|
||||
"""
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
# How far before the end of the file to start reading packets. ffprobe
|
||||
# seeks there instead of scanning the whole file, which keeps this fast on
|
||||
# multi-hour recordings. If a track ends earlier than this window (so no
|
||||
# packets for it turn up), the whole track is scanned instead.
|
||||
TAIL_WINDOW_SEC = 30.0
|
||||
|
||||
# Start- and end-based estimates are compared against this. AAC encoder
|
||||
# delay alone typically puts the start ~20-45ms off, so anything much
|
||||
# tighter would warn on perfectly ordinary files.
|
||||
DISAGREEMENT_TOLERANCE_MS = 50.0
|
||||
|
||||
|
||||
class StreamEdgesError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class StreamEdges:
|
||||
video_start: float
|
||||
video_end: float
|
||||
audio_start: float
|
||||
audio_end: float
|
||||
|
||||
@property
|
||||
def end_delay_ms(self) -> float:
|
||||
return (self.video_end - self.audio_end) * 1000.0
|
||||
|
||||
@property
|
||||
def start_delay_ms(self) -> float:
|
||||
return (self.video_start - self.audio_start) * 1000.0
|
||||
|
||||
@property
|
||||
def estimates_disagree(self) -> bool:
|
||||
return abs(self.end_delay_ms - self.start_delay_ms) > DISAGREEMENT_TOLERANCE_MS
|
||||
|
||||
|
||||
def _ffprobe(args: List[str]) -> str:
|
||||
result = subprocess.run(["ffprobe", "-v", "error", *args], capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
raise StreamEdgesError(f"ffprobe failed: {result.stderr.strip()}")
|
||||
return result.stdout
|
||||
|
||||
|
||||
def _parse_float(value: str) -> Optional[float]:
|
||||
try:
|
||||
return float(value)
|
||||
except ValueError:
|
||||
return None # "N/A"
|
||||
|
||||
|
||||
def _format_duration(input_path: str) -> Optional[float]:
|
||||
out = _ffprobe(["-show_entries", "format=duration", "-of", "csv=p=0", input_path])
|
||||
return _parse_float(out.strip())
|
||||
|
||||
|
||||
def _stream_start(input_path: str, stream_specifier: str) -> float:
|
||||
out = _ffprobe([
|
||||
"-select_streams", stream_specifier,
|
||||
"-show_entries", "stream=start_time", "-of", "csv=p=0", input_path,
|
||||
])
|
||||
start = _parse_float(out.strip())
|
||||
return start if start is not None else 0.0
|
||||
|
||||
|
||||
def _packets(input_path: str, stream_specifier: str, read_from: Optional[float]) -> List[Tuple[float, Optional[float]]]:
|
||||
args = ["-select_streams", stream_specifier]
|
||||
if read_from is not None:
|
||||
args += ["-read_intervals", f"{read_from:.3f}%"]
|
||||
args += ["-show_entries", "packet=pts_time,duration_time", "-of", "csv=p=0", input_path]
|
||||
packets = []
|
||||
for line in _ffprobe(args).splitlines():
|
||||
fields = line.strip().rstrip(",").split(",")
|
||||
pts = _parse_float(fields[0]) if fields and fields[0] else None
|
||||
if pts is None:
|
||||
continue
|
||||
duration = _parse_float(fields[1]) if len(fields) > 1 else None
|
||||
packets.append((pts, duration))
|
||||
return packets
|
||||
|
||||
|
||||
def _stream_end(input_path: str, stream_specifier: str, file_duration: Optional[float]) -> float:
|
||||
"""End of the last packet (pts + duration), not the container's stored
|
||||
duration -- that's frequently missing (e.g. per-stream in MKV) or
|
||||
rounded, and this needs to be accurate to well under a frame."""
|
||||
read_from = None
|
||||
if file_duration is not None and file_duration > TAIL_WINDOW_SEC:
|
||||
read_from = file_duration - TAIL_WINDOW_SEC
|
||||
packets = _packets(input_path, stream_specifier, read_from)
|
||||
if not packets and read_from is not None:
|
||||
packets = _packets(input_path, stream_specifier, None)
|
||||
if not packets:
|
||||
raise StreamEdgesError(f"no packets found for stream {stream_specifier}")
|
||||
|
||||
# Max rather than last: video with B-frames isn't stored in pts order.
|
||||
ends = [pts + duration for pts, duration in packets if duration is not None]
|
||||
if ends:
|
||||
return max(ends)
|
||||
# No per-packet durations (some containers): fall back to the gap
|
||||
# between the last two packets as the final packet's duration.
|
||||
pts_sorted = sorted(pts for pts, _ in packets)
|
||||
step = pts_sorted[-1] - pts_sorted[-2] if len(pts_sorted) > 1 else 0.0
|
||||
return pts_sorted[-1] + step
|
||||
|
||||
|
||||
def probe_stream_edges(
|
||||
input_path: str,
|
||||
video_stream_index: Optional[int] = None,
|
||||
audio_stream_index: Optional[int] = None,
|
||||
) -> StreamEdges:
|
||||
"""Stream indices are ffmpeg global indices (mpv's `ff-index`), so the
|
||||
estimate uses the same tracks the player is showing -- see
|
||||
clap_detection.py for why that matters on multi-track files."""
|
||||
video_spec = f"{video_stream_index}" if video_stream_index is not None else "v:0"
|
||||
audio_spec = f"{audio_stream_index}" if audio_stream_index is not None else "a:0"
|
||||
file_duration = _format_duration(input_path)
|
||||
return StreamEdges(
|
||||
video_start=_stream_start(input_path, video_spec),
|
||||
video_end=_stream_end(input_path, video_spec, file_duration),
|
||||
audio_start=_stream_start(input_path, audio_spec),
|
||||
audio_end=_stream_end(input_path, audio_spec, file_duration),
|
||||
)
|
||||
@@ -31,3 +31,10 @@ validating marker capture, offset math, and lossless export.
|
||||
|
||||
Frame at exactly t=5.000s is frame index 150 (30fps) — useful for checking
|
||||
frame-step precision lands exactly on it.
|
||||
|
||||
- `sync_test_MULTITRACK.mp4` — 10s, blue video, two AAC audio tracks: track
|
||||
index 1 (default) has a beep at t=3.000s, track index 2 (not default) has
|
||||
a beep at t=7.000s. Used to verify clap detection follows whichever audio
|
||||
track mpv is actually playing (its `ff-index`), not whichever one
|
||||
ffmpeg's own default-stream selection would pick -- see
|
||||
scripts/smoke_test_multitrack_clap_detection.py.
|
||||
|
||||
Binary file not shown.
Reference in New Issue
Block a user