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 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._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) # Transport row: play/pause, frame step, seek slider, position label transport = QHBoxLayout() self.load_button = QPushButton("Load File…") self.load_button.clicked.connect(self._on_load_clicked) transport.addWidget(self.load_button) self.play_button = QPushButton("Play") self.play_button.clicked.connect(self._toggle_play) transport.addWidget(self.play_button) self.prev_frame_button = QPushButton("⏮ Frame (,)") self.prev_frame_button.clicked.connect(lambda: self._step_frame(-1)) transport.addWidget(self.prev_frame_button) self.next_frame_button = QPushButton("Frame ⏭ (.)") self.next_frame_button.clicked.connect(lambda: self._step_frame(1)) transport.addWidget(self.next_frame_button) self.mute_checkbox = QCheckBox("Mute audio") self.mute_checkbox.toggled.connect(self._on_mute_toggled) transport.addWidget(self.mute_checkbox) layout.addLayout(transport) 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 = QSlider(Qt.Orientation.Horizontal) self.seek_slider.setRange(0, 0) self.seek_slider.sliderPressed.connect(lambda: setattr(self, "_slider_being_dragged", True)) 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) # Clap/onset auto-detect row detect_group = QGroupBox("Clap detection") detect_layout = QHBoxLayout(detect_group) self.detect_all_button = QPushButton("Detect All Claps (whole file)") self.detect_all_button.setEnabled(False) self.detect_all_button.clicked.connect(self._on_detect_all_clicked) detect_layout.addWidget(self.detect_all_button) self.detect_next_button = QPushButton("Detect Next Clap (from cursor)") self.detect_next_button.setEnabled(False) self.detect_next_button.clicked.connect(self._on_detect_next_clicked) detect_layout.addWidget(self.detect_next_button) self.detect_status_label = QLabel("") detect_layout.addWidget(self.detect_status_label, stretch=1) layout.addWidget(detect_group) # 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.") layout.addWidget(self.log_view) self.setCentralWidget(central) def _build_menu(self): view_menu = self.menuBar().addMenu("&View") self.toggle_log_action = QAction("Show ffmpeg Log", self, checkable=True, checked=True) self.toggle_log_action.toggled.connect(self.log_view.setVisible) view_menu.addAction(self.toggle_log_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) # -- 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): 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.detect_status_label.setText("") self.detect_all_button.setEnabled(True) self.detect_next_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 _on_slider_released(self): target_sec = self.seek_slider.value() / 1000.0 self.player.seek(target_sec) 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_button.setEnabled(False) self.detect_next_button.setEnabled(False) self.detect_status_label.setText(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_button.setEnabled(True) self.detect_next_button.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.detect_status_label.setText("No claps found.") self.log_view.appendPlainText("Clap detection: no onsets found.") return self.detect_status_label.setText(f"Found {len(events)} clap(s) — click a marker above the timeline to jump to it.") 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_button.setEnabled(True) self.detect_next_button.setEnabled(True) self.detect_status_label.setText("Detection failed (see log).") 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()