751 lines
26 KiB
Python
751 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Verify sync: mpv video player (separate window) + 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 (separate window, IPC-controlled)
|
|
- 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
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import os, sys, math, argparse, subprocess, json, shutil, threading, tempfile, warnings, signal, locale, socket, time, socketserver
|
|
|
|
os.environ["LC_NUMERIC"] = "C"
|
|
locale.setlocale(locale.LC_NUMERIC, "C")
|
|
|
|
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
|
|
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
|
|
|
|
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
|
|
|
|
|
|
# ─── mpv IPC controller ───
|
|
|
|
class MpvController:
|
|
"""Control mpv as a standalone process via JSON IPC socket."""
|
|
|
|
def __init__(self, video_path: str):
|
|
self._sock_path = f"/tmp/mpv-verify-{os.getpid()}.sock"
|
|
self._cmd_id = 0
|
|
self._lock = threading.Lock()
|
|
self._proc = None
|
|
self._sock = None
|
|
|
|
# Remove stale socket
|
|
if os.path.exists(self._sock_path):
|
|
os.unlink(self._sock_path)
|
|
|
|
# Launch mpv with IPC server
|
|
self._proc = subprocess.Popen([
|
|
"mpv",
|
|
"--no-terminal",
|
|
"--keep-open=yes",
|
|
"--pause",
|
|
"--no-audio",
|
|
f"--input-ipc-server={self._sock_path}",
|
|
video_path,
|
|
], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
# Wait for socket to appear
|
|
for _ in range(50):
|
|
if os.path.exists(self._sock_path):
|
|
break
|
|
time.sleep(0.05)
|
|
|
|
self._connect()
|
|
|
|
def _connect(self):
|
|
self._sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self._sock.connect(self._sock_path)
|
|
self._sock.settimeout(0.5)
|
|
# Read any welcome message
|
|
try:
|
|
self._sock.recv(4096)
|
|
except socket.timeout:
|
|
pass
|
|
|
|
def _send(self, command: list, timeout: float = 2.0) -> dict | None:
|
|
with self._lock:
|
|
self._cmd_id += 1
|
|
msg = json.dumps({"command": command, "request_id": self._cmd_id}) + "\n"
|
|
self._sock.sendall(msg.encode())
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
data = self._sock.recv(4096)
|
|
if not data:
|
|
break
|
|
for line in data.decode().strip().split("\n"):
|
|
try:
|
|
resp = json.loads(line)
|
|
if resp.get("request_id") == self._cmd_id:
|
|
return resp
|
|
except json.JSONDecodeError:
|
|
continue
|
|
except socket.timeout:
|
|
continue
|
|
return None
|
|
|
|
def get_property(self, prop: str):
|
|
resp = self._send(["get_property", prop])
|
|
if resp and resp.get("error") == "success":
|
|
return resp.get("data")
|
|
return None
|
|
|
|
def set_property(self, prop: str, value):
|
|
self._send(["set_property", prop, value])
|
|
|
|
@property
|
|
def time_pos(self) -> float:
|
|
v = self.get_property("time-pos")
|
|
return float(v) if v is not None else 0.0
|
|
|
|
@property
|
|
def duration(self) -> float:
|
|
v = self.get_property("duration")
|
|
return float(v) if v is not None else 0.0
|
|
|
|
@property
|
|
def paused(self) -> bool:
|
|
v = self.get_property("pause")
|
|
return bool(v) if v is not None else True
|
|
|
|
def play(self):
|
|
self.set_property("pause", False)
|
|
|
|
def pause(self):
|
|
self.set_property("pause", True)
|
|
|
|
def seek(self, seconds: float):
|
|
self._send(["seek", seconds, "absolute"])
|
|
|
|
def stop(self):
|
|
try:
|
|
self._send(["stop"], timeout=0.5)
|
|
except Exception:
|
|
pass
|
|
|
|
def terminate(self):
|
|
try:
|
|
self._send(["quit"], timeout=0.5)
|
|
except Exception:
|
|
pass
|
|
if self._proc:
|
|
try:
|
|
self._proc.wait(timeout=2)
|
|
except subprocess.TimeoutExpired:
|
|
self._proc.kill()
|
|
if self._sock:
|
|
try:
|
|
self._sock.close()
|
|
except Exception:
|
|
pass
|
|
if os.path.exists(self._sock_path):
|
|
try:
|
|
os.unlink(self._sock_path)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ─── 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
|
|
self.mpv: MpvController | None = None
|
|
|
|
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, 700)
|
|
|
|
# Load audio
|
|
print(f"Loading audio A: {wav_a_path}")
|
|
self.a_orig, _ = cached_load_wav(wav_a_path, target_sr=sr, _loader=load_wav)
|
|
print(f"Loading audio B: {wav_b_path}")
|
|
self.b_orig, _ = cached_load_wav(wav_b_path, target_sr=sr, _loader=load_wav)
|
|
|
|
# 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()
|
|
|
|
# Start mpv in its own window
|
|
QtCore.QTimer.singleShot(0, self._start_mpv)
|
|
|
|
# Sync timer: polls mpv position and updates waveforms
|
|
self.sync_timer = QtCore.QTimer()
|
|
self.sync_timer.setInterval(33) # ~30fps
|
|
self.sync_timer.timeout.connect(self._sync_from_mpv)
|
|
self.sync_timer.start()
|
|
|
|
def _setup_ui(self):
|
|
central = QtWidgets.QWidget()
|
|
self.setCentralWidget(central)
|
|
lay = QtWidgets.QVBoxLayout(central)
|
|
lay.setSpacing(2)
|
|
|
|
# ── 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 _start_mpv(self):
|
|
"""Launch mpv in its own window, controlled via IPC socket."""
|
|
print(f"Launching mpv: {self.video_path}")
|
|
try:
|
|
self.mpv = MpvController(self.video_path)
|
|
vid_dur = self.mpv.duration
|
|
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"
|
|
)
|
|
)
|
|
except Exception as e:
|
|
print(f"ERROR: failed to start mpv: {e}", file=sys.stderr)
|
|
self.info_label.setText(self.info_label.text() + " | mpv: FAILED")
|
|
|
|
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_mpv(self):
|
|
"""Poll mpv position and update waveforms."""
|
|
if not self.mpv:
|
|
return
|
|
try:
|
|
t = self.mpv.time_pos
|
|
except Exception:
|
|
t = 0
|
|
if t > self.duration:
|
|
t = self.duration
|
|
self._set_pos(t)
|
|
|
|
def toggle_play(self):
|
|
if not self.mpv:
|
|
return
|
|
try:
|
|
if not self.mpv.paused:
|
|
self.stop_play()
|
|
else:
|
|
self.start_play()
|
|
except Exception:
|
|
self.start_play()
|
|
|
|
def start_play(self):
|
|
if self.duration <= 0 or not self.mpv:
|
|
return
|
|
|
|
# Start video playback in mpv
|
|
self.mpv.play()
|
|
|
|
# Start audio mix via sounddevice
|
|
t = self.mpv.time_pos
|
|
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")
|
|
|
|
def stop_play(self):
|
|
if self.mpv:
|
|
self.mpv.pause()
|
|
self.audio_player.stop()
|
|
self.playing_a_b = False
|
|
self.btn_play.setText("Play")
|
|
|
|
def _on_seek_start(self):
|
|
self._seeking = True
|
|
if self.mpv:
|
|
self._was_playing = not self.mpv.paused
|
|
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)
|
|
if self.mpv:
|
|
self.mpv.seek(t)
|
|
|
|
def _on_seek_end(self):
|
|
frac = self.seek.value() / 1_000_000.0
|
|
t = frac * self.duration
|
|
if self.mpv:
|
|
self.mpv.seek(t)
|
|
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):
|
|
if self.mpv:
|
|
self.mpv.terminate()
|
|
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
|
|
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)
|
|
signal.signal(signal.SIGINT, signal.SIG_DFL)
|
|
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()
|