monster/festifun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'lane/a-media'

Browse Source
This commit is contained in:
m3ultra 2026-07-16 09:01:59 +10:00
commit e28dc78e06
5 changed files with 839 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

388
backend/festival4d/audio_sync.py
View File

@ -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

121
backend/festival4d/ingest.py
View File

@ -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

240
backend/tests/test_audio_sync.py Normal file
View File

@ -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

93
backend/tests/test_ingest.py Normal file
View File

@ -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

24
plan/status/lane-A.md Normal file
View File

@ -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).