import os import sys import tempfile from pathlib import Path from typing import Optional 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, QTabWidget, ) from player_widget import PlayerWidget 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 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 ExportWorker(QThread): finished_ok = pyqtSignal(object) # ExportResult finished_err = pyqtSignal(str, str) # message, stderr 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, 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) 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): super().__init__() self.input_path = input_path self.start_offset = start_offset self.max_results = max_results def run(self): try: events = detect_claps( self.input_path, start_offset=self.start_offset, max_results=self.max_results, ) self.finished_ok.emit(events) except (ClapDetectionUnavailable, RuntimeError) as e: self.finished_err.emit(str(e)) class MainWindow(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("A/V Sync Marker") self.resize(960, 720) self.input_path: Optional[str] = None self.video_mark: Optional[float] = None self.audio_mark: Optional[float] = None self.offset: Optional[SyncOffset] = None self._export_worker: Optional[ExportWorker] = None self._detect_worker: Optional[DetectWorker] = None self._preview_worker: Optional[ExportWorker] = None self._preview_temp_path: Optional[str] = None # currently loaded in preview_player self._pending_preview_path: Optional[str] = None # being built by _preview_worker self._slider_being_dragged = False self.player = PlayerWidget() self.player.setMinimumHeight(400) self.player.fileDropped.connect(self._load) self.player.rightClicked.connect(self._toggle_play) # Independent second mpv instance for the Preview tab (Kdenlive-style # project/clip monitor tabs) -- kept alive for the whole session # rather than created per-preview, so switching tabs back and forth # doesn't tear down and rebuild an mpv instance each time. self.preview_player = PlayerWidget() self.preview_player.setMinimumHeight(320) self.preview_player.rightClicked.connect(self._toggle_preview_play) self._build_ui() self._build_menu() self._wire_shortcuts() self._position_timer = QTimer(self) self._position_timer.setInterval(100) self._position_timer.timeout.connect(self._update_position_ui) self._position_timer.start() # -- UI construction --------------------------------------------------- def _build_ui(self): central = QWidget() layout = QVBoxLayout(central) self.video_tabs = QTabWidget() self.video_tabs.addTab(self.player, "Source") preview_tab = QWidget() preview_tab_layout = QVBoxLayout(preview_tab) preview_tab_layout.setContentsMargins(0, 0, 0, 0) preview_tab_layout.addWidget(self.preview_player, stretch=1) preview_controls = QHBoxLayout() self.preview_play_button = QPushButton("Play") self.preview_play_button.clicked.connect(self._toggle_preview_play) preview_controls.addWidget(self.preview_play_button) self.preview_position_label = QLabel("00:00.000") preview_controls.addWidget(self.preview_position_label) preview_controls.addStretch(1) preview_tab_layout.addLayout(preview_controls) self.video_tabs.addTab(preview_tab, "Preview") layout.addWidget(self.video_tabs, 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) # Marker row marker_group = QGroupBox("Sync markers") marker_layout = QHBoxLayout(marker_group) 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) 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) layout.addWidget(marker_group) # 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) export_layout.addWidget(QLabel("Override (ms):")) self.override_spinbox = QDoubleSpinBox() self.override_spinbox.setRange(-600000, 600000) self.override_spinbox.setDecimals(0) 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) export_layout.addWidget(self.export_button) layout.addWidget(export_group) self.log_view = QPlainTextEdit() self.log_view.setReadOnly(True) self.log_view.setMaximumHeight(120) self.log_view.setPlaceholderText("ffmpeg output / warnings will appear here after export.") self.log_view.setVisible(False) layout.addWidget(self.log_view) self.setCentralWidget(central) def _build_menu(self): file_menu = self.menuBar().addMenu("&File") load_action = QAction("Load File…", self) load_action.triggered.connect(self._on_load_clicked) file_menu.addAction(load_action) view_menu = self.menuBar().addMenu("&View") self.toggle_log_action = QAction("Show ffmpeg Log", self, checkable=True, checked=False) self.toggle_log_action.toggled.connect(self.log_view.setVisible) view_menu.addAction(self.toggle_log_action) tools_menu = self.menuBar().addMenu("&Tools") self.detect_all_action = QAction("Detect All Claps", self) self.detect_all_action.setEnabled(False) self.detect_all_action.triggered.connect(self._on_detect_all_clicked) tools_menu.addAction(self.detect_all_action) self.detect_next_action = QAction("Detect Next Clap", self) self.detect_next_action.setEnabled(False) self.detect_next_action.triggered.connect(self._on_detect_next_clicked) tools_menu.addAction(self.detect_next_action) 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) # 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)) # -- transport ----------------------------------------------------- def _on_load_clicked(self): path, _ = QFileDialog.getOpenFileName(self, "Load video file", "", "Video files (*.mp4 *.mov *.mkv *.avi *.webm);;All files (*)") if path: 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. self.statusBar().showMessage(f"Loading {Path(path).name}…") QApplication.processEvents() self.input_path = path self.player.load(path) self.video_mark = None self.audio_mark = None self.offset = None self.video_mark_label.setText("video: —") 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.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.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) # 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 _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 _toggle_preview_play(self): if self.preview_player.mpv.pause: self.preview_player.play() self.preview_play_button.setText("Pause") else: self.preview_player.pause() self.preview_play_button.setText("Play") 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() def _update_position_ui(self): if self.input_path is not None: 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) if self._preview_temp_path is not None: self.preview_position_label.setText(format_timestamp(self.preview_player.get_position())) # -- markers / offset ----------------------------------------------- def _mark_video(self): self.video_mark = self.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_label.setText(f"audio: {format_timestamp(self.audio_mark)}") self._recompute_offset() def _recompute_offset(self): 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.override_spinbox.setEnabled(True) self.export_button.setEnabled(True) self.preview_button.setEnabled(True) # -- 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. center = (self.video_mark + self.audio_mark) / 2.0 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 player 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_player.load(new_path) self.preview_player.play() self.preview_play_button.setText("Pause") 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() self._start_detection(start_offset=cursor, max_results=1, label=f"Scanning from {format_timestamp(cursor)} for next clap…") 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.statusBar().showMessage(label) self._detect_worker = DetectWorker(self.input_path, start_offset=start_offset, max_results=max_results) 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) existing = set(self.marker_strip.markers()) existing.update(e.time_sec for e in events) self.marker_strip.set_markers(list(existing)) if not events: self.statusBar().showMessage("No claps found.", 5000) self.log_view.appendPlainText("Clap detection: no onsets found.") return self.statusBar().showMessage(f"Found {len(events)} clap(s) — click a marker above the timeline to jump to it.", 8000) for e in events: self.log_view.appendPlainText(f"Clap detected at {format_timestamp(e.time_sec)} (strength {e.strength:.2f})") 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.statusBar().showMessage("Detection failed (see log).", 8000) self.log_view.appendPlainText(f"Clap detection FAILED: {message}") # -- export ----------------------------------------------------- def _on_export_clicked(self): if self.input_path is None or self.offset is None: return 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}")) output_path, _ = QFileDialog.getSaveFileName(self, "Export synced file", suggested) if not output_path: return self.export_button.setEnabled(False) self.export_button.setText("Exporting…") self.log_view.appendPlainText(f"Exporting with delay_ms={delay_ms:.0f} …") self._export_worker = ExportWorker(self.input_path, delay_ms, output_path) self._export_worker.finished_ok.connect(self._on_export_done) self._export_worker.finished_err.connect(self._on_export_failed) self._export_worker.start() def _on_export_done(self, result: ExportResult): self.export_button.setEnabled(True) self.export_button.setText("Export…") self.log_view.appendPlainText("Export finished.") if result.stderr.strip(): # ffmpeg logs things like "sps_id out of range" on stream-copy # of screen-recording-style sources; harmless, just surfaced # for visibility rather than treated as failure. self.log_view.appendPlainText("ffmpeg log:\n" + result.stderr) QMessageBox.information(self, "Export complete", "Export finished successfully.") def _on_export_failed(self, message: str, stderr: str): self.export_button.setEnabled(True) self.export_button.setText("Export…") self.log_view.appendPlainText(f"Export FAILED: {message}\n{stderr}") QMessageBox.critical(self, "Export failed", f"{message}\n\n{stderr[-2000:]}") def closeEvent(self, event): # 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.player.shutdown() self.preview_player.shutdown() if self._preview_temp_path and os.path.exists(self._preview_temp_path): os.unlink(self._preview_temp_path) super().closeEvent(event) def main(): app = QApplication(sys.argv) window = MainWindow() window.show() if len(sys.argv) > 1: window._load(sys.argv[1]) sys.exit(app.exec()) if __name__ == "__main__": main()