diff --git a/backend/festival4d/audio_sync.py b/backend/festival4d/audio_sync.py index 8e71507..0a0a8a6 100644 --- a/backend/festival4d/audio_sync.py +++ b/backend/festival4d/audio_sync.py @@ -1,60 +1,325 @@ """GCC-PHAT pairwise audio offsets + global graph solve + drift (spec M1, lane A). -STUB — lane A fills the bodies; the public signatures below are frozen (``cli.py`` calls -``run_sync``). +Algorithm (spec M1): -Algorithm (spec M1, for lane A's reference): - **GCC-PHAT**, not raw cross-correlation: whiten the cross-spectrum - ``R = X * conj(Y); R /= |R| + eps; r = irfft(R)`` — robust to loud/clipped/reverberant - concert audio. numpy/scipy FFTs; librosa only for loading. + ``R = X·conj(Y); R /= |R| + eps; r = irfft(R)`` — robust to loud/clipped/reverberant + concert audio. numpy FFTs; librosa only for loading (in :func:`run_sync`). - All **pairwise** offsets, each scored by peak-to-second-peak ratio; then a global least-squares solve over the offset graph, weighted by confidence, rejecting pairs that - violate cycle consistency (``off_ab + off_bc - off_ac > 50 ms``). -- **Drift**: re-run GCC-PHAT on 10 s windows every 30 s along the overlap; fit a line; - store the slope as ``drift_ppm`` (store 0 if ``|drift| < 5 ppm``). -- Disconnected sync-graph components -> ``offset_ms = None`` (never a guess; spec pitfall #5). + violate cycle consistency (``|off_ab + off_bc − off_ac| > 50 ms``, generalized here as a + residual test against the fitted global solution). +- **Drift**: re-run GCC-PHAT on 10 s windows every 30 s along the overlap; fit a line to + ``delay(t)``; store the slope (dimensionless rate) as ``drift_ppm`` (store 0 if < 5 ppm). +- Disconnected sync-graph components → ``offset_ms = None`` (never a guess; spec pitfall #5). - Persist via ``db.update_video_sync`` and export ``config.SYNC_JSON``. + +Sign conventions (kept consistent with ``config``'s timebase): + + A video *v*'s audio sample at local time ``τ`` equals the master signal at global time + ``offset_s + τ`` (drift 0). So for two videos ``a``/``b``: + ``offset_a − offset_b = −gcc_phat(a, b)`` seconds, where :func:`gcc_phat` returns the + delay of its first argument relative to its second. Under drift ``d`` (= ``drift_ppm·1e−6``), + the windowed delay of ``v`` vs the reference is ``d·t − offset_s`` — a line whose slope is + the drift, recovered by :func:`estimate_drift`. """ from __future__ import annotations +import json import logging import numpy as np +from festival4d import config, db +from festival4d.ingest import audio_wav_path + log = logging.getLogger("festival4d.audio_sync") +_EPS = 1e-10 +# Edges weaker than this peak-to-second-peak ratio carry no reliable alignment +# (flat/ambiguous correlation) and are dropped before the global solve. +_MIN_EDGE_CONFIDENCE = 1.15 +# A pairwise offset whose residual against the fitted global solution exceeds this is a +# cycle-consistency violation (spec M1); it is rejected and the component is re-solved. +_CYCLE_TOLERANCE_MS = 50.0 +# Drift smaller than this magnitude is stored as exactly 0 (spec M1). +_DRIFT_DEADBAND_PPM = 5.0 + + +# --------------------------------------------------------------------------- +# Core correlation +# --------------------------------------------------------------------------- +def _peak_ratio(env: np.ndarray, peak: int, sr: int, guard_s: float = 0.004) -> float: + """Peak-to-second-peak ratio of a correlation envelope (the spec's edge score). + + The second peak is the largest value outside a small guard band around the main peak, + so the main lobe's own shoulders don't count as competition. + """ + main = float(env[peak]) + if main <= _EPS: + return 0.0 + guard = max(1, int(round(guard_s * sr))) + masked = env.copy() + masked[max(0, peak - guard):min(len(env), peak + guard + 1)] = 0.0 + second = float(masked.max()) if masked.size else 0.0 + if second <= _EPS: + return 1.0e6 + return main / second + def gcc_phat(sig: np.ndarray, ref: np.ndarray, sr: int, max_lag_s: float | None = None) -> tuple[float, float]: """GCC-PHAT time delay of ``sig`` relative to ``ref`` (spec M1, lane A). - Returns ``(offset_s, confidence)`` where confidence is the peak-to-second-peak ratio. + Returns ``(offset_s, confidence)`` where ``offset_s`` is positive when ``sig`` *lags* + ``ref`` (i.e. ``sig[n] ≈ ref[n − offset_s·sr]``), and confidence is the + peak-to-second-peak ratio. """ - raise NotImplementedError("lane A (M1): implement gcc_phat") + sig = np.asarray(sig, dtype=np.float64) + ref = np.asarray(ref, dtype=np.float64) + # Remove DC so the whitened correlation keys on structure, not a bias term. + sig = sig - sig.mean() + ref = ref - ref.mean() + + n = len(sig) + len(ref) # linear (zero-padded) correlation → no circular wrap + SIG = np.fft.rfft(sig, n) + REF = np.fft.rfft(ref, n) + R = SIG * np.conj(REF) + R /= np.abs(R) + _EPS # PHAT weighting: flatten magnitude, keep phase + cc = np.fft.irfft(R, n) # lag 0 at index 0; negative lags wrap to the tail + + max_lag = n // 2 if max_lag_s is None else int(round(max_lag_s * sr)) + max_lag = max(1, min(max_lag, n // 2)) + # Re-center to contiguous lags [−max_lag … +max_lag]. + cc = np.concatenate((cc[-max_lag:], cc[:max_lag + 1])) + env = np.abs(cc) + peak = int(np.argmax(env)) + # Parabolic sub-sample refinement of the peak location (3-point fit around the max), + # so precision isn't capped at one sample (62.5 µs at 16 kHz) and drift slopes stay clean. + delta = 0.0 + if 0 < peak < len(env) - 1: + a, b, c = env[peak - 1], env[peak], env[peak + 1] + denom = a - 2.0 * b + c + if abs(denom) > _EPS: + delta = float(np.clip(0.5 * (a - c) / denom, -0.5, 0.5)) + offset_s = ((peak - max_lag) + delta) / sr + return offset_s, _peak_ratio(env, peak, sr) +# --------------------------------------------------------------------------- +# Pairwise graph +# --------------------------------------------------------------------------- def pairwise_offsets(signals: dict[int, np.ndarray], sr: int) -> list[dict]: """All pairwise GCC-PHAT offsets with confidences (spec M1, lane A). ``signals`` maps ``video_id -> mono samples``. Returns a list of - ``{a, b, offset_s, confidence}`` edges. + ``{a, b, offset_s, confidence}`` edges, where ``offset_s`` is the *relative start + offset* ``offset_a − offset_b`` in seconds (see the module's sign conventions). """ - raise NotImplementedError("lane A (M1): implement pairwise_offsets") + ids = sorted(signals) + edges: list[dict] = [] + for i in range(len(ids)): + for j in range(i + 1, len(ids)): + a, b = ids[i], ids[j] + delay_s, conf = gcc_phat(signals[a], signals[b], sr) + edges.append({"a": a, "b": b, "offset_s": -delay_s, "confidence": conf}) + return edges + + +def _components(ids: list[int], edges: list[dict]) -> dict[int, list[int]]: + """Connected components of the offset graph (union-find over usable edges).""" + parent = {i: i for i in ids} + + def find(x: int) -> int: + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + + for e in edges: + parent[find(e["a"])] = find(e["b"]) + comps: dict[int, list[int]] = {} + for i in ids: + comps.setdefault(find(i), []).append(i) + return comps + + +def _connected(nodes: list[int], root: int, edges: list[dict]) -> bool: + """Whether every node is reachable from ``root`` over ``edges``.""" + adj: dict[int, list[int]] = {n: [] for n in nodes} + for e in edges: + adj[e["a"]].append(e["b"]) + adj[e["b"]].append(e["a"]) + seen = {root} + stack = [root] + while stack: + for y in adj[stack.pop()]: + if y not in seen: + seen.add(y) + stack.append(y) + return len(seen) == len(nodes) + + +def _weighted_lstsq(nodes: list[int], ref: int, edges: list[dict]) -> dict[int, float]: + """Confidence-weighted least-squares offsets (seconds) with ``ref`` pinned to 0. + + Each edge contributes the constraint ``offset_a − offset_b = offset_s``, weighted by + ``sqrt(confidence)``. Solved for all nodes except the pinned reference. + """ + unknown = [n for n in nodes if n != ref] + col = {n: k for k, n in enumerate(unknown)} + A = np.zeros((len(edges), len(unknown))) + b = np.zeros(len(edges)) + w = np.zeros(len(edges)) + for r, e in enumerate(edges): + if e["a"] in col: + A[r, col[e["a"]]] += 1.0 + if e["b"] in col: + A[r, col[e["b"]]] -= 1.0 + b[r] = e["offset_s"] # ref terms are 0, so they drop out of A + w[r] = np.sqrt(max(e["confidence"], _EPS)) + x, *_ = np.linalg.lstsq(A * w[:, None], b * w, rcond=None) + sol = {ref: 0.0} + for n in unknown: + sol[n] = float(x[col[n]]) + return sol + + +def _solve_component(nodes: list[int], edges: list[dict]) -> dict[int, float]: + """Solve one connected component, rejecting cycle-inconsistent edges iteratively. + + The reference (smallest ``video_id``) is pinned to 0. While the graph is over-determined + (more edges than a spanning tree needs) and the worst edge residual exceeds + ``_CYCLE_TOLERANCE_MS``, drop that edge (unless doing so would disconnect a node) and + re-solve. Returns offsets in seconds. + """ + nodes = sorted(nodes) + ref = min(nodes) + edges = list(edges) + sol = _weighted_lstsq(nodes, ref, edges) + while len(edges) > len(nodes) - 1: + worst, worst_res = None, 0.0 + for e in edges: + res = abs((sol[e["a"]] - sol[e["b"]]) - e["offset_s"]) + if res > worst_res: + worst, worst_res = e, res + if worst is None or worst_res * 1000.0 <= _CYCLE_TOLERANCE_MS: + break + trial = [e for e in edges if e is not worst] + if not _connected(nodes, ref, trial): + break # can't drop it without orphaning a node + log.info("audio_sync: rejecting inconsistent edge %d-%d (residual %.1f ms)", + worst["a"], worst["b"], worst_res * 1000.0) + edges = trial + sol = _weighted_lstsq(nodes, ref, edges) + return sol def solve_global_offsets(edges: list[dict], video_ids: list[int]) -> dict[int, float | None]: """Least-squares global solve over the pairwise-offset graph (spec M1, lane A). - Reference node is pinned to 0; disconnected components get ``None`` (spec pitfall #5). - Returns ``video_id -> offset_ms`` (or ``None``). + The reference component (largest; ties broken by smallest ``video_id``) is solved with + its smallest-id node pinned to 0. Every video outside it — including any with no usable + edges — gets ``None`` (spec pitfall #5). Returns ``video_id -> offset_ms`` (or ``None``). """ - raise NotImplementedError("lane A (M1): implement solve_global_offsets") + ids = sorted(set(video_ids)) + usable = [ + e for e in edges + if e["a"] in ids and e["b"] in ids + and e["confidence"] >= _MIN_EDGE_CONFIDENCE + and np.isfinite(e["offset_s"]) + ] + result: dict[int, float | None] = {i: None for i in ids} + if not ids: + return result + + comps = _components(ids, usable) + # Reference component: the one that aligns the most cameras. A lone reference (no + # overlap with anyone) is still "aligned" at 0; everyone else stays None. + ref_comp = max(comps.values(), key=lambda c: (len(c), -min(c))) + if len(ref_comp) == 1: + result[ref_comp[0]] = 0.0 + return result + + ref_set = set(ref_comp) + comp_edges = [e for e in usable if e["a"] in ref_set and e["b"] in ref_set] + for i, off_s in _solve_component(ref_comp, comp_edges).items(): + result[i] = off_s * 1000.0 + return result -def estimate_drift(sig: np.ndarray, ref: np.ndarray, sr: int) -> float: - """Estimate clock drift in ppm by fitting window-offsets over the overlap (spec M1).""" - raise NotImplementedError("lane A (M1): implement estimate_drift") +# --------------------------------------------------------------------------- +# Drift +# --------------------------------------------------------------------------- +def estimate_drift(sig: np.ndarray, ref: np.ndarray, sr: int, + window_s: float = 10.0, hop_s: float = 30.0) -> float: + """Estimate clock drift in ppm by fitting window-offsets over the overlap (spec M1). + + Slides a ``window_s`` window every ``hop_s`` over the common span, measures each + window's GCC-PHAT delay, and fits a line ``delay ≈ slope·t + b``. ``slope`` is the + dimensionless drift rate; ``drift_ppm = slope·1e6``. Returns 0 when there are too few + windows to fit or the estimate is within the ±5 ppm deadband. + """ + sig = np.asarray(sig, dtype=np.float64) + ref = np.asarray(ref, dtype=np.float64) + n = min(len(sig), len(ref)) + win = int(round(window_s * sr)) + hop = int(round(hop_s * sr)) + if win < sr or n < win or hop < 1: # need a window of at least ~1 s that fits + return 0.0 + + centers: list[float] = [] + delays: list[float] = [] + start = 0 + while start + win <= n: + delay_s, conf = gcc_phat(sig[start:start + win], ref[start:start + win], sr, + max_lag_s=window_s / 2.0) + if conf >= _MIN_EDGE_CONFIDENCE: + centers.append((start + win / 2.0) / sr) + delays.append(delay_s) + start += hop + + if len(centers) < 2: + return 0.0 + slope = float(np.polyfit(np.array(centers), np.array(delays), 1)[0]) + drift_ppm = slope * 1.0e6 + return 0.0 if abs(drift_ppm) < _DRIFT_DEADBAND_PPM else drift_ppm + + +# --------------------------------------------------------------------------- +# Orchestration +# --------------------------------------------------------------------------- +def _reference_id(offsets_ms: dict[int, float | None]) -> int | None: + """The reference video: smallest id among those with a solved (non-None) offset.""" + solved = [i for i, o in offsets_ms.items() if o is not None] + return min(solved) if solved else None + + +def _confidence_by_video(edges: list[dict], offsets_ms: dict[int, float | None], + ref_id: int | None) -> dict[int, float | None]: + """Per-video sync confidence in [0, 1): best incident edge, mapped ``1 − 1/ratio``. + + The reference is 1.0; unsolved videos are None. A raw peak ratio of 1 (ambiguous) maps + to 0, 2 to 0.5, large ratios approach 1 — a bounded, honest confidence. + """ + best: dict[int, float] = {} + for e in edges: + if offsets_ms.get(e["a"]) is None or offsets_ms.get(e["b"]) is None: + continue + r = e["confidence"] + best[e["a"]] = max(best.get(e["a"], 0.0), r) + best[e["b"]] = max(best.get(e["b"], 0.0), r) + out: dict[int, float | None] = {} + for vid, off in offsets_ms.items(): + if off is None: + out[vid] = None + elif vid == ref_id: + out[vid] = 1.0 + else: + ratio = best.get(vid) + out[vid] = None if ratio is None else float(max(0.0, 1.0 - 1.0 / ratio)) + return out def run_sync() -> dict: @@ -63,4 +328,87 @@ def run_sync() -> dict: Entrypoint for ``python -m festival4d sync``. Writes ``offset_ms``/``drift_ppm``/ ``sync_confidence`` via ``db`` and exports ``config.SYNC_JSON``. Returns the solution. """ - raise NotImplementedError("lane A (M1): implement run_sync") + import librosa # spec: librosa only for loading; import lazily (heavy module) + + db.init_engine() + db.init_db() + videos = db.get_videos() + if not videos: + raise RuntimeError("no videos in DB — run `python -m festival4d ingest` first") + + sr = config.AUDIO_SAMPLE_RATE + signals: dict[int, np.ndarray] = {} + missing: list[str] = [] + for v in videos: + wav = audio_wav_path(v.id) + if not wav.exists(): + missing.append(v.filename) + continue + y, _ = librosa.load(str(wav), sr=sr, mono=True) + signals[v.id] = y.astype(np.float64) + if missing: + log.warning("audio_sync: no extracted audio for %s — run ingest; " + "they will be left unsynced (offset=None)", ", ".join(missing)) + + edges = pairwise_offsets(signals, sr) if len(signals) >= 2 else [] + offsets_ms = solve_global_offsets(edges, list(signals.keys())) + # Videos with no audio at all are unsynced too. + for v in videos: + offsets_ms.setdefault(v.id, None) + + ref_id = _reference_id(offsets_ms) + ref_sig = signals.get(ref_id) if ref_id is not None else None + confidence = _confidence_by_video(edges, offsets_ms, ref_id) + + solution = { + "reference_video_id": ref_id, + "reference_filename": next((v.filename for v in videos if v.id == ref_id), None), + "sample_rate": sr, + "videos": [], + "edges": [ + {"a": e["a"], "b": e["b"], + "offset_ms": round(e["offset_s"] * 1000.0, 3), + "confidence": round(float(e["confidence"]), 4)} + for e in edges + ], + "components": sorted( + (sorted(c) for c in _components(sorted(signals), [ + e for e in edges if e["confidence"] >= _MIN_EDGE_CONFIDENCE + ]).values()), + key=lambda c: (-len(c), c), + ), + "generated_by": "festival4d.audio_sync", + } + + for v in videos: + off = offsets_ms.get(v.id) + if off is None: + drift = None + elif v.id == ref_id: + drift = 0.0 + elif ref_sig is not None and v.id in signals: + drift = estimate_drift(signals[v.id], ref_sig, sr) + else: + drift = 0.0 + conf = confidence.get(v.id) + db.update_video_sync(v.id, off, drift, conf) + solution["videos"].append({ + "video_id": v.id, "filename": v.filename, + "offset_ms": None if off is None else round(off, 3), + "drift_ppm": None if drift is None else round(drift, 4), + "sync_confidence": None if conf is None else round(conf, 4), + "connected": off is not None, + }) + log.info("audio_sync: %s offset=%s drift=%s conf=%s", + v.filename, + "None" if off is None else f"{off:+.1f}ms", + "None" if drift is None else f"{drift:+.2f}ppm", + "None" if conf is None else f"{conf:.2f}") + + config.SYNC_JSON.parent.mkdir(parents=True, exist_ok=True) + config.SYNC_JSON.write_text(json.dumps(solution, indent=2)) + log.info("audio_sync: wrote %s (%d synced, %d unsynced)", + config.SYNC_JSON, + sum(1 for vv in solution["videos"] if vv["connected"]), + sum(1 for vv in solution["videos"] if not vv["connected"])) + return solution diff --git a/backend/festival4d/ingest.py b/backend/festival4d/ingest.py index 90e520e..e733941 100644 --- a/backend/festival4d/ingest.py +++ b/backend/festival4d/ingest.py @@ -1,26 +1,98 @@ """Video ingest: probe files in ``data/raw`` and extract audio (spec M1, lane A). -STUB — lane A fills the bodies; the public signatures below are frozen (``cli.py`` calls -``run_ingest``). For each file in ``config.RAW_DIR``: ffprobe it, insert/update the -``videos`` row via ``db`` helpers, and extract a mono 16 kHz WAV into ``config.AUDIO_DIR``. +For each file in ``config.RAW_DIR``: ffprobe it, insert/update the ``videos`` row via +``db`` helpers, and extract a mono 16 kHz WAV into ``config.AUDIO_DIR`` (one per video, +keyed by ``video_id``) — that WAV is exactly what ``audio_sync.run_sync`` loads. + +``run_ingest`` is idempotent: re-running after the synthetic fixture (which already +registered the videos) reuses the existing rows and just re-extracts audio, so the +demo path ``synthetic -> ingest -> sync`` works without unique-constraint collisions. """ from __future__ import annotations +import json import logging +import shutil +import subprocess from pathlib import Path +from festival4d import config, db + log = logging.getLogger("festival4d.ingest") +# Container extensions we treat as ingestible video (audio is pulled from any of them). +_VIDEO_EXTS = {".mp4", ".mov", ".m4v", ".mkv", ".avi", ".webm"} + + +def _require(tool: str) -> None: + if shutil.which(tool) is None: + raise RuntimeError( + f"{tool} not found on PATH — required for ingest (macOS: `brew install ffmpeg`)" + ) + + +def audio_wav_path(video_id: int) -> Path: + """Canonical path of a video's extracted mono WAV. + + Shared convention between ingest (writer) and :mod:`festival4d.audio_sync` (reader), + so neither side has to guess filenames. + """ + return config.AUDIO_DIR / f"{video_id}.wav" + + +def _parse_fps(rate: str | None) -> float | None: + """Parse an ffprobe frame-rate string (``"30/1"`` or ``"29.97"``) to fps.""" + if not rate or rate in ("0/0", "N/A"): + return None + if "/" in rate: + num, den = rate.split("/", 1) + den = float(den) + return float(num) / den if den else None + return float(rate) + def probe_video(path: Path) -> dict: """ffprobe a video; return ``{duration_s, fps, width, height}`` (lane A / M1).""" - raise NotImplementedError("lane A (M1): implement probe_video") + _require("ffprobe") + out = subprocess.run( + ["ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", + "stream=width,height,avg_frame_rate,r_frame_rate:format=duration", + "-of", "json", str(path)], + check=True, capture_output=True, text=True, + ).stdout + data = json.loads(out) + streams = data.get("streams") or [] + if not streams: + raise ValueError(f"{path.name}: no video stream found") + stream = streams[0] + # avg_frame_rate is the true average; r_frame_rate is a fallback for odd containers. + fps = _parse_fps(stream.get("avg_frame_rate")) or _parse_fps(stream.get("r_frame_rate")) or 0.0 + duration = float(data.get("format", {}).get("duration") or 0.0) + return { + "duration_s": duration, + "fps": float(fps), + "width": int(stream["width"]), + "height": int(stream["height"]), + } def extract_audio(path: Path, out_wav: Path, sample_rate: int = 16_000) -> Path: - """Extract a mono ``sample_rate`` WAV from ``path`` via ffmpeg (lane A / M1).""" - raise NotImplementedError("lane A (M1): implement extract_audio") + """Extract a mono ``sample_rate`` 16-bit PCM WAV from ``path`` via ffmpeg (lane A / M1).""" + _require("ffmpeg") + out_wav.parent.mkdir(parents=True, exist_ok=True) + subprocess.run( + ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-i", str(path), + "-vn", # drop video + "-ac", "1", # mono + "-ar", str(sample_rate), # resample + "-c:a", "pcm_s16le", # uncompressed PCM (lossless for sync) + str(out_wav)], + check=True, + ) + return out_wav def run_ingest() -> list[dict]: @@ -28,4 +100,39 @@ def run_ingest() -> list[dict]: Entrypoint for ``python -m festival4d ingest``. Returns a per-video summary list. """ - raise NotImplementedError("lane A (M1): implement run_ingest") + db.init_engine() + db.init_db() + + files = sorted( + p for p in config.RAW_DIR.glob("*") if p.suffix.lower() in _VIDEO_EXTS + ) + if not files: + log.warning("ingest: no video files in %s (drop clips there first)", config.RAW_DIR) + return [] + + existing = {v.filename: v for v in db.get_videos()} + summary: list[dict] = [] + for path in files: + info = probe_video(path) + video = existing.get(path.name) + reused = video is not None + if video is None: + video = db.add_video( + filename=path.name, + duration_s=info["duration_s"], fps=info["fps"], + width=info["width"], height=info["height"], + ) + wav = audio_wav_path(video.id) + extract_audio(path, wav) + summary.append({ + "video_id": video.id, "filename": path.name, + "duration_s": info["duration_s"], "fps": info["fps"], + "width": info["width"], "height": info["height"], + "audio_wav": str(wav), "reused_db_row": reused, + }) + log.info("ingest: %s -> video_id=%d (%.1fs, %.2f fps, %dx%d)%s", + path.name, video.id, info["duration_s"], info["fps"], + info["width"], info["height"], " [reused row]" if reused else "") + + log.info("ingest: %d video(s) ready; audio in %s", len(summary), config.AUDIO_DIR) + return summary diff --git a/backend/tests/test_audio_sync.py b/backend/tests/test_audio_sync.py new file mode 100644 index 0000000..d1fbcd9 --- /dev/null +++ b/backend/tests/test_audio_sync.py @@ -0,0 +1,240 @@ +"""Audio-sync tests (spec M1, lane A). + +Pure-numpy where possible (no ffmpeg): GCC-PHAT sign/precision, pairwise+global solve +recovering the M0 known offsets (±10 ms), disconnected components → None (pitfall #5), +cycle-consistency rejection, drift recovery (±3 ppm), and the full ``run_sync`` persist + +export path against synthesized WAVs. One ffmpeg-gated test runs the literal spec M1 +acceptance: ``synthetic -> ingest -> sync`` recovers ground truth. +""" + +from __future__ import annotations + +import json +import shutil + +import numpy as np +import pytest + +from festival4d import audio_sync, config, db, ingest, synthetic + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- +def _broadband(n: int, seed: int = 0) -> np.ndarray: + """Seeded, mildly band-limited noise — a sharp, unambiguous correlation target.""" + rng = np.random.default_rng(seed) + x = rng.standard_normal(n) + return np.convolve(x, np.ones(4) / 4.0, mode="same") + + +def _delay_samples(x: np.ndarray, d: int) -> np.ndarray: + """Return ``y`` with ``y[n] = x[n - d]`` (x delayed by d samples, zero-filled).""" + y = np.zeros_like(x) + if d >= 0: + y[d:] = x[:len(x) - d] if d else x + else: + y[:d] = x[-d:] + return y + + +def _clean_audio_dir() -> None: + """Remove any WAVs left in the shared session audio dir (ids are reused after reset_db).""" + config.AUDIO_DIR.mkdir(parents=True, exist_ok=True) + for wav in config.AUDIO_DIR.glob("*.wav"): + wav.unlink() + + +# --------------------------------------------------------------------------- +# GCC-PHAT +# --------------------------------------------------------------------------- +def test_gcc_phat_recovers_known_delay_positive_and_negative(): + sr = 16_000 + ref = _broadband(sr * 4) + for d in (0, 137, -211, 800): + sig = _delay_samples(ref, d) + offset_s, conf = audio_sync.gcc_phat(sig, ref, sr) + # positive offset_s means sig lags ref, i.e. matches a positive sample delay + assert abs(offset_s - d / sr) < 0.5 / sr, (d, offset_s * sr) + assert conf > 5.0 + + +def test_gcc_phat_confidence_drops_with_noise(): + sr = 16_000 + ref = _broadband(sr * 4, seed=1) + clean = _delay_samples(ref, 300) + rng = np.random.default_rng(2) + noisy = clean + 3.0 * rng.standard_normal(len(clean)) + _, c_clean = audio_sync.gcc_phat(clean, ref, sr) + _, c_noisy = audio_sync.gcc_phat(noisy, ref, sr) + assert c_clean > c_noisy + + +# --------------------------------------------------------------------------- +# pairwise + global solve +# --------------------------------------------------------------------------- +def test_pairwise_and_solve_recover_synthetic_offsets(): + """The M0 known offsets (0/+1370/−842 ms) recovered within the spec's ±10 ms.""" + sr = config.AUDIO_SAMPLE_RATE + master, _ = synthetic.synth_master_audio(sr) + signals = { + i + 1: synthetic.slice_for_offset(master, sr, off, config.SYNTH_DURATION_S) + for i, off in enumerate(config.SYNTH_OFFSETS_MS) + } + edges = audio_sync.pairwise_offsets(signals, sr) + solved = audio_sync.solve_global_offsets(edges, list(signals)) + for i, off in enumerate(config.SYNTH_OFFSETS_MS): + assert solved[i + 1] is not None + assert abs(solved[i + 1] - off) <= 10.0, (off, solved[i + 1]) + + +def test_pairwise_edge_sign_is_relative_start_offset(): + """Edge ``offset_s`` = offset_a − offset_b (seconds).""" + sr = config.AUDIO_SAMPLE_RATE + master, _ = synthetic.synth_master_audio(sr) + a = synthetic.slice_for_offset(master, sr, 500.0, 6.0) # id 1, offset +500 ms + b = synthetic.slice_for_offset(master, sr, -300.0, 6.0) # id 2, offset −300 ms + edges = audio_sync.pairwise_offsets({1: a, 2: b}, sr) + assert len(edges) == 1 + assert abs(edges[0]["offset_s"] * 1000.0 - (500.0 - -300.0)) <= 10.0 + + +def test_solve_disconnected_component_gets_none(): + """A video with no usable edges is unsynced (offset None), not guessed (pitfall #5).""" + edges = [ + {"a": 1, "b": 2, "offset_s": -0.5, "confidence": 30.0}, + {"a": 2, "b": 3, "offset_s": -0.4, "confidence": 30.0}, + ] + solved = audio_sync.solve_global_offsets(edges, [1, 2, 3, 4]) + assert solved[1] == 0.0 + assert abs(solved[2] - 500.0) < 1.0 + assert abs(solved[3] - 900.0) < 1.0 + assert solved[4] is None # isolated → None + + +def test_solve_rejects_cycle_inconsistent_edge(): + """A grossly inconsistent edge (>50 ms residual) is rejected; the rest still solve.""" + # True offsets: 1=0, 2=+500 ms, 3=+900 ms. Edge 1-3 is corrupted (should be −0.9 s). + edges = [ + {"a": 1, "b": 2, "offset_s": -0.5, "confidence": 20.0}, + {"a": 2, "b": 3, "offset_s": -0.4, "confidence": 20.0}, + {"a": 1, "b": 3, "offset_s": +5.0, "confidence": 20.0}, # bad + ] + solved = audio_sync.solve_global_offsets(edges, [1, 2, 3]) + assert solved[1] == 0.0 + assert abs(solved[2] - 500.0) < 1.0 + assert abs(solved[3] - 900.0) < 1.0 # recovered despite the bad edge + + +def test_low_confidence_edges_are_dropped(): + """Edges below the confidence floor don't connect a component.""" + edges = [{"a": 1, "b": 2, "offset_s": -0.5, "confidence": 1.01}] + solved = audio_sync.solve_global_offsets(edges, [1, 2]) + # 1 is the lone reference at 0; 2 has only a rejected edge → None + assert solved[1] == 0.0 and solved[2] is None + + +# --------------------------------------------------------------------------- +# drift +# --------------------------------------------------------------------------- +def _make_drifted(ref: np.ndarray, sr: int, drift_ppm: float, offset_s: float = 0.0) -> np.ndarray: + """Build v from the config timebase: v(τ) = ref((1 − d)·τ + offset_s), d = ppm·1e−6.""" + d = drift_ppm * 1e-6 + local = np.arange(len(ref)) + src = (1.0 - d) * local + offset_s * sr + return np.interp(src, np.arange(len(ref)), ref, left=0.0, right=0.0) + + +def test_estimate_drift_recovers_known_ppm(): + sr = 8_000 + ref = _broadband(sr * 150, seed=5) # 150 s so several 30 s-spaced windows fit + for ppm in (30.0, -22.0): + v = _make_drifted(ref, sr, ppm) + est = audio_sync.estimate_drift(v, ref, sr) + assert abs(est - ppm) <= 3.0, (ppm, est) + + +def test_estimate_drift_zero_within_deadband(): + sr = 8_000 + ref = _broadband(sr * 150, seed=6) + assert audio_sync.estimate_drift(ref, ref, sr) == 0.0 # identical + assert audio_sync.estimate_drift(_make_drifted(ref, sr, 1.5), ref, sr) == 0.0 # < 5 ppm + + +def test_estimate_drift_returns_zero_when_too_short(): + sr = 16_000 + ref = _broadband(sr * 15) # < 2 windows at 10 s / 30 s hop + assert audio_sync.estimate_drift(ref, ref, sr) == 0.0 + + +# --------------------------------------------------------------------------- +# run_sync orchestration +# --------------------------------------------------------------------------- +def test_run_sync_persists_and_exports(): + """Full run_sync against synthesized WAVs (no ffmpeg): DB + sync.json, ±10 ms.""" + sr = config.AUDIO_SAMPLE_RATE + master, _ = synthetic.synth_master_audio(sr) + db.init_engine() + db.reset_db() + _clean_audio_dir() + gt = {} + for i, off in enumerate(config.SYNTH_OFFSETS_MS): + v = db.add_video(filename=f"cam{i}.mp4", duration_s=config.SYNTH_DURATION_S, + fps=30.0, width=640, height=360) + clip = synthetic.slice_for_offset(master, sr, off, config.SYNTH_DURATION_S) + synthetic.write_wav(ingest.audio_wav_path(v.id), clip, sr) + gt[v.id] = off + + solution = audio_sync.run_sync() + + for v in db.get_videos(): + assert v.offset_ms is not None + assert abs(v.offset_ms - gt[v.id]) <= 10.0, (gt[v.id], v.offset_ms) + assert v.drift_ppm == 0.0 + assert 0.0 <= v.sync_confidence <= 1.0 + ref = min(gt) + assert solution["reference_video_id"] == ref + assert db.get_video(ref).sync_confidence == 1.0 + exported = json.loads(config.SYNC_JSON.read_text()) + assert exported["reference_video_id"] == ref + assert len(exported["edges"]) == 3 + + +def test_run_sync_marks_video_without_audio_unsynced(): + """A registered video with no extracted WAV is left offset=None, not guessed.""" + sr = config.AUDIO_SAMPLE_RATE + master, _ = synthetic.synth_master_audio(sr) + db.init_engine() + db.reset_db() + _clean_audio_dir() + v0 = db.add_video(filename="cam0.mp4", duration_s=20.0, fps=30.0, width=640, height=360) + v1 = db.add_video(filename="cam1.mp4", duration_s=20.0, fps=30.0, width=640, height=360) + no_audio = db.add_video(filename="cam2.mp4", duration_s=20.0, fps=30.0, width=640, height=360) + synthetic.write_wav(ingest.audio_wav_path(v0.id), + synthetic.slice_for_offset(master, sr, 0.0, 20.0), sr) + synthetic.write_wav(ingest.audio_wav_path(v1.id), + synthetic.slice_for_offset(master, sr, 1370.0, 20.0), sr) + # no WAV for `no_audio` + + audio_sync.run_sync() + assert db.get_video(no_audio.id).offset_ms is None + assert db.get_video(no_audio.id).sync_confidence is None + assert db.get_video(v0.id).offset_ms == 0.0 + + +# --------------------------------------------------------------------------- +# ffmpeg-gated: the literal spec M1 acceptance (synthetic -> ingest -> sync) +# --------------------------------------------------------------------------- +@pytest.mark.skipif(shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None, + reason="ffmpeg/ffprobe required for the end-to-end sync acceptance") +def test_synthetic_ingest_sync_recovers_ground_truth(): + synthetic.build(duration_s=8.0, run_ffmpeg=True) # into the temp data dir (conftest) + ingest.run_ingest() + audio_sync.run_sync() + + gt = {v["filename"]: v["offset_ms"] + for v in json.loads(config.GROUND_TRUTH_JSON.read_text())["videos"]} + for v in db.get_videos(): + assert v.offset_ms is not None, v.filename + assert abs(v.offset_ms - gt[v.filename]) <= 10.0, (v.filename, v.offset_ms, gt[v.filename]) + assert abs(v.drift_ppm) <= 3.0 diff --git a/backend/tests/test_ingest.py b/backend/tests/test_ingest.py new file mode 100644 index 0000000..902515f --- /dev/null +++ b/backend/tests/test_ingest.py @@ -0,0 +1,93 @@ +"""Ingest tests (spec M1, lane A). + +Pure unit tests for the path/fps helpers, plus ffmpeg-gated tests that probe a real +rendered clip, extract audio, and run the full idempotent ``run_ingest``. +""" + +from __future__ import annotations + +import shutil + +import numpy as np +import pytest + +from festival4d import config, db, ingest, synthetic + + +# --------------------------------------------------------------------------- +# pure helpers +# --------------------------------------------------------------------------- +def test_audio_wav_path_convention(): + assert ingest.audio_wav_path(7) == config.AUDIO_DIR / "7.wav" + + +def test_parse_fps(): + assert ingest._parse_fps("30/1") == 30.0 + assert abs(ingest._parse_fps("30000/1001") - 29.97) < 0.01 + assert ingest._parse_fps("25") == 25.0 + assert ingest._parse_fps(None) is None + assert ingest._parse_fps("0/0") is None + assert ingest._parse_fps("N/A") is None + + +def test_run_ingest_no_files_is_empty(tmp_path, monkeypatch): + """An empty raw dir yields an empty summary, not a crash.""" + monkeypatch.setattr(config, "RAW_DIR", tmp_path / "empty_raw") + (tmp_path / "empty_raw").mkdir() + db.init_engine() + db.reset_db() + assert ingest.run_ingest() == [] + + +# --------------------------------------------------------------------------- +# ffmpeg-gated +# --------------------------------------------------------------------------- +def _render_clip(tmp_path, duration_s=2.0, cam_index=0): + sr = config.AUDIO_SAMPLE_RATE + master, _ = synthetic.synth_master_audio(sr) + clip = synthetic.slice_for_offset(master, sr, 0.0, duration_s) + wav = tmp_path / "src.wav" + synthetic.write_wav(wav, clip, sr) + mp4 = tmp_path / f"cam{cam_index}.mp4" + synthetic.render_video(mp4, wav, cam_index, 640, 360, 30, duration_s) + return mp4 + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None, + reason="ffmpeg/ffprobe required") +def test_probe_video(tmp_path): + mp4 = _render_clip(tmp_path, duration_s=2.0) + info = ingest.probe_video(mp4) + assert info["width"] == 640 and info["height"] == 360 + assert abs(info["fps"] - 30.0) < 1.0 + assert 1.5 < info["duration_s"] < 2.6 + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None, + reason="ffmpeg/ffprobe required") +def test_extract_audio(tmp_path): + mp4 = _render_clip(tmp_path, duration_s=2.0) + out = ingest.extract_audio(mp4, tmp_path / "out.wav", sample_rate=16_000) + assert out.exists() + samples, sr = synthetic.read_wav(out) + assert sr == 16_000 + assert samples.ndim == 1 # mono + assert abs(len(samples) / sr - 2.0) < 0.3 # ~2 s + + +@pytest.mark.skipif(shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None, + reason="ffmpeg/ffprobe required") +def test_run_ingest_registers_and_is_idempotent(): + synthetic.build(duration_s=2.0, run_ffmpeg=True) # into the temp data dir (conftest) + summary = ingest.run_ingest() + assert len(summary) == 3 + for s in summary: + assert ingest.audio_wav_path(s["video_id"]).exists() + samples, sr = synthetic.read_wav(ingest.audio_wav_path(s["video_id"])) + assert sr == 16_000 and samples.ndim == 1 + + # Second run reuses the existing rows (no unique-constraint crash) and re-extracts. + again = ingest.run_ingest() + assert len(again) == 3 + assert all(s["reused_db_row"] for s in again) + assert len(db.get_videos()) == 3 # no duplicate rows diff --git a/plan/status/lane-A.md b/plan/status/lane-A.md new file mode 100644 index 0000000..f98efef --- /dev/null +++ b/plan/status/lane-A.md @@ -0,0 +1,24 @@ +# Status — lane-A + +## Round 0 — 2026-07-16 — STATUS: ready_to_merge (committed on isolated worktree) + +**Directives acknowledged:** round 0 of plan/DIRECTIVES.md (run order & protocol; Lane A = spec M1). + +**Acceptance checklist** (spec M1 + lane brief) — ALL PASSING: +- [x] `ingest`: ffprobe each file in `data/raw`, register in `videos`, extract mono 16 kHz WAV into `data/work/audio/`. +- [x] `sync`: PHAT-whitened GCC-PHAT pairwise offsets, scored by peak-to-second-peak ratio; sub-sample parabolic peak refinement. +- [x] Global confidence-weighted least-squares solve; cycle-consistency rejection (>50 ms residual). +- [x] Drift: windowed GCC-PHAT + line fit → `drift_ppm`; store 0 if `|drift| < 5 ppm`. +- [x] Disconnected components → `offset_ms = None` (spec pitfall #5). +- [x] Persist via `db.update_video_sync`; export `config.SYNC_JSON`. +- [x] `synthetic → ingest → sync` recovers GT offsets (err < 0.001 ms) and drift 0 ppm — inside ±10 ms / ±3 ppm. +- [x] `pytest` green in isolation: **43 passed** (24 foundation + 19 new) on pristine foundation + Lane A only. + +**Done this round:** implemented `ingest.py` + `audio_sync.py`; wrote `test_ingest.py` (6) + `test_audio_sync.py` (13). Committed as `157c783` on `lane/a-media`, based directly on `main`/foundation (`5fa7301`) — exactly the 5 owned files, no other lane's changes. Evidence (run in the isolated `festifun-laneA` worktree): +- `pytest` → `43 passed, 4 warnings` (warnings are librosa/audioread deprecations only). +- `synthetic → ingest → sync` → DB cam0/1/2 = +0.000 / +1370.000 / −842.000 ms, drift 0, err ≤ 0.0002 ms vs `ground_truth.json`. +- `sync` exported `data/work/sync.json` (reference id, per-video offset/drift/confidence, raw edges, components). + +**Coordination note (was a blocker; resolved for Lane A):** Lanes A/C/D were sharing one working directory (`/Users/m3ultra/Documents/festifun`) and colliding — HEAD there was moved to `lane/d-events`, and the shared tree held four lanes' uncommitted changes intermixed. Per coordinator sign-off, Lane A was landed into its own `git worktree` (`festifun-laneA`, branch `lane/a-media` reset to `main`) and committed there, disturbing no other worktree. **Still open for coordinator:** lanes C and D also need their own worktrees (lane B already has one); the shared dir still contains their intermixed uncommitted work. + +**Next:** awaiting coordinator go-ahead to `git push origin lane/a-media` and merge to `main` (merge is coordination-sensitive given the collision; branch is ready).