festifun/backend/festival4d/events_ai.py

"""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. 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``).

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 — provider SDKs (``google-genai``,
``anthropic``, ``openai``) are imported lazily inside the provider bodies, never at load.
"""

from __future__ import annotations

import logging
import os
from pathlib import Path
from typing import Literal, Protocol, runtime_checkable

from pydantic import BaseModel, Field

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


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."""
        ...


# ---------------------------------------------------------------------------
# Provider stubs (lane D fills bodies; classifiers import their SDK lazily).
# ---------------------------------------------------------------------------
class GeminiClassifier:
    """Default provider — Gemini 2.5 Flash, native video input (``GEMINI_API_KEY``)."""

    model = "gemini-2.5-flash"

    def __init__(self) -> None:
        raise NotImplementedError("lane D (M7): implement GeminiClassifier")

    def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification:
        raise NotImplementedError("lane D (M7): implement GeminiClassifier.classify")


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

    model = "claude-opus-4-8"

    def __init__(self) -> None:
        raise NotImplementedError("lane D (M7): implement ClaudeClassifier")

    def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification:
        raise NotImplementedError("lane D (M7): implement ClaudeClassifier.classify")


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.
    """

    def __init__(self) -> None:
        raise NotImplementedError("lane D (M7): implement LocalClassifier")

    def classify(self, clip_path: Path, frames: list[Path]) -> MomentClassification:
        raise NotImplementedError("lane D (M7): implement LocalClassifier.classify")


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.
    """
    _ = name or os.environ.get("FESTIVAL4D_CLASSIFIER", "gemini")
    raise NotImplementedError("lane D (M7): implement get_classifier")


# ---------------------------------------------------------------------------
# Candidate detection + orchestration (lane D fills bodies).
# ---------------------------------------------------------------------------
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).
    """
    raise NotImplementedError("lane D (M7): implement detect_candidates")


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


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).
    """
    raise NotImplementedError("lane D (M7): implement run_events")