#!/usr/bin/env python3 """ Verify sync: QMediaPlayer 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 QMediaPlayer/QVideoWidget - 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, time, threading, tempfile import numpy as np try: import tomllib except ModuleNotFoundError: import tomli as tomllib try: import pyqtgraph as pg from PyQt6 import QtWidgets, QtCore, QtGui, QtMultimedia, QtMultimediaWidgets HAS_PG = True except ImportError as e: print("ERROR: pyqtgraph/PyQt6 not installed — run make install", file=sys.stderr) sys.exit(1) try: import sounddevice as sd HAS_SD = True except ImportError: HAS_SD = False try: import soundfile as sf HAS_SF = True except ImportError: HAS_SF = False from sync_audio import 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 # our custom audio mix playing 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, _ = load_wav(wav_a_path, target_sr=sr) print(f"Loading audio B: {wav_b_path}") self.b_orig, _ = 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() # Set up QMediaPlayer for video self._setup_video_player() def _setup_ui(self): central = QtWidgets.QWidget() self.setCentralWidget(central) lay = QtWidgets.QVBoxLayout(central) lay.setSpacing(2) # ── Video display (QWidget container) ── 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) # Disable auto-range on all plots to prevent overflow on linked axes self.p_comb.getViewBox().setAutoVisible(y=False) self.p_a.getViewBox().setAutoVisible(y=False) self.p_b.getViewBox().setAutoVisible(y=False) 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_video_player(self): """Set up QMediaPlayer with QVideoWidget for smooth video playback.""" self.media_player = QtMultimedia.QMediaPlayer() self.video_output = QtMultimediaWidgets.QVideoWidget() # Place QVideoWidget inside our container layout = QtWidgets.QVBoxLayout(self.video_container) layout.setContentsMargins(0, 0, 0, 0) layout.addWidget(self.video_output) self.media_player.setVideoOutput(self.video_output) # Mute the video player's own audio — we play audio via sounddevice self.audio_output = QtMultimedia.QAudioOutput() self.audio_output.setMuted(True) self.media_player.setAudioOutput(self.audio_output) # Load the video file from PyQt6.QtCore import QUrl self.media_player.setSource(QUrl.fromLocalFile(os.path.abspath(self.video_path))) # Timer to sync waveform playheads with video position self.sync_timer = QtCore.QTimer() self.sync_timer.setInterval(33) # ~30fps self.sync_timer.timeout.connect(self._sync_from_video) self.media_player.playbackStateChanged.connect(self._on_playback_state) # Wait for media loaded to get actual duration self.media_player.durationChanged.connect(self._on_duration_changed) def _on_duration_changed(self, dur_ms): """QMediaPlayer reports actual video duration in ms.""" vid_dur = dur_ms / 1000.0 if vid_dur > 0: self.info_label.setText( self.info_label.text().replace( f"Audio duration: {self.duration:.1f}s", f"Video: {vid_dur:.1f}s" ) ) def _on_playback_state(self, state): if state == QtMultimedia.QMediaPlayer.PlaybackState.StoppedState: self.sync_timer.stop() self.playing_a_b = False self.btn_play.setText("Play") self.audio_player.stop() 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 video position.""" pos_ms = self.media_player.position() t = pos_ms / 1000.0 if t > self.duration: t = self.duration self._set_pos(t) def toggle_play(self): state = self.media_player.playbackState() if state == QtMultimedia.QMediaPlayer.PlaybackState.PlayingState: self.stop_play() else: self.start_play() def start_play(self): if self.duration <= 0: return # Start video playback self.media_player.play() # Start audio mix via sounddevice t = self.media_player.position() / 1000.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.media_player.pause() 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 was_playing = self.media_player.playbackState() == QtMultimedia.QMediaPlayer.PlaybackState.PlayingState if was_playing: self.stop_play() self._was_playing = was_playing def _on_seek_move(self, value): frac = value / 1_000_000.0 t = frac * self.duration self._set_pos(t) self.media_player.setPosition(int(t * 1000)) def _on_seek_end(self): frac = self.seek.value() / 1_000_000.0 t = frac * self.duration self.media_player.setPosition(int(t * 1000)) 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): self.media_player.stop() 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 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()