709 lines
25 KiB
Python
709 lines
25 KiB
Python
#!/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()
|