festifun/backend/festival4d/audio_features.py

"""Audio features: beats & onsets (spec M10, lane E).

Contract (frozen at foundation2 merge — phase-5 contract #1):

:func:`run_features` loads the **reference** video's ingest WAV (16 kHz mono from
``config.AUDIO_DIR``; the reference timeline *is* ``t_global``), runs beat/tempo tracking
and onset-strength detection (librosa is already a dependency), and writes
``config.BEATS_JSON``::

    {"tempo_bpm": float | null,
     "beats_s": [float, ...],
     "onsets": [{"t_global_s": float, "strength": float 0..1}, ...],
     "generated_by": "festival4d.audio_features"}

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="<i2").astype(np.float64) / 32767.0
    if channels > 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)."""
    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),
    }