diff --git a/backend/festival4d/events_ai.py b/backend/festival4d/events_ai.py index 47e0e23..57b6b84 100644 --- a/backend/festival4d/events_ai.py +++ b/backend/festival4d/events_ai.py @@ -4,9 +4,9 @@ Two stages: cheap audio candidates first, then a vision LLM on candidates only. This module defines the **FROZEN classifier contract** — :class:`MomentClassification` (Pydantic model) and :class:`MomentClassifier` (Protocol) — exactly as spec M7. Lanes and -integration depend on these shapes. Everything else here is a STUB that lane D fills: -candidate detection, the provider classes, provider selection, and the ``run_events`` -entrypoint (called by both ``cli.py`` and ``api.py``'s ``POST /api/events/detect``). +integration depend on these shapes. Lane D fills the rest: candidate detection, the provider +classes, provider selection, and the ``run_events`` entrypoint (called by both ``cli.py`` and +``api.py``'s ``POST /api/events/detect``). Provider selection (lane D): ``FESTIVAL4D_CLASSIFIER=gemini|claude|local`` (default ``gemini``). Inputs prepared once per candidate: a ~3 s MP4 clip around ``t_global`` from @@ -15,19 +15,31 @@ behavior: catch exceptions per candidate (one failure never aborts the batch); i selected provider is unconfigured, run candidate detection only (``event_type='candidate'``) and log that classification was skipped. -The module imports with only ``pydantic`` present — provider SDKs (``google-genai``, -``anthropic``, ``openai``) are imported lazily inside the provider bodies, never at load. +The module imports with only ``pydantic`` present — heavy numeric libs (``numpy``, +``librosa``, ``scipy``, ``soundfile``) and provider SDKs (``google-genai``, ``anthropic``, +``openai``) are imported lazily inside the functions that need them, never at load. + +DB note (lane D): ``db.py`` (frozen) exposes ``add_event`` and ``clear_events`` but *no* +update-event helper, so instead of "insert audio_auto candidate, later update to ai" this +module inserts each event **once in its final state** — ``source='ai'`` with the classified +type when a provider succeeds, else ``source='audio_auto'`` with ``event_type='candidate'``. +The end state in the DB is identical to the spec's two-step description. """ from __future__ import annotations +import base64 import logging import os +import shutil +import subprocess from pathlib import Path from typing import Literal, Protocol, runtime_checkable from pydantic import BaseModel, Field +from festival4d import config, db + log = logging.getLogger("festival4d.events_ai") # The closed set of moment classes (spec M7). Lane C colors timeline markers by these. @@ -40,6 +52,25 @@ EVENT_TYPES: tuple[str, ...] = ( "artist_moment", "light_show", "quiet_moment", "other", ) +# --------------------------------------------------------------------------- +# Tuning constants (lane D). Detection is verified against the synthetic fixture's +# ground-truth pulse times (spec M0/M7); see backend/tests/test_events_ai.py. +# --------------------------------------------------------------------------- +STFT_HOP = 512 # librosa frame hop (samples) +STFT_FRAME = 2048 # RMS window length (samples) +NEIGHBORHOOD_S = 10.0 # a candidate must be the max over this centered window +PEAK_PERCENTILE = 90.0 # ...and above this percentile of the detection function +MERGE_S = 1.0 # merge peaks closer than this, keeping the strongest +MIN_SIGNAL_STD = 1e-3 # AC-RMS floor (~−60 dBFS): silence AND constant/DC have ~0 std +CANDIDATE_DURATION_S = 1.0 # stored event duration (a moment, not an interval) +DEFAULT_WINDOW_S = 6.0 # window width when run_events is given only t_global_s + +CLIP_SECONDS = 3.0 # length of the classifier clip (spec M7: ~3 s) +NUM_FRAMES = 6 # sampled JPEGs for providers without video input +FRAME_LONG_EDGE = 768 # max long-edge px for sampled frames (spec M7: ≤768) +CLIP_MAX_HEIGHT = 720 # clip capped at ≤720p (spec M7) +AUDIO_EXTRACT_SR = 48_000 # sample rate when extracting reference audio via ffmpeg + class MomentClassification(BaseModel): """Structured classifier output (FROZEN CONTRACT, spec M7).""" @@ -58,8 +89,17 @@ class MomentClassifier(Protocol): ... +_CLASSIFY_PROMPT = ( + "This is a ~3-second clip from a concert, captured around a detected audio spike. " + "Classify the single most salient moment in it. Choose event_type from: " + "bass_drop, pyro, confetti, crowd_wave, artist_moment, light_show, quiet_moment, other. " + "Give a confidence in [0, 1] and one short human-readable sentence describing what happens." +) + + # --------------------------------------------------------------------------- -# Provider stubs (lane D fills bodies; classifiers import their SDK lazily). +# Provider classes (lane D). Each imports its SDK lazily so the module loads with +# only pydantic present; construction reads the key/endpoint from the environment. # --------------------------------------------------------------------------- class GeminiClassifier: """Default provider — Gemini 2.5 Flash, native video input (``GEMINI_API_KEY``).""" @@ -67,78 +107,439 @@ class GeminiClassifier: model = "gemini-2.5-flash" def __init__(self) -> None: - raise NotImplementedError("lane D (M7): implement GeminiClassifier") + from google import genai + + self._genai = genai + self._client = genai.Client() # reads GEMINI_API_KEY from the environment def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: - raise NotImplementedError("lane D (M7): implement GeminiClassifier.classify") + genai = self._genai + clip_bytes = Path(clip_path).read_bytes() + resp = self._client.models.generate_content( + model=self.model, + contents=[ + genai.types.Part.from_bytes(data=clip_bytes, mime_type="video/mp4"), + _CLASSIFY_PROMPT, + ], + config={ + "response_mime_type": "application/json", + "response_schema": MomentClassification, + }, + ) + return MomentClassification.model_validate_json(resp.text) class ClaudeClassifier: """Optional provider — Claude Opus 4.8, 6 frames as images (``ANTHROPIC_API_KEY``). Uses ``client.messages.parse(..., thinking={'type': 'adaptive'}, - output_format=MomentClassification)``; do **not** pass ``temperature`` (rejected on - Opus 4.8). Model id ``claude-opus-4-8`` (never date-suffixed). + output_format=MomentClassification)`` -> ``resp.parsed_output``; does **not** pass + ``temperature`` (rejected on Opus 4.8). Model id ``claude-opus-4-8`` (never date-suffixed). """ model = "claude-opus-4-8" def __init__(self) -> None: - raise NotImplementedError("lane D (M7): implement ClaudeClassifier") + import anthropic + + self._client = anthropic.Anthropic() # reads ANTHROPIC_API_KEY from the environment def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: - raise NotImplementedError("lane D (M7): implement ClaudeClassifier.classify") + content: list[dict] = [] + for fp in frames[:NUM_FRAMES]: + data = base64.standard_b64encode(Path(fp).read_bytes()).decode("ascii") + content.append({ + "type": "image", + "source": {"type": "base64", "media_type": "image/jpeg", "data": data}, + }) + content.append({"type": "text", "text": _CLASSIFY_PROMPT}) + resp = self._client.messages.parse( + model=self.model, + max_tokens=1024, + thinking={"type": "adaptive"}, + output_format=MomentClassification, + messages=[{"role": "user", "content": content}], + ) + return resp.parsed_output class LocalClassifier: """Offline/free provider — any OpenAI-compatible vision endpoint (Ollama/LM Studio/ OpenRouter). Config: ``FESTIVAL4D_OPENAI_BASE_URL``, ``FESTIVAL4D_OPENAI_KEY``, ``FESTIVAL4D_OPENAI_MODEL``. Sends the 6 frames as image parts; validates JSON with one - retry on failure. + retry on failure (local models are sloppier about JSON). """ def __init__(self) -> None: - raise NotImplementedError("lane D (M7): implement LocalClassifier") + from openai import OpenAI + + base_url = os.environ["FESTIVAL4D_OPENAI_BASE_URL"] + # Local servers (Ollama/LM Studio) accept any non-empty key; OpenRouter needs a real one. + api_key = os.environ.get("FESTIVAL4D_OPENAI_KEY") or "not-needed" + self._model = os.environ["FESTIVAL4D_OPENAI_MODEL"] + self._client = OpenAI(base_url=base_url, api_key=api_key) def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: - raise NotImplementedError("lane D (M7): implement LocalClassifier.classify") + from pydantic import ValidationError + + schema = MomentClassification.model_json_schema() + content: list[dict] = [{ + "type": "text", + "text": ( + _CLASSIFY_PROMPT + + " Respond with ONLY a JSON object matching this schema: " + + repr(schema) + ), + }] + for fp in frames[:NUM_FRAMES]: + b64 = base64.standard_b64encode(Path(fp).read_bytes()).decode("ascii") + content.append({ + "type": "image_url", + "image_url": {"url": f"data:image/jpeg;base64,{b64}"}, + }) + messages = [{"role": "user", "content": content}] + + last_exc: Exception | None = None + for attempt in range(2): # one retry on validation failure + resp = self._client.chat.completions.create( + model=self._model, + messages=messages, + response_format={"type": "json_object"}, + ) + text = resp.choices[0].message.content or "" + try: + return MomentClassification.model_validate_json(text) + except ValidationError as exc: + last_exc = exc + log.warning("events: local classifier returned invalid JSON (attempt %d)", + attempt + 1) + assert last_exc is not None + raise last_exc + + +_PROVIDERS: dict[str, tuple[type, tuple[str, ...]]] = { + # name -> (class, required environment variables) + "gemini": (GeminiClassifier, ("GEMINI_API_KEY",)), + "claude": (ClaudeClassifier, ("ANTHROPIC_API_KEY",)), + "local": (LocalClassifier, ("FESTIVAL4D_OPENAI_BASE_URL", "FESTIVAL4D_OPENAI_MODEL")), +} def get_classifier(name: str | None = None) -> MomentClassifier | None: """Select a provider by name or ``FESTIVAL4D_CLASSIFIER`` (default ``gemini``). Returns a ready classifier, or ``None`` if the selected provider is unconfigured (no - key/endpoint) — callers then run candidates-only. STUB — lane D implements selection. + key/endpoint) or fails to initialize — callers then run candidates-only. """ - _ = name or os.environ.get("FESTIVAL4D_CLASSIFIER", "gemini") - raise NotImplementedError("lane D (M7): implement get_classifier") + name = (name or os.environ.get("FESTIVAL4D_CLASSIFIER") or "gemini").lower() + entry = _PROVIDERS.get(name) + if entry is None: + log.warning("events: unknown classifier %r; running candidates-only", name) + return None + cls, required = entry + missing = [var for var in required if not os.environ.get(var)] + if missing: + log.info("events: provider %r unconfigured (missing %s); classification skipped, " + "storing candidates only", name, ", ".join(missing)) + return None + try: + return cls() + except Exception: # SDK import/construction failure -> degrade, don't crash the batch + log.exception("events: failed to initialize %r classifier; running candidates-only", name) + return None # --------------------------------------------------------------------------- -# Candidate detection + orchestration (lane D fills bodies). +# Candidate detection (spec M7). Verified against the fixture ground-truth pulses. # --------------------------------------------------------------------------- def detect_candidates(audio: "object", sr: int) -> list[float]: """Audio candidate moments on the reference track (spec M7, lane D). - Compute RMS energy + spectral-flux onset strength (librosa); return the times - (reference-local seconds) of peaks that are local maxima over a 10 s neighborhood and - above the 90th percentile. Verified against the fixture's ground-truth pulse times - (±0.5 s). + Compute RMS energy and spectral-flux onset strength (librosa), combine into a single + detection function (product — bangs dominate in both, so the product isolates them + cleanly from the regular beat floor), and return the times (reference-local seconds) of + peaks that are local maxima over a ~10 s neighborhood and above the 90th percentile. + Verified against the fixture's ground-truth pulse times (±0.5 s). """ - raise NotImplementedError("lane D (M7): implement detect_candidates") + import librosa + import numpy as np + from scipy.ndimage import maximum_filter1d + + y = np.asarray(audio, dtype=np.float32).reshape(-1) + if y.size < STFT_FRAME: + return [] + # A reference track with no real moments (silence, constant/DC, or a bare noise floor) + # has ~0 AC energy. The percentile threshold below is purely *relative*, so on such a + # flat signal it would flag ~1 spurious peak per second; gate on AC-RMS (std) first — + # this catches silence and DC alike (peak amplitude does not: a DC offset has a large + # peak but no transients, only STFT-boundary artifacts). + if float(np.std(y)) < MIN_SIGNAL_STD: + return [] + + rms = librosa.feature.rms(y=y, hop_length=STFT_HOP, frame_length=STFT_FRAME)[0] + flux = librosa.onset.onset_strength(y=y, sr=sr, hop_length=STFT_HOP) + n = min(len(rms), len(flux)) + if n == 0: + return [] + rms, flux = rms[:n], flux[:n] + eps = 1e-12 + strength = (rms / (rms.max() + eps)) * (flux / (flux.max() + eps)) + # No transient structure at all (e.g. constant/DC audio -> flat detection function). + if float(strength.max()) <= eps: + return [] + + times = librosa.frames_to_time(np.arange(n), sr=sr, hop_length=STFT_HOP) + win = max(1, int(round(NEIGHBORHOOD_S * sr / STFT_HOP))) + if win % 2 == 0: + win += 1 + # A frame is a local maximum if it equals the max over its centered ~10 s window. Because + # the fixture's bangs are >5 s apart, each is the unique max in its own window — a + # min-distance peak picker would instead merge the 7 s-spaced bangs and drop one. + is_local_max = strength >= (maximum_filter1d(strength, size=win, mode="nearest") - eps) + threshold = float(np.percentile(strength, PEAK_PERCENTILE)) + idx = np.where(is_local_max & (strength >= threshold))[0] + + # Greedy merge of near-duplicate peaks (plateaus), keeping the strongest. + peaks: list[float] = [] + for i in sorted(idx.tolist(), key=lambda j: -strength[j]): + t = float(times[i]) + if all(abs(t - p) >= MERGE_S for p in peaks): + peaks.append(t) + return sorted(peaks) + + +# --------------------------------------------------------------------------- +# Reference-audio loading + classifier-input preparation (ffmpeg). +# --------------------------------------------------------------------------- +def _select_reference(videos: list) -> "object | None": + """The reference video (spec §2: offset 0). Fall back to the smallest |offset|.""" + 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 _load_reference_audio(video) -> "tuple[object, int] | None": + """Load the reference video's audio as mono float samples. + + Independent of lane A: extract the audio straight from the raw video with ffmpeg (a + reused ingest WAV would couple us to lane A's naming). Returns ``(samples, sr)`` or + ``None`` if the raw video or ffmpeg is unavailable. + """ + import tempfile + + import numpy as np + import soundfile as sf + + raw = config.RAW_DIR / video.filename + if not raw.exists(): + log.warning("events: raw video %s not found under %s", video.filename, config.RAW_DIR) + return None + if shutil.which("ffmpeg") is None: + log.warning("events: ffmpeg not on PATH; cannot extract reference audio") + return None + + with tempfile.TemporaryDirectory() as tmp: + wav = Path(tmp) / "ref.wav" + cmd = [ + "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-i", str(raw), "-vn", "-ac", "1", "-ar", str(AUDIO_EXTRACT_SR), + "-f", "wav", str(wav), + ] + try: + subprocess.run(cmd, check=True, capture_output=True) + except subprocess.CalledProcessError as exc: + log.warning("events: ffmpeg audio extraction failed: %s", exc.stderr.decode("utf-8", "ignore")) + return None + samples, sr = sf.read(str(wav), dtype="float32", always_2d=False) + if getattr(samples, "ndim", 1) > 1: + samples = samples.mean(axis=1) + return np.asarray(samples, dtype=np.float32), int(sr) + + +def _select_best_video(videos: list, t_global_s: float) -> "tuple[object, float] | None": + """The video whose 3 s window around ``t_global`` is best-covered (most centered).""" + best = None + best_score = None + best_t = 0.0 + for v in videos: + t_video = config.t_video_from_global(t_global_s, v.offset_ms or 0.0, v.drift_ppm or 0.0) + margin = min(t_video, (v.duration_s or 0.0) - t_video) # ≥ CLIP_SECONDS/2 => fully covers + if best_score is None or margin > best_score: + best_score, best, best_t = margin, v, t_video + if best is None: + return None + return best, best_t + + +def _events_workdir() -> Path: + d = config.WORK_DIR / "events" + d.mkdir(parents=True, exist_ok=True) + return d def prepare_inputs(t_global_s: float) -> tuple[Path, list[Path]]: - """Prepare classifier inputs for a candidate: a ~3 s MP4 clip + 6 sampled JPEGs (M7).""" - raise NotImplementedError("lane D (M7): implement prepare_inputs") + """Prepare classifier inputs for a candidate: a ~3 s MP4 clip + up to 6 sampled JPEGs (M7). + + Picks the best-covering video, then uses ffmpeg to cut a ≤720p clip (audio kept, for + Gemini's native-video path) and sample 6 frames (≤768 px long edge). Raises if ffmpeg + or a usable source video is missing — ``run_events`` isolates that per candidate. + """ + if shutil.which("ffmpeg") is None: + raise RuntimeError("ffmpeg not found on PATH — required to prepare classifier inputs") + + videos = db.get_videos() + picked = _select_best_video(videos, t_global_s) + if picked is None: + raise RuntimeError("no videos available to prepare classifier inputs") + video, t_video = picked + raw = config.RAW_DIR / video.filename + if not raw.exists(): + raise RuntimeError(f"source video {raw} not found") + + duration = video.duration_s or (t_video + CLIP_SECONDS) + start = max(0.0, min(t_video - CLIP_SECONDS / 2.0, max(0.0, duration - CLIP_SECONDS))) + + workdir = _events_workdir() + tag = f"cand_{t_global_s:0.3f}".replace("-", "m") + clip_path = workdir / f"{tag}.mp4" + frames_dir = workdir / tag + if frames_dir.exists(): + for old in frames_dir.glob("frame_*.jpg"): + old.unlink() + frames_dir.mkdir(parents=True, exist_ok=True) + + # 3 s clip, downscaled only if taller than 720p, audio preserved, faststart for inline bytes. + subprocess.run( + ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-ss", f"{start:.3f}", "-i", str(raw), "-t", f"{CLIP_SECONDS:.3f}", + "-vf", f"scale=-2:'min({CLIP_MAX_HEIGHT},ih)'", + "-c:v", "libx264", "-preset", "veryfast", "-crf", "28", "-pix_fmt", "yuv420p", + "-c:a", "aac", "-b:a", "96k", "-movflags", "+faststart", str(clip_path)], + check=True, capture_output=True, + ) + # 6 frames evenly across the 3 s window, long edge ≤ 768 px. + scale = (f"scale='if(gt(iw,ih),min({FRAME_LONG_EDGE},iw),-2)':" + f"'if(gt(iw,ih),-2,min({FRAME_LONG_EDGE},ih))'") + fps = NUM_FRAMES / CLIP_SECONDS + subprocess.run( + ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-ss", f"{start:.3f}", "-i", str(raw), "-t", f"{CLIP_SECONDS:.3f}", + "-vf", f"fps={fps},{scale}", "-frames:v", str(NUM_FRAMES), + str(frames_dir / "frame_%02d.jpg")], + check=True, capture_output=True, + ) + frames = sorted(frames_dir.glob("frame_*.jpg")) + return clip_path, frames + + +# --------------------------------------------------------------------------- +# Orchestration (spec M7). Entrypoint for the CLI + POST /api/events/detect. +# --------------------------------------------------------------------------- +def _classify_one(t_global_s: float, classifier: "MomentClassifier | None") -> tuple[str, str, float | None, str | None]: + """Resolve one candidate to (event_type, source, confidence, description). + + No classifier -> a bare ``audio_auto`` candidate. Otherwise classify, isolating any + failure (SDK error, bad clip) by falling back to a candidate — one failure never aborts + the batch (spec M7). + """ + if classifier is None: + return "candidate", "audio_auto", None, None + try: + clip_path, frames = prepare_inputs(t_global_s) + result = classifier.classify(clip_path, frames) + return result.event_type, "ai", result.confidence, result.description + except Exception: + log.exception("events: classification failed at t_global=%.3f s; storing as candidate", + t_global_s) + return "candidate", "audio_auto", None, None def run_events(t_global_s: float | None = None, window_s: float | None = None) -> dict: """Detect audio candidates and (if a provider is configured) classify them. - Entrypoint for ``python -m festival4d events`` and ``POST /api/events/detect``. Inserts - events via ``db`` (``source='audio_auto'``; classified ones updated to ``source='ai'``). - Optionally restrict to a single window around ``t_global_s`` of width ``window_s``. - Returns a summary dict. Per-candidate exceptions are isolated (spec M7). + Entrypoint for ``python -m festival4d events`` and ``POST /api/events/detect``. Detects + on the reference track, maps candidate times to ``t_global``, then inserts events via + ``db`` (``source='audio_auto'`` for bare candidates, ``source='ai'`` for classified ones; + see the module docstring on why this is a single insert per event). Per-candidate + exceptions are isolated. Returns a summary dict. + + Whole-track detection (``t_global_s is None``) is authoritative: it first clears prior + machine-generated events (``audio_auto`` + ``ai``) while preserving user edits. A windowed + call (``t_global_s`` given, optional ``window_s``) only *adds* candidates inside the + window and clears nothing, so it never disturbs the rest of the timeline. """ - raise NotImplementedError("lane D (M7): implement run_events") + db.init_engine() + db.init_db() + + selected = (os.environ.get("FESTIVAL4D_CLASSIFIER") or "gemini").lower() + videos = db.get_videos() + reference = _select_reference(videos) + if reference is None: + log.warning("events: no videos registered; run `synthetic` or `ingest` first") + return _summary(selected, False, 0, 0, 0, None, note="no videos registered") + + loaded = _load_reference_audio(reference) + if loaded is None: + return _summary(selected, False, 0, 0, 0, None, + note="no reference audio (missing raw video or ffmpeg)") + audio, sr = loaded + + local_times = detect_candidates(audio, sr) + off = reference.offset_ms or 0.0 + drift = reference.drift_ppm or 0.0 + cand_globals = [config.t_global_from_video(t, off, drift) for t in local_times] + + window = None + if t_global_s is not None: + w = window_s if window_s is not None else DEFAULT_WINDOW_S + lo, hi = t_global_s - w / 2.0, t_global_s + w / 2.0 + cand_globals = [t for t in cand_globals if lo <= t <= hi] + window = {"t_global_s": t_global_s, "window_s": w} + else: + # Whole-track detection replaces the machine timeline; user events are kept. + db.clear_events("audio_auto") + db.clear_events("ai") + + classifier = get_classifier(selected) + n_ai = n_auto = 0 + for t_global in cand_globals: + event_type, source, confidence, description = _classify_one(t_global, classifier) + db.add_event( + t_global_s=t_global, + event_type=event_type, + source=source, + duration_s=CANDIDATE_DURATION_S, + confidence=confidence, + description=description, + ) + if source == "ai": + n_ai += 1 + else: + n_auto += 1 + + summary = _summary(selected, classifier is not None, len(cand_globals), n_ai, n_auto, window) + log.info("events: reference=%s candidates=%d classified=%d candidates_only=%d provider=%s", + reference.filename, len(cand_globals), n_ai, n_auto, + selected if classifier is not None else f"{selected} (unconfigured)") + return summary + + +def _summary(provider: str, configured: bool, candidates: int, classified: int, + candidates_only: int, window: dict | None, note: str | None = None) -> dict: + out = { + "provider": provider, + "classifier_configured": configured, + "candidates": candidates, + "classified": classified, + "candidates_only": candidates_only, + "window": window, + } + if note is not None: + out["note"] = note + return out diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index baf5d3a..2aea65e 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -91,10 +91,16 @@ def test_cors_allows_vite_origin(client): assert resp.headers.get("access-control-allow-origin") == "http://localhost:5173" -def test_detect_degrades_gracefully(client): - data = client.post("/api/events/detect", json={}).json() - assert "note" in data and "events" in data # stubbed lane D -> 200 with note - assert client.post("/api/events/detect", json={}).status_code == 200 +def test_detect_events_endpoint(client): + # Lane D (M7) has landed: POST /api/events/detect runs real detection and returns + # {result, events} (no more stub "note"). result is the run_events summary. + resp = client.post("/api/events/detect", json={}) + assert resp.status_code == 200 + data = resp.json() + assert "result" in data and "events" in data + assert {"candidates", "classified", "candidates_only", "classifier_configured"} <= set(data["result"]) + assert isinstance(data["events"], list) + assert client.post("/api/events/detect", json={}).status_code == 200 # idempotent def test_annotation_stored(client): diff --git a/backend/tests/test_events_ai.py b/backend/tests/test_events_ai.py new file mode 100644 index 0000000..bb845c5 --- /dev/null +++ b/backend/tests/test_events_ai.py @@ -0,0 +1,288 @@ +"""Tests for events_ai.py (lane D / spec M7). + +No test hits a real API or ffmpeg: candidate detection runs on the synthetic fixture's +in-memory audio, and the classifier orchestration is exercised with stub classifiers and a +monkeypatched input-preparer. This mirrors spec M7's acceptance: + - candidates land within ±0.5 s of the ground-truth pulse times, and + - the two-stage loop (detect -> classify) works independent of any provider. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from festival4d import config, db, events_ai, synthetic +from festival4d.events_ai import ( + ClaudeClassifier, + GeminiClassifier, + LocalClassifier, + MomentClassification, + MomentClassifier, + detect_candidates, + get_classifier, + run_events, +) + +SR = config.SYNTH_AUDIO_SR + + +# --------------------------------------------------------------------------- +# Helpers / stub classifiers (no network) +# --------------------------------------------------------------------------- +def _reference_audio(): + """cam0's audio: the master signal sliced at offset 0 (local == global time).""" + master, gt = synthetic.synth_master_audio(SR) + audio = synthetic.slice_for_offset(master, SR, 0.0, config.SYNTH_DURATION_S) + return audio, gt["pulse_times_global_s"] + + +class StubClassifier: + """Always returns the same valid classification; counts calls.""" + + def __init__(self, event_type: str = "bass_drop") -> None: + self.calls = 0 + self.event_type = event_type + + def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: + self.calls += 1 + return MomentClassification( + event_type=self.event_type, confidence=0.9, description="stub classification" + ) + + +class FlakyClassifier: + """Raises on a chosen call to exercise per-candidate exception isolation.""" + + def __init__(self, fail_on_call: int = 2) -> None: + self.calls = 0 + self.fail_on_call = fail_on_call + + def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: + self.calls += 1 + if self.calls == self.fail_on_call: + raise RuntimeError("simulated provider failure") + return MomentClassification(event_type="pyro", confidence=0.6, description="ok") + + +def _fresh_reference_db(offset_ms: float = 0.0) -> None: + """A clean DB with a single reference video (matching cam0).""" + db.init_engine() + db.reset_db() + db.add_video( + filename="cam0.mp4", duration_s=config.SYNTH_DURATION_S, fps=config.SYNTH_FPS, + width=config.SYNTH_VIDEO_W, height=config.SYNTH_VIDEO_H, + offset_ms=offset_ms, drift_ppm=0.0, sync_confidence=1.0, + ) + + +def _stub_run(monkeypatch, classifier): + """Wire run_events to run real detection on fixture audio + a stub classifier.""" + audio, _ = _reference_audio() + monkeypatch.setattr(events_ai, "_load_reference_audio", lambda video: (audio, SR)) + monkeypatch.setattr(events_ai, "get_classifier", lambda name=None: classifier) + monkeypatch.setattr( + events_ai, "prepare_inputs", + lambda t: (Path("/nonexistent/clip.mp4"), [Path("/nonexistent/frame_00.jpg")]), + ) + + +# --------------------------------------------------------------------------- +# Candidate detection (the acceptance-critical numeric test) +# --------------------------------------------------------------------------- +def test_detect_candidates_matches_ground_truth(): + audio, pulses = _reference_audio() + cands = detect_candidates(audio, SR) + + # Recall: every ground-truth pulse has a candidate within ±0.5 s. + for pulse in pulses: + nearest = min(abs(c - pulse) for c in cands) + assert nearest < 0.5, f"pulse {pulse}s unmatched; nearest candidate {nearest:.3f}s away" + + # Precision: no spurious candidates — each maps to a distinct ground-truth pulse. + for c in cands: + nearest = min(abs(c - pulse) for pulse in pulses) + assert nearest < 0.5, f"spurious candidate at {c:.3f}s ({nearest:.3f}s from any pulse)" + assert len(cands) == len(pulses), (cands, pulses) + + +def test_detect_candidates_short_audio_is_empty(): + import numpy as np + + assert detect_candidates(np.zeros(16, dtype=np.float32), SR) == [] + + +def test_detect_candidates_silent_or_flat_audio_is_empty(): + """Silence / DC / a bare noise floor must yield NO candidates, never fabricate events.""" + import numpy as np + + n = SR * 6 + assert detect_candidates(np.zeros(n, dtype=np.float32), SR) == [] # digital silence + assert detect_candidates(np.full(n, 0.5, dtype=np.float32), SR) == [] # constant / DC + rng = np.random.default_rng(0) + assert detect_candidates((1e-4 * rng.standard_normal(n)).astype(np.float32), SR) == [] # noise floor + + +# --------------------------------------------------------------------------- +# Frozen contract sanity (MomentClassification / MomentClassifier) +# --------------------------------------------------------------------------- +def test_moment_classification_validates_bounds_and_enum(): + ok = MomentClassification(event_type="pyro", confidence=0.5, description="x") + assert ok.event_type == "pyro" + with pytest.raises(ValidationError): + MomentClassification(event_type="pyro", confidence=1.5, description="x") # >1 + with pytest.raises(ValidationError): + MomentClassification(event_type="not_a_type", confidence=0.5, description="x") + + +def test_stub_satisfies_classifier_protocol(): + assert isinstance(StubClassifier(), MomentClassifier) + + +# --------------------------------------------------------------------------- +# Provider selection + degradation (no network — constructors don't call out) +# --------------------------------------------------------------------------- +def test_get_classifier_unconfigured_returns_none(monkeypatch): + for var in ("GEMINI_API_KEY", "ANTHROPIC_API_KEY", + "FESTIVAL4D_OPENAI_BASE_URL", "FESTIVAL4D_OPENAI_MODEL", "FESTIVAL4D_OPENAI_KEY"): + monkeypatch.delenv(var, raising=False) + assert get_classifier("gemini") is None + assert get_classifier("claude") is None + assert get_classifier("local") is None + + +def test_get_classifier_unknown_provider_returns_none(): + assert get_classifier("does-not-exist") is None + + +def test_get_classifier_defaults_to_gemini(monkeypatch): + monkeypatch.delenv("FESTIVAL4D_CLASSIFIER", raising=False) + monkeypatch.setenv("GEMINI_API_KEY", "test-key") + assert isinstance(get_classifier(), GeminiClassifier) + + +def test_get_classifier_local_when_configured(monkeypatch): + monkeypatch.setenv("FESTIVAL4D_OPENAI_BASE_URL", "http://localhost:11434/v1") + monkeypatch.setenv("FESTIVAL4D_OPENAI_MODEL", "qwen2.5-vl") + monkeypatch.delenv("FESTIVAL4D_OPENAI_KEY", raising=False) + clf = get_classifier("local") + assert isinstance(clf, LocalClassifier) + assert isinstance(clf, MomentClassifier) + + +def test_get_classifier_claude_when_configured(monkeypatch): + monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key") + assert isinstance(get_classifier("claude"), ClaudeClassifier) + + +# --------------------------------------------------------------------------- +# Orchestration loop (stub classifier — no API, no ffmpeg) +# --------------------------------------------------------------------------- +def test_run_events_classifies_all_candidates(monkeypatch): + _fresh_reference_db() + stub = StubClassifier(event_type="bass_drop") + _stub_run(monkeypatch, stub) + + result = run_events() + + assert result["candidates"] == 3 + assert result["classified"] == 3 + assert result["candidates_only"] == 0 + assert result["classifier_configured"] is True + assert stub.calls == 3 + + events = db.get_events() + assert len(events) == 3 + assert all(e.source == "ai" for e in events) + assert all(e.event_type == "bass_drop" for e in events) + # Candidate times map to t_global (reference offset 0) near the fixture pulses. + times = sorted(e.t_global_s for e in events) + for got, pulse in zip(times, (3.0, 10.0, 17.0)): + assert abs(got - pulse) < 0.5 + + +def test_run_events_unconfigured_stores_candidates_only(monkeypatch): + _fresh_reference_db() + audio, _ = _reference_audio() + monkeypatch.setattr(events_ai, "_load_reference_audio", lambda video: (audio, SR)) + monkeypatch.setattr(events_ai, "get_classifier", lambda name=None: None) + # Degraded path must never prepare inputs. A *raised* tripwire would be swallowed by + # _classify_one's `except Exception`; record calls with a side effect instead. + prepared: list = [] + monkeypatch.setattr(events_ai, "prepare_inputs", lambda t: prepared.append(t)) + + result = run_events() + + assert prepared == [] # prepare_inputs was never called when unconfigured + + assert result["candidates"] == 3 + assert result["classified"] == 0 + assert result["candidates_only"] == 3 + assert result["classifier_configured"] is False + + events = db.get_events() + assert len(events) == 3 + assert all(e.source == "audio_auto" for e in events) + assert all(e.event_type == "candidate" for e in events) + + +def test_run_events_isolates_per_candidate_failure(monkeypatch): + _fresh_reference_db() + flaky = FlakyClassifier(fail_on_call=2) # 2nd candidate fails + _stub_run(monkeypatch, flaky) + + result = run_events() + + assert result["candidates"] == 3 + assert result["classified"] == 2 # two succeeded + assert result["candidates_only"] == 1 # the failed one fell back to a candidate + assert flaky.calls == 3 # the batch was not aborted + + events = db.get_events() + assert sum(e.source == "ai" for e in events) == 2 + assert sum(e.source == "audio_auto" for e in events) == 1 + + +def test_run_events_whole_track_replaces_machine_events_keeps_user(monkeypatch): + _fresh_reference_db() + # Pre-existing events: a stale AI one and a user correction. + db.add_event(t_global_s=99.0, event_type="other", source="ai", description="stale") + db.add_event(t_global_s=5.0, event_type="crowd_wave", source="user", description="mine") + _stub_run(monkeypatch, StubClassifier()) + + run_events() + + events = db.get_events() + user = [e for e in events if e.source == "user"] + ai = [e for e in events if e.source == "ai"] + assert len(user) == 1 and user[0].description == "mine" # user edit preserved + assert len(ai) == 3 # fresh detection, stale AI gone + assert not any(e.description == "stale" for e in events) + + +def test_run_events_windowed_adds_in_window_and_clears_nothing(monkeypatch): + _fresh_reference_db() + # A machine event OUTSIDE the window must survive — windowed detection is purely additive. + db.add_event(t_global_s=3.0, event_type="bass_drop", source="ai", description="pre-existing") + _stub_run(monkeypatch, StubClassifier()) + + result = run_events(t_global_s=10.0, window_s=4.0) # window [8, 12] -> only the ~10 s pulse + + assert result["candidates"] == 1 + assert result["window"] == {"t_global_s": 10.0, "window_s": 4.0} + events = db.get_events() + assert any(e.description == "pre-existing" for e in events) # out-of-window event preserved + assert len(events) == 2 + new = [e for e in events if e.description != "pre-existing"] + assert len(new) == 1 and abs(new[0].t_global_s - 10.0) < 0.5 + + +def test_run_events_no_videos_degrades(monkeypatch): + db.init_engine() + db.reset_db() + result = run_events() + assert result["candidates"] == 0 + assert "note" in result + assert db.get_events() == [] diff --git a/plan/CHANGE_REQUESTS.md b/plan/CHANGE_REQUESTS.md index d40d429..10c583e 100644 --- a/plan/CHANGE_REQUESTS.md +++ b/plan/CHANGE_REQUESTS.md @@ -18,3 +18,18 @@ Instead append an entry here and work around it locally; the integration agent a --- + +### CR-1 — lane D — 2026-07-16 — Update obsolete stub assertion in test_api.py +- **File:** `backend/tests/test_api.py::test_detect_degrades_gracefully` +- **Problem:** That foundation test asserted the *stub* contract of `POST /api/events/detect` + (`"note" in data`), which was correct only while lane D was unimplemented. With M7 landed, + `run_events` no longer raises `NotImplementedError`, so `api.py` returns the real + `{result, events}` shape (spec M3) and the stub assertion fails. +- **Proposed change:** rename to `test_detect_events_endpoint` and assert the real landed + contract (200; `result` is the run_events summary with candidates/classified/… keys; `events` + is a list; idempotent on repeat). +- **Workaround in place:** applied the minimal edit directly (test file, not a frozen contract + file). The frozen `api.py` app is **unchanged**. Only lane D's owned modules + this obsolete + stub assertion were touched. Full suite green (89 passed). +- **Decision:** (integration) — informational; the change is a direct, unavoidable consequence + of lane D landing. Confirm the assertion matches the intended M3 detect-response shape. diff --git a/plan/status/lane-D.md b/plan/status/lane-D.md new file mode 100644 index 0000000..05899b6 --- /dev/null +++ b/plan/status/lane-D.md @@ -0,0 +1,59 @@ +# Status — lane-D + +## Round 1 — 2026-07-16 — STATUS: ready_to_merge + +**Adversarial review (self-run, 5 lenses → verify):** 3 findings, all confirmed & fixed — +(1) `detect_candidates` fabricated ~1 candidate/s on silent/DC/noise audio → added an AC-RMS +(std) floor `MIN_SIGNAL_STD`; silence/DC/noise now return `[]`, fixture unchanged [3,10,17]; +(2) degradation test's raised tripwire was swallowed by `except Exception` → switched to a +call-recorder (`assert prepared == []`); (3) windowed test ran on a fresh DB → now seeds an +out-of-window `ai` event and asserts it survives. `pytest` **90 passed** after fixes. + + +**Directives acknowledged:** round 0 of plan/DIRECTIVES.md (foundation merged; lanes may start; +lane D is small + fully independent; merge when acceptance passes — no waiting on siblings). + +**Acceptance checklist** (plan/lane-D-events.md + spec M7) — implementation complete, verified: +- [x] `detect_candidates` finds candidates within ±0.5 s of fixture ground-truth pulses + [3.0, 10.0, 17.0]. Evidence: `test_events_ai.py::test_detect_candidates_matches_ground_truth` + (recall+precision+count). RMS×spectral-flux, centered-10 s-window local-max, ≥ P90. +- [x] `MomentClassification` + `MomentClassifier` frozen contract preserved verbatim; import + of events_ai pulls in ZERO heavy libs/SDKs (all lazy). Evidence: import-cost probe → `[]`. +- [x] Three providers implemented against the installed SDKs (verified surfaces): Gemini + (google-genai 2.11.0, `Part.from_bytes(data,mime_type)` + `generate_content` structured + output), Claude (anthropic 0.116.0, `messages.parse(output_format=…, thinking=adaptive, + max_tokens)` → `parsed_output`, NO temperature, model `claude-opus-4-8`), Local + (openai 2.45.0, `chat.completions.create` json_object, 6 image_url parts, 1 retry). +- [x] Provider selection `FESTIVAL4D_CLASSIFIER=gemini|claude|local` (default gemini); + unconfigured → candidates-only + log line. Evidence: `test_get_classifier_*` (5 tests). +- [x] Per-candidate exception isolation. Evidence: `test_run_events_isolates_per_candidate_failure` + (2 ai + 1 fallback candidate; batch of 3 completes despite one raising). +- [x] `run_events` fills the frozen entrypoint (cli.py + api.py dispatch); writes via db helpers + only (insert-once in final state — db.py has no update-event helper; end state matches spec). +- [x] `pytest` green with NO network: **89 passed** across 5 consecutive full-suite runs. +- [x] Real-path end-to-end (ffmpeg-decoded AAC audio, not in-memory): isolated 20 s fixture → + `run_events()` degraded → candidates at 3.008/10.016/17.013 s (≤0.016 s error), exactly 3, + no spurious. Evidence: scratchpad/e2e_verify.py → RESULT PASS. + +**Done this round:** implemented `backend/festival4d/events_ai.py` (candidate detection + 3 +providers + get_classifier + prepare_inputs + run_events orchestration) and +`backend/tests/test_events_ai.py` (15 tests). Detection algorithm locked via a scratch +experiment before coding. Confirmed Claude call shape via claude-api skill and all three SDK +surfaces via introspection. Currently running a 5-lens adversarial review workflow +(spec-compliance / provider-SDK / detection-orchestration / contract-ownership / test-quality) +before merge. + +**Coordinator note (out-of-strict-ownership edit):** updated `backend/tests/test_api.py` +`test_detect_degrades_gracefully` → `test_detect_events_endpoint`. That foundation test asserted +the STUB behavior (`"note" in data`) of `POST /api/events/detect`; my landing replaces the stub +with real detection returning `{result, events}`. The frozen `api.py` app is UNCHANGED — only the +now-obsolete stub assertion was updated. Flagged here for transparency; see plan/CHANGE_REQUESTS.md CR-1. + +**Note on shared tree:** this working directory holds all four lanes' uncommitted work +concurrently (A/B/C/D). I will stage and commit ONLY my owned files. One transient +`test_audio_sync.py` (lane A) failure was observed once mid-run (concurrent write) but the suite +is deterministically green (89/89 ×5); not caused by lane D. + +**Blockers / questions for coordinator:** none. + +**Next:** address any confirmed review findings, then merge `lane/d-events` → main.