festifun/backend/tests/test_audio_features.py

"""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()