562 lines
23 KiB
Python
562 lines
23 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
|
|
|
|
|
|
|
|
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):
|
|
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.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)
|
|
|
|
# waveform A
|
|
self.p1 = self.gw.addPlot(row=0, 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=1, 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.p2.setXLink(self.p1)
|
|
|
|
# xcorr
|
|
self.p3 = self.gw.addPlot(row=2, 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)
|
|
|
|
# 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("Verify (30s)")
|
|
ctr.addWidget(btn_verify)
|
|
|
|
btn_apply = QtWidgets.QPushButton("Apply sync")
|
|
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()
|
|
self.play_dur.setRange(0.1, 30.0)
|
|
self.play_dur.setDecimals(1)
|
|
self.play_dur.setSingleStep(0.5)
|
|
self.play_dur.setValue(2.0)
|
|
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)
|
|
|
|
# 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()
|
|
|
|
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)
|
|
# we keep t_b but offset x
|
|
shift = lag_sec
|
|
# pyqtgraph setPos shifts items; we recreate x
|
|
# Instead shift data: plot t_b + lag
|
|
self.curve_b.setData(self.t_b + shift, self.b_plot)
|
|
self.update_labels()
|
|
self._updating = False
|
|
|
|
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)
|
|
# 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",
|
|
"--lag", f"{self.lag_sec:.6f}", "--mode", "short"]
|
|
print(f"Launching: {' '.join(cmd)}")
|
|
env = os.environ.copy()
|
|
env["LC_NUMERIC"] = "C"
|
|
env["QT_QPA_PLATFORM"] = "xcb"
|
|
subprocess.Popen(cmd, cwd=os.path.dirname(os.path.abspath(__file__)), env=env)
|
|
|
|
def on_apply_sync(self):
|
|
"""Run apply_sync.py on full files, then launch verify_sync.py in full mode."""
|
|
import subprocess
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
apply_script = os.path.join(script_dir, "apply_sync.py")
|
|
verify_script = os.path.join(script_dir, "verify_sync.py")
|
|
|
|
# Run apply_sync
|
|
cmd = [sys.executable, apply_script, "--config", "config.toml",
|
|
"--lag", f"{self.lag_sec:.6f}"]
|
|
print(f"Running: {' '.join(cmd)}")
|
|
r = subprocess.run(cmd, cwd=script_dir)
|
|
if r.returncode != 0:
|
|
print(f"apply_sync failed with code {r.returncode}", file=sys.stderr)
|
|
return
|
|
|
|
# Launch verify in full mode
|
|
cmd2 = [sys.executable, verify_script, "--config", "config.toml",
|
|
"--lag", f"{self.lag_sec:.6f}", "--mode", "full"]
|
|
print(f"Launching: {' '.join(cmd2)}")
|
|
env2 = os.environ.copy()
|
|
env2["LC_NUMERIC"] = "C"
|
|
env2["QT_QPA_PLATFORM"] = "xcb"
|
|
subprocess.Popen(cmd2, cwd=script_dir, env=env2)
|
|
|
|
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":
|
|
sd.play(a_seg, samplerate=sr)
|
|
elif which == "b":
|
|
sd.play(b_seg, samplerate=sr)
|
|
elif which == "mix":
|
|
mix = (a_seg + b_seg) / 2.0
|
|
sd.play(mix, samplerate=sr)
|
|
except Exception as e:
|
|
print(f"playback failed {e}", file=sys.stderr)
|
|
|
|
def stop_play(self):
|
|
if HAS_SD:
|
|
sd.stop()
|
|
|
|
|
|
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, _, _, 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)
|
|
w.show()
|
|
sys.exit(app.exec())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|