777 lines
32 KiB
Python
777 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
PyQtGraph GUI to visualize audio cross-correlation and interactively tune lag.
|
|
|
|
Features (per user request):
|
|
- uv venv via Makefile, torch auto GPU (ROCm/CUDA) with cpu fallback
|
|
- config.toml for file_a/file_b, duration/start (extract duration is source of truth)
|
|
- PyQtGraph waveforms (decimated for 4M samples) + xcorr plot
|
|
- QSlider + QDoubleSpinBox numeric entry for lag (seconds + samples)
|
|
- Volume slider (sounddevice optional, playback with lag applied, silence padded for delay)
|
|
|
|
Run: make ast or .venv/bin/python sync_audio_test.py --config config.toml
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import os, sys, math, argparse, signal
|
|
import numpy as np
|
|
|
|
# allow tomllib (py3.11+) with tomli fallback
|
|
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)
|
|
print(e, 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, correlate, device_info, lag_to_seconds
|
|
from waveform_cache import cached_load_wav
|
|
|
|
|
|
class ApplySyncWorker(QtCore.QThread):
|
|
"""Background thread for apply_sync.py — runs ffmpeg and parses frame progress."""
|
|
progress = QtCore.pyqtSignal(int, int) # frames_done, frames_total
|
|
status = QtCore.pyqtSignal(str)
|
|
finished = QtCore.pyqtSignal(bool) # success
|
|
|
|
def __init__(self, lag_sec: float, video_a: str, video_b: str, config: str = "config.toml"):
|
|
super().__init__()
|
|
self.lag_sec = lag_sec
|
|
self.video_a = video_a
|
|
self.video_b = video_b
|
|
self.config = config
|
|
self._abort = False
|
|
|
|
def abort(self):
|
|
self._abort = True
|
|
|
|
@staticmethod
|
|
def _parse_rate(s: str) -> float:
|
|
s = (s or "").strip()
|
|
if not s or s in ("0/0", "0"):
|
|
return 0.0
|
|
if "/" in s:
|
|
try:
|
|
n, d = s.split("/")
|
|
return float(n) / float(d) if float(d) != 0 else 0.0
|
|
except Exception:
|
|
return 0.0
|
|
try:
|
|
return float(s)
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
@classmethod
|
|
def probe_video_total(cls, path: str) -> tuple[int, int]:
|
|
"""Fast total frames + duration ms for file A (video is stream-copied).
|
|
|
|
Uses duration x fps metadata — no slow -count_frames decode.
|
|
Returns (total_frames, total_ms), 0s on failure.
|
|
"""
|
|
import json as _json
|
|
import subprocess as _sp
|
|
try:
|
|
r = _sp.run(
|
|
["ffprobe", "-v", "error", "-select_streams", "v:0",
|
|
"-show_entries", "stream=avg_frame_rate,r_frame_rate,nb_frames,duration",
|
|
"-show_entries", "format=duration",
|
|
"-of", "json", path],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
if r.returncode != 0:
|
|
return 0, 0
|
|
data = _json.loads(r.stdout)
|
|
streams = data.get("streams", [{}])
|
|
s = streams[0] if streams else {}
|
|
fmt = data.get("format", {})
|
|
fps = cls._parse_rate(str(s.get("avg_frame_rate", "")))
|
|
if fps <= 0:
|
|
fps = cls._parse_rate(str(s.get("r_frame_rate", "")))
|
|
dur = 0.0
|
|
for cand in (s.get("duration"), fmt.get("duration")):
|
|
try:
|
|
dur = float(cand)
|
|
if dur > 0:
|
|
break
|
|
except (TypeError, ValueError):
|
|
continue
|
|
total_ms = int(round(dur * 1000)) if dur > 0 else 0
|
|
total_frames = 0
|
|
if dur > 0 and fps > 0:
|
|
total_frames = int(round(dur * fps))
|
|
if total_frames <= 0:
|
|
try:
|
|
total_frames = int(s.get("nb_frames") or 0)
|
|
except (TypeError, ValueError):
|
|
total_frames = 0
|
|
return max(0, total_frames), max(0, total_ms)
|
|
except Exception:
|
|
return 0, 0
|
|
|
|
def run(self):
|
|
import subprocess
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
apply_script = os.path.join(script_dir, "apply_sync.py")
|
|
|
|
cmd = [sys.executable, apply_script, "--config", self.config,
|
|
"--a", self.video_a, "--b", self.video_b,
|
|
"--lag", f"{self.lag_sec:.6f}", "--progress"]
|
|
self.status.emit("starting apply_sync...")
|
|
|
|
total_frames, _total_ms = self.probe_video_total(self.video_a)
|
|
|
|
# Single merged pipe: apply_sync forwards ffmpeg -progress key=value
|
|
# lines on stdout; human text has no "=" and is skipped for progress.
|
|
proc = subprocess.Popen(
|
|
cmd, cwd=script_dir,
|
|
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
|
universal_newlines=True, bufsize=1
|
|
)
|
|
|
|
frames_done = 0
|
|
last_line = ""
|
|
assert proc.stdout is not None
|
|
for line in proc.stdout:
|
|
if self._abort:
|
|
proc.kill()
|
|
self.finished.emit(False)
|
|
return
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
last_line = line
|
|
if line.startswith("frame="):
|
|
try:
|
|
frames_done = int(line.split("=", 1)[1].strip())
|
|
except (ValueError, IndexError):
|
|
pass
|
|
self.progress.emit(frames_done, total_frames)
|
|
elif line.startswith("progress=end"):
|
|
self.progress.emit(total_frames or frames_done,
|
|
total_frames or frames_done)
|
|
elif "=" not in line and ("error" in line.lower() or "failed" in line.lower()):
|
|
self.status.emit(line)
|
|
|
|
rc = proc.wait()
|
|
# Keep ffmpeg's final summary line visible in the launching terminal.
|
|
if last_line:
|
|
print(f"[apply_sync] {last_line}", flush=True)
|
|
self.finished.emit(rc == 0)
|
|
|
|
|
|
|
|
def load_config(path: str) -> dict:
|
|
if not os.path.exists(path):
|
|
return {}
|
|
with open(path, "rb") as f:
|
|
return tomllib.load(f)
|
|
|
|
|
|
def decimate_for_plot(arr: np.ndarray, max_pts: int = 120_000) -> tuple[np.ndarray, int]:
|
|
if len(arr) <= max_pts:
|
|
return arr, 1
|
|
stride = int(math.ceil(len(arr) / max_pts))
|
|
return arr[::stride], stride
|
|
|
|
def decimate_with_peak(arr: np.ndarray, peak_idx: int, max_pts: int = 200_000) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Decimate array for plotting using max-pooling to preserve all peaks."""
|
|
n = len(arr)
|
|
if n <= max_pts:
|
|
return arr, np.arange(n, dtype=np.float64)
|
|
stride = int(math.ceil(n / max_pts))
|
|
abs_arr = np.abs(arr)
|
|
n_blocks = (n + stride - 1) // stride
|
|
out = np.empty(n_blocks, dtype=arr.dtype)
|
|
idx_out = np.empty(n_blocks, dtype=np.float64)
|
|
for i in range(n_blocks):
|
|
start = i * stride
|
|
end = min(start + stride, n)
|
|
block_abs = abs_arr[start:end]
|
|
local_peak = int(np.argmax(block_abs))
|
|
out[i] = arr[start + local_peak]
|
|
idx_out[i] = float(start + local_peak)
|
|
return out, idx_out
|
|
|
|
|
|
def find_wavs_from_config(cfg: dict) -> tuple[str, str, int, float, float, int]:
|
|
audio_cfg = cfg.get("audio", {})
|
|
extract_cfg = cfg.get("extract", {})
|
|
sync_cfg = cfg.get("sync", {})
|
|
sr = int(audio_cfg.get("sample_rate", 48000))
|
|
volume = int(audio_cfg.get("volume"))
|
|
# priority: explicit wav paths else derived from extracted dir
|
|
a_wav = audio_cfg.get("file_a_wav")
|
|
b_wav = audio_cfg.get("file_b_wav")
|
|
# fallback to extract files
|
|
out_dir = extract_cfg.get("out_dir", "extracted")
|
|
file_a = extract_cfg.get("file_a")
|
|
file_b = extract_cfg.get("file_b")
|
|
if not a_wav and file_a:
|
|
stem = os.path.splitext(os.path.basename(file_a))[0]
|
|
cand = os.path.join(out_dir, f"{stem}_audio.wav")
|
|
if os.path.exists(cand):
|
|
a_wav = cand
|
|
else:
|
|
a_wav = cand # keep for error msg
|
|
if not b_wav and file_b:
|
|
stem = os.path.splitext(os.path.basename(file_b))[0]
|
|
cand = os.path.join(out_dir, f"{stem}_audio.wav")
|
|
if os.path.exists(cand):
|
|
b_wav = cand
|
|
else:
|
|
b_wav = cand
|
|
# also allow sync-specific override
|
|
duration = extract_cfg.get("duration", 90)
|
|
start = float(extract_cfg.get("start", 0.0))
|
|
# duration_sec comment: only duration needed is off the extract — we respect that
|
|
return a_wav, b_wav, sr, float(start), float(duration), volume
|
|
|
|
|
|
class SyncWindow(QtWidgets.QMainWindow):
|
|
def __init__(self, a_path: str, b_path: str, sr: int = 48000, method: str = "phat", device: str = "auto",
|
|
video_a: str = "", video_b: str = "", out_dir: str = "output", volume: int = 30,
|
|
duration: float = 90.0):
|
|
super().__init__()
|
|
self.setWindowTitle(f"movie_sync — audio sync test — {device_info()}")
|
|
self.resize(1400, 900)
|
|
self.a_path = a_path
|
|
self.b_path = b_path
|
|
self.video_a = video_a
|
|
self.video_b = video_b
|
|
self.out_dir = out_dir
|
|
self.sr = sr
|
|
self.duration = duration
|
|
self.method = method
|
|
self.device_pref = device
|
|
self._updating = False
|
|
|
|
# load
|
|
print(f"Loading {a_path}")
|
|
self.a, self.sr_a = cached_load_wav(a_path, target_sr=sr, _loader=load_wav)
|
|
print(f"Loading {b_path}")
|
|
self.b, self.sr_b = cached_load_wav(b_path, target_sr=sr, _loader=load_wav)
|
|
# keep originals for lag apply
|
|
self.a_orig = self.a.copy()
|
|
self.b_orig = self.b.copy()
|
|
|
|
# initial correlate
|
|
self.lag_samples, self.peak, self.corr = correlate(self.a, self.b, sr=sr, method=method, device=device)
|
|
self.lag_sec = self.lag_samples / sr
|
|
print(f"initial lag {self.lag_samples} samples = {self.lag_sec:.6f}s peak={self.peak:.4f}")
|
|
|
|
# decimated views
|
|
self.a_plot, self.stride_a = decimate_for_plot(self.a)
|
|
self.b_plot, self.stride_b = decimate_for_plot(self.b)
|
|
self.t_a = np.arange(len(self.a_plot)) * self.stride_a / sr
|
|
self.t_b = np.arange(len(self.b_plot)) * self.stride_b / sr
|
|
|
|
# xcorr display — corr is already fftshifted
|
|
N = len(self.corr)
|
|
# shifted lags: -N//2 .. N//2-1
|
|
self.corr_lags = np.arange(N) - N//2
|
|
self.corr_lags_sec = self.corr_lags / sr
|
|
|
|
central = QtWidgets.QWidget()
|
|
self.setCentralWidget(central)
|
|
lay = QtWidgets.QVBoxLayout(central)
|
|
|
|
# info bar
|
|
info = QtWidgets.QLabel(f"A: {os.path.basename(a_path)} ({len(self.a)/sr:.1f}s) B: {os.path.basename(b_path)} ({len(self.b)/sr:.1f}s) sr={sr} method={method} {device_info()}")
|
|
info.setWordWrap(True)
|
|
lay.addWidget(info)
|
|
|
|
# plots
|
|
self.gw = pg.GraphicsLayoutWidget()
|
|
lay.addWidget(self.gw, stretch=1)
|
|
|
|
# overlay: A and B on same axes
|
|
self.p0 = self.gw.addPlot(row=0, col=0, title="Overlay A (teal) + B (red)")
|
|
self.p0.showGrid(x=True, y=True)
|
|
self.p0.setLabel("bottom", "Time", "s")
|
|
self.curve_a_ov = self.p0.plot(self.t_a, self.a_plot, pen=pg.mkPen("#4cc9f0", width=1), name="A")
|
|
self.curve_b_ov = self.p0.plot(self.t_b, self.b_plot, pen=pg.mkPen("#f72585", width=1, style=QtCore.Qt.PenStyle.DashLine), name="B")
|
|
self.p0.addLegend(offset=(60, 10))
|
|
self.p0.setMouseEnabled(x=True, y=False)
|
|
|
|
# waveform A
|
|
self.p1 = self.gw.addPlot(row=1, col=0, title="Waveform A (reference)")
|
|
self.p1.showGrid(x=True, y=True)
|
|
self.p1.setLabel("bottom", "Time", "s")
|
|
self.curve_a = self.p1.plot(self.t_a, self.a_plot, pen=pg.mkPen("#4cc9f0", width=1))
|
|
self.p1.setMouseEnabled(x=True, y=False)
|
|
|
|
# waveform B with lag
|
|
self.p2 = self.gw.addPlot(row=2, col=0, title=f"Waveform B (shifted by lag)")
|
|
self.p2.showGrid(x=True, y=True)
|
|
self.p2.setLabel("bottom", "Time", "s")
|
|
self.curve_b = self.p2.plot(self.t_b, self.b_plot, pen=pg.mkPen("#f72585", width=1))
|
|
self.p2.setMouseEnabled(x=True, y=False)
|
|
# link x
|
|
self.p1.setXLink(self.p0)
|
|
self.p2.setXLink(self.p0)
|
|
|
|
# xcorr
|
|
self.p3 = self.gw.addPlot(row=3, col=0, title="Cross-correlation (GCC-PHAT)" if method=="phat" else "Cross-correlation")
|
|
self.p3.showGrid(x=True, y=True)
|
|
self.p3.setLabel("bottom", "Lag", "s")
|
|
# decimate corr with peak awareness
|
|
peak_in_shifted = int(np.argmax(np.abs(self.corr)))
|
|
corr_dec, corr_idx = decimate_with_peak(self.corr, peak_in_shifted, max_pts=200_000)
|
|
corr_lags_dec = self.corr_lags_sec[corr_idx.astype(int)]
|
|
self.curve_corr = self.p3.plot(corr_lags_dec, np.abs(corr_dec), pen=pg.mkPen("#ffd166", width=1))
|
|
self.peak_line = pg.InfiniteLine(pos=self.lag_sec, angle=90, pen=pg.mkPen("#06d6a0", width=2, style=QtCore.Qt.PenStyle.DashLine))
|
|
self.p3.addItem(self.peak_line)
|
|
self.p3.setXRange(self.lag_sec-2, self.lag_sec+2)
|
|
|
|
self.xcorr_val_label = QtWidgets.QLabel()
|
|
self.xcorr_val_label.setStyleSheet("color: #ffd166; font-size: 12px;")
|
|
lay.addWidget(self.xcorr_val_label)
|
|
|
|
# controls
|
|
ctr = QtWidgets.QHBoxLayout()
|
|
lay.addLayout(ctr)
|
|
|
|
ctr.addWidget(QtWidgets.QLabel("Lag (s):"))
|
|
self.slider = QtWidgets.QSlider(QtCore.Qt.Orientation.Horizontal)
|
|
self.slider.setRange(-10000, 10000) # will map to seconds via scale
|
|
# range +/-10s in ms
|
|
self.slider.setMinimum(int(-10 * 1000))
|
|
self.slider.setMaximum(int(10 * 1000))
|
|
self.slider.setValue(int(self.lag_sec * 1000))
|
|
self.slider.setTickInterval(100)
|
|
ctr.addWidget(self.slider, stretch=1)
|
|
|
|
self.spin_sec = QtWidgets.QDoubleSpinBox()
|
|
self.spin_sec.setRange(-60, 60)
|
|
self.spin_sec.setDecimals(6)
|
|
self.spin_sec.setSingleStep(0.001)
|
|
self.spin_sec.setValue(self.lag_sec)
|
|
self.spin_sec.setSuffix(" s")
|
|
self.spin_sec.setKeyboardTracking(False)
|
|
ctr.addWidget(self.spin_sec)
|
|
|
|
self.spin_samp = QtWidgets.QSpinBox()
|
|
self.spin_samp.setRange(-10_000_000, 10_000_000)
|
|
self.spin_samp.setSingleStep(1)
|
|
self.spin_samp.setValue(self.lag_samples)
|
|
self.spin_samp.setSuffix(" samp")
|
|
ctr.addWidget(self.spin_samp)
|
|
|
|
btn_auto = QtWidgets.QPushButton("Auto peak")
|
|
ctr.addWidget(btn_auto)
|
|
|
|
btn_swap = QtWidgets.QPushButton("Swap A/B")
|
|
ctr.addWidget(btn_swap)
|
|
|
|
btn_export = QtWidgets.QPushButton("Export data")
|
|
ctr.addWidget(btn_export)
|
|
|
|
btn_verify = QtWidgets.QPushButton(f"Verify ({duration:g}s)")
|
|
ctr.addWidget(btn_verify)
|
|
|
|
btn_apply = QtWidgets.QPushButton("Apply sync")
|
|
self.btn_apply = btn_apply
|
|
ctr.addWidget(btn_apply)
|
|
|
|
# volume + playback
|
|
ctr2 = QtWidgets.QHBoxLayout()
|
|
lay.addLayout(ctr2)
|
|
ctr2.addWidget(QtWidgets.QLabel("Volume:"))
|
|
self.vol = QtWidgets.QSlider(QtCore.Qt.Orientation.Horizontal)
|
|
self.vol.setRange(0, 100)
|
|
self.vol.setValue(volume)
|
|
self.vol.setMaximumWidth(200)
|
|
ctr2.addWidget(self.vol)
|
|
self.vol_label = QtWidgets.QLabel(f"{volume}%")
|
|
ctr2.addWidget(self.vol_label)
|
|
|
|
self.btn_play_a = QtWidgets.QPushButton("Play A")
|
|
self.btn_play_b = QtWidgets.QPushButton("Play B+lag")
|
|
self.btn_play_mix = QtWidgets.QPushButton("Play Mix")
|
|
self.btn_stop = QtWidgets.QPushButton("Stop")
|
|
|
|
ctr2.addWidget(QtWidgets.QLabel("Duration:"))
|
|
self.play_dur = QtWidgets.QDoubleSpinBox()
|
|
max_dur = min(self.duration, len(self.a_orig) / self.sr)
|
|
self.play_dur.setRange(0.1, max(max_dur, 0.1))
|
|
self.play_dur.setDecimals(1)
|
|
self.play_dur.setSingleStep(0.5)
|
|
self.play_dur.setValue(min(2.0, max_dur))
|
|
self.play_dur.setSuffix("s")
|
|
self.play_dur.setMaximumWidth(80)
|
|
ctr2.addWidget(self.play_dur)
|
|
|
|
for b in [self.btn_play_a, self.btn_play_b, self.btn_play_mix, self.btn_stop]:
|
|
ctr2.addWidget(b)
|
|
if not HAS_SD:
|
|
for b in [self.btn_play_a, self.btn_play_b, self.btn_play_mix, self.btn_stop]:
|
|
b.setEnabled(False)
|
|
b.setToolTip("sounddevice not installed — run make install (optional)")
|
|
ctr2.addWidget(QtWidgets.QLabel("(sounddevice missing — playback disabled)"))
|
|
|
|
self.ffmpeg_label = QtWidgets.QLabel()
|
|
self.ffmpeg_label.setWordWrap(True)
|
|
self.ffmpeg_label.setTextInteractionFlags(QtCore.Qt.TextInteractionFlag.TextSelectableByMouse)
|
|
lay.addWidget(self.ffmpeg_label)
|
|
|
|
# ── Apply-sync progress (full-width bottom row) ──
|
|
apply_row = QtWidgets.QHBoxLayout()
|
|
lay.addLayout(apply_row)
|
|
self.apply_progress = QtWidgets.QProgressBar()
|
|
self.apply_progress.setRange(0, 100)
|
|
self.apply_progress.setValue(0)
|
|
self.apply_progress.setTextVisible(True)
|
|
self.apply_progress.setVisible(False)
|
|
self.apply_progress.setSizePolicy(
|
|
QtWidgets.QSizePolicy.Policy.Expanding,
|
|
QtWidgets.QSizePolicy.Policy.Fixed)
|
|
apply_row.addWidget(self.apply_progress, stretch=1)
|
|
|
|
self.apply_status = QtWidgets.QLabel("")
|
|
self.apply_status.setStyleSheet("color: #aaa; font-size: 11px;")
|
|
self.apply_status.setWordWrap(True)
|
|
apply_row.addWidget(self.apply_status, stretch=0)
|
|
|
|
# signals
|
|
self.slider.valueChanged.connect(self.on_slider)
|
|
self.spin_sec.valueChanged.connect(self.on_spin_sec)
|
|
self.spin_samp.valueChanged.connect(self.on_spin_samp)
|
|
self.vol.valueChanged.connect(lambda v: self.vol_label.setText(f"{v}%"))
|
|
btn_auto.clicked.connect(self.on_auto)
|
|
btn_swap.clicked.connect(self.on_swap)
|
|
btn_export.clicked.connect(self.on_export)
|
|
btn_verify.clicked.connect(self.on_verify)
|
|
btn_apply.clicked.connect(self.on_apply_sync)
|
|
self.btn_play_a.clicked.connect(lambda: self.play("a"))
|
|
self.btn_play_b.clicked.connect(lambda: self.play("b"))
|
|
self.btn_play_mix.clicked.connect(lambda: self.play("mix"))
|
|
self.btn_stop.clicked.connect(self.stop_play)
|
|
|
|
self.update_labels()
|
|
self._update_xcorr_label()
|
|
|
|
def _set_lag(self, lag_sec: float, source: str = ""):
|
|
if self._updating:
|
|
return
|
|
self._updating = True
|
|
lag_sec = float(np.clip(lag_sec, -60, 60))
|
|
lag_samp = int(round(lag_sec * self.sr))
|
|
self.lag_sec = lag_sec
|
|
self.lag_samples = lag_samp
|
|
# update widgets
|
|
self.slider.blockSignals(True); self.slider.setValue(int(lag_sec*1000)); self.slider.blockSignals(False)
|
|
self.spin_sec.blockSignals(True); self.spin_sec.setValue(lag_sec); self.spin_sec.blockSignals(False)
|
|
self.spin_samp.blockSignals(True); self.spin_samp.setValue(lag_samp); self.spin_samp.blockSignals(False)
|
|
self.peak_line.setPos(lag_sec)
|
|
# shift B curve visually by lag (pad with silence not needed for visualize, just shift x)
|
|
shift = lag_sec
|
|
self.curve_b.setData(self.t_b + shift, self.b_plot)
|
|
self.curve_b_ov.setData(self.t_b + shift, self.b_plot)
|
|
self._update_xcorr_label()
|
|
self.update_labels()
|
|
self._updating = False
|
|
|
|
def _update_xcorr_label(self):
|
|
N = len(self.corr)
|
|
idx = N // 2 + self.lag_samples
|
|
if 0 <= idx < N:
|
|
val = np.abs(self.corr[idx])
|
|
peak_val = np.abs(self.peak)
|
|
pct = (val / peak_val * 100) if peak_val > 0 else 0
|
|
self.xcorr_val_label.setText(
|
|
f"Xcorr at lag: {val:.6f} (peak: {peak_val:.6f} = {pct:.1f}%)"
|
|
)
|
|
else:
|
|
self.xcorr_val_label.setText("Xcorr at lag: (out of range)")
|
|
|
|
def update_labels(self):
|
|
lag_ms = self.lag_sec * 1000
|
|
stem = os.path.splitext(os.path.basename(self.video_a))[0]
|
|
out_file = os.path.join(self.out_dir, f"{stem}_synced.mkv")
|
|
script = os.path.basename(__file__) if not self.video_a else "apply_sync.py"
|
|
cmd = (f'python {script} --config config.toml --lag {self.lag_sec:.6f}')
|
|
if lag_ms >= 0:
|
|
note = f"B delayed by {self.lag_sec:.6f}s ({self.lag_samples} samp) → pad B start with silence {lag_ms:.0f} ms"
|
|
else:
|
|
note = f"B earlier by {abs(self.lag_sec):.6f}s → pad A / use -itsoffset {abs(lag_ms):.0f}ms"
|
|
note += f"\nOutput: {out_file}"
|
|
self.ffmpeg_label.setText(f"{note}\n{cmd}")
|
|
|
|
# slots
|
|
def on_slider(self, v):
|
|
self._set_lag(v/1000.0, "slider")
|
|
|
|
def on_spin_sec(self, v):
|
|
self._set_lag(float(v), "sec")
|
|
|
|
def on_spin_samp(self, v):
|
|
self._set_lag(int(v)/self.sr, "samp")
|
|
|
|
def on_auto(self):
|
|
# recompute peak
|
|
lag, peak, corr = correlate(self.a_orig, self.b_orig, sr=self.sr, method=self.method, device=self.device_pref)
|
|
self.corr = corr
|
|
# update corr curve? re-decimate
|
|
self._set_lag(lag / self.sr, "auto")
|
|
print(f"auto lag {lag} samp {lag/self.sr:.6f}s peak={peak:.4f}")
|
|
|
|
def on_swap(self):
|
|
self.a_orig, self.b_orig = self.b_orig, self.a_orig
|
|
self.a, self.b = self.a_orig.copy(), self.b_orig.copy()
|
|
self.a_plot, _ = decimate_for_plot(self.a)
|
|
self.b_plot, _ = decimate_for_plot(self.b)
|
|
self.curve_a.setData(self.t_a, self.a_plot)
|
|
self.curve_a_ov.setData(self.t_a, self.a_plot)
|
|
# recalc
|
|
lag, _, corr = correlate(self.a, self.b, sr=self.sr, method=self.method, device=self.device_pref)
|
|
self.corr = corr
|
|
self._set_lag(lag/self.sr, "swap")
|
|
|
|
def on_export(self):
|
|
import json
|
|
path, _ = QtWidgets.QFileDialog.getSaveFileName(
|
|
self, "Export sync data", "sync_data.json", "JSON (*.json)")
|
|
if not path:
|
|
return
|
|
# export decimated correlation (full res is too large)
|
|
peak_in_shifted = int(np.argmax(np.abs(self.corr)))
|
|
corr_dec, corr_idx = decimate_with_peak(self.corr, peak_in_shifted, max_pts=200_000)
|
|
corr_lags_sec = self.corr_lags_sec[corr_idx.astype(int)]
|
|
export = {
|
|
"file_a": self.a_path,
|
|
"file_b": self.b_path,
|
|
"sr": self.sr,
|
|
"method": self.method,
|
|
"device": self.device_pref,
|
|
"lag_samples": self.lag_samples,
|
|
"lag_sec": self.lag_sec,
|
|
"peak_value": float(self.peak),
|
|
"a_samples": len(self.a_orig),
|
|
"a_duration_sec": len(self.a_orig) / self.sr,
|
|
"b_samples": len(self.b_orig),
|
|
"b_duration_sec": len(self.b_orig) / self.sr,
|
|
"corr_peak_index": peak_in_shifted,
|
|
"corr_length": len(self.corr),
|
|
# decimated correlation for plotting (not full 4M points)
|
|
"corr_lags_sec": corr_lags_sec.tolist(),
|
|
"corr_magnitude": np.abs(corr_dec).tolist(),
|
|
# waveform stats (full res waveforms too large for JSON)
|
|
"a_stats": {
|
|
"mean": float(self.a_orig.mean()),
|
|
"std": float(self.a_orig.std()),
|
|
"min": float(self.a_orig.min()),
|
|
"max": float(self.a_orig.max()),
|
|
},
|
|
"b_stats": {
|
|
"mean": float(self.b_orig.mean()),
|
|
"std": float(self.b_orig.std()),
|
|
"min": float(self.b_orig.min()),
|
|
"max": float(self.b_orig.max()),
|
|
},
|
|
}
|
|
with open(path, "w") as f:
|
|
json.dump(export, f, indent=2)
|
|
print(f"exported to {path}")
|
|
self.setWindowTitle(f"movie_sync — exported to {os.path.basename(path)}")
|
|
|
|
def on_verify(self):
|
|
"""Launch verify_sync.py in short (30s) mode with extracted parts."""
|
|
import subprocess
|
|
script = os.path.join(os.path.dirname(__file__), "verify_sync.py")
|
|
cmd = [sys.executable, script, "--config", "config.toml",
|
|
"--a", self.video_a, "--b", self.video_b,
|
|
"--wav-a", self.a_path, "--wav-b", self.b_path,
|
|
"--lag", f"{self.lag_sec:.6f}", "--mode", "short"]
|
|
print(f"Launching: {' '.join(cmd)}")
|
|
subprocess.Popen(cmd, cwd=os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
def on_apply_sync(self):
|
|
"""Run apply_sync.py on a background thread with progress tracking."""
|
|
self.btn_apply.setEnabled(False)
|
|
self.apply_progress.setVisible(True)
|
|
self.apply_progress.setRange(0, 0) # indeterminate until we get frame count
|
|
self.apply_progress.setValue(0)
|
|
self.apply_status.setText("starting...")
|
|
|
|
self._apply_worker = ApplySyncWorker(
|
|
self.lag_sec, self.video_a, self.video_b
|
|
)
|
|
self._apply_worker.progress.connect(self._on_apply_progress)
|
|
self._apply_worker.status.connect(lambda s: self.apply_status.setText(s))
|
|
self._apply_worker.finished.connect(self._on_apply_done)
|
|
self._apply_worker.start()
|
|
|
|
def _on_apply_progress(self, frames_done: int, frames_total: int):
|
|
if frames_total > 0:
|
|
done = min(frames_done, frames_total)
|
|
self.apply_progress.setRange(0, frames_total)
|
|
self.apply_progress.setValue(done)
|
|
pct = done * 100 // frames_total
|
|
self.apply_progress.setFormat(f"{pct}% — frame {done}/{frames_total}")
|
|
self.apply_status.setText(f"encoding frame {done}/{frames_total} ({pct}%)")
|
|
else:
|
|
self.apply_progress.setRange(0, 0) # indeterminate
|
|
self.apply_progress.setFormat(f"frame {frames_done}...")
|
|
self.apply_status.setText(f"encoding frame {frames_done}...")
|
|
|
|
def _on_apply_done(self, success: bool):
|
|
self.btn_apply.setEnabled(True)
|
|
self.apply_progress.setVisible(False)
|
|
if success:
|
|
self.apply_status.setText("done")
|
|
# Launch verify in full mode
|
|
import subprocess
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
verify_script = os.path.join(script_dir, "verify_sync.py")
|
|
cmd = [sys.executable, verify_script, "--config", "config.toml",
|
|
"--a", self.video_a, "--b", self.video_b,
|
|
"--lag", f"{self.lag_sec:.6f}", "--mode", "full"]
|
|
print(f"Launching: {' '.join(cmd)}")
|
|
subprocess.Popen(cmd, cwd=script_dir)
|
|
else:
|
|
self.apply_status.setText("apply_sync failed")
|
|
|
|
def play(self, which: str):
|
|
if not HAS_SD:
|
|
return
|
|
vol = self.vol.value()/100.0
|
|
sr = self.sr
|
|
dur_samp = int(self.play_dur.value() * sr)
|
|
start_samp = int(10 * sr)
|
|
if start_samp + dur_samp > len(self.a_orig):
|
|
start_samp = 0
|
|
a_seg = self.a_orig[start_samp:start_samp+dur_samp] * vol
|
|
# apply lag with silence padding (no sound for delay) — Option A
|
|
lag = self.lag_samples
|
|
# For playback mix, we need to align: if B delayed (lag>0), pad B front with silence
|
|
# For preview we shift B segment accordingly
|
|
b_start = start_samp
|
|
if lag > 0:
|
|
# B delayed => B segment should be taken earlier to align? For playback we pad B with lag silence, so effective B at time t is original at t-lag
|
|
idx = b_start - lag
|
|
if idx < 0:
|
|
b_seg = np.concatenate([np.zeros(-idx, dtype=np.float32), self.b_orig[0:b_start + dur_samp - (-idx) if b_start+dur_samp - (-idx)>0 else 0]])
|
|
# slice to dur
|
|
b_seg = b_seg[:dur_samp]
|
|
if len(b_seg) < dur_samp:
|
|
b_seg = np.concatenate([b_seg, np.zeros(dur_samp-len(b_seg), dtype=np.float32)])
|
|
else:
|
|
b_seg = self.b_orig[idx:idx+dur_samp]
|
|
elif lag < 0:
|
|
idx = b_start - lag # lag negative => plus
|
|
b_seg = self.b_orig[idx:idx+dur_samp] if idx+dur_samp < len(self.b_orig) else np.zeros(dur_samp, dtype=np.float32)
|
|
if len(b_seg) < dur_samp:
|
|
b_seg = np.concatenate([b_seg, np.zeros(dur_samp-len(b_seg), dtype=np.float32)])
|
|
else:
|
|
b_seg = self.b_orig[b_start:b_start+dur_samp]
|
|
b_seg = b_seg * vol
|
|
try:
|
|
sd.stop()
|
|
if which == "a":
|
|
# A on left, silence on right
|
|
stereo = np.column_stack([a_seg, np.zeros_like(a_seg)])
|
|
sd.play(stereo, samplerate=sr)
|
|
elif which == "b":
|
|
# Silence on left, B on right
|
|
stereo = np.column_stack([np.zeros_like(b_seg), b_seg])
|
|
sd.play(stereo, samplerate=sr)
|
|
elif which == "mix":
|
|
# A on left, B on right (stereo split)
|
|
ml = min(len(a_seg), len(b_seg))
|
|
stereo = np.column_stack([a_seg[:ml], b_seg[:ml]])
|
|
sd.play(stereo, samplerate=sr)
|
|
except Exception as e:
|
|
print(f"playback failed {e}", file=sys.stderr)
|
|
|
|
def stop_play(self):
|
|
if HAS_SD:
|
|
sd.stop()
|
|
|
|
def closeEvent(self, event):
|
|
worker = getattr(self, "_apply_worker", None)
|
|
if worker is not None and worker.isRunning():
|
|
worker.abort()
|
|
worker.wait(3000)
|
|
super().closeEvent(event)
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--config", default="config.toml")
|
|
ap.add_argument("--a", default=None)
|
|
ap.add_argument("--b", default=None)
|
|
ap.add_argument("--device", default=None, choices=["auto","cpu","cuda"])
|
|
ap.add_argument("--method", default=None, choices=["phat","standard"])
|
|
args = ap.parse_args()
|
|
|
|
cfg = load_config(args.config)
|
|
cfg_a, cfg_b, sr, _, duration, volume = find_wavs_from_config(cfg)
|
|
a_path = args.a or cfg_a
|
|
b_path = args.b or cfg_b
|
|
method = args.method or cfg.get("sync", {}).get("method", "phat")
|
|
device = args.device or cfg.get("sync", {}).get("use_gpu", "auto")
|
|
# map toml "use_gpu" to device
|
|
if device == "auto":
|
|
device = "auto"
|
|
elif device == "cpu":
|
|
device = "cpu"
|
|
elif device in ("cuda", "gpu"):
|
|
device = "cuda"
|
|
|
|
video_a = cfg.get("extract", {}).get("file_a", "")
|
|
video_b = cfg.get("extract", {}).get("file_b", "")
|
|
out_dir = cfg.get("sync", {}).get("out_dir", "output")
|
|
|
|
if not a_path or not b_path:
|
|
print(f"ERROR: no input wavs. Set file_a/file_b in {args.config} or --a/--b", file=sys.stderr)
|
|
sys.exit(1)
|
|
if not os.path.exists(a_path):
|
|
print(f"ERROR: A not found {a_path}", file=sys.stderr); sys.exit(1)
|
|
if not os.path.exists(b_path):
|
|
print(f"ERROR: B not found {b_path}", file=sys.stderr); sys.exit(1)
|
|
|
|
app = QtWidgets.QApplication(sys.argv)
|
|
signal.signal(signal.SIGINT, signal.SIG_DFL)
|
|
# Dark theme: Fusion style + dark palette (works without kvantum)
|
|
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")
|
|
w = SyncWindow(a_path, b_path, sr=sr, method=method, device=device,
|
|
video_a=video_a, video_b=video_b, out_dir=out_dir, volume=volume,
|
|
duration=duration)
|
|
w.show()
|
|
sys.exit(app.exec())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|