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, QDoubleSpinBox, QFileDialog, QMessageBox, QPlainTextEdit, QGroupBox, QTabWidget, QDialog, QDialogButtonBox, QFormLayout, ) 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, DEFAULT_THRESHOLD, DEFAULT_MIN_INTERVAL, ) 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") 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) 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, 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, 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: events = detect_claps( 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 MainWindow(QMainWindow): def __init__(self): super().__init__() 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 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_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 # 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() 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 _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) 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_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) 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 + 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) 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): # 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._active_panel().seek_relative(-5.0)) QShortcut(QKeySequence(Qt.Key.Key_Right), self, activated=lambda: self._active_panel().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): # 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.source_panel.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() 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) # 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): self.source_panel.update_position() self.preview_panel.update_position() # -- markers / offset ----------------------------------------------- def _mark_video(self): 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.source_panel.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 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_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.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, 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.source_panel.marker_strip.markers()) existing.update(e.time_sec for e in events) self.source_panel.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.detect_next_button.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) # 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 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.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) 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()