diff --git a/backend/festival4d/director.py b/backend/festival4d/director.py index afb7a60..4e6753a 100644 --- a/backend/festival4d/director.py +++ b/backend/festival4d/director.py @@ -1,4 +1,4 @@ -"""Auto-director: events -> keyframed camera path (spec M11, lane E). STUB — lane E fills. +"""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:: @@ -15,11 +15,138 @@ for each, choose the registered camera with a pose nearest the event time; keyfr 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).""" - raise NotImplementedError("auto-director (lane E / M11)") + 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} diff --git a/backend/tests/test_director.py b/backend/tests/test_director.py new file mode 100644 index 0000000..61ed877 --- /dev/null +++ b/backend/tests/test_director.py @@ -0,0 +1,242 @@ +"""M11 acceptance tests (lane E): deterministic auto-director -> frozen camPath JSON. + +Runs on the synthetic fixture WITHOUT ffmpeg (poses + events + DB only): generate_path +never touches media files. The synthetic seed events have known times/confidences, so +event selection, keyframe placement, beat snapping, and the pose conversion are all +checked against ground truth. The quaternion test locks the [x,y,z,w] Three.js order via +the frozen geometry.colmap_to_threejs — this repo's #1 historical bug source. +""" + +from __future__ import annotations + +import json +import math + +import numpy as np +import pytest + +# Synthetic seed events (synthetic.SEED_EVENTS): (t_global, type, confidence) +# 3.0 bass_drop .94 | 6.0 crowd_wave .72 | 8.0 quiet .61 | 10.0 pyro .88 +# 13.0 light_show .90 | 17.0 confetti .81 | 19.0 artist .77 +TOP3_TIMES = [3.0, 10.0, 13.0] # confidences .94, .88, .90 -> top 3 +KEYFRAME_KEYS = {"t_global", "pos", "quat", "fov"} + + +@pytest.fixture() +def project(): + """Fresh synthetic project (no ffmpeg) per test — several tests mutate the DB.""" + from festival4d import config, synthetic + + synthetic.build(run_ffmpeg=False) + config.BEATS_JSON.unlink(missing_ok=True) # unsnapped by default; tests opt in + yield config + config.BEATS_JSON.unlink(missing_ok=True) + + +# --------------------------------------------------------------------------- +# Structure lock: the frozen camPath shape (contract #2). +# --------------------------------------------------------------------------- +def test_output_matches_frozen_campath_shape(project): + from festival4d import director + + path = director.generate_path() + assert set(path) == {"version", "keyframes"} # exact-set lock (no stray fields on success) + assert path["version"] == 1 + assert len(path["keyframes"]) > 0 + for kf in path["keyframes"]: + assert set(kf) == KEYFRAME_KEYS # exact-set lock per keyframe + assert isinstance(kf["t_global"], float) + assert len(kf["pos"]) == 3 and all(isinstance(v, float) for v in kf["pos"]) + assert len(kf["quat"]) == 4 and all(isinstance(v, float) for v in kf["quat"]) + assert abs(math.hypot(*kf["quat"][:3], kf["quat"][3]) - 1.0) < 1e-6 # unit quaternion + assert 10.0 < kf["fov"] < 120.0 + times = [kf["t_global"] for kf in path["keyframes"]] + assert times == sorted(times) + assert len(set(times)) == len(times) # strictly monotone (dedup after snapping) + # JSON-serializable end to end (the API returns this dict verbatim). + json.dumps(path) + + +def test_covers_the_top_n_events_with_lead_and_duration(project): + from festival4d import director + + path = director.generate_path(top_n=3, lead_s=2.0) + times = [kf["t_global"] for kf in path["keyframes"]] + # Each top-3 event contributes keys at t-2.0 and t+duration (1.0 in the fixture): + # 3.0 -> {1.0, 4.0}; 10.0 -> {8.0, 11.0}; 13.0 -> {11.0, 14.0}. The shared 11.0 + # collapses to one keyframe -> exactly 5, and every event is bracketed. + assert times == [1.0, 4.0, 8.0, 11.0, 14.0] + for t_event in TOP3_TIMES: + assert any(t <= t_event for t in times) and any(t >= t_event for t in times) + + +def test_full_top_n_covers_every_event(project): + from festival4d import director + + path = director.generate_path(top_n=8, lead_s=2.0) # fixture has 7 events + times = [kf["t_global"] for kf in path["keyframes"]] + for t_event in [3.0, 6.0, 8.0, 10.0, 13.0, 17.0, 19.0]: + assert min(abs(t - (t_event - 2.0)) for t in times) < 1e-9 + assert min(abs(t - (t_event + 1.0)) for t in times) < 1e-9 + + +def test_confidence_ties_break_to_the_earlier_event(project): + from festival4d import db, director + + db.clear_events() + db.add_event(t_global_s=15.0, event_type="pyro", source="ai", duration_s=1.0, confidence=0.9) + db.add_event(t_global_s=5.0, event_type="pyro", source="ai", duration_s=1.0, confidence=0.9) + path = director.generate_path(top_n=1, lead_s=2.0) + assert [kf["t_global"] for kf in path["keyframes"]] == [3.0, 6.0] # the t=5 event won + + +def test_lead_time_clamps_at_zero(project): + from festival4d import db, director + + db.clear_events() + db.add_event(t_global_s=0.5, event_type="pyro", source="ai", duration_s=1.0, confidence=0.9) + path = director.generate_path(top_n=1, lead_s=2.0) + assert [kf["t_global"] for kf in path["keyframes"]] == [0.0, 1.5] + + +# --------------------------------------------------------------------------- +# Pose conversion: through the frozen geometry.colmap_to_threejs, quat order [x,y,z,w]. +# --------------------------------------------------------------------------- +def test_keyframe_pose_matches_frozen_conversion(project): + from festival4d import db, director + from festival4d.geometry import colmap_to_threejs, mat_to_quat, quat_to_mat + + path = director.generate_path(top_n=1, lead_s=2.0) # top event: bass_drop @ 3.0 + kf = path["keyframes"][0] # t_global = 1.0 + + # Every synthetic event time is an exact cam0 pose time (offset 0), so the chosen + # camera is video 1 and the keyframe pose is cam0's registered row at t_video_s=1.0. + pose = next(p for p in db.get_poses(1) if abs(p.t_video_s - 1.0) < 1e-9) + expected_pos, expected_rot = colmap_to_threejs( + [pose.qw, pose.qx, pose.qy, pose.qz], [pose.tx, pose.ty, pose.tz] + ) + assert np.allclose(kf["pos"], expected_pos, atol=1e-9) + + # Quaternion order lock: reordering [x,y,z,w] -> [w,x,y,z] must reproduce the Three.js + # world rotation matrix (up to the quaternion double cover). + x, y, z, w = kf["quat"] + assert np.allclose(quat_to_mat([w, x, y, z]), expected_rot, atol=1e-9) + expected_q = mat_to_quat(expected_rot) # [w,x,y,z], canonical w >= 0 + q_out = np.array([w, x, y, z]) + if np.dot(q_out, expected_q) < 0: + q_out = -q_out + assert np.allclose(q_out, expected_q, atol=1e-9) + + # FOV from the chosen camera's intrinsics: 2*atan(height / (2*fy)) in degrees. + video = db.get_video(1) + expected_fov = math.degrees(2.0 * math.atan(video.height / (2.0 * pose.fy))) + assert kf["fov"] == pytest.approx(expected_fov, abs=1e-9) + + +def test_unregistered_poses_are_excluded(project): + from festival4d import db, director + + # Mark every cam0 pose unregistered: the director must cut to cam1/cam2 instead. + poses = db.get_poses(1) + db.set_poses(1, [ + {"frame_idx": p.frame_idx, "t_video_s": p.t_video_s, + "qw": p.qw, "qx": p.qx, "qy": p.qy, "qz": p.qz, + "tx": p.tx, "ty": p.ty, "tz": p.tz, + "fx": p.fx, "fy": p.fy, "cx": p.cx, "cy": p.cy, "registered": False} + for p in poses + ]) + path = director.generate_path(top_n=1, lead_s=2.0) + assert len(path["keyframes"]) == 2 + cam0_centers = { + tuple(np.round(np.asarray(_center(p)), 6)) for p in poses + } + for kf in path["keyframes"]: + assert tuple(np.round(kf["pos"], 6)) not in cam0_centers + + +def _center(pose): + from festival4d.geometry import colmap_to_threejs + + pos, _ = colmap_to_threejs([pose.qw, pose.qx, pose.qy, pose.qz], [pose.tx, pose.ty, pose.tz]) + return pos + + +# --------------------------------------------------------------------------- +# Beat snapping (M10 -> M11), and degradation without it. +# --------------------------------------------------------------------------- +def test_keyframes_snap_to_beats_when_beats_exist(project): + from festival4d import director + + grid = [round(k * 0.75, 4) for k in range(0, 28)] # 0.75 s grid != the keyframe times + project.BEATS_JSON.parent.mkdir(parents=True, exist_ok=True) + project.BEATS_JSON.write_text(json.dumps({ + "tempo_bpm": 80.0, "beats_s": grid, "onsets": [], + "generated_by": "festival4d.audio_features", + })) + path = director.generate_path(top_n=3, lead_s=2.0) + times = [kf["t_global"] for kf in path["keyframes"]] + assert len(times) > 0 + assert all(any(abs(t - b) < 1e-9 for b in grid) for t in times) # every key ON a beat + assert times != [1.0, 4.0, 8.0, 11.0, 14.0] # actually moved vs the unsnapped grid + + +def test_no_beats_file_means_no_snapping(project): + from festival4d import director + + assert not project.BEATS_JSON.exists() + times = [kf["t_global"] for kf in director.generate_path(top_n=3, lead_s=2.0)["keyframes"]] + assert times == [1.0, 4.0, 8.0, 11.0, 14.0] + + +def test_corrupt_beats_file_degrades_to_unsnapped(project): + from festival4d import director + + project.BEATS_JSON.parent.mkdir(parents=True, exist_ok=True) + project.BEATS_JSON.write_text("{ not json") + times = [kf["t_global"] for kf in director.generate_path(top_n=3, lead_s=2.0)["keyframes"]] + assert times == [1.0, 4.0, 8.0, 11.0, 14.0] + + +# --------------------------------------------------------------------------- +# Degradation: empty inputs -> valid empty path (still camPath.fromJSON-safe). +# --------------------------------------------------------------------------- +def test_no_events_yields_valid_empty_path(project): + from festival4d import db, director + + db.clear_events() + path = director.generate_path() + assert path["version"] == 1 and path["keyframes"] == [] + assert "events" in path["note"] + + +def test_no_registered_poses_yields_valid_empty_path_with_note(project): + from festival4d import db, director + + for video in db.get_videos(): + db.set_poses(video.id, []) + path = director.generate_path() + assert path["version"] == 1 and path["keyframes"] == [] + assert "poses" in path["note"] + + +# --------------------------------------------------------------------------- +# Dispatch: the API route and CLI now run the real implementation. +# --------------------------------------------------------------------------- +def test_api_director_route_returns_real_path(project): + from fastapi.testclient import TestClient + + from festival4d import api + + with TestClient(api.app) as client: + data = client.post("/api/director", json={"top_n": 3, "lead_s": 2.0}).json() + assert data["version"] == 1 + assert [kf["t_global"] for kf in data["keyframes"]] == [1.0, 4.0, 8.0, 11.0, 14.0] + assert all(set(kf) == KEYFRAME_KEYS for kf in data["keyframes"]) + + +def test_cli_direct_prints_campath_json(project, capsys): + from festival4d import cli + + assert cli.main(["direct", "--top-n", "2", "--lead-s", "1.0"]) == 0 + printed = json.loads(capsys.readouterr().out) + assert printed["version"] == 1 and len(printed["keyframes"]) > 0 diff --git a/frontend/src/director.js b/frontend/src/director.js index 0ab6e91..93b9c74 100644 --- a/frontend/src/director.js +++ b/frontend/src/director.js @@ -1,16 +1,70 @@ -// Auto-director UI (spec M11, lane E). STUB — lane E fills THIS FILE ONLY. +// Auto-director UI (spec M11, lane E). // -// Pre-wired by foundation2: the hidden 🎬 button (#btn-director in index.html), and this -// module's import + init() call in main.js. main.js does NOT wire the click — you do, -// inside init(). -// -// When implementing: unhide the button (el.style.display = ""), set `ready = true`; on click -// POST {top_n?, lead_s?} to `${state.apiBase}/api/director` and load the response with -// camPath.fromJSON(...) — never edit camPath.js (frozen). The response is ALWAYS a valid -// camPath JSON; zero keyframes means "no events yet / stubbed" — surface that in the UI, -// don't crash (pitfall #5). Play it via the existing #btn-path flow. +// Pre-wired by foundation2: the hidden 🎬 button (#btn-director in index.html) and this +// module's import + init() call in main.js. This module wires the click: POST /api/director +// -> the response is ALWAYS a valid camPath JSON (frozen contract #2) -> load it with +// camPath.fromJSON and play it via the existing #btn-path flow (never edit camPath.js). +// Zero keyframes means "no events / no poses yet" — surfaced on the button, never a crash +// (pitfall #5). Failures degrade the same way; the button always recovers. + +import { state } from "./state.js"; +import { camPath } from "./camPath.js"; + +const BTN_LABEL = "🎬 Auto-path"; +let flashTimer = null; + +/** Show a transient message on the button, then restore the label. */ +function flash(btn, text, ms = 2200) { + btn.textContent = text; + clearTimeout(flashTimer); + flashTimer = setTimeout(() => (btn.textContent = BTN_LABEL), ms); +} + +async function requestPath(btn) { + btn.disabled = true; + btn.textContent = "🎬 …"; + try { + const resp = await fetch(`${state.apiBase}/api/director`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({}), // server defaults: top_n=8, lead_s=2.0 + }); + if (!resp.ok) throw new Error(`HTTP ${resp.status}`); + const path = await resp.json(); + + if (!Array.isArray(path.keyframes) || path.keyframes.length === 0) { + // Valid-empty path: no events/poses yet (or still stubbed). Surface, don't crash. + console.info("[director] empty path from /api/director", path.note ?? ""); + flash(btn, "🎬 no events yet"); + return; + } + + camPath.fromJSON(path); // replaces any existing keyframes; updates the + Key counter + flash(btn, `🎬 ${path.keyframes.length} keys`); + + // Play through the existing #btn-path flow so its UI state (⏹ label, transport + // auto-start) stays consistent with a manually loaded path. + const playBtn = document.getElementById("btn-path"); + if (playBtn && !camPath.playing && path.keyframes.length >= 2) playBtn.click(); + } catch (err) { + console.error("[director] /api/director failed", err); + flash(btn, "🎬 error — see console"); + } finally { + btn.disabled = false; + } +} export const director = { - ready: false, // flips to true when the lane implements this module - init() {}, // ponytail: stub — lane E replaces the body + ready: false, + + init() { + const btn = document.getElementById("btn-director"); + if (!btn) return; // markup missing: stay hidden, degrade silently + btn.addEventListener("click", () => { + if (btn.disabled) return; + requestPath(btn); + }); + btn.style.display = ""; // unhide: this lane has landed + this.ready = true; + }, };