import sys 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, ) 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): super().__init__() self.input_path = input_path self.delay_ms = delay_ms self.output_path = output_path def run(self): try: result: ExportResult = export_synced(self.input_path, self.delay_ms, self.output_path) 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._slider_being_dragged = False self.player = PlayerWidget() self.player.setMinimumHeight(400) self.player.fileDropped.connect(self._load) self.player.rightClicked.connect(self._toggle_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) 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) # 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) layout.addWidget(marker_group) # Offset + export row export_group = QGroupBox("Offset 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.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.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_next_action.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): if self.input_path is None: return self.player.pause() self.play_button.setText("Play") 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 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) # -- 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) # -- 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_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_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_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): self.player.shutdown() 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()