677 lines
29 KiB
Python
677 lines
29 KiB
Python
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,
|
||
)
|
||
from stream_edges import probe_stream_edges, StreamEdges, StreamEdgesError
|
||
|
||
|
||
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 EstimateWorker(QThread):
|
||
finished_ok = pyqtSignal(object) # StreamEdges
|
||
finished_err = pyqtSignal(str)
|
||
|
||
def __init__(self, input_path, video_stream_index=None, audio_stream_index=None):
|
||
super().__init__()
|
||
self.input_path = input_path
|
||
self.video_stream_index = video_stream_index
|
||
self.audio_stream_index = audio_stream_index
|
||
|
||
def run(self):
|
||
try:
|
||
edges = probe_stream_edges(
|
||
self.input_path,
|
||
video_stream_index=self.video_stream_index,
|
||
audio_stream_index=self.audio_stream_index,
|
||
)
|
||
self.finished_ok.emit(edges)
|
||
except StreamEdgesError 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
|
||
# The offset currently feeding Preview/Export -- derived either from
|
||
# the two markers or from the stream-ends estimate, whichever was
|
||
# applied last.
|
||
self.offset: Optional[SyncOffset] = None
|
||
self._export_worker: Optional[ExportWorker] = None
|
||
self._detect_worker: Optional[DetectWorker] = None
|
||
self._estimate_worker: Optional[EstimateWorker] = 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)
|
||
|
||
# Re-applies the marker-derived offset after the stream-ends
|
||
# estimate replaced it, without having to re-mark an event.
|
||
self.use_marker_offset_button = QPushButton("Use Marker Offset")
|
||
self.use_marker_offset_button.setEnabled(False)
|
||
self.use_marker_offset_button.clicked.connect(self._recompute_offset)
|
||
marker_layout.addWidget(self.use_marker_offset_button)
|
||
|
||
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)
|
||
|
||
self.estimate_button = QPushButton("Estimate from Stream Ends")
|
||
self.estimate_button.setEnabled(False)
|
||
self.estimate_button.clicked.connect(self._on_estimate_clicked)
|
||
export_layout.addWidget(self.estimate_button)
|
||
|
||
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()
|
||
|
||
self.estimate_action = QAction("Estimate Offset from Stream Ends", self)
|
||
self.estimate_action.setEnabled(False)
|
||
self.estimate_action.triggered.connect(self._on_estimate_clicked)
|
||
tools_menu.addAction(self.estimate_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.use_marker_offset_button.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)
|
||
self.estimate_action.setEnabled(True)
|
||
self.estimate_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._apply_offset(self.offset, self.offset.describe())
|
||
self.use_marker_offset_button.setEnabled(True)
|
||
|
||
def _apply_offset(self, offset: SyncOffset, label: str):
|
||
"""Shared by both offset sources (markers, stream-ends estimate):
|
||
whichever is applied last is what Preview/Export use."""
|
||
self.offset = offset
|
||
self.offset_label.setText(label)
|
||
self.override_spinbox.setValue(offset.delay_ms)
|
||
self.override_spinbox.setEnabled(True)
|
||
self.export_button.setEnabled(True)
|
||
self.preview_button.setEnabled(True)
|
||
|
||
# -- stream-ends estimate ----------------------------------------------
|
||
|
||
def _current_video_stream_index(self) -> Optional[int]:
|
||
for track in self.source_panel.player.mpv.track_list:
|
||
if track.get("type") == "video" and track.get("selected"):
|
||
return track.get("ff-index")
|
||
return None
|
||
|
||
def _on_estimate_clicked(self):
|
||
if self.input_path is None:
|
||
return
|
||
self.estimate_action.setEnabled(False)
|
||
self.estimate_button.setEnabled(False)
|
||
self.statusBar().showMessage("Estimating offset from stream ends…")
|
||
|
||
self._estimate_worker = EstimateWorker(
|
||
self.input_path,
|
||
video_stream_index=self._current_video_stream_index(),
|
||
audio_stream_index=self._current_audio_stream_index(),
|
||
)
|
||
self._estimate_worker.finished_ok.connect(self._on_estimate_done)
|
||
self._estimate_worker.finished_err.connect(self._on_estimate_failed)
|
||
self._estimate_worker.start()
|
||
|
||
def _on_estimate_done(self, edges: StreamEdges):
|
||
self.estimate_action.setEnabled(True)
|
||
self.estimate_button.setEnabled(True)
|
||
|
||
offset = SyncOffset(delay_ms=round(edges.end_delay_ms))
|
||
self._apply_offset(offset, f"{offset.describe()} — estimated from stream ends")
|
||
|
||
self.log_view.appendPlainText(
|
||
f"Stream-ends estimate: video {edges.video_start:.3f}–{edges.video_end:.3f}s, "
|
||
f"audio {edges.audio_start:.3f}–{edges.audio_end:.3f}s → "
|
||
f"end-based {edges.end_delay_ms:+.0f} ms, start-based {edges.start_delay_ms:+.0f} ms"
|
||
)
|
||
if edges.estimates_disagree:
|
||
message = (
|
||
f"Estimated {offset.delay_ms:+.0f} ms from stream ends, but stream starts suggest "
|
||
f"{edges.start_delay_ms:+.0f} ms — verify with Preview."
|
||
)
|
||
self.log_view.appendPlainText("WARNING: " + message)
|
||
else:
|
||
message = f"Estimated {offset.delay_ms:+.0f} ms from stream ends (assumes both tracks stopped recording together) — verify with Preview."
|
||
self.statusBar().showMessage(message, 10000)
|
||
|
||
def _on_estimate_failed(self, message: str):
|
||
self.estimate_action.setEnabled(True)
|
||
self.estimate_button.setEnabled(True)
|
||
self.statusBar().showMessage("Offset estimate failed (see log).", 8000)
|
||
self.log_view.appendPlainText(f"Stream-ends estimate FAILED: {message}")
|
||
|
||
# -- 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.
|
||
#
|
||
# An estimated offset may have no markers behind it; center on the
|
||
# Source playhead instead, so the user can park it on a sync event
|
||
# (or any stretch of speech) before previewing.
|
||
if self.video_mark is not None and self.audio_mark is not None:
|
||
center = (self.video_mark + self.audio_mark) / 2.0
|
||
else:
|
||
center = self.source_panel.player.get_position()
|
||
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()
|