festifun/backend/festival4d/director.py

"""Auto-director: events -> keyframed camera path (spec M11, lane E).

Contract (frozen at foundation2 merge — phase-5 contract #2): :func:`generate_path`
returns a **camPath-compatible** dict::

    {"version": 1,
     "keyframes": [{"t_global": float, "pos": [x, y, z], "quat": [x, y, z, w], "fov": float}]}

Poses are **Three.js scene space**, quaternion order **[x, y, z, w]** — convert from the
COLMAP rows via :func:`festival4d.geometry.colmap_to_threejs`, never inline (pitfall #1).

Deterministic v1 rules — no ML: take the top-N events by confidence (ties -> earlier);
for each, choose the registered camera with a pose nearest the event time; keyframes at
``t_event - lead_s`` and ``t_event + duration``; when ``config.BEATS_JSON`` exists, snap
keyframe times to the nearest beat (no beats file -> no snapping — degrade, don't block);
FOV from that camera's intrinsics (``2 * atan(height / (2 * fy))`` in degrees, the same
formula as ``scene3d.snapTo``).

Degradation (house style): no events -> valid empty path; no registered poses -> valid
empty path with a clear ``note`` (the response always round-trips through
``camPath.fromJSON``); no beats.json -> same path, just unsnapped. Never a crash.
"""

from __future__ import annotations

import json
import logging
import math

from festival4d import config, db
from festival4d.geometry import colmap_to_threejs, mat_to_quat

log = logging.getLogger("festival4d.director")

DEFAULT_EVENT_DURATION_S = 1.0  # used when an event has no duration_s


def _empty(note: str | None = None) -> dict:
    path: dict = {"version": 1, "keyframes": []}
    if note:
        path["note"] = note
        log.info("director: %s -> empty path", note)
    return path


def _load_beats() -> list[float]:
    """The beat grid from beats.json, if valid; otherwise [] (no snapping — pitfall #5)."""
    if not config.BEATS_JSON.exists():
        return []
    try:
        beats = json.loads(config.BEATS_JSON.read_text()).get("beats_s", [])
        return sorted(float(b) for b in beats)
    except (ValueError, TypeError, AttributeError) as exc:
        log.warning("director: unreadable beats.json (%s) — skipping beat snap", exc)
        return []


def _snap_to_beat(t: float, beats: list[float]) -> float:
    """Nearest beat to t (beats sorted ascending); t unchanged when the grid is empty."""
    if not beats:
        return t
    return min(beats, key=lambda b: abs(b - t))


def _nearest_pose(poses: list, t_video: float):
    """The registered pose nearest a local time, and its |dt|. Poses are t_video_s-sorted."""
    best = min(poses, key=lambda p: abs(p.t_video_s - t_video))
    return best, abs(best.t_video_s - t_video)


def _keyframe(pose, video, t_global: float) -> dict:
    """Build one frozen-shape keyframe from a COLMAP pose row (contract #2).

    The COLMAP [w,x,y,z] world->camera pose goes through the frozen
    ``geometry.colmap_to_threejs`` (never inline — pitfall #1); the resulting Three.js
    world rotation matrix becomes the camPath quaternion, reordered to Three.js [x,y,z,w].
    """
    position, rotation = colmap_to_threejs(
        [pose.qw, pose.qx, pose.qy, pose.qz], [pose.tx, pose.ty, pose.tz]
    )
    qw, qx, qy, qz = mat_to_quat(rotation)  # [w,x,y,z] of the *Three.js* world rotation
    fov_deg = math.degrees(2.0 * math.atan(video.height / (2.0 * pose.fy)))  # scene3d.snapTo
    return {
        "t_global": float(t_global),
        "pos": [float(v) for v in position],
        "quat": [float(qx), float(qy), float(qz), float(qw)],  # Three.js order [x,y,z,w]
        "fov": float(fov_deg),
    }


def generate_path(top_n: int = 8, lead_s: float = 2.0) -> dict:
    """Build a camPath JSON dict from the detected events (see module doc)."""
    events = db.get_events()
    if not events:
        return _empty("no events to direct (run `python -m festival4d events`)")

    videos = db.get_videos()
    # Registered poses per video (interpolated rows are excluded: the director only cuts
    # to cameras that were actually solved at that moment).
    tracks: dict[int, list] = {}
    for video in videos:
        registered = [p for p in db.get_poses(video.id) if p.registered]
        if registered:
            tracks[video.id] = registered
    if not tracks:
        return _empty("no registered camera poses (run `python -m festival4d reconstruct`)")
    video_by_id = {v.id: v for v in videos}

    # Top-N events by confidence, ties -> earlier; then shoot them in chronological order.
    ranked = sorted(events, key=lambda e: (-(e.confidence or 0.0), e.t_global_s, e.id))
    chosen = sorted(ranked[: max(0, int(top_n))], key=lambda e: (e.t_global_s, e.id))

    beats = _load_beats()

    keyframes: list[dict] = []
    for event in chosen:
        # The registered camera whose pose track comes closest to the event time.
        best_id, best_dt = None, None
        for vid, poses in sorted(tracks.items()):  # sorted -> deterministic tie-break
            video = video_by_id[vid]
            tv_event = config.t_video_from_global(
                event.t_global_s, video.offset_ms or 0.0, video.drift_ppm or 0.0
            )
            _, dt = _nearest_pose(poses, tv_event)
            if best_dt is None or dt < best_dt:
                best_id, best_dt = vid, dt
        video = video_by_id[best_id]
        poses = tracks[best_id]

        duration = event.duration_s if event.duration_s is not None else DEFAULT_EVENT_DURATION_S
        for t_key in (event.t_global_s - lead_s, event.t_global_s + duration):
            t_key = _snap_to_beat(max(0.0, t_key), beats)
            tv_key = config.t_video_from_global(
                t_key, video.offset_ms or 0.0, video.drift_ppm or 0.0
            )
            pose, _ = _nearest_pose(poses, tv_key)
            keyframes.append(_keyframe(pose, video, t_key))

    # Chronological, and collapse keyframes that landed on the same instant (adjacent events
    # snapping to a shared beat) — camPath needs a monotone timeline.
    keyframes.sort(key=lambda k: k["t_global"])
    deduped: list[dict] = []
    for kf in keyframes:
        if deduped and abs(kf["t_global"] - deduped[-1]["t_global"]) < 1e-6:
            continue
        deduped.append(kf)

    log.info(
        "director: %d event(s) -> %d keyframe(s)%s",
        len(chosen), len(deduped), " (beat-snapped)" if beats else "",
    )
    return {"version": 1, "keyframes": deduped}