#!/usr/bin/env python3 """ Verify sync: video player + dual waveform display for A/B audio comparison. Modes: --mode short : extracted 30s parts (video_a + wav_a + wav_b) --mode full : synced output MKV (extracts audio streams from it) Features: - Video playback with timeline seek bar - Audio playback: A only, B only, or A+B mix - Combined waveform overlay (A and B in different colors) - Two separate waveform graphs with playhead - Zoomable waveforms (Ctrl+scroll) with linked x-axes - Toggle A/B on/off, volume, play/pause 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 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_video_info(path: str) -> dict: try: r = subprocess.run( ["ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,r_frame_rate,duration", "-show_entries", "format=duration", "-of", "json", path], capture_output=True, text=True, timeout=10 ) data = json.loads(r.stdout) stream = data.get("streams", [{}])[0] fmt = data.get("format", {}) fps_str = stream.get("r_frame_rate", "24000/1001") if "/" in fps_str: num, den = fps_str.split("/") fps = float(num) / float(den) else: fps = float(fps_str) duration = float(fmt.get("duration", stream.get("duration", 0))) return { "width": int(stream.get("width", 1920)), "height": int(stream.get("height", 1080)), "fps": fps, "duration": duration, } except Exception: return {"width": 1920, "height": 1080, "fps": 24000/1001, "duration": 0} def probe_audio_streams(path: str) -> list[dict]: """Probe audio streams for full mode.""" 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", "und"), "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: """Extract a single audio stream from MKV to mono WAV.""" 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 extract_frame(video_path: str, timestamp: float, width: int, height: int) -> bytes | None: """Extract a single frame at timestamp as raw RGB bytes.""" try: r = subprocess.run( ["ffmpeg", "-ss", f"{timestamp:.4f}", "-i", video_path, "-vframes", "1", "-vf", f"scale={width}:{height}", "-f", "rawvideo", "-pix_fmt", "rgb24", "-v", "error", "-"], capture_output=True, timeout=5 ) if r.returncode != 0 or not r.stdout: return None return r.stdout except Exception: return None def raw_to_qimage(raw: bytes, width: int, height: int) -> QtGui.QImage | None: if len(raw) < width * height * 3: return None img = QtGui.QImage(raw, width, height, width * 3, QtGui.QImage.Format.Format_RGB888) return img.copy() def apply_lag(a: np.ndarray, b: np.ndarray, lag_samples: int) -> tuple[np.ndarray, np.ndarray]: """Align A and B with lag applied (B delayed by lag_samples).""" 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 # ─── Zoomable plot widget ─── class ZoomPlot(pg.PlotWidget): """PlotWidget with Ctrl+scroll zoom on x-axis, right-drag to pan.""" def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.setMouseEnabled(x=True, y=False) # Enable wheel zoom on x-axis only self.getViewBox().setMouseEnabled(x=True, y=False) def wheelEvent(self, event): delta = event.angleDelta().y() factor = 1.1 if delta > 0 else 1 / 1.1 vb = self.getViewBox() # zoom x-axis around cursor mouse_pos = vb.mapSceneToView(event.position()) x_range = vb.viewRange()[0] new_width = (x_range[1] - x_range[0]) / factor center = mouse_pos.x() vb.setXRange(center - new_width / 2, center + new_width / 2, padding=0) event.accept() # ─── 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): 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 = AudioPlayer() 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 # Video info vinfo = probe_video_info(video_path) self.v_width = vinfo["width"] self.v_height = vinfo["height"] self.v_fps = vinfo["fps"] self.v_duration = vinfo["duration"] if vinfo["duration"] > 0 else self.duration # Playback state self.playing = False self.play_start_time = 0.0 self.play_offset = 0.0 self.timer = QtCore.QTimer() self.timer.setInterval(33) self.timer.timeout.connect(self._on_tick) self._seeking = False self._last_frame_t = -1.0 self._setup_ui() self._draw_waveforms() self._update_frame(0.0) def _setup_ui(self): central = QtWidgets.QWidget() self.setCentralWidget(central) lay = QtWidgets.QVBoxLayout(central) lay.setSpacing(2) # ── Video display ── self.video_label = QtWidgets.QLabel() self.video_label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter) self.video_label.setMinimumHeight(300) self.video_label.setStyleSheet("background: #000;") lay.addWidget(self.video_label, stretch=4) # ── Timeline seek bar (video editor style) ── 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) # ── 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)) 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)) # Link all x-axes for synchronized zoom/pan self.p_a.setXLink(self.p_comb) self.p_b.setXLink(self.p_comb) lay.addWidget(self.gw_dual, stretch=2) # ── Controls row ── ctrl = QtWidgets.QHBoxLayout() lay.addLayout(ctrl) # Play mode 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) # Volume ctrl.addWidget(QtWidgets.QLabel("Vol:")) self.vol_slider = QtWidgets.QSlider(QtCore.Qt.Orientation.Horizontal) self.vol_slider.setRange(0, 100) self.vol_slider.setValue(70) self.vol_slider.setMaximumWidth(100) ctrl.addWidget(self.vol_slider) self.vol_label = QtWidgets.QLabel("70%") 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) # Toggle A/B 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) # Reset zoom button 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() # Info 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"Video: {self.v_duration:.1f}s" ) self.info_label.setStyleSheet("color: #aaa; font-size: 11px;") ctrl.addWidget(self.info_label) 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) # Set initial view to full duration self.p_comb.setXRange(0, self.duration, padding=0) def _update_frame(self, t: float): t = max(0, min(t, self.v_duration)) # Avoid re-extracting same frame if abs(t - self._last_frame_t) < 0.04: return self._last_frame_t = t raw = extract_frame(self.video_path, t, self.v_width, self.v_height) if raw: img = raw_to_qimage(raw, self.v_width, self.v_height) if img: pixmap = QtGui.QPixmap.fromImage(img) scaled = pixmap.scaled(self.video_label.size(), QtCore.Qt.AspectRatioMode.KeepAspectRatio, QtCore.Qt.TransformationMode.SmoothTransformation) self.video_label.setPixmap(scaled) 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 _current_pos(self) -> float: if self.playing: elapsed = time.time() - self.play_start_time return min(self.play_offset + elapsed, self.duration) return self.play_offset def _set_pos(self, t: float): t = max(0, min(t, self.duration)) self.play_offset = t 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 _on_tick(self): if not self.playing: return t = self._current_pos() if t >= self.duration: self.stop_play() return self._set_pos(t) # Update video frame (limit to ~8fps) frame_interval = 1.0 / 8.0 now = time.time() if int(t / frame_interval) != int((t - 0.033) / frame_interval): self._update_frame(t) def toggle_play(self): if self.playing: self.stop_play() else: self.start_play() def start_play(self): if self.duration <= 0: return pos = self._current_pos() if pos >= self.duration: pos = 0.0 start_samp = int(pos * self.sr) 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.play(mix, self.sr) self.play_offset = pos self.play_start_time = time.time() self.playing = True self.btn_play.setText("Pause") self.timer.start() def stop_play(self): self.audio.stop() self.play_offset = self._current_pos() self.playing = False self.btn_play.setText("Play") self.timer.stop() def _on_seek_start(self): self._seeking = True was_playing = self.playing 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._update_frame(t) def _on_seek_end(self): frac = self.seek.value() / 1_000_000.0 t = frac * self.duration self.play_offset = t self._set_pos(t) self._update_frame(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.setXRange(0, self.duration, padding=0) def closeEvent(self, event): self.audio.stop() self.timer.stop() # Clean up temp files if any 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)) out_dir = args.out or sync_cfg.get("out_dir", "output") lag_sec = args.lag if args.mode == "full": # Full mode: use synced MKV, extract audio streams from it synced = args.synced if not synced: # Auto-derive from config 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) # Extract both audio streams to temp WAVs 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: # Short mode: use original video + extracted WAVs 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) w.show() sys.exit(app.exec()) if __name__ == "__main__": main()