#!/usr/bin/env python3 """ Verify sync: mpv video player + dual waveform display. Modes: --mode short : extracted 30s parts (video_a + wav_a + wav_b) --mode full : synced output MKV (extracts audio streams from it) Features: - Smooth video playback via mpv (handles all codecs natively) - Audio playback: A only, B only, or A+B mix via sounddevice - Combined waveform overlay (A and B in different colors) - Two separate waveform graphs with playhead - Zoomable waveforms (scroll) with linked x-axes - Timeline seek bar synced to video position - Toggle A/B on/off, volume Usage: make verify LAG=0.042 python verify_sync.py --config config.toml --lag 0.042 --mode short python verify_sync.py --synced output/movie_synced.mkv --lag 0.042 --mode full """ from __future__ import annotations import os, sys, math, argparse, subprocess, json, shutil, threading, tempfile, warnings import numpy as np warnings.filterwarnings("ignore", category=RuntimeWarning, module="pyqtgraph") try: import tomllib except ModuleNotFoundError: import tomli as tomllib try: import pyqtgraph as pg from PyQt6 import QtWidgets, QtCore, QtGui HAS_PG = True except ImportError as e: print("ERROR: pyqtgraph/PyQt6 not installed — run make install", file=sys.stderr) sys.exit(1) try: import mpv HAS_MPV = True except ImportError: HAS_MPV = False print("ERROR: python-mpv not installed — pip install python-mpv", file=sys.stderr) sys.exit(1) try: import sounddevice as sd HAS_SD = True except ImportError: HAS_SD = False from sync_audio import load_wav from waveform_cache import cached_load_wav # ─── Helpers ─── def load_config(path: str) -> dict: if not os.path.exists(path): return {} with open(path, "rb") as f: return tomllib.load(f) def probe_audio_streams(path: str) -> list[dict]: try: r = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=index,codec_name,channels,sample_rate," "stream_tags=language,title", "-of", "json", path], capture_output=True, text=True, timeout=10 ) if r.returncode != 0: return [] data = json.loads(r.stdout) streams = [] for s in data.get("streams", []): tags = {k.lower(): v for k, v in (s.get("tags") or {}).items()} sr = s.get("sample_rate") streams.append({ "index": s.get("index", 0), "codec": s.get("codec_name", ""), "channels": s.get("channels", 0), "sample_rate": int(sr) if sr else 0, "language": tags.get("language", ""), "title": tags.get("title", ""), }) return streams except Exception: return [] def extract_audio_from_mkv(mkv_path: str, stream_index: int, out_wav: str, sr: int = 48000) -> bool: try: r = subprocess.run( ["ffmpeg", "-y", "-i", mkv_path, "-map", f"0:a:{stream_index}", "-acodec", "pcm_f32le", "-ar", str(sr), "-ac", "1", "-v", "error", out_wav], capture_output=True, timeout=300 ) return r.returncode == 0 except Exception: return False def apply_lag(a: np.ndarray, b: np.ndarray, lag_samples: int) -> tuple[np.ndarray, np.ndarray]: if lag_samples >= 0: a_out = a[lag_samples:] b_out = np.concatenate([np.zeros(lag_samples, dtype=np.float32), b]) else: pad = -lag_samples a_out = np.concatenate([np.zeros(pad, dtype=np.float32), a]) b_out = b[pad:] ml = min(len(a_out), len(b_out)) return a_out[:ml], b_out[:ml] def apply_dark_theme(app: QtWidgets.QApplication): app.setStyle("Fusion") dark = QtGui.QPalette() dark.setColor(QtGui.QPalette.ColorRole.Window, QtGui.QColor(45, 45, 45)) dark.setColor(QtGui.QPalette.ColorRole.WindowText, QtGui.QColor(220, 220, 220)) dark.setColor(QtGui.QPalette.ColorRole.Base, QtGui.QColor(30, 30, 30)) dark.setColor(QtGui.QPalette.ColorRole.AlternateBase, QtGui.QColor(45, 45, 45)) dark.setColor(QtGui.QPalette.ColorRole.ToolTipBase, QtGui.QColor(45, 45, 45)) dark.setColor(QtGui.QPalette.ColorRole.ToolTipText, QtGui.QColor(220, 220, 220)) dark.setColor(QtGui.QPalette.ColorRole.Text, QtGui.QColor(220, 220, 220)) dark.setColor(QtGui.QPalette.ColorRole.Button, QtGui.QColor(45, 45, 45)) dark.setColor(QtGui.QPalette.ColorRole.ButtonText, QtGui.QColor(220, 220, 220)) dark.setColor(QtGui.QPalette.ColorRole.BrightText, QtGui.QColor(255, 50, 50)) dark.setColor(QtGui.QPalette.ColorRole.Link, QtGui.QColor(100, 180, 255)) dark.setColor(QtGui.QPalette.ColorRole.Highlight, QtGui.QColor(60, 120, 200)) dark.setColor(QtGui.QPalette.ColorRole.HighlightedText, QtGui.QColor(255, 255, 255)) dark.setColor(QtGui.QPalette.ColorGroup.Disabled, QtGui.QPalette.ColorRole.Text, QtGui.QColor(120, 120, 120)) dark.setColor(QtGui.QPalette.ColorGroup.Disabled, QtGui.QPalette.ColorRole.ButtonText, QtGui.QColor(120, 120, 120)) app.setPalette(dark) pg.setConfigOptions(antialias=True, background="#1e1e1e", foreground="#ddd") # ─── Audio playback ─── class AudioPlayer: def __init__(self): self.playing = False self._lock = threading.Lock() def play(self, data: np.ndarray, sr: int): if not HAS_SD: return with self._lock: sd.stop() self.playing = True try: sd.play(data, samplerate=sr, blocking=False) except Exception as e: print(f"audio play error: {e}", file=sys.stderr) self.playing = False def stop(self): if HAS_SD: with self._lock: sd.stop() self.playing = False # ─── Main Window ─── class VerifyWindow(QtWidgets.QMainWindow): def __init__(self, video_path: str, wav_a_path: str, wav_b_path: str, sr: int = 48000, lag_sec: float = 0.0, mode: str = "short", out_dir: str = "output", temp_dir: str | None = None, volume: int = 70): super().__init__() self.video_path = video_path self.wav_a_path = wav_a_path self.wav_b_path = wav_b_path self.sr = sr self.lag_sec = lag_sec self.lag_samples = int(round(lag_sec * sr)) self.mode = mode self.out_dir = out_dir self.temp_dir = temp_dir self.audio_player = AudioPlayer() self.default_volume = volume self.duration = 0.0 self.playing_a_b = False title = "verify_sync" if mode == "short": title += " — 30s preview" else: title += " — full length" title += f" — {os.path.basename(video_path)}" self.setWindowTitle(title) self.resize(1400, 950) # Load audio print(f"Loading audio A: {wav_a_path}") self.a_orig, _ = cached_load_wav(wav_a_path, target_sr=sr) print(f"Loading audio B: {wav_b_path}") self.b_orig, _ = cached_load_wav(wav_b_path, target_sr=sr) # Apply lag alignment self.a_aligned, self.b_aligned = apply_lag(self.a_orig, self.b_orig, self.lag_samples) self.duration = len(self.a_aligned) / sr self._setup_ui() self._draw_waveforms() self._setup_mpv() def _setup_ui(self): central = QtWidgets.QWidget() self.setCentralWidget(central) lay = QtWidgets.QVBoxLayout(central) lay.setSpacing(2) # ── Video display (QWidget container for mpv) ── self.video_container = QtWidgets.QWidget() self.video_container.setMinimumHeight(300) self.video_container.setStyleSheet("background: #000;") lay.addWidget(self.video_container, stretch=4) # ── Timeline seek bar ── time_row = QtWidgets.QHBoxLayout() lay.addLayout(time_row) self.btn_play = QtWidgets.QPushButton("Play") self.btn_play.setFixedWidth(60) time_row.addWidget(self.btn_play) self.btn_stop = QtWidgets.QPushButton("Stop") self.btn_stop.setFixedWidth(60) time_row.addWidget(self.btn_stop) self.seek = QtWidgets.QSlider(QtCore.Qt.Orientation.Horizontal) self.seek.setRange(0, 1_000_000) self.seek.setValue(0) time_row.addWidget(self.seek, stretch=1) self.time_label = QtWidgets.QLabel("00:00:00.000 / 00:00:00.000") self.time_label.setMinimumWidth(220) self.time_label.setStyleSheet("font-family: monospace; font-size: 12px;") time_row.addWidget(self.time_label) self.btn_play.clicked.connect(self.toggle_play) self.btn_stop.clicked.connect(self.stop_play) self.seek.sliderPressed.connect(self._on_seek_start) self.seek.sliderMoved.connect(self._on_seek_move) self.seek.sliderReleased.connect(self._on_seek_end) self._seeking = False # ── Combined waveform overlay ── self.gw_comb = pg.GraphicsLayoutWidget() self.gw_comb.setMinimumHeight(100) self.p_comb = self.gw_comb.addPlot(title="A (teal) + B (red) overlay") self.p_comb.showGrid(x=True, y=True, alpha=0.3) self.p_comb.setLabel("bottom", "Time", "s") self.curve_a_comb = self.p_comb.plot(pen=pg.mkPen("#4ecdc4", width=1), name="A") self.curve_b_comb = self.p_comb.plot(pen=pg.mkPen("#ff6b6b", width=1, style=QtCore.Qt.PenStyle.DashLine), name="B") self.playhead_comb = self.p_comb.addLine(x=0, pen=pg.mkPen("#ffff00", width=2, style=QtCore.Qt.PenStyle.DashLine)) self.p_comb.addLegend(offset=(60, 10)) lay.addWidget(self.gw_comb, stretch=1) # ── Two separate waveforms ── self.gw_dual = pg.GraphicsLayoutWidget() self.gw_dual.setMinimumHeight(200) self.p_a = self.gw_dual.addPlot(row=0, col=0, title="Audio A") self.p_a.showGrid(x=True, y=True, alpha=0.3) self.p_a.setLabel("bottom", "Time", "s") self.curve_a_sep = self.p_a.plot(pen=pg.mkPen("#4ecdc4", width=1)) self.playhead_a = self.p_a.addLine(x=0, pen=pg.mkPen("#ffff00", width=2, style=QtCore.Qt.PenStyle.DashLine)) self.p_b = self.gw_dual.addPlot(row=1, col=0, title="Audio B (+delay)") self.p_b.showGrid(x=True, y=True, alpha=0.3) self.p_b.setLabel("bottom", "Time", "s") self.curve_b_sep = self.p_b.plot(pen=pg.mkPen("#ff6b6b", width=1)) self.playhead_b = self.p_b.addLine(x=0, pen=pg.mkPen("#ffff00", width=2, style=QtCore.Qt.PenStyle.DashLine)) self.p_a.setXLink(self.p_comb) self.p_b.setXLink(self.p_comb) for p in [self.p_comb, self.p_a, self.p_b]: p.enableAutoRange(axis='x', enable=False) p.enableAutoRange(axis='y', enable=False) lay.addWidget(self.gw_dual, stretch=2) # ── Controls row ── ctrl = QtWidgets.QHBoxLayout() lay.addLayout(ctrl) ctrl.addWidget(QtWidgets.QLabel("Play:")) self.play_mode = QtWidgets.QComboBox() self.play_mode.addItems(["A+B mix", "A only", "B only"]) self.play_mode.setCurrentIndex(0) self.play_mode.setFixedWidth(90) ctrl.addWidget(self.play_mode) ctrl.addWidget(QtWidgets.QLabel("Vol:")) self.vol_slider = QtWidgets.QSlider(QtCore.Qt.Orientation.Horizontal) self.vol_slider.setRange(0, 100) self.vol_slider.setValue(self.default_volume) self.vol_slider.setMaximumWidth(100) ctrl.addWidget(self.vol_slider) self.vol_label = QtWidgets.QLabel(f"{self.default_volume}%") self.vol_label.setMinimumWidth(35) ctrl.addWidget(self.vol_label) self.vol_slider.valueChanged.connect(lambda v: self.vol_label.setText(f"{v}%")) ctrl.addSpacing(10) self.chk_a = QtWidgets.QCheckBox("Show A") self.chk_a.setChecked(True) self.chk_a.toggled.connect(self._update_visibility) ctrl.addWidget(self.chk_a) self.chk_b = QtWidgets.QCheckBox("Show B") self.chk_b.setChecked(True) self.chk_b.toggled.connect(self._update_visibility) ctrl.addWidget(self.chk_b) ctrl.addSpacing(10) btn_reset_zoom = QtWidgets.QPushButton("Reset zoom") btn_reset_zoom.setFixedWidth(90) btn_reset_zoom.clicked.connect(self._reset_zoom) ctrl.addWidget(btn_reset_zoom) ctrl.addStretch() mode_str = "30s preview" if self.mode == "short" else "full length" self.info_label = QtWidgets.QLabel( f"[{mode_str}] Lag: {self.lag_sec:+.6f}s ({self.lag_samples} samp) | " f"A: {len(self.a_orig)/self.sr:.1f}s | B: {len(self.b_orig)/self.sr:.1f}s | " f"Audio duration: {self.duration:.1f}s" ) self.info_label.setStyleSheet("color: #aaa; font-size: 11px;") ctrl.addWidget(self.info_label) def _setup_mpv(self): """Set up mpv player embedded in the video container widget.""" self.mpv_player = mpv.MPV( wid=str(int(self.video_container.winId())), input_default_bindings=False, input_vo_keyboard=False, cursor_autohide=False, keep_open=True, osc=False, osd_bar=False, ) # Mute mpv audio — we play audio via sounddevice self.mpv_player.audio = "no" self.mpv_player.play(os.path.abspath(self.video_path)) # Timer to sync waveform playheads with mpv position self.sync_timer = QtCore.QTimer() self.sync_timer.setInterval(33) # ~30fps self.sync_timer.timeout.connect(self._sync_from_video) # Wait for mpv to load duration self._duration_timer = QtCore.QTimer() self._duration_timer.setInterval(100) self._duration_timer.timeout.connect(self._wait_for_duration) self._duration_timer.start() def _wait_for_duration(self): """Poll mpv until duration is available.""" try: dur = self.mpv_player.duration if dur and dur > 0: self._duration_timer.stop() vid_dur = float(dur) self.info_label.setText( self.info_label.text().replace( f"Audio duration: {self.duration:.1f}s", f"Video: {vid_dur:.1f}s" ) ) except Exception: pass def _draw_waveforms(self): max_pts = 200_000 if self.mode == "full" else 80_000 def decimate(arr): if len(arr) <= max_pts: return arr, np.arange(len(arr), dtype=np.float64) / self.sr stride = int(math.ceil(len(arr) / max_pts)) dec = arr[::stride] t = np.arange(0, len(arr), stride).astype(np.float64) / self.sr return dec, t a_dec, t_a = decimate(self.a_aligned) b_dec, t_b = decimate(self.b_aligned) self.curve_a_comb.setData(t_a, a_dec) self.curve_b_comb.setData(t_b, b_dec) self.curve_a_sep.setData(t_a, a_dec) self.curve_b_sep.setData(t_b, b_dec) y_max = max(np.max(np.abs(a_dec)), np.max(np.abs(b_dec))) * 1.1 if y_max > 0: self.p_comb.setYRange(-y_max, y_max) self.p_a.setYRange(-y_max, y_max) self.p_b.setYRange(-y_max, y_max) for p in [self.p_comb, self.p_a, self.p_b]: p.setXRange(0, self.duration, padding=0) p.getViewBox().setLimits(xMin=0, xMax=self.duration) def _set_pos(self, t: float): t = max(0, min(t, self.duration)) self.playhead_comb.setPos(t) self.playhead_a.setPos(t) self.playhead_b.setPos(t) if not self._seeking: frac = t / self.duration if self.duration > 0 else 0 self.seek.blockSignals(True) self.seek.setValue(int(frac * 1_000_000)) self.seek.blockSignals(False) self.time_label.setText(f"{self._fmt_time(t)} / {self._fmt_time(self.duration)}") def _fmt_time(self, t: float) -> str: h = int(t // 3600) m = int((t % 3600) // 60) s = t % 60 return f"{h:02d}:{m:02d}:{s:06.3f}" def _sync_from_video(self): """Timer callback: sync waveform playheads and seek bar to mpv position.""" try: t = float(self.mpv_player.time_pos or 0) except Exception: t = 0 if t > self.duration: t = self.duration self._set_pos(t) def toggle_play(self): try: paused = self.mpv_player.pause if paused is False: self.stop_play() else: self.start_play() except Exception: self.start_play() def start_play(self): if self.duration <= 0: return # Start video playback self.mpv_player.pause = False # Start audio mix via sounddevice try: t = float(self.mpv_player.time_pos or 0) except Exception: t = 0 start_samp = int(t * self.sr) if start_samp >= len(self.a_aligned): start_samp = 0 a_seg = self.a_aligned[start_samp:] b_seg = self.b_aligned[start_samp:] vol = self.vol_slider.value() / 100.0 mode = self.play_mode.currentIndex() if mode == 0: # A+B mix ml = min(len(a_seg), len(b_seg)) mix = ((a_seg[:ml] + b_seg[:ml]) / 2.0) * vol elif mode == 1: # A only mix = a_seg * vol else: # B only mix = b_seg * vol self.audio_player.play(mix, self.sr) self.playing_a_b = True self.btn_play.setText("Pause") self.sync_timer.start() def stop_play(self): self.mpv_player.pause = True self.audio_player.stop() self.playing_a_b = False self.btn_play.setText("Play") self.sync_timer.stop() def _on_seek_start(self): self._seeking = True try: paused = self.mpv_player.pause self._was_playing = (paused is False) except Exception: self._was_playing = False if self._was_playing: self.stop_play() def _on_seek_move(self, value): frac = value / 1_000_000.0 t = frac * self.duration self._set_pos(t) self.mpv_player.seek(t, reference='absolute') def _on_seek_end(self): frac = self.seek.value() / 1_000_000.0 t = frac * self.duration self.mpv_player.seek(t, reference='absolute') self._set_pos(t) self._seeking = False if getattr(self, '_was_playing', False): self.start_play() def _update_visibility(self): self.curve_a_comb.setVisible(self.chk_a.isChecked()) self.curve_b_comb.setVisible(self.chk_b.isChecked()) self.curve_a_sep.setVisible(self.chk_a.isChecked()) self.curve_b_sep.setVisible(self.chk_b.isChecked()) def _reset_zoom(self): self.p_comb.enableAutoRange(axis='x', enable=False) self.p_comb.setXRange(0, self.duration, padding=0) def closeEvent(self, event): try: self.mpv_player.stop() self.mpv_player.terminate() except Exception: pass self.audio_player.stop() self.sync_timer.stop() if self.temp_dir and os.path.isdir(self.temp_dir): try: shutil.rmtree(self.temp_dir) except Exception: pass super().closeEvent(event) def main(): ap = argparse.ArgumentParser(description="Verify sync: video + dual waveform playback") ap.add_argument("--config", default="config.toml") ap.add_argument("--a", default=None, help="Video file A") ap.add_argument("--b", default=None, help="Video file B") ap.add_argument("--synced", default=None, help="Synced MKV output (full mode)") ap.add_argument("--lag", type=float, default=0.0, help="Lag in seconds") ap.add_argument("--mode", choices=["short", "full"], default="short", help="short = extracted 30s parts, full = synced MKV output") ap.add_argument("--out", default=None, help="Output dir") args = ap.parse_args() cfg = load_config(args.config) ext = cfg.get("extract", {}) sync_cfg = cfg.get("sync", {}) audio_cfg = cfg.get("audio", {}) sr = int(audio_cfg.get("sample_rate", 48000)) volume = int(audio_cfg.get("volume", 70)) out_dir = args.out or sync_cfg.get("out_dir", "output") lag_sec = args.lag if args.mode == "full": synced = args.synced if not synced: video_a = ext.get("file_a", "") stem = os.path.splitext(os.path.basename(video_a))[0] synced = os.path.join(out_dir, f"{stem}_synced.mkv") if not os.path.exists(synced): print(f"ERROR: synced MKV not found: {synced}", file=sys.stderr) sys.exit(1) print(f"Synced MKV: {synced}") streams = probe_audio_streams(synced) print(f"Audio streams in synced MKV: {len(streams)}") for i, s in enumerate(streams): print(f" [{i}] {s['language']} {s['codec'].upper()} {s['channels']}ch") if len(streams) < 2: print("ERROR: synced MKV needs at least 2 audio streams (A + B)", file=sys.stderr) sys.exit(1) temp_dir = tempfile.mkdtemp(prefix="verify_sync_") wav_a = os.path.join(temp_dir, "audio_a.wav") wav_b = os.path.join(temp_dir, "audio_b.wav") print(f"Extracting audio stream 0 (A) to {wav_a}...") if not extract_audio_from_mkv(synced, 0, wav_a, sr): print("ERROR: failed to extract audio stream 0", file=sys.stderr) sys.exit(1) print(f"Extracting audio stream 1 (B) to {wav_b}...") if not extract_audio_from_mkv(synced, 1, wav_b, sr): print("ERROR: failed to extract audio stream 1", file=sys.stderr) sys.exit(1) video_path = synced else: video_a = args.a or ext.get("file_a", "") video_b = args.b or ext.get("file_b", "") out_dir_extract = ext.get("out_dir", "extracted") def find_wav(video_path: str) -> str: if not video_path: return "" stem = os.path.splitext(os.path.basename(video_path))[0] cand = os.path.join(out_dir_extract, f"{stem}_audio.wav") return cand if os.path.exists(cand) else cand wav_a = find_wav(video_a) wav_b = find_wav(video_b) if not video_a or not os.path.exists(video_a): print(f"ERROR: video A not found: {video_a}", file=sys.stderr); sys.exit(1) if not wav_a or not os.path.exists(wav_a): print(f"ERROR: wav A not found: {wav_a}", file=sys.stderr); sys.exit(1) if not wav_b or not os.path.exists(wav_b): print(f"ERROR: wav B not found: {wav_b}", file=sys.stderr); sys.exit(1) video_path = video_a # In short mode, use the extracted FFV1 video instead of the original HEVC source if args.mode == "short": stem = os.path.splitext(os.path.basename(video_a))[0] extracted_video = os.path.join(out_dir_extract, f"{stem}_video.mkv") if os.path.exists(extracted_video): video_path = extracted_video print(f"Using extracted video: {video_path}") else: print(f"WARNING: extracted video not found ({extracted_video}), using original", file=sys.stderr) temp_dir = None print(f"Mode: {args.mode}") print(f"Video: {video_path}") print(f"WAV A: {wav_a}") print(f"WAV B: {wav_b}") print(f"Lag: {lag_sec:+.6f}s") app = QtWidgets.QApplication(sys.argv) apply_dark_theme(app) w = VerifyWindow(video_path, wav_a, wav_b, sr=sr, lag_sec=lag_sec, mode=args.mode, out_dir=out_dir, temp_dir=temp_dir, volume=volume) w.show() sys.exit(app.exec()) if __name__ == "__main__": main()