ThePendulum
5a6c4e6e84
- Clap detection moved from inline buttons to a Tools menu, plain labels (no more '(whole file)' / '(from cursor)' parentheticals) -- status messages moved to the window's status bar since the dedicated label widget went away with the button row. - ClickableSlider subclass: clicking anywhere along the seek track jumps straight there instead of QSlider's default page-by-a-step behavior; reposition-then-delegate-to-super keeps native drag behavior intact. - Dragging now seeks live via sliderMoved instead of waiting for sliderReleased. - PlayerWidget accepts drag-and-drop of a local file to load it. Verified with real Qt events (QMouseEvent clicks, a QDropEvent, a real mouse click via xdotool opening the Tools menu on an actual running window) rather than just calling the underlying methods directly.
425 lines
16 KiB
Python
425 lines
16 KiB
Python
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._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 = 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)
|
|
|
|
# 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)
|
|
|
|
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)
|
|
|
|
# -- 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.statusBar().clearMessage()
|
|
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 _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()
|