M11: auto-director — top-N events -> beat-snapped camPath (backend + UI)

generate_path() fills the foundation2 stub with the deterministic v1 rules:
top-N events by confidence (ties -> earlier, then shot chronologically);
per event, the registered camera whose pose track is nearest the event
time; keyframes at t_event - lead_s (clamped >= 0) and t_event + duration;
nearest-beat snapping when beats.json exists (missing/corrupt file -> no
snapping); FOV = 2*atan(h/(2*fy)) like scene3d.snapTo. Every pose goes
through the frozen geometry.colmap_to_threejs and the camPath quaternion
is emitted in Three.js [x,y,z,w] order (pitfall #1 — locked by test).
Degradation: no events / no registered poses -> valid empty camPath with a
note; the response always round-trips through camPath.fromJSON.

frontend/src/director.js wires the pre-wired hidden 🎬 button: POST
/api/director, camPath.fromJSON(response), auto-play via the existing
#btn-path flow; empty path or fetch failure surfaces on the button and
recovers — never a crash.

Acceptance (test_director.py, 14 tests): structure-locked against the
frozen camPath shape (exact key sets, unit quats, monotone times);
top-3 grid == [1,4,8,11,14] incl. shared-keyframe dedupe; conversion
matches colmap_to_threejs + mat_to_quat exactly; snapping puts every key
on the beat grid; unregistered poses excluded; API route + CLI dispatch
real output. Suite 148 passed; frontend build clean.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
type-two 2026-07-17 17:51:52 +10:00
parent 0f8685b49e
commit 52e891204d
3 changed files with 437 additions and 14 deletions

View File

@ -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}

View File

@ -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

View File

@ -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;
},
};