1103 lines
45 KiB
Python
1103 lines
45 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Video cross-correlation lab — compare multiple video alignment methods.
|
||
|
||
Reads ONLY files from the extract out_dir (extracted/{stem}_proxy.mkv preferred,
|
||
extracted/{stem}_video.mkv fallback). NEVER touches the original sources
|
||
(network mounts) — config is used only to derive filenames and settings.
|
||
|
||
Methods (share one decode + letterbox crop + grid, then one batched FFT):
|
||
M1-global-luma mean luma per frame — naive 1D baseline (usually fails)
|
||
M2-grid-diff-3x3 frame-diff of 3x3 zone means (option b)
|
||
M3-pix-diff-sum per-pixel frame-diff z-scored, sum of per-pixel xcorr
|
||
(== true 2D space x time correlation, main proposal)
|
||
M4-pix-diff-median same signal, median over pixels (robust to logos/mismatches)
|
||
M5-pix-diff-phat same signal, PHAT whitening per pixel (codec-spectrum-robust)
|
||
M6-cut-spikes scene-change impulse trains (grade/quality independent)
|
||
M7-pix-raw-sum z-scored frames, no diff — isolates the diff effect
|
||
|
||
Run GUI: make vst or .venv/bin/python sync_video_test.py --config config.toml
|
||
Headless: .venv/bin/python sync_video_test.py --no-gui [--audio-lag 0.042] [--export v.json]
|
||
|
||
Lag sign convention is identical to sync_audio.correlate() (same estimator,
|
||
same axis mapping), so lag_sec feeds apply_sync --lag directly.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import math
|
||
import json
|
||
import time
|
||
import hashlib
|
||
import argparse
|
||
import subprocess
|
||
from types import SimpleNamespace
|
||
|
||
import numpy as np
|
||
|
||
ROOT = os.path.dirname(os.path.abspath(__file__))
|
||
CACHE_DIR = os.path.join(ROOT, "cache", "videos")
|
||
|
||
try:
|
||
import tomllib
|
||
except ModuleNotFoundError:
|
||
import tomli as tomllib
|
||
|
||
METHOD_NAMES = [
|
||
"M1-global-luma",
|
||
"M2-grid-diff-3x3",
|
||
"M3-pix-diff-sum",
|
||
"M4-pix-diff-median",
|
||
"M5-pix-diff-phat",
|
||
"M6-cut-spikes",
|
||
"M7-pix-raw-sum",
|
||
]
|
||
METHOD_COLORS = {
|
||
"M1-global-luma": "#8d99ae",
|
||
"M2-grid-diff-3x3": "#f72585",
|
||
"M3-pix-diff-sum": "#4cc9f0",
|
||
"M4-pix-diff-median": "#f8961e",
|
||
"M5-pix-diff-phat": "#b5179e",
|
||
"M6-cut-spikes": "#06d6a0",
|
||
"M7-pix-raw-sum": "#ffd166",
|
||
}
|
||
|
||
|
||
# ─────────────────────────── config ───────────────────────────
|
||
|
||
def load_config(path: str) -> dict:
|
||
if not os.path.exists(path):
|
||
return {}
|
||
with open(path, "rb") as f:
|
||
return tomllib.load(f)
|
||
|
||
|
||
def _parse_rate(s) -> float | None:
|
||
s = str(s).strip()
|
||
if not s or s in ("0/0", "0"):
|
||
return None
|
||
if "/" in s:
|
||
try:
|
||
n, d = s.split("/")
|
||
return float(n) / float(d) if float(d) != 0 else None
|
||
except ValueError:
|
||
return None
|
||
try:
|
||
return float(s)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _derive_stems(paths: list[str]) -> list[str]:
|
||
"""Same pure-path logic as extract.py: filenames differ -> use them;
|
||
identical -> walk up dirs until they diverge."""
|
||
basenames = [os.path.splitext(os.path.basename(p))[0] for p in paths]
|
||
if len(paths) == 1 or basenames[0] != basenames[1]:
|
||
return basenames
|
||
parts_a = os.path.normpath(paths[0]).split(os.sep)
|
||
parts_b = os.path.normpath(paths[1]).split(os.sep)
|
||
common = 0
|
||
for a, b in zip(reversed(parts_a), reversed(parts_b)):
|
||
if a == b:
|
||
common += 1
|
||
else:
|
||
break
|
||
rel_a = parts_a[len(parts_a) - common:]
|
||
rel_b = parts_b[len(parts_b) - common:]
|
||
stem_a = os.path.join(*rel_a) if len(rel_a) > 1 else rel_a[0]
|
||
stem_b = os.path.join(*rel_b) if len(rel_b) > 1 else rel_b[0]
|
||
return [stem_a, stem_b]
|
||
|
||
|
||
def find_videos(cfg: dict) -> tuple[str, str, float]:
|
||
"""Resolve the two extracted video files. Only ever looks inside out_dir."""
|
||
ex = cfg.get("extract", {})
|
||
vc = cfg.get("video", {})
|
||
out_dir = ex.get("out_dir", "extracted")
|
||
if not os.path.isabs(out_dir):
|
||
out_dir = os.path.join(ROOT, out_dir)
|
||
prefer_proxy = bool(vc.get("prefer_proxy", True))
|
||
fps_default = _parse_rate(ex.get("target_fps", "24000/1001")) or (24000 / 1001)
|
||
|
||
a = vc.get("file_a_mkv")
|
||
b = vc.get("file_b_mkv")
|
||
for v in (a, b):
|
||
if v and not os.path.isabs(v):
|
||
if a == v:
|
||
a = os.path.join(ROOT, v)
|
||
if b == v:
|
||
b = os.path.join(ROOT, v)
|
||
if (not a or not b) and ex.get("file_a") and ex.get("file_b"):
|
||
stems = _derive_stems([ex["file_a"], ex["file_b"]])
|
||
cands = []
|
||
for stem in stems:
|
||
master = os.path.join(out_dir, f"{stem}_video.mkv")
|
||
proxy = os.path.join(out_dir, f"{stem}_proxy.mkv")
|
||
cands.append(proxy if (prefer_proxy and os.path.exists(proxy)) else master)
|
||
a = a or cands[0]
|
||
b = b or cands[1]
|
||
if not a or not b:
|
||
print("ERROR: cannot resolve extracted videos. Set [extract] file_a/file_b or "
|
||
"[video] file_a_mkv/file_b_mkv in config.toml.", file=sys.stderr)
|
||
sys.exit(1)
|
||
for p in (a, b):
|
||
if not os.path.exists(p):
|
||
print(f"ERROR: not found: {p}\n run: make extract (or make vst)", file=sys.stderr)
|
||
sys.exit(1)
|
||
return a, b, fps_default
|
||
|
||
|
||
# ─────────────────────────── decode + cache ───────────────────────────
|
||
|
||
def probe_video(path: str) -> tuple[int | None, int | None, float | None]:
|
||
try:
|
||
r = subprocess.run(
|
||
["ffprobe", "-v", "error", "-select_streams", "v:0",
|
||
"-show_entries", "stream=width,height,avg_frame_rate,r_frame_rate",
|
||
"-of", "json", path],
|
||
capture_output=True, text=True, timeout=20)
|
||
s = json.loads(r.stdout)["streams"][0]
|
||
fps = _parse_rate(s.get("avg_frame_rate")) or _parse_rate(s.get("r_frame_rate"))
|
||
return int(s["width"]), int(s["height"]), fps
|
||
except Exception:
|
||
return None, None, None
|
||
|
||
|
||
def decode_gray(path: str, width: int, fps_fallback: float) -> tuple[np.ndarray, float]:
|
||
"""Decode via ffmpeg rawvideo pipe to float32 gray (T,H,W) in [0,1]."""
|
||
w_src, h_src, fps = probe_video(path)
|
||
fps = fps or fps_fallback
|
||
if w_src and h_src and width and w_src > width:
|
||
w = width
|
||
h = max(2, int(round(h_src * width / w_src / 2)) * 2)
|
||
else:
|
||
w, h = w_src, h_src
|
||
cmd = ["ffmpeg", "-v", "error", "-i", path,
|
||
"-vf", f"format=gray,scale={w}:{h}",
|
||
"-f", "rawvideo", "-pix_fmt", "gray", "-"]
|
||
proc = subprocess.run(cmd, capture_output=True)
|
||
if proc.returncode != 0 or not proc.stdout:
|
||
raise RuntimeError(f"ffmpeg decode failed for {path}: {proc.stderr.decode(errors='replace')[-300:]}")
|
||
fsz = w * h
|
||
T = len(proc.stdout) // fsz
|
||
frames = np.frombuffer(proc.stdout, dtype=np.uint8, count=T * fsz).reshape(T, h, w)
|
||
return frames.astype(np.float32) / 255.0, fps
|
||
|
||
|
||
def _cache_key(path: str, width: int) -> str:
|
||
st = os.stat(path)
|
||
raw = f"{os.path.realpath(path)}|{st.st_mtime_ns}|{st.st_size}|{width}"
|
||
return hashlib.sha256(raw.encode()).hexdigest()[:16]
|
||
|
||
|
||
def cached_decode(path: str, width: int, fps_fallback: float) -> tuple[np.ndarray, float]:
|
||
key = _cache_key(path, width)
|
||
npz = os.path.join(CACHE_DIR, f"{key}.npz")
|
||
if os.path.exists(npz):
|
||
try:
|
||
with np.load(npz) as z:
|
||
return z["frames"].astype(np.float32) / 255.0, float(z["fps"])
|
||
except Exception:
|
||
pass
|
||
frames, fps = decode_gray(path, width, fps_fallback)
|
||
os.makedirs(CACHE_DIR, exist_ok=True)
|
||
np.savez(npz, frames=(frames * 255.0).astype(np.uint8), fps=np.array(fps))
|
||
return frames, fps
|
||
|
||
|
||
# ─────────────────────────── bar crop + grid ───────────────────────────
|
||
|
||
def content_box(frames: np.ndarray, black: float = 0.09, static: float = 0.03,
|
||
max_frac: float = 0.45) -> tuple[int, int, int, int] | None:
|
||
"""Bounding box after trimming OUTER rows/cols that are near-black AND
|
||
temporally static across the whole clip (letterbox/pillarbox)."""
|
||
mean_t = frames.mean(axis=0)
|
||
std_t = frames.std(axis=0)
|
||
row_ok = ~((mean_t.mean(axis=1) < black) & (std_t.mean(axis=1) < static))
|
||
col_ok = ~((mean_t.mean(axis=0) < black) & (std_t.mean(axis=0) < static))
|
||
rows = np.flatnonzero(row_ok)
|
||
cols = np.flatnonzero(col_ok)
|
||
if len(rows) < 2 or len(cols) < 2:
|
||
return None
|
||
r0, r1 = int(rows[0]), int(rows[-1]) + 1
|
||
c0, c1 = int(cols[0]), int(cols[-1]) + 1
|
||
H, W = mean_t.shape
|
||
if min(r0, H - r1) > max_frac * H or min(c0, W - c1) > max_frac * W:
|
||
return None
|
||
return r0, r1, c0, c1
|
||
|
||
|
||
def block_avg(frames: np.ndarray, gh: int, gw: int) -> np.ndarray:
|
||
"""Block-average (T,H,W) -> (T,gh,gw) with non-integer ratios (mean per bin)."""
|
||
T, H, W = frames.shape
|
||
rb = np.unique(np.linspace(0, H, gh + 1).astype(int))
|
||
rb[0], rb[-1] = 0, H
|
||
cb = np.unique(np.linspace(0, W, gw + 1).astype(int))
|
||
cb[0], cb[-1] = 0, W
|
||
if len(rb) > 1:
|
||
s = np.add.reduceat(frames, rb[:-1], axis=1)
|
||
else:
|
||
s = frames.sum(axis=1, keepdims=True)
|
||
if len(cb) > 1:
|
||
s = np.add.reduceat(s, cb[:-1], axis=2)
|
||
else:
|
||
s = s.sum(axis=2, keepdims=True)
|
||
hs = (rb[1:] - rb[:-1]).astype(np.float32)
|
||
ws = (cb[1:] - cb[:-1]).astype(np.float32)
|
||
return (s / (hs[:, None] * ws[None, :])).astype(np.float32)
|
||
|
||
|
||
def zscore_rows(x: np.ndarray, clip: float | None = None) -> np.ndarray:
|
||
m = x.mean(axis=1, keepdims=True)
|
||
s = x.std(axis=1, keepdims=True)
|
||
z = (x - m) / np.maximum(s, 1e-6)
|
||
if clip:
|
||
z = np.clip(z, -clip, clip)
|
||
return z.astype(np.float32)
|
||
|
||
|
||
# ─────────────────────────── method signals ───────────────────────────
|
||
|
||
def _pix_mask(ga: np.ndarray, gb: np.ndarray, thr: float = 0.004) -> np.ndarray:
|
||
"""Valid pixels: temporal luma std above threshold in BOTH clips
|
||
(kills letterbox, static logos, flat credits)."""
|
||
va = ga.std(axis=0).reshape(-1)
|
||
vb = gb.std(axis=0).reshape(-1)
|
||
m = (va > thr) & (vb > thr)
|
||
if int(m.sum()) < 20:
|
||
return np.ones(va.shape, dtype=bool)
|
||
return m
|
||
|
||
|
||
def _runmed(x: np.ndarray, k: int) -> np.ndarray:
|
||
h = k // 2
|
||
pad = np.pad(x, h, mode="edge")
|
||
return np.median(np.lib.stride_tricks.sliding_window_view(pad, k), axis=1)
|
||
|
||
|
||
def _spike_signal(frames: np.ndarray, up: int = 4) -> np.ndarray:
|
||
"""Scene-change impulse train: peaks of |frame diff| energy above a running
|
||
median baseline, zero-stuffed x `up` and gaussian-smoothed."""
|
||
e = np.abs(np.diff(frames.reshape(frames.shape[0], -1), axis=0)).mean(axis=1)
|
||
if len(e) < 8:
|
||
return np.zeros(max(8, len(e) * up), dtype=np.float32)
|
||
r = e - _runmed(e, 31)
|
||
mad = np.median(np.abs(r - np.median(r))) * 1.4826
|
||
thr = max(6.0 * mad, 0.01)
|
||
keep: list[int] = []
|
||
for i in np.flatnonzero(r > thr):
|
||
lo, hi = max(0, i - 3), min(len(r), i + 4)
|
||
if r[i] >= r[lo:hi].max():
|
||
if not keep or i - keep[-1] >= 4:
|
||
keep.append(int(i))
|
||
s = np.zeros(len(e), dtype=np.float64)
|
||
if keep:
|
||
s[keep] = r[keep]
|
||
s4 = np.zeros(len(s) * up, dtype=np.float64)
|
||
s4[::up] = s
|
||
sig = 2.0
|
||
n = 6
|
||
g = np.exp(-0.5 * (np.arange(-n, n + 1) / sig) ** 2)
|
||
g /= g.sum()
|
||
s4 = np.convolve(s4, g, mode="same")
|
||
return s4.astype(np.float32)
|
||
|
||
|
||
def build_method_signals(ca: np.ndarray, cb: np.ndarray, gh: int, gw: int) -> dict:
|
||
"""Returns name -> {sa, sb, agg, phat, up}; sa/sb are (P,T) row-zscored."""
|
||
ga = block_avg(ca, gh, gw)
|
||
gb = block_avg(cb, gh, gw)
|
||
mask = _pix_mask(ga, gb)
|
||
specs: dict[str, dict] = {}
|
||
|
||
va = ca.mean(axis=(1, 2), dtype=np.float32)
|
||
vb = cb.mean(axis=(1, 2), dtype=np.float32)
|
||
specs["M1-global-luma"] = dict(
|
||
sa=zscore_rows(va[None, :]), sb=zscore_rows(vb[None, :]), agg="sum", phat=False, up=1)
|
||
|
||
za = np.diff(block_avg(ca, 3, 3).reshape(ca.shape[0], -1).T, axis=1)
|
||
zb = np.diff(block_avg(cb, 3, 3).reshape(cb.shape[0], -1).T, axis=1)
|
||
specs["M2-grid-diff-3x3"] = dict(
|
||
sa=zscore_rows(za, clip=2.0), sb=zscore_rows(zb, clip=2.0), agg="sum", phat=False, up=1)
|
||
|
||
fa = np.diff(ga.reshape(ga.shape[0], -1).T, axis=1)
|
||
fb = np.diff(gb.reshape(gb.shape[0], -1).T, axis=1)
|
||
sa_d = zscore_rows(fa, clip=2.0)[mask].astype(np.float32)
|
||
sb_d = zscore_rows(fb, clip=2.0)[mask].astype(np.float32)
|
||
specs["M3-pix-diff-sum"] = dict(sa=sa_d, sb=sb_d, agg="sum", phat=False, up=1)
|
||
specs["M4-pix-diff-median"] = dict(sa=sa_d, sb=sb_d, agg="median", phat=False, up=1)
|
||
specs["M5-pix-diff-phat"] = dict(sa=sa_d, sb=sb_d, agg="sum", phat=True, up=1)
|
||
|
||
spa = _spike_signal(ca)
|
||
spb = _spike_signal(cb)
|
||
specs["M6-cut-spikes"] = dict(
|
||
sa=zscore_rows(spa[None, :]), sb=zscore_rows(spb[None, :]), agg="sum", phat=False, up=4)
|
||
|
||
ra = ga.reshape(ga.shape[0], -1).T
|
||
rb_ = gb.reshape(gb.shape[0], -1).T
|
||
specs["M7-pix-raw-sum"] = dict(
|
||
sa=zscore_rows(ra)[mask].astype(np.float32),
|
||
sb=zscore_rows(rb_)[mask].astype(np.float32), agg="sum", phat=False, up=1)
|
||
|
||
return specs
|
||
|
||
|
||
# ─────────────────────────── xcorr core ───────────────────────────
|
||
|
||
def xcorr_pix(sa: np.ndarray, sb: np.ndarray, agg: str = "sum",
|
||
phat: bool = False) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
|
||
"""Batched per-pixel linear xcorr, summed/median over pixels, overlap-normalized.
|
||
Sum of per-pixel 1D xcorr == 2D space x time correlation.
|
||
Returns (curve, lags, per_pixel_curves)."""
|
||
Ta, Tb = sa.shape[1], sb.shape[1]
|
||
N = 1 << max(1, (Ta + Tb - 2)).bit_length()
|
||
Fa = np.fft.rfft(sa, n=N, axis=1)
|
||
Fb = np.fft.rfft(sb, n=N, axis=1)
|
||
R = Fa * np.conj(Fb)
|
||
if phat:
|
||
R = R / (np.abs(R) + 1e-9)
|
||
c = np.fft.irfft(R, n=N, axis=1)
|
||
maxlag = min(Ta, Tb) - 1
|
||
lags = np.arange(-maxlag, maxlag + 1)
|
||
idx = np.where(lags < 0, lags + N, lags)
|
||
lo = np.maximum(0, lags)
|
||
hi = np.minimum(Ta, Tb + lags)
|
||
overlap = np.maximum(hi - lo, 1).astype(np.float64)
|
||
floor = max(1.0, 0.2 * min(Ta, Tb))
|
||
cp = c[:, idx] / np.maximum(overlap, floor)[None, :]
|
||
if agg == "median":
|
||
curve = np.median(cp, axis=0)
|
||
else:
|
||
curve = cp.mean(axis=0)
|
||
return curve.astype(np.float32), lags, cp.astype(np.float32)
|
||
|
||
|
||
def find_peak(curve: np.ndarray, lags: np.ndarray,
|
||
max_lag_units: int = 0) -> tuple[int, float, float]:
|
||
if max_lag_units and max_lag_units < int(np.abs(lags).max()):
|
||
sel = np.flatnonzero(np.abs(lags) <= max_lag_units)
|
||
j = int(sel[int(np.argmax(curve[sel]))])
|
||
else:
|
||
j = int(np.argmax(curve))
|
||
lag = float(lags[j])
|
||
if 0 < j < len(curve) - 1:
|
||
a, b0, c0 = (float(curve[j - 1]), float(curve[j]), float(curve[j + 1]))
|
||
den = a - 2.0 * b0 + c0
|
||
if abs(den) > 1e-12:
|
||
lag += float(np.clip(0.5 * (a - c0) / den, -0.5, 0.5))
|
||
return j, lag, float(curve[j])
|
||
|
||
|
||
def pad_slice(arr: np.ndarray, a: int, b: int) -> np.ndarray:
|
||
P = arr.shape[0]
|
||
out = np.zeros((P, max(0, b - a)), dtype=arr.dtype)
|
||
s0, e0 = max(0, a), min(arr.shape[1], b)
|
||
if e0 > s0:
|
||
out[:, s0 - a:e0 - a] = arr[:, s0:e0]
|
||
return out
|
||
|
||
|
||
def chunk_fit(sa: np.ndarray, sb: np.ndarray, agg: str, phat: bool,
|
||
lag0_units: float, n_chunks: int = 6, margin_units: int = 48) -> dict | None:
|
||
"""Per-chunk lag estimates + robust linear fit: lag(t) = intercept + slope*t.
|
||
slope ~ 0 confirms a pure constant offset (no speed mismatch)."""
|
||
Ta = sa.shape[1]
|
||
L0 = int(round(lag0_units))
|
||
xs, ys = [], []
|
||
edges = np.linspace(0, Ta, n_chunks + 1).astype(int)
|
||
for i in range(n_chunks):
|
||
t0, t1 = int(edges[i]), int(edges[i + 1])
|
||
if t1 - t0 < 4:
|
||
continue
|
||
wa = sa[:, t0:t1]
|
||
bstart = t0 - L0 - margin_units
|
||
wb = pad_slice(sb, bstart, t1 - L0 + margin_units)
|
||
if wb.shape[1] < 8:
|
||
continue
|
||
curve, lags, _ = xcorr_pix(wa, wb, agg=agg, phat=phat)
|
||
sel = np.flatnonzero(np.abs(lags) <= margin_units + 1)
|
||
if len(sel) == 0:
|
||
continue
|
||
j = int(sel[int(np.argmax(curve[sel]))])
|
||
L_units = float(lags[j]) + t0 - bstart
|
||
xs.append(0.5 * (t0 + t1))
|
||
ys.append(L_units)
|
||
if len(xs) < 3:
|
||
return None
|
||
x = np.asarray(xs, dtype=np.float64)
|
||
y = np.asarray(ys, dtype=np.float64)
|
||
slopes = []
|
||
for i in range(len(x)):
|
||
for j in range(i + 1, len(x)):
|
||
dx = x[j] - x[i]
|
||
if dx > 0:
|
||
slopes.append((y[j] - y[i]) / dx)
|
||
slope = float(np.median(slopes)) if slopes else 0.0
|
||
intercept = float(np.median(y - slope * x))
|
||
k = np.array([slope, intercept], dtype=np.float64)
|
||
res = y - (slope * x + intercept)
|
||
return {
|
||
"intercept_units": float(k[1]),
|
||
"slope_units_per_unit": float(k[0]),
|
||
"resid_units": float(np.sqrt(np.mean(res ** 2))),
|
||
"n_chunks": len(xs),
|
||
"chunks": [{"t_units": float(xm), "lag_units": float(yv)} for xm, yv in zip(x, y)],
|
||
}
|
||
|
||
|
||
def zone_energy(frames: np.ndarray) -> np.ndarray:
|
||
z = block_avg(frames, 3, 3)
|
||
return np.abs(np.diff(z, axis=0)).mean(axis=(1, 2)).astype(np.float32)
|
||
|
||
|
||
# ─────────────────────────── analysis ───────────────────────────
|
||
|
||
def run_analysis(args, cfg: dict) -> SimpleNamespace:
|
||
t0 = time.time()
|
||
ex = cfg.get("extract", {})
|
||
vc = cfg.get("video", {})
|
||
|
||
a_path, b_path, fps_default = find_videos(cfg)
|
||
if args.a:
|
||
a_path = args.a if os.path.isabs(args.a) else os.path.join(ROOT, args.a)
|
||
if args.b:
|
||
b_path = args.b if os.path.isabs(args.b) else os.path.join(ROOT, args.b)
|
||
|
||
width = args.width or int(vc.get("decode_width", 192))
|
||
grid_s = args.grid or str(vc.get("grid", "32x18"))
|
||
try:
|
||
gw, gh = (int(v) for v in grid_s.lower().split("x"))
|
||
except Exception:
|
||
sys.exit(f"ERROR: bad --grid {grid_s!r} (want WxH like 32x18)")
|
||
use_bars = not args.no_bars and bool(vc.get("bars", True))
|
||
max_lag = args.max_lag if args.max_lag is not None else float(vc.get("max_lag", 0.0))
|
||
|
||
print(f"[video] A: {a_path}")
|
||
print(f"[video] B: {b_path}")
|
||
print(f"[video] decode width={width} grid={gw}x{gh} bars={use_bars} "
|
||
f"max_lag={'full' if max_lag <= 0 else f'{max_lag}s'}")
|
||
|
||
ca, fps_a = cached_decode(a_path, width, fps_default)
|
||
cb, fps_b = cached_decode(b_path, width, fps_default)
|
||
if abs(fps_a - fps_b) > 0.02:
|
||
print(f"WARNING: clip fps differ A={fps_a:.4f} B={fps_b:.4f} — "
|
||
f"results assume both CFR-normalized; use chunk fit to check drift")
|
||
fps = fps_a
|
||
print(f"[video] decoded: A {ca.shape} @ {fps_a:.4f} fps, "
|
||
f"B {cb.shape} @ {fps_b:.4f} fps ({time.time()-t0:.1f}s incl. cache)")
|
||
|
||
boxes = {}
|
||
for tag, clip in (("a", "A"), ("b", "B")):
|
||
box = content_box(ca if tag == "a" else cb) if use_bars else None
|
||
boxes[tag] = box
|
||
if use_bars:
|
||
print(f"[bars] {clip}: {'crop rows/cols ' + str(box) if box else 'no static bars detected'}")
|
||
if use_bars:
|
||
if boxes["a"]:
|
||
r0, r1, c0, c1 = boxes["a"]
|
||
ca = ca[:, r0:r1, c0:c1]
|
||
if boxes["b"]:
|
||
r0, r1, c0, c1 = boxes["b"]
|
||
cb = cb[:, r0:r1, c0:c1]
|
||
|
||
specs = build_method_signals(ca, cb, gh, gw)
|
||
|
||
results = {}
|
||
t1 = time.time()
|
||
for name in METHOD_NAMES:
|
||
sp = specs[name]
|
||
curve, lags, cp = xcorr_pix(sp["sa"], sp["sb"], agg=sp["agg"], phat=sp["phat"])
|
||
j, lag_u, pk = find_peak(curve, lags, int(round(max_lag * fps * sp["up"])))
|
||
results[name] = {
|
||
"curve": curve, "lags": lags, "cp": cp,
|
||
"peak_index": j,
|
||
"lag_units": lag_u,
|
||
"lag_frames": lag_u / sp["up"],
|
||
"lag_sec": lag_u / sp["up"] / fps,
|
||
"peak": pk,
|
||
"up": sp["up"],
|
||
"n_pix": int(sp["sa"].shape[0]),
|
||
"fit": None,
|
||
}
|
||
print(f"[video] xcorr done for {len(results)} methods in {time.time()-t1:.2f}s "
|
||
f"({int(specs['M3-pix-diff-sum']['sa'].shape[0])} valid pixels)")
|
||
|
||
return SimpleNamespace(
|
||
a_path=a_path, b_path=b_path, fps=fps, fps_a=fps_a, fps_b=fps_b,
|
||
width=width, gh=gh, gw=gw, bars=use_bars, boxes=boxes, max_lag=max_lag,
|
||
ca=ca, cb=cb, specs=specs, results=results,
|
||
audio_lag=args.audio_lag,
|
||
)
|
||
|
||
|
||
# ─────────────────────────── reporting ───────────────────────────
|
||
|
||
def report_table(ctx: SimpleNamespace, methods: list[str] | None = None,
|
||
run_fits: bool = False) -> list[dict]:
|
||
methods = methods or METHOD_NAMES
|
||
rows = []
|
||
lags = [ctx.results[m]["lag_frames"] for m in methods]
|
||
med = float(np.median(lags)) if lags else 0.0
|
||
for m in methods:
|
||
r = ctx.results[m]
|
||
row = {
|
||
"method": m,
|
||
"lag_frames": round(r["lag_frames"], 3),
|
||
"lag_sec": round(r["lag_sec"], 6),
|
||
"peak": round(r["peak"], 4),
|
||
"agree_med_frames": round(r["lag_frames"] - med, 3),
|
||
}
|
||
if run_fits:
|
||
sp = ctx.specs[m]
|
||
up = ctx.results[m]["up"]
|
||
margin = int(round(2.0 * ctx.fps * up))
|
||
n_chunks = int(np.clip(round(sp["sa"].shape[1] / max(1.0, 4.0 * ctx.fps * up)), 3, 8))
|
||
fit = chunk_fit(sp["sa"], sp["sb"], sp["agg"], sp["phat"], r["lag_units"],
|
||
n_chunks=n_chunks, margin_units=margin)
|
||
r["fit"] = fit
|
||
if fit:
|
||
up = ctx.results[m]["up"]
|
||
resid_fr = fit["resid_units"] / up
|
||
row["fit_intercept_sec"] = round(fit["intercept_units"] / (ctx.fps * up), 6)
|
||
row["fit_slope_ppm"] = round(fit["slope_units_per_unit"] * 1e6, 2)
|
||
row["fit_resid_frames"] = round(resid_fr, 2)
|
||
row["fit_ok"] = bool(resid_fr <= 2.0)
|
||
rows.append(row)
|
||
return rows
|
||
|
||
|
||
def print_report(ctx: SimpleNamespace):
|
||
rows = report_table(ctx, run_fits=True)
|
||
hdr = (f"{'method':<20}{'lag_frames':>12}{'lag_sec':>12}{'peak':>10}"
|
||
f"{'Δmedian(fr)':>13}{'fit_int(s)':>12}{'slope(ppm)':>12}{'fit_res(fr)':>12}")
|
||
print("\n" + hdr)
|
||
print("-" * len(hdr))
|
||
for row in rows:
|
||
resid = row.get('fit_resid_frames', float('nan'))
|
||
mark = "" if row.get("fit_ok", True) else "*"
|
||
print(f"{row['method']:<20}{row['lag_frames']:>12}{row['lag_sec']:>12}"
|
||
f"{row['peak']:>10}{row['agree_med_frames']:>13}"
|
||
f"{row.get('fit_intercept_sec', float('nan')):>12}"
|
||
f"{row.get('fit_slope_ppm', float('nan')):>12}"
|
||
f"{resid:>11}{mark}")
|
||
if ctx.audio_lag is not None:
|
||
print(f"\n[audio reference] lag_sec = {ctx.audio_lag:.6f} "
|
||
f"(= {ctx.audio_lag * ctx.fps:.2f} frames)")
|
||
for row in rows:
|
||
d = row["lag_sec"] - ctx.audio_lag
|
||
print(f" {row['method']:<20} Δaudio = {d:+.6f}s ({d * ctx.fps:+.2f} fr)")
|
||
|
||
|
||
def export_json(ctx: SimpleNamespace, path: str):
|
||
data = {
|
||
"file_a": ctx.a_path,
|
||
"file_b": ctx.b_path,
|
||
"fps": ctx.fps,
|
||
"fps_a": ctx.fps_a,
|
||
"fps_b": ctx.fps_b,
|
||
"grid": f"{ctx.gw}x{ctx.gh}",
|
||
"bars_crop_a": list(ctx.boxes["a"]) if ctx.boxes["a"] else None,
|
||
"bars_crop_b": list(ctx.boxes["b"]) if ctx.boxes["b"] else None,
|
||
"max_lag_sec": ctx.max_lag,
|
||
"audio_lag_sec": ctx.audio_lag,
|
||
"convention": "lag sign identical to sync_audio.correlate() -> apply_sync --lag",
|
||
"methods": report_table(ctx, run_fits=True),
|
||
}
|
||
with open(path, "w") as f:
|
||
json.dump(data, f, indent=2)
|
||
print(f"exported → {path}")
|
||
|
||
|
||
# ─────────────────────────── GUI ───────────────────────────
|
||
|
||
def apply_dark_theme(app):
|
||
from PyQt6 import QtGui
|
||
app.setStyle("Fusion")
|
||
dark = QtGui.QPalette()
|
||
for role, col in [
|
||
("Window", (45, 45, 45)), ("WindowText", (220, 220, 220)),
|
||
("Base", (30, 30, 30)), ("AlternateBase", (45, 45, 45)),
|
||
("ToolTipBase", (45, 45, 45)), ("ToolTipText", (220, 220, 220)),
|
||
("Text", (220, 220, 220)), ("Button", (45, 45, 45)),
|
||
("ButtonText", (220, 220, 220)), ("BrightText", (255, 50, 50)),
|
||
("Link", (100, 180, 255)), ("Highlight", (60, 120, 200)),
|
||
("HighlightedText", (255, 255, 255)),
|
||
]:
|
||
dark.setColor(getattr(QtGui.QPalette.ColorRole, role), QtGui.QColor(*col))
|
||
app.setPalette(dark)
|
||
|
||
|
||
class VideoSyncWindow:
|
||
def __init__(self, ctx: SimpleNamespace, args=None, cfg: dict | None = None):
|
||
from PyQt6 import QtWidgets, QtCore, QtGui
|
||
import pyqtgraph as pg
|
||
self.Qt = (QtWidgets, QtCore, QtGui)
|
||
self.pg = pg
|
||
self.args = args
|
||
self.cfg = cfg if cfg is not None else {}
|
||
|
||
self.ctx = ctx
|
||
self.enabled = {m: True for m in METHOD_NAMES}
|
||
self.active = "M3-pix-diff-sum"
|
||
self.lag_frames = ctx.results[self.active]["lag_frames"]
|
||
self._updating = False
|
||
self.Ta, self.Tb = ctx.ca.shape[0], ctx.cb.shape[0]
|
||
self.scrub_t = min(int(self.Ta * 0.4), self.Ta - 1)
|
||
|
||
QtWidgets, QtCore, QtGui = self.Qt
|
||
app = QtWidgets.QApplication.instance() or QtWidgets.QApplication(sys.argv)
|
||
apply_dark_theme(app)
|
||
pg.setConfigOptions(antialias=True, background="#1e1e1e", foreground="#ddd")
|
||
|
||
w = QtWidgets.QMainWindow()
|
||
self.w = w
|
||
w.setWindowTitle(f"movie_sync — video sync test — {ctx.gw}x{ctx.gh} grid")
|
||
w.resize(1500, 1000)
|
||
central = QtWidgets.QWidget()
|
||
w.setCentralWidget(central)
|
||
lay = QtWidgets.QVBoxLayout(central)
|
||
|
||
info = QtWidgets.QLabel(
|
||
f"A: {os.path.basename(ctx.a_path)} ({self.Ta} fr) "
|
||
f"B: {os.path.basename(ctx.b_path)} ({self.Tb} fr) fps={ctx.fps:.4f} "
|
||
f"grid={ctx.gw}x{ctx.gh} bars={'on' if ctx.bars else 'off'} "
|
||
f"lag convention = sync_audio.correlate()")
|
||
info.setWordWrap(True)
|
||
lay.addWidget(info)
|
||
|
||
self.gw_widget = pg.GraphicsLayoutWidget()
|
||
lay.addWidget(self.gw_widget, stretch=1)
|
||
|
||
# row 0: xcorr overlay
|
||
self.p_x = self.gw_widget.addPlot(row=0, col=0, title="Cross-correlation (peak-normalized, all methods)")
|
||
self.p_x.showGrid(x=True, y=True)
|
||
self.p_x.setLabel("bottom", "Lag", "s")
|
||
self.p_x.addLegend(offset=(60, 10))
|
||
self.p_x.setMouseEnabled(x=True, y=False)
|
||
self.curves = {}
|
||
for m in METHOD_NAMES:
|
||
self.curves[m] = self.p_x.plot(pen=pg.mkPen(METHOD_COLORS[m], width=1), name=m)
|
||
self.peak_line = pg.InfiniteLine(pos=0, angle=90,
|
||
pen=pg.mkPen("#06d6a0", width=2, style=QtCore.Qt.PenStyle.DashLine))
|
||
self.p_x.addItem(self.peak_line)
|
||
if ctx.audio_lag is not None:
|
||
ref = pg.InfiniteLine(pos=ctx.audio_lag, angle=90,
|
||
pen=pg.mkPen("#ff006e", width=2, style=QtCore.Qt.PenStyle.DotLine))
|
||
self.p_x.addItem(ref)
|
||
self.p_x.plot([], [], pen=pg.mkPen("#ff006e"), name=f"audio lag {ctx.audio_lag:.4f}s")
|
||
|
||
# row 1: heatmap
|
||
self.p_h = self.gw_widget.addPlot(row=1, col=0, title="Per-pixel xcorr map (active method): lag × pixel")
|
||
self.p_h.setLabel("bottom", "Lag", "s")
|
||
self.p_h.setLabel("left", "pixel")
|
||
self.img = pg.ImageItem(axisOrder="col-major")
|
||
self.p_h.addItem(self.img)
|
||
self.heat_line = pg.InfiniteLine(pos=0, angle=90,
|
||
pen=pg.mkPen("#06d6a0", width=1, style=QtCore.Qt.PenStyle.DashLine))
|
||
self.p_h.addItem(self.heat_line)
|
||
self.p_h.setMouseEnabled(x=True, y=False)
|
||
|
||
# row 2: 3x3 zone motion energy
|
||
self.p_z = self.gw_widget.addPlot(row=2, col=0, title="3×3 zone motion energy: A vs B(lag)")
|
||
self.p_z.showGrid(x=True, y=True)
|
||
self.p_z.setLabel("bottom", "Time", "s")
|
||
self.cz_a = self.p_z.plot(pen=pg.mkPen("#4cc9f0", width=1), name="A")
|
||
self.cz_b = self.p_z.plot(pen=pg.mkPen("#f72585", width=1,
|
||
style=QtCore.Qt.PenStyle.DashLine), name="B(lag)")
|
||
self.p_z.addLegend(offset=(60, 10))
|
||
self.p_z.setMouseEnabled(x=True, y=False)
|
||
|
||
# row 3: scrub view
|
||
self.p_s = self.gw_widget.addPlot(row=3, col=0, title="Frame scrub: A(t) | B(t at lag)")
|
||
self.p_s.hideAxis("left")
|
||
self.p_s.hideAxis("bottom")
|
||
self.imA = pg.ImageItem(axisOrder="row-major")
|
||
self.imB = pg.ImageItem(axisOrder="row-major")
|
||
self.p_s.addItem(self.imA)
|
||
self.p_s.addItem(self.imB)
|
||
self.lblA = pg.TextItem("A", color="#4cc9f0")
|
||
self.lblB = pg.TextItem("B", color="#f72585")
|
||
self.p_s.addItem(self.lblA)
|
||
self.p_s.addItem(self.lblB)
|
||
|
||
# method checkboxes + settings
|
||
row1 = QtWidgets.QHBoxLayout()
|
||
lay.addLayout(row1)
|
||
row1.addWidget(QtWidgets.QLabel("Methods:"))
|
||
self.cbs = {}
|
||
for m in METHOD_NAMES:
|
||
cb = QtWidgets.QCheckBox(m.split("-", 1)[0])
|
||
cb.setToolTip(m)
|
||
cb.setChecked(self.enabled[m])
|
||
cb.setStyleSheet(f"color: {METHOD_COLORS[m]};")
|
||
cb.toggled.connect(lambda on, mm=m: self.on_toggle(mm, on))
|
||
row1.addWidget(cb)
|
||
self.cbs[m] = cb
|
||
row1.addSpacing(16)
|
||
row1.addWidget(QtWidgets.QLabel("Grid:"))
|
||
self.combo_grid = QtWidgets.QComboBox()
|
||
for g in ("16x9", "24x14", "32x18", "48x27", "64x36"):
|
||
self.combo_grid.addItem(g)
|
||
self.combo_grid.setCurrentText(f"{ctx.gw}x{ctx.gh}")
|
||
self.combo_grid.currentTextChanged.connect(self.on_settings)
|
||
row1.addWidget(self.combo_grid)
|
||
self.chk_bars = QtWidgets.QCheckBox("bars-crop")
|
||
self.chk_bars.setChecked(ctx.bars)
|
||
self.chk_bars.toggled.connect(self.on_settings)
|
||
row1.addWidget(self.chk_bars)
|
||
row1.addWidget(QtWidgets.QLabel("max|lag|:"))
|
||
self.spin_maxlag = QtWidgets.QDoubleSpinBox()
|
||
self.spin_maxlag.setRange(0, 120)
|
||
self.spin_maxlag.setDecimals(1)
|
||
self.spin_maxlag.setSingleStep(1.0)
|
||
self.spin_maxlag.setValue(ctx.max_lag)
|
||
self.spin_maxlag.setSuffix(" s")
|
||
self.spin_maxlag.setToolTip("0 = full range")
|
||
self.spin_maxlag.setKeyboardTracking(False)
|
||
self.spin_maxlag.valueChanged.connect(self.on_maxlag)
|
||
row1.addWidget(self.spin_maxlag)
|
||
row1.addStretch(1)
|
||
|
||
# lag controls
|
||
row2 = QtWidgets.QHBoxLayout()
|
||
lay.addLayout(row2)
|
||
row2.addWidget(QtWidgets.QLabel("Lag (fr):"))
|
||
cap = int(min(self.Ta, self.Tb) * 0.45)
|
||
self.slider = QtWidgets.QSlider(QtCore.Qt.Orientation.Horizontal)
|
||
self.slider.setRange(-cap, cap)
|
||
self.slider.setValue(int(round(self.lag_frames)))
|
||
self.slider.setTickInterval(10)
|
||
row2.addWidget(self.slider, stretch=1)
|
||
self.spin_fr = QtWidgets.QDoubleSpinBox()
|
||
self.spin_fr.setRange(-cap, cap)
|
||
self.spin_fr.setDecimals(3)
|
||
self.spin_fr.setSingleStep(0.1)
|
||
self.spin_fr.setValue(self.lag_frames)
|
||
self.spin_fr.setSuffix(" fr")
|
||
self.spin_fr.setKeyboardTracking(False)
|
||
row2.addWidget(self.spin_fr)
|
||
self.spin_sec = QtWidgets.QDoubleSpinBox()
|
||
self.spin_sec.setRange(-60, 60)
|
||
self.spin_sec.setDecimals(6)
|
||
self.spin_sec.setSingleStep(1.0 / ctx.fps)
|
||
self.spin_sec.setValue(self.lag_frames / ctx.fps)
|
||
self.spin_sec.setSuffix(" s")
|
||
self.spin_sec.setKeyboardTracking(False)
|
||
row2.addWidget(self.spin_sec)
|
||
btn_auto = QtWidgets.QPushButton("Auto peak")
|
||
btn_fit = QtWidgets.QPushButton("Chunk fit")
|
||
btn_export = QtWidgets.QPushButton("Export data")
|
||
row2.addWidget(btn_auto)
|
||
row2.addWidget(btn_fit)
|
||
row2.addWidget(btn_export)
|
||
self.lbl_x = QtWidgets.QLabel()
|
||
self.lbl_x.setStyleSheet("color:#ffd166;")
|
||
row2.addWidget(self.lbl_x)
|
||
|
||
# results table
|
||
self.table = QtWidgets.QTableWidget(len(METHOD_NAMES), 5)
|
||
self.table.setHorizontalHeaderLabels(["Method", "Lag (fr)", "Lag (s)", "Peak", "Δmedian (fr)"])
|
||
self.table.verticalHeader().setVisible(False)
|
||
self.table.setMaximumHeight(180)
|
||
self.table.setSelectionBehavior(QtWidgets.QAbstractItemView.SelectionBehavior.SelectRows)
|
||
lay.addWidget(self.table)
|
||
|
||
# scrub slider + status
|
||
row3 = QtWidgets.QHBoxLayout()
|
||
lay.addLayout(row3)
|
||
row3.addWidget(QtWidgets.QLabel("Scrub:"))
|
||
self.scrub = QtWidgets.QSlider(QtCore.Qt.Orientation.Horizontal)
|
||
self.scrub.setRange(0, max(0, self.Ta - 1))
|
||
self.scrub.setValue(self.scrub_t)
|
||
row3.addWidget(self.scrub, stretch=1)
|
||
self.lbl_mad = QtWidgets.QLabel("")
|
||
self.lbl_mad.setStyleSheet("color:#aaa;")
|
||
row3.addWidget(self.lbl_mad)
|
||
self.status = QtWidgets.QLabel("ready")
|
||
self.status.setStyleSheet("color:#aaa; font-size: 11px;")
|
||
lay.addWidget(self.status)
|
||
|
||
# signals
|
||
self.slider.valueChanged.connect(lambda v: self._set_lag(float(v), "slider"))
|
||
self.spin_fr.valueChanged.connect(lambda v: self._set_lag(float(v), "fr"))
|
||
self.spin_sec.valueChanged.connect(lambda v: self._set_lag(float(v) * ctx.fps, "sec"))
|
||
self.scrub.valueChanged.connect(self.on_scrub)
|
||
btn_auto.clicked.connect(self.on_auto)
|
||
btn_fit.clicked.connect(self.on_fit)
|
||
btn_export.clicked.connect(self.on_export)
|
||
self.table.cellPressed.connect(self.on_table_pick)
|
||
|
||
self.plot_all()
|
||
self._set_lag(self.lag_frames, "init")
|
||
w.show()
|
||
|
||
# ── compute/plot ──
|
||
|
||
def plot_all(self):
|
||
ctx = self.ctx
|
||
for m in METHOD_NAMES:
|
||
r = ctx.results[m]
|
||
if not self.enabled[m]:
|
||
self.curves[m].setData([], [])
|
||
continue
|
||
lag_s = r["lags"] / (r["up"] * ctx.fps)
|
||
c = r["curve"]
|
||
k = max(1, len(c) // 4000)
|
||
self.curves[m].setData(lag_s[::k], c[::k] / max(1e-12, float(np.max(c))))
|
||
self.update_heat()
|
||
self.update_table()
|
||
self.update_zones()
|
||
self.update_scrub()
|
||
span = max(2.0, abs(self.lag_frames) / ctx.fps * 2 + 2)
|
||
self.p_x.setXRange(self.lag_sec - span, self.lag_sec + span, padding=0)
|
||
|
||
def update_heat(self):
|
||
ctx = self.ctx
|
||
r = ctx.results[self.active]
|
||
cp = r["cp"]
|
||
k = max(1, cp.shape[1] // 1200)
|
||
img = np.ascontiguousarray(cp[:, ::k].T) # (lags', pixels) col-major x=lag
|
||
v0 = float(np.percentile(img, 1))
|
||
v1 = float(max(np.max(img), v0 + 1e-6))
|
||
self.img.setImage(img, levels=(v0, v1), autoLevels=False)
|
||
lag_s = r["lags"][::k] / (r["up"] * ctx.fps)
|
||
self.p_h.setXRange(self.lag_sec - 2, self.lag_sec + 2, padding=0)
|
||
self.p_h.setYRange(0, img.shape[1], padding=0)
|
||
self.img.setRect(lag_s[0], 0, lag_s[-1] - lag_s[0], img.shape[1])
|
||
self.p_h.setTitle(f"Per-pixel xcorr map ({self.active}): lag × pixel")
|
||
|
||
def update_table(self):
|
||
ctx = self.ctx
|
||
rows = report_table(ctx)
|
||
med = float(np.median([r["lag_frames"] for r in rows]))
|
||
for i, row in enumerate(rows):
|
||
m = row["method"]
|
||
items = [row["method"], f"{row['lag_frames']:.3f}", f"{row['lag_sec']:.6f}",
|
||
f"{row['peak']:.4f}", f"{row['agree_med_frames']:+.2f}"]
|
||
for j, txt in enumerate(items):
|
||
it = self.table.item(i, j) or self.Qt[0].QTableWidgetItem()
|
||
it.setText(txt)
|
||
if m == self.active:
|
||
it.setBackground(self.Qt[2].QColor(60, 120, 200).darker(150))
|
||
if not self.enabled[m]:
|
||
it.setForeground(self.Qt[2].QColor(110, 110, 110))
|
||
self.table.setItem(i, j, it)
|
||
|
||
def update_zones(self):
|
||
ctx = self.ctx
|
||
if getattr(self, "_za", None) is None:
|
||
self._za = zone_energy(ctx.ca)
|
||
self._zb = zone_energy(ctx.cb)
|
||
ta = np.arange(len(self._za)) / ctx.fps
|
||
tb = (np.arange(len(self._zb)) - self.lag_frames) / ctx.fps
|
||
self.cz_a.setData(ta, self._za)
|
||
self.cz_b.setData(tb, self._zb)
|
||
|
||
def update_scrub(self):
|
||
ctx = self.ctx
|
||
t = self.scrub_t
|
||
tB_f = t - self.lag_frames
|
||
iB0 = int(math.floor(tB_f))
|
||
fr = tB_f - iB0
|
||
wA = ctx.ca.shape[2]
|
||
self.imA.setImage(ctx.ca[t])
|
||
if 0 <= iB0 and self.Tb >= 2:
|
||
if iB0 + 1 < self.Tb:
|
||
b = (1.0 - fr) * ctx.cb[iB0] + fr * ctx.cb[iB0 + 1]
|
||
elif iB0 < self.Tb:
|
||
b = ctx.cb[iB0]
|
||
else:
|
||
b = np.zeros_like(ctx.cb[0])
|
||
else:
|
||
b = np.zeros_like(ctx.cb[0])
|
||
self.imB.setImage(b)
|
||
self.imB.setPos(wA + 8, 0)
|
||
self.lblA.setPos(0, -8)
|
||
self.lblB.setPos(wA + 8, -8)
|
||
self.lblA.setText(f"A t={t/ctx.fps:.3f}s (fr {t})")
|
||
self.lblB.setText(f"B t={tB_f/ctx.fps:.3f}s (fr {tB_f:.2f})")
|
||
from_ga = block_avg(ctx.ca[t:t + 1], ctx.gh, ctx.gw)[0]
|
||
from_gb = block_avg(b[np.newaxis], ctx.gh, ctx.gw)[0]
|
||
mad = float(np.abs(from_ga - from_gb).mean())
|
||
self.lbl_mad.setText(
|
||
f"pair mean|Δ| = {mad * 100:.2f}% of full range (active method: {self.active})")
|
||
|
||
# ── state ──
|
||
|
||
@property
|
||
def lag_sec(self):
|
||
return self.lag_frames / self.ctx.fps
|
||
|
||
def _set_lag(self, lag_frames: float, source: str = ""):
|
||
if self._updating:
|
||
return
|
||
self._updating = True
|
||
self.lag_frames = float(lag_frames)
|
||
self.slider.blockSignals(True)
|
||
self.slider.setValue(int(round(lag_frames)))
|
||
self.slider.blockSignals(False)
|
||
self.spin_fr.blockSignals(True)
|
||
self.spin_fr.setValue(lag_frames)
|
||
self.spin_fr.blockSignals(False)
|
||
self.spin_sec.blockSignals(True)
|
||
self.spin_sec.setValue(lag_frames / self.ctx.fps)
|
||
self.spin_sec.blockSignals(False)
|
||
self.peak_line.setPos(self.lag_sec)
|
||
self.heat_line.setPos(self.lag_sec)
|
||
r = self.ctx.results[self.active]
|
||
c = r["curve"]
|
||
lag_units = lag_frames * r["up"]
|
||
idx = int(round(lag_units - r["lags"][0]))
|
||
if 0 <= idx < len(c):
|
||
v = float(c[idx])
|
||
pk = float(np.max(c))
|
||
self.lbl_x.setText(f"xcorr@lag: {v:.4f} / peak {pk:.4f} = {v / pk * 100:.1f}%")
|
||
self.update_zones()
|
||
self.update_scrub()
|
||
self._updating = False
|
||
|
||
# ── slots ──
|
||
|
||
def on_toggle(self, m: str, on: bool):
|
||
self.enabled[m] = on
|
||
if on and not any(self.enabled.values()):
|
||
self.enabled[m] = True
|
||
self.cbs[m].setChecked(True)
|
||
return
|
||
self.plot_all()
|
||
|
||
def on_table_pick(self, row: int, col: int):
|
||
m = METHOD_NAMES[row]
|
||
self.active = m
|
||
self.update_heat()
|
||
self.update_table()
|
||
self._set_lag(self.lag_frames, "table")
|
||
|
||
def on_settings(self, *args):
|
||
self.reload("settings changed")
|
||
|
||
def on_maxlag(self, v: float):
|
||
self.ctx.max_lag = float(v)
|
||
for name in METHOD_NAMES:
|
||
r = self.ctx.results[name]
|
||
j, lag_u, pk = find_peak(r["curve"], r["lags"], int(round(self.ctx.max_lag * self.ctx.fps * r["up"])))
|
||
r.update(peak_index=j, lag_units=lag_u, lag_frames=lag_u / r["up"], lag_sec=lag_u / r["up"] / self.ctx.fps, peak=pk)
|
||
self.plot_all()
|
||
self._set_lag(self.ctx.results[self.active]["lag_frames"], "maxlag")
|
||
|
||
def reload(self, why: str = ""):
|
||
"""Re-run analysis after grid/bars change (decode comes from cache)."""
|
||
if self.args is None:
|
||
return
|
||
self.args.grid = self.combo_grid.currentText()
|
||
self.args.no_bars = not self.chk_bars.isChecked()
|
||
self.args.max_lag = float(self.spin_maxlag.value())
|
||
try:
|
||
self.ctx = run_analysis(self.args, self.cfg)
|
||
except SystemExit:
|
||
self.status.setText(f"reload failed ({why})")
|
||
return
|
||
self.Ta, self.Tb = self.ctx.ca.shape[0], self.ctx.cb.shape[0]
|
||
self._za = None
|
||
cap = int(min(self.Ta, self.Tb) * 0.45)
|
||
self.slider.blockSignals(True)
|
||
self.slider.setRange(-cap, cap)
|
||
self.slider.blockSignals(False)
|
||
self.spin_fr.blockSignals(True)
|
||
self.spin_fr.setRange(-cap, cap)
|
||
self.spin_fr.blockSignals(False)
|
||
self.scrub.blockSignals(True)
|
||
self.scrub.setRange(0, max(0, self.Ta - 1))
|
||
self.scrub_t = min(self.scrub_t, self.Ta - 1)
|
||
self.scrub.setValue(self.scrub_t)
|
||
self.scrub.blockSignals(False)
|
||
self.plot_all()
|
||
self._set_lag(self.ctx.results[self.active]["lag_frames"], f"reload:{why}")
|
||
self.status.setText(f"recomputed ({why})")
|
||
|
||
def on_auto(self):
|
||
r = self.ctx.results[self.active]
|
||
self._set_lag(r["lag_frames"], "auto")
|
||
self.status.setText(f"auto peak {self.active}: {r['lag_frames']:.3f} fr "
|
||
f"= {r['lag_sec']:.6f}s (peak {r['peak']:.4f})")
|
||
|
||
def on_fit(self):
|
||
ctx = self.ctx
|
||
sp = ctx.specs[self.active]
|
||
r = ctx.results[self.active]
|
||
margin = int(round(2.0 * ctx.fps * r["up"]))
|
||
n_chunks = int(np.clip(round(sp["sa"].shape[1] / max(1.0, 4.0 * ctx.fps * r["up"])), 3, 8))
|
||
fit = chunk_fit(sp["sa"], sp["sb"], sp["agg"], sp["phat"], r["lag_units"],
|
||
n_chunks=n_chunks, margin_units=margin)
|
||
r["fit"] = fit
|
||
if not fit:
|
||
self.status.setText("chunk fit: not enough chunks")
|
||
return
|
||
inter = fit["intercept_units"] / (ctx.fps * r["up"])
|
||
slope = fit["slope_units_per_unit"]
|
||
resid_fr = fit["resid_units"] / r["up"]
|
||
flag = "" if resid_fr <= 2.0 else " [high residual — slope unreliable]"
|
||
self.status.setText(
|
||
f"chunk fit [{self.active}]: offset {inter:+.6f}s slope {slope * 1e6:+.1f} ppm "
|
||
f"(Δfps {slope * ctx.fps:+.6f}) resid ±{resid_fr:.2f}fr "
|
||
f"chunks={fit['n_chunks']}{flag}")
|
||
|
||
def on_export(self):
|
||
path, _ = self.Qt[0].QFileDialog.getSaveFileName(
|
||
self.w, "Export video sync data", "video_sync_data.json", "JSON (*.json)")
|
||
if not path:
|
||
return
|
||
export_json(self.ctx, path)
|
||
|
||
def on_scrub(self, v: int):
|
||
self.scrub_t = int(v)
|
||
self.update_scrub()
|
||
|
||
|
||
# ─────────────────────────── main ───────────────────────────
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="video cross-correlation lab (extracted files only)")
|
||
ap.add_argument("--config", default="config.toml")
|
||
ap.add_argument("--a", default=None, help="override A (path inside extracted/)")
|
||
ap.add_argument("--b", default=None, help="override B (path inside extracted/)")
|
||
ap.add_argument("--width", type=int, default=None, help="decode width px (default 192)")
|
||
ap.add_argument("--grid", default=None, help="correlation grid WxH (default config or 32x18)")
|
||
ap.add_argument("--max-lag", type=float, default=None, help="search half-window seconds (0 = full)")
|
||
ap.add_argument("--no-bars", action="store_true", help="disable letterbox auto-crop")
|
||
ap.add_argument("--audio-lag", type=float, default=None,
|
||
help="known audio lag in seconds (reference line / Δ column)")
|
||
ap.add_argument("--export", default=None, help="write JSON report to this path (implies --no-gui)")
|
||
ap.add_argument("--no-gui", action="store_true", help="headless: print table and exit")
|
||
args = ap.parse_args()
|
||
|
||
cfg_path = args.config if os.path.isabs(args.config) else os.path.join(ROOT, args.config)
|
||
cfg = load_config(cfg_path)
|
||
ctx = run_analysis(args, cfg)
|
||
|
||
if args.no_gui or args.export:
|
||
print_report(ctx)
|
||
if args.export:
|
||
path = args.export if os.path.isabs(args.export) else os.path.join(ROOT, args.export)
|
||
export_json(ctx, path)
|
||
return
|
||
|
||
try:
|
||
from PyQt6 import QtWidgets # noqa: F401
|
||
import pyqtgraph as pg # noqa: F401
|
||
except ImportError as e:
|
||
print(f"ERROR: pyqtgraph/PyQt6 not installed — run make install (or use --no-gui)", file=sys.stderr)
|
||
print(e, file=sys.stderr)
|
||
sys.exit(1)
|
||
import signal as _signal
|
||
app = QtWidgets.QApplication(sys.argv)
|
||
_signal.signal(_signal.SIGINT, _signal.SIG_DFL)
|
||
win = VideoSyncWindow(ctx, args, cfg)
|
||
sys.exit(app.exec())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|