"""Moment detection + pluggable AI classification (spec M7, lane D). 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. 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 the best-covering video and 6 sampled JPEGs for providers without video input. Common behavior: catch exceptions per candidate (one failure never aborts the batch); if the selected provider is unconfigured, run candidate detection only (``event_type='candidate'``) and log that classification was skipped. 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. EventType = Literal[ "bass_drop", "pyro", "confetti", "crowd_wave", "artist_moment", "light_show", "quiet_moment", "other", ] EVENT_TYPES: tuple[str, ...] = ( "bass_drop", "pyro", "confetti", "crowd_wave", "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).""" event_type: EventType confidence: float = Field(ge=0.0, le=1.0) # 0..1 description: str # one sentence, human-readable @runtime_checkable class MomentClassifier(Protocol): """Interchangeable classifier provider (FROZEN CONTRACT, spec M7).""" def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: """Classify one candidate moment from a short clip + sampled frames.""" ... _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 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 flash tier, native video input (``GEMINI_API_KEY``). Model is overridable via ``FESTIVAL4D_GEMINI_MODEL``. Default was ``gemini-2.5-flash`` until Google retired it for new API keys (404 "no longer available to new users", field-found 2026-07-16); ``gemini-3.1-flash-lite`` is the cheapest current model verified to accept inline video + JSON-schema structured output. """ def __init__(self) -> None: from google import genai self.model = os.environ.get("FESTIVAL4D_GEMINI_MODEL", "gemini-3.1-flash-lite") self._genai = genai self._client = genai.Client() # reads GEMINI_API_KEY from the environment def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: 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)`` -> ``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: import anthropic self._client = anthropic.Anthropic() # reads ANTHROPIC_API_KEY from the environment def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification: 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 (local models are sloppier about JSON). """ def __init__(self) -> None: 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: 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) or fails to initialize — callers then run candidates-only. """ 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 (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 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). """ 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 + 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``. 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. """ 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