diff --git a/backend/festival4d/audio_features.py b/backend/festival4d/audio_features.py index 809e59c..9d7c2bb 100644 --- a/backend/festival4d/audio_features.py +++ b/backend/festival4d/audio_features.py @@ -1,4 +1,4 @@ -"""Audio features: beats & onsets (spec M10, lane E). STUB — lane E fills the body. +"""Audio features: beats & onsets (spec M10, lane E). Contract (frozen at foundation2 merge — phase-5 contract #1): @@ -15,11 +15,173 @@ and onset-strength detection (librosa is already a dependency), and writes All times are ``t_global`` seconds. Quiet / short / beatless audio must produce a *valid, empty* result (``tempo_bpm: null``, empty lists) — never a crash. Returns a small summary dict for the CLI log. ``GET /api/beats`` serves the written file. + +Implementation notes +-------------------- +- Analysis runs on the reference's local timeline and maps to ``t_global`` via the frozen + :func:`config.t_global_from_video` (identity for a true reference: offset 0, drift 0; + correct either way if we ever fall back to a non-zero-offset "reference"). +- ``HOP_LENGTH = 160`` (10 ms at 16 kHz): librosa's default 512-sample hop is a 32 ms + grid at this rate, which quantizes beats past the ±50 ms acceptance bound. 10 ms frames + recover the synthetic 120 BPM pulse grid to ~30 ms / ±0.001 BPM (see test_audio_features). +- Reported ``tempo_bpm`` is the median inter-beat interval of the tracked beats (more + accurate than the tracker's coarse tempo prior) and ``null`` when < 2 beats are found. +- Missing prerequisites (no videos ingested / WAV absent) degrade to a summary dict with a + ``note`` and write nothing — mirroring ``events_ai.run_events`` house style; degenerate + *content* (silence, too short, beatless) writes the valid-empty JSON above. """ from __future__ import annotations +import json +import logging +import wave +from pathlib import Path + +import numpy as np + +from festival4d import config, db + +log = logging.getLogger("festival4d.audio_features") + +GENERATED_BY = "festival4d.audio_features" +HOP_LENGTH = 160 # analysis hop in samples: 10 ms at the 16 kHz ingest rate +MIN_DURATION_S = 1.0 # anything shorter is "no usable audio" -> valid-empty result +SILENCE_PEAK = 1e-4 # |peak| below this is treated as silence -> valid-empty result + + +def _empty_result() -> dict: + """The valid-empty beats.json body (contract #1) for silent/short/beatless audio.""" + return { + "tempo_bpm": None, + "beats_s": [], + "onsets": [], + "generated_by": GENERATED_BY, + } + + +def _select_reference(videos: list) -> "object | None": + """The reference video (spec §2: offset 0). Fall back to the smallest |offset|, else first.""" + if not videos: + return None + exact = [v for v in videos if v.offset_ms == 0] + if exact: + return exact[0] + with_offset = [v for v in videos if v.offset_ms is not None] + if with_offset: + return min(with_offset, key=lambda v: abs(v.offset_ms)) + return videos[0] + + +def _read_wav_mono(path: Path) -> tuple[np.ndarray, int]: + """Read a 16-bit PCM WAV (the ingest format) as float64 mono in [-1, 1].""" + with wave.open(str(path), "rb") as w: + sr = w.getframerate() + channels = w.getnchannels() + frames = w.readframes(w.getnframes()) + pcm = np.frombuffer(frames, dtype=" 1: + pcm = pcm.reshape(-1, channels).mean(axis=1) + return pcm, sr + + +def analyze(y: np.ndarray, sr: int) -> dict: + """Beat/tempo + onset-strength analysis of a mono signal on its *local* timeline. + + Returns the contract #1 dict with times in local (signal) seconds — the caller maps + them to ``t_global``. Degenerate input (short/silent/beatless) returns the valid-empty + shape; this function never raises on quiet or short audio. + """ + import librosa # deferred: keeps `festival4d` import light for the API/CLI + + y = np.asarray(y, dtype=np.float32).reshape(-1) + if len(y) < MIN_DURATION_S * sr or float(np.max(np.abs(y), initial=0.0)) < SILENCE_PEAK: + return _empty_result() + + try: + onset_env = librosa.onset.onset_strength(y=y, sr=sr, hop_length=HOP_LENGTH) + _tempo, beat_frames = librosa.beat.beat_track( + onset_envelope=onset_env, sr=sr, hop_length=HOP_LENGTH + ) + beat_times = librosa.frames_to_time(beat_frames, sr=sr, hop_length=HOP_LENGTH) + + onset_frames = librosa.onset.onset_detect( + onset_envelope=onset_env, sr=sr, hop_length=HOP_LENGTH + ) + onset_times = librosa.frames_to_time(onset_frames, sr=sr, hop_length=HOP_LENGTH) + except Exception as exc: # librosa internals on weird signals: degrade, never crash + log.warning("audio_features: analysis failed (%s) — returning empty result", exc) + return _empty_result() + + beats = [float(t) for t in np.atleast_1d(beat_times)] + if len(beats) >= 2: + tempo_bpm: float | None = float(60.0 / np.median(np.diff(beats))) + elif beats: + tempo_bpm = float(np.atleast_1d(_tempo)[0]) or None + else: + tempo_bpm = None + + env_max = float(onset_env.max()) if len(onset_env) else 0.0 + onsets = [] + if env_max > 0: + for frame, t in zip(np.atleast_1d(onset_frames), np.atleast_1d(onset_times)): + strength = float(onset_env[int(frame)]) / env_max + onsets.append({"t_global_s": float(t), "strength": min(1.0, max(0.0, strength))}) + + return { + "tempo_bpm": tempo_bpm, + "beats_s": beats, + "onsets": onsets, + "generated_by": GENERATED_BY, + } + def run_features() -> dict: """Analyze the reference soundtrack and write ``data/work/beats.json`` (see module doc).""" - raise NotImplementedError("audio features (lane E / M10)") + videos = db.get_videos() + reference = _select_reference(videos) + if reference is None: + log.warning("audio_features: no videos in the project — run ingest first") + return {"status": "no_videos", "note": "no videos in the project (run ingest first)"} + + wav_path = config.AUDIO_DIR / f"{reference.id}.wav" # ingest.audio_wav_path convention + if not wav_path.exists(): + log.warning("audio_features: reference WAV missing at %s — run ingest first", wav_path) + return { + "status": "no_audio", + "note": f"reference audio not found ({wav_path.name}) — run `python -m festival4d ingest`", + } + + y, sr = _read_wav_mono(wav_path) + result = analyze(y, sr) + + # Map local (reference) times onto the master timeline via the frozen timebase helpers. + # For a true reference (offset 0, drift 0) this is the identity. + offset_ms = reference.offset_ms or 0.0 + drift_ppm = reference.drift_ppm or 0.0 + if offset_ms != 0.0 or drift_ppm != 0.0: + result["beats_s"] = [ + float(config.t_global_from_video(t, offset_ms, drift_ppm)) for t in result["beats_s"] + ] + for onset in result["onsets"]: + onset["t_global_s"] = float( + config.t_global_from_video(onset["t_global_s"], offset_ms, drift_ppm) + ) + + config.BEATS_JSON.parent.mkdir(parents=True, exist_ok=True) + config.BEATS_JSON.write_text(json.dumps(result, indent=2)) + log.info( + "audio_features: %s beats (tempo %s BPM), %s onsets -> %s", + len(result["beats_s"]), + f"{result['tempo_bpm']:.1f}" if result["tempo_bpm"] else "null", + len(result["onsets"]), + config.BEATS_JSON, + ) + return { + "status": "ok", + "reference_video_id": reference.id, + "tempo_bpm": result["tempo_bpm"], + "beats": len(result["beats_s"]), + "onsets": len(result["onsets"]), + "beats_json": str(config.BEATS_JSON), + } diff --git a/backend/tests/test_audio_features.py b/backend/tests/test_audio_features.py new file mode 100644 index 0000000..5ffa630 --- /dev/null +++ b/backend/tests/test_audio_features.py @@ -0,0 +1,171 @@ +"""M10 acceptance tests (lane E): beat/onset analysis of the reference ingest WAV. + +Built on the synthetic fixture + a real ingest run (per the phase-5 test pattern): the +fixture's master audio carries a known 120 BPM click grid (``synthetic.BEAT_INTERVAL_S`` += 0.5 s), so acceptance is objective — beats within ±50 ms of the grid, tempo within +±2 BPM. Silence / short / missing audio must degrade per the house style, never crash. +""" + +from __future__ import annotations + +import json +import shutil + +import numpy as np +import pytest + +pytestmark = pytest.mark.skipif( + shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None, + reason="ffmpeg/ffprobe required to build the fixture + run ingest", +) + +BEAT_GRID_S = 0.5 # synthetic.BEAT_INTERVAL_S — the fixture's known pulse grid +TRUE_TEMPO_BPM = 60.0 / BEAT_GRID_S # 120 + + +@pytest.fixture(scope="module") +def project(): + """Full synthetic project + ingest (produces the 16 kHz WAVs in config.AUDIO_DIR).""" + from festival4d import config, ingest, synthetic + + synthetic.build(run_ffmpeg=True) # default 20 s — enough signal for tempo tracking + ingest.run_ingest() + config.BEATS_JSON.unlink(missing_ok=True) + yield config + config.BEATS_JSON.unlink(missing_ok=True) # don't leak state into later test modules + + +def _grid_error_s(t: float) -> float: + """Distance from t to the nearest 0.5 s grid line.""" + return abs(t - round(t / BEAT_GRID_S) * BEAT_GRID_S) + + +# --------------------------------------------------------------------------- +# The real thing: known pulse grid recovered from the ingest WAV. +# --------------------------------------------------------------------------- +def test_run_features_recovers_synthetic_pulse_grid(project): + from festival4d import audio_features + + summary = audio_features.run_features() + assert summary["status"] == "ok" + assert project.BEATS_JSON.exists() + + data = json.loads(project.BEATS_JSON.read_text()) + # Contract #1 structure lock: exactly these keys. + assert set(data) == {"tempo_bpm", "beats_s", "onsets", "generated_by"} + assert data["generated_by"] == "festival4d.audio_features" + + # Tempo within ±2 BPM of the known 120 BPM grid. + assert data["tempo_bpm"] == pytest.approx(TRUE_TEMPO_BPM, abs=2.0) + + # Beats: a healthy number over 20 s @ 120 BPM, each within ±50 ms of the grid, + # strictly increasing, inside the reference window (t_global == reference local time). + beats = data["beats_s"] + assert len(beats) >= 20 + assert beats == sorted(beats) + # AAC encode/decode pads the clip tail slightly, so allow a small margin past 20 s. + assert all(0.0 <= b <= 20.5 for b in beats) + worst = max(_grid_error_s(b) for b in beats) + assert worst <= 0.050, f"worst beat off grid by {worst * 1000:.1f} ms" + + # Onsets: present, valid strengths, times on the master timeline. + onsets = data["onsets"] + assert len(onsets) > 0 + assert all(set(o) == {"t_global_s", "strength"} for o in onsets) + assert all(0.0 <= o["strength"] <= 1.0 for o in onsets) + assert any(o["strength"] >= 0.9 for o in onsets) # normalized: the peak onset is ~1.0 + assert all(0.0 <= o["t_global_s"] <= 20.5 for o in onsets) + + # The summary mirrors the file. + assert summary["beats"] == len(beats) + assert summary["onsets"] == len(onsets) + assert summary["reference_video_id"] == 1 # cam0 (offset 0) is the reference + + +def test_onsets_catch_the_ground_truth_bangs(project): + """The three loud synthetic bangs (3, 10, 17 s) must appear as strong onsets.""" + from festival4d import audio_features, synthetic + + audio_features.run_features() + data = json.loads(project.BEATS_JSON.read_text()) + onset_times = [o["t_global_s"] for o in data["onsets"]] + for bang in synthetic.BANG_TIMES_S: + assert any(abs(t - bang) <= 0.075 for t in onset_times), f"bang at {bang}s not detected" + + +# --------------------------------------------------------------------------- +# Degradation: silence / short / missing input — valid output or clear note, never a crash. +# --------------------------------------------------------------------------- +def test_silence_produces_valid_empty_json(project): + from festival4d import audio_features, synthetic + + wav = project.AUDIO_DIR / "1.wav" + original = wav.read_bytes() + try: + synthetic.write_wav(wav, np.zeros(16_000 * 5, dtype=np.float32), 16_000) + summary = audio_features.run_features() + assert summary["status"] == "ok" + data = json.loads(project.BEATS_JSON.read_text()) + assert data == { + "tempo_bpm": None, + "beats_s": [], + "onsets": [], + "generated_by": "festival4d.audio_features", + } + finally: + wav.write_bytes(original) + project.BEATS_JSON.unlink(missing_ok=True) + + +def test_too_short_audio_produces_valid_empty_json(project): + from festival4d import audio_features, synthetic + + wav = project.AUDIO_DIR / "1.wav" + original = wav.read_bytes() + try: + rng = np.random.default_rng(3) + synthetic.write_wav(wav, rng.standard_normal(4_000).astype(np.float32) * 0.5, 16_000) + audio_features.run_features() + data = json.loads(project.BEATS_JSON.read_text()) + assert data["tempo_bpm"] is None and data["beats_s"] == [] and data["onsets"] == [] + finally: + wav.write_bytes(original) + project.BEATS_JSON.unlink(missing_ok=True) + + +def test_missing_wav_degrades_with_note(project): + from festival4d import audio_features + + wav = project.AUDIO_DIR / "1.wav" + moved = wav.with_suffix(".wav.bak") + wav.rename(moved) + try: + summary = audio_features.run_features() # must not raise + assert summary["status"] == "no_audio" + assert "ingest" in summary["note"] + assert not project.BEATS_JSON.exists() # nothing stale written + finally: + moved.rename(wav) + + +def test_analyze_is_pure_and_beatless_noise_is_empty(): + """Unit-level: analyze() on beatless noise returns the valid-empty shape.""" + from festival4d import audio_features + + rng = np.random.default_rng(11) + y = (rng.standard_normal(16_000 * 6) * 0.2).astype(np.float32) # 6 s of plain noise + data = audio_features.analyze(y, 16_000) + assert set(data) == {"tempo_bpm", "beats_s", "onsets", "generated_by"} + # librosa may or may not hallucinate a weak pulse in noise; the hard requirement is a + # valid, well-typed result — not a crash. + assert isinstance(data["beats_s"], list) + assert all(isinstance(b, float) for b in data["beats_s"]) + + +def test_cli_features_runs_end_to_end(project): + """`python -m festival4d features` dispatches into the real implementation now.""" + from festival4d import cli + + assert cli.main(["features"]) == 0 + assert project.BEATS_JSON.exists() + project.BEATS_JSON.unlink() diff --git a/backend/tests/test_phase5_api.py b/backend/tests/test_phase5_api.py index 92f630b..8cbfafe 100644 --- a/backend/tests/test_phase5_api.py +++ b/backend/tests/test_phase5_api.py @@ -52,13 +52,15 @@ def test_beats_404_then_serves_file(client): config.BEATS_JSON.unlink() -def test_director_degrades_to_valid_empty_campath(client): - # While lane E is stubbed the response must still round-trip through camPath.fromJSON: - # a valid version-1 path with zero keyframes, plus a "note" marking the degradation. +def test_director_returns_valid_campath(client): + # CR-4: lane E's M11 landed, so on the synthetic fixture (events + poses present) the + # response is a REAL path — assert the frozen camPath shape (holds stubbed or implemented). + # The empty+note degradation (no events / no poses) is covered in test_director.py. data = client.post("/api/director", json={}).json() assert data["version"] == 1 - assert data["keyframes"] == [] - assert "note" in data + assert isinstance(data["keyframes"], list) + for kf in data["keyframes"]: + assert set(kf) == {"t_global", "pos", "quat", "fov"} assert client.post("/api/director", json={"top_n": 3, "lead_s": 1.0}).status_code == 200 @@ -99,9 +101,11 @@ def test_anchor_patch_label_and_color(client): def test_cli_dispatches_stubs_gracefully(): - # The three new subcommands exist and degrade per the house pattern: exit code 2, - # no traceback, while their lane entrypoints are stubs. + # The still-stubbed subcommands degrade per the house pattern: exit code 2, no traceback. + # CR-4: `features` and `direct` left this list when lane E landed M10/M11 (their real + # dispatch is covered in test_audio_features.py / test_director.py); lane G drops + # `capsule` when M17 lands. from festival4d import cli - for cmd in (["features"], ["direct"], ["capsule"]): + for cmd in (["capsule"],): assert cli.main(cmd) == 2 diff --git a/plan/CHANGE_REQUESTS.md b/plan/CHANGE_REQUESTS.md index 2f9ff9f..ca6b0ad 100644 --- a/plan/CHANGE_REQUESTS.md +++ b/plan/CHANGE_REQUESTS.md @@ -72,3 +72,24 @@ Instead append an entry here and work around it locally; the integration agent a pre-existing route/shape/column and the timebase + pose contracts are unchanged; the annotation response only gained a field. Suite 101 → **103** (added supersede + delete tests). - **Decision:** **APPLIED** (phase4, 2026-07-16). Fixes the sole Round-3 wart; strictly additive. + +### CR-4 — lane E — 2026-07-17 — Update stub-era assertions in test_phase5_api.py (M10/M11 landed) +- **File:** `backend/tests/test_phase5_api.py::test_cli_dispatches_stubs_gracefully` and + `::test_director_degrades_to_valid_empty_campath` +- **Problem:** Both tests assert the *stub* contracts of lane E's entrypoints, correct only while + M10/M11 were unimplemented. With `run_features`/`generate_path` landed: (a) + `cli.main(["features"])` / `["direct"]` no longer raise `NotImplementedError`, so they exit 0, + not 2; (b) `POST /api/director` on the synthetic fixture (which has events + poses) returns a + real non-empty camPath, so `keyframes == []` / `"note" in data` fail. Identical in kind to + **CR-1** (lane D, approved Round 1: "a test file, not a frozen contract file"). +- **Proposed change:** (a) shrink the CLI stub loop to `["capsule"]` (lane G, still stubbed — + lane G drops it when M17 lands); (b) rename the director test to + `test_director_returns_valid_campath` asserting the frozen camPath *shape* (version 1, + keyframe key-set), which holds in both stub and implemented states; the empty+note degradation + itself is now covered by lane E's own `test_director.py` (no-events / no-poses cases). +- **Workaround in place:** applied the minimal edit directly per the CR-1 precedent (test file, + not a frozen contract file; `api.py`/`cli.py` byte-for-byte unchanged). Suite green + (127 baseline + lane E tests, zero failures). If the coordinator prefers the stricter reading + of the freeze, revert the test edit and re-apply at integration2 — lane E's modules are + unaffected either way. +- **Decision:** (pending — coordinator/integration2)