foundation3 WIP checkpoint (steps 2-5,7): DB/API/CLI/config/capsule + tracker stubs

Incomplete: step 1 (fixture markers + track_truth.json), step 6 (frontend wiring),
step 8 (tests) not done; suite not yet run. Salvaged from interrupted prior session.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
type-two 2026-07-17 20:55:29 +10:00
parent 56e569fd7a
commit 6a703c314d
9 changed files with 793 additions and 17 deletions

View File

@ -23,6 +23,17 @@ Phase 5 (foundation2; shapes frozen in plan/20-phase5.md contracts):
DELETE /api/paths/{id} DELETE /api/paths/{id}
PATCH /api/anchors/{id} {label?, color?} -> updated anchor dict PATCH /api/anchors/{id} {label?, color?} -> updated anchor dict
Phase 6 (foundation3 / M18; shapes frozen in plan/30-phase6.md contract #2):
GET /api/tracks -> [{id, marker_key, label, color, points:[{t_global_s,x,y,z,
quality,views}]}] (points ordered by t_global_s)
PATCH /api/tracks/{id} {label?, color?} -> updated track dict (404 on missing)
DELETE /api/tracks/{id} -> {deleted: id} (404 on missing)
POST /api/tracks/solve -> {result, tracks[, note]} runs lane H's detect->solve
(tracker_solve.run_solve); while lane H is stubbed it
degrades to a VALID empty result + "note" (house pattern).
manifest gains "has_tracks": bool.
Source videos in ``data/raw`` are mounted at ``/media`` via Starlette ``StaticFiles``, Source videos in ``data/raw`` are mounted at ``/media`` via Starlette ``StaticFiles``,
which supports HTTP Range (required for ``<video>`` seeking; verify ``curl -H "Range: which supports HTTP Range (required for ``<video>`` seeking; verify ``curl -H "Range:
bytes=0-100"`` -> 206). Stubbed lane entrypoints (events detection, M8 annotation bytes=0-100"`` -> 206). Stubbed lane entrypoints (events detection, M8 annotation
@ -118,6 +129,11 @@ class PathIn(BaseModel):
path_json: str = Field(alias="json") path_json: str = Field(alias="json")
class TrackPatch(BaseModel):
label: str | None = None
color: str | None = None
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Serialization helpers # Serialization helpers
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -162,6 +178,24 @@ def _event_dict(e) -> dict:
} }
def _track_dict(t) -> dict:
"""A track with its points inlined (phase 6 contract #2), ordered by t_global_s."""
return {
"id": t.id,
"marker_key": t.marker_key,
"label": t.label,
"color": t.color,
"points": [
{
"t_global_s": p.t_global_s,
"x": p.x, "y": p.y, "z": p.z,
"quality": p.quality, "views": p.views,
}
for p in db.get_track_points(t.id)
],
}
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Routes # Routes
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -183,6 +217,8 @@ def manifest() -> dict:
"has_capture": capture.capture_enabled(), "has_capture": capture.capture_enabled(),
"has_poses": db.has_poses(), "has_poses": db.has_poses(),
"has_splat": config.SPLAT_PLY.exists(), "has_splat": config.SPLAT_PLY.exists(),
"has_tracks": db.has_tracks(), # phase 6 (M18): any solved friend track ⇒ frontend loads them
# Always False from the live server; the capsule (M17) bakes a manifest with true, # Always False from the live server; the capsule (M17) bakes a manifest with true,
# which makes the frontend hide all write UI (phase-5 contract #5). # which makes the frontend hide all write UI (phase-5 contract #5).
"capsule": False, "capsule": False,
@ -465,3 +501,52 @@ def delete_anchor(anchor_id: int) -> dict:
if not db.delete_anchor(anchor_id): if not db.delete_anchor(anchor_id):
raise HTTPException(status_code=404, detail=f"no anchor with id={anchor_id}") raise HTTPException(status_code=404, detail=f"no anchor with id={anchor_id}")
return {"deleted": anchor_id} return {"deleted": anchor_id}
# ---------------------------------------------------------------------------
# Phase 6 (foundation3 / M18). Friend tracks. Shapes frozen in plan/30-phase6.md contract #2.
# ---------------------------------------------------------------------------
@app.get("/api/tracks")
def get_tracks() -> list[dict]:
"""All friend tracks with their points inlined (contract #2)."""
return [_track_dict(t) for t in db.get_tracks()]
@app.patch("/api/tracks/{track_id}")
def patch_track(track_id: int, payload: TrackPatch) -> dict:
"""Rename / recolor a track in place (M21 friends panel)."""
try:
track = db.patch_track(track_id, label=payload.label, color=payload.color)
except KeyError:
raise HTTPException(status_code=404, detail=f"no track with id={track_id}")
return _track_dict(track)
@app.delete("/api/tracks/{track_id}")
def delete_track(track_id: int) -> dict:
"""Delete a track and its points (M21)."""
if not db.delete_track(track_id):
raise HTTPException(status_code=404, detail=f"no track with id={track_id}")
return {"deleted": track_id}
@app.post("/api/tracks/solve")
def solve_tracks() -> dict:
"""Run marker detection + track solving (M19+M20) and return the resulting tracks.
Dispatches into lane H's ``tracker_solve.run_solve`` (which itself consumes
``tracker_detect``'s detections). While lane H is stubbed both raise
``NotImplementedError``; we catch it and degrade to a VALID empty result + a note, so the
frozen route never 500s the house pattern (mirrors ``POST /api/events/detect``)."""
from festival4d import tracker_solve
try:
result = tracker_solve.run_solve()
except NotImplementedError as exc:
log.info("track solving not implemented yet: %s", exc)
return {
"result": None,
"note": "friend-track solving not implemented yet (lane H / M19-M20)",
"tracks": [_track_dict(t) for t in db.get_tracks()],
}
return {"result": result, "tracks": [_track_dict(t) for t in db.get_tracks()]}

View File

@ -6,7 +6,7 @@ Contract: :func:`build_capsule` writes a self-contained bundle to ``out_dir`` (d
- copy ``frontend/dist`` (error clearly if missing tell the user to ``npm run build``); - copy ``frontend/dist`` (error clearly if missing tell the user to ``npm run build``);
- copy media from ``config.RAW_DIR``; - copy media from ``config.RAW_DIR``;
- bake every read-only API response to files at the *same relative paths* - bake every read-only API response to files at the *same relative paths*
(``api/manifest``, ``api/events``, ``api/anchors``, ``api/beats``, (``api/manifest``, ``api/events``, ``api/anchors``, ``api/tracks``, ``api/beats``,
``api/videos/{id}/poses``, ``api/pointcloud``, ``api/splat``, ``api/audio/{id}`` ``api/videos/{id}/poses``, ``api/pointcloud``, ``api/splat``, ``api/audio/{id}``
extracting audio via ``ingest.extract_audio_hq`` if not yet cached); extracting audio via ``ingest.extract_audio_hq`` if not yet cached);
- inject ``<script>window.__F4D_API_BASE__=""</script>`` into the copied ``index.html``; - inject ``<script>window.__F4D_API_BASE__=""</script>`` into the copied ``index.html``;
@ -265,6 +265,8 @@ def build_capsule(out_dir: str | Path | None = None,
_bake_json(api_dir / "manifest", manifest) _bake_json(api_dir / "manifest", manifest)
_bake_json(api_dir / "events", api.get_events()) _bake_json(api_dir / "events", api.get_events())
_bake_json(api_dir / "anchors", api.get_anchors()) _bake_json(api_dir / "anchors", api.get_anchors())
# Friend tracks (M18/M20) — read-only in a capsule so ribbons + follow still work zero-backend.
_bake_json(api_dir / "tracks", api.get_tracks())
for v in videos: for v in videos:
_bake_json(api_dir / "videos" / str(v.id) / "poses", api.video_poses(v.id)) _bake_json(api_dir / "videos" / str(v.id) / "poses", api.video_poses(v.id))

View File

@ -5,7 +5,7 @@ to (in ``ingest.py``, ``audio_sync.py``, ``sfm.py``, ``events_ai.py``); they nev
edit this file or ``api.py``. edit this file or ``api.py``.
Subcommands: ``synthetic | ingest | sync | reconstruct | events | features | direct | Subcommands: ``synthetic | ingest | sync | reconstruct | events | features | direct |
capsule | serve``. track | capsule | serve``.
Every subcommand dispatches to a single lane entrypoint. While a lane is still a stub, Every subcommand dispatches to a single lane entrypoint. While a lane is still a stub,
its entrypoint raises :class:`NotImplementedError`; we catch that here and print a clear its entrypoint raises :class:`NotImplementedError`; we catch that here and print a clear
@ -76,6 +76,23 @@ def _cmd_direct(args: argparse.Namespace) -> int:
return 0 return 0
def _cmd_track(args: argparse.Namespace) -> int:
"""Detect wearable markers and solve them into 3D friend tracks (lane H / M19-M20).
Runs detect -> solve by default; ``--detect-only`` / ``--solve-only`` run one stage.
Both stages dispatch into lane-H stubs, so while unimplemented this prints the
"not implemented yet" message and exits 2 (same as every other stubbed subcommand)."""
from festival4d import tracker_detect, tracker_solve
if not args.solve_only:
result = tracker_detect.run_detect()
log.info("track detect: %s", result)
if not args.detect_only:
result = tracker_solve.run_solve()
log.info("track solve: %s", result)
return 0
def _cmd_capsule(args: argparse.Namespace) -> int: def _cmd_capsule(args: argparse.Namespace) -> int:
from festival4d import capsule from festival4d import capsule
@ -134,6 +151,14 @@ def build_parser() -> argparse.ArgumentParser:
help="seconds of lead-in before each event (default 2.0)") help="seconds of lead-in before each event (default 2.0)")
p_dir.set_defaults(func=_cmd_direct) p_dir.set_defaults(func=_cmd_direct)
p_trk = sub.add_parser("track", help="detect markers + solve friend tracks (lane H / M19-M20)")
trk_mode = p_trk.add_mutually_exclusive_group()
trk_mode.add_argument("--detect-only", dest="detect_only", action="store_true",
help="run marker detection only (write detections.json)")
trk_mode.add_argument("--solve-only", dest="solve_only", action="store_true",
help="run track solving only (from an existing detections.json)")
p_trk.set_defaults(func=_cmd_track, detect_only=False, solve_only=False)
p_cap = sub.add_parser("capsule", help="bake a zero-backend shareable bundle (lane G / M17)") p_cap = sub.add_parser("capsule", help="bake a zero-backend shareable bundle (lane G / M17)")
p_cap.add_argument("--out", default=None, help="output dir (default data/capsule/)") p_cap.add_argument("--out", default=None, help="output dir (default data/capsule/)")
p_cap.set_defaults(func=_cmd_capsule) p_cap.set_defaults(func=_cmd_capsule)

View File

@ -46,6 +46,8 @@ POINTS_PLY = WORK_DIR / "points.ply" # reconstructed / synthetic point cl
SPLAT_PLY = WORK_DIR / "splat.ply" # optional 3DGS splat (trained externally, e.g. via MODELBEAST — see docs/modelbeast-crossover.md); viewer prefers it over the point cloud SPLAT_PLY = WORK_DIR / "splat.ply" # optional 3DGS splat (trained externally, e.g. via MODELBEAST — see docs/modelbeast-crossover.md); viewer prefers it over the point cloud
SYNC_JSON = WORK_DIR / "sync.json" # exported sync solution (lane A) SYNC_JSON = WORK_DIR / "sync.json" # exported sync solution (lane A)
GROUND_TRUTH_JSON = WORK_DIR / "ground_truth.json" # synthetic fixture ground truth GROUND_TRUTH_JSON = WORK_DIR / "ground_truth.json" # synthetic fixture ground truth
DETECTIONS_JSON = WORK_DIR / "detections.json" # marker detections (phase 6 / M19, contract #3)
TRACK_TRUTH_JSON = WORK_DIR / "track_truth.json" # synthetic marker ground truth (phase 6 / M18)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Constants # Constants

View File

@ -13,6 +13,8 @@ Schema (spec §2)::
events(id, t_global_s, duration_s, event_type, confidence, description, source) events(id, t_global_s, duration_s, event_type, confidence, description, source)
annotations(id, video_id FK, t_video_s, x0,y0,x1,y1, resolved_anchor_id FK NULL) annotations(id, video_id FK, t_video_s, x0,y0,x1,y1, resolved_anchor_id FK NULL)
paths(id, name, created_at, path_json) # phase 5 (M16): saved camPath JSON, as text paths(id, name, created_at, path_json) # phase 5 (M16): saved camPath JSON, as text
tracks(id, marker_key, label, color, created_at) # phase 6 (M18/M20): friend tracks
track_points(id, track_id FK, t_global_s, x,y,z, quality, views) # ordered by t_global_s
Engine management: :func:`init_engine` (re)binds the module to a SQLite file. It defaults Engine management: :func:`init_engine` (re)binds the module to a SQLite file. It defaults
to ``config.DB_PATH`` but tests point it at a temp file. Helpers open and commit their own to ``config.DB_PATH`` but tests point it at a temp file. Helpers open and commit their own
@ -143,6 +145,43 @@ class SavedPath(Base):
path_json: Mapped[str] = mapped_column(String, nullable=False) path_json: Mapped[str] = mapped_column(String, nullable=False)
class Track(Base):
"""A friend track (phase 6 contract #2): a wearable marker's identity + metadata.
``marker_key`` is the detector-emitted identity ``"hue:<opencv_hue>"`` for a color
marker or ``"code:<id>"`` for a blink badge (see ``tracker_detect``). The per-timestep
3D path lives in :class:`TrackPoint`. ``label`` / ``color`` are user-editable.
"""
__tablename__ = "tracks"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
marker_key: Mapped[str] = mapped_column(String, nullable=False)
label: Mapped[str | None] = mapped_column(String, nullable=True)
color: Mapped[str | None] = mapped_column(String, nullable=True)
created_at: Mapped[str] = mapped_column(String, nullable=False)
class TrackPoint(Base):
"""One solved position of a track at a master-timeline instant (phase 6 contract #2).
Coordinates are **Three.js scene space**. ``views`` = number of cameras used to solve the
point (1 single-view ground-plane fallback, contract #5); ``quality`` is 0..1.
Points are always read ordered by ``t_global_s``.
"""
__tablename__ = "track_points"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
track_id: Mapped[int] = mapped_column(ForeignKey("tracks.id"), nullable=False, index=True)
t_global_s: Mapped[float] = mapped_column(Float, nullable=False)
x: Mapped[float] = mapped_column(Float, nullable=False)
y: Mapped[float] = mapped_column(Float, nullable=False)
z: Mapped[float] = mapped_column(Float, nullable=False)
quality: Mapped[float | None] = mapped_column(Float, nullable=True)
views: Mapped[int | None] = mapped_column(Integer, nullable=True)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Engine / session management # Engine / session management
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -508,6 +547,107 @@ def delete_path(path_id: int) -> bool:
return True return True
# ---------------------------------------------------------------------------
# Friend tracks (phase 6 / M18 contract #2). Written by the solver (lane H / M20),
# read by the API + frontend. TrackPoints are always returned ordered by t_global_s.
# ---------------------------------------------------------------------------
def add_track(marker_key: str, label: str | None = None, color: str | None = None) -> Track:
"""Insert a track row (no points yet) and return it."""
with session_scope() as s:
track = Track(marker_key=marker_key, label=label, color=color, created_at=_now_iso())
s.add(track)
s.flush()
s.refresh(track)
return track
def get_tracks() -> list[Track]:
"""All tracks, ordered by id."""
with session_scope() as s:
return list(s.scalars(select(Track).order_by(Track.id)))
def get_track(track_id: int) -> Track | None:
with session_scope() as s:
return s.get(Track, track_id)
def get_track_points(track_id: int) -> list[TrackPoint]:
"""A track's points, ordered by ``t_global_s`` (contract #2)."""
with session_scope() as s:
return list(
s.scalars(
select(TrackPoint)
.where(TrackPoint.track_id == track_id)
.order_by(TrackPoint.t_global_s)
)
)
def set_track_points(track_id: int, points: Iterable[dict]) -> int:
"""Replace all points for a track with ``points`` (solver output, M20). Returns the count.
Each dict needs ``t_global_s, x, y, z`` and optionally ``quality`` (default None) and
``views`` (default None). Atomic: the delete + inserts share one transaction, so a caller
that raises mid-way never leaves a half-written track. Raises ``KeyError`` if the track
is unknown.
"""
rows = list(points)
with session_scope() as s:
if s.get(Track, track_id) is None:
raise KeyError(f"no track with id={track_id}")
s.execute(delete(TrackPoint).where(TrackPoint.track_id == track_id))
for p in rows:
s.add(
TrackPoint(
track_id=track_id,
t_global_s=float(p["t_global_s"]),
x=float(p["x"]), y=float(p["y"]), z=float(p["z"]),
quality=None if p.get("quality") is None else float(p["quality"]),
views=None if p.get("views") is None else int(p["views"]),
)
)
return len(rows)
def patch_track(track_id: int, label: str | None = None, color: str | None = None) -> Track:
"""Rename / recolor a track (M21 panel). Only provided fields change.
Raises ``KeyError`` if the id is unknown.
"""
with session_scope() as s:
track = s.get(Track, track_id)
if track is None:
raise KeyError(f"no track with id={track_id}")
if label is not None:
track.label = label
if color is not None:
track.color = color
s.flush()
s.refresh(track)
return track
def delete_track(track_id: int) -> bool:
"""Delete a track and its points. Returns ``True`` if a row was deleted, else ``False``.
SQLite doesn't cascade by default, so the points are removed explicitly.
"""
with session_scope() as s:
track = s.get(Track, track_id)
if track is None:
return False
s.execute(delete(TrackPoint).where(TrackPoint.track_id == track_id))
s.delete(track)
return True
def has_tracks() -> bool:
"""Whether any track exists (drives ``manifest.has_tracks``)."""
with session_scope() as s:
return s.scalar(select(Track.id).limit(1)) is not None
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Annotations # Annotations
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------

View File

@ -11,11 +11,17 @@ testable without real footage:
- **Point cloud**: random points on the stage box + ground plane, written to - **Point cloud**: random points on the stage box + ground plane, written to
``points.ply`` in the exact binary-little-endian format the real COLMAP path produces. ``points.ply`` in the exact binary-little-endian format the real COLMAP path produces.
- **Seeded events + anchors** so the timeline and overlays have data immediately. - **Seeded events + anchors** so the timeline and overlays have data immediately.
- **Two moving markers** (phase 6 / M18) composited into every camera's video *through the
same camera projection the poses use*, plus ``track_truth.json`` the 3D ground truth lane
H recovers. Marker A is a solid magenta disc (color mode, ``hue:150``); marker B is a
blinking bright disc speaking the frozen OOK protocol with ID 5 (``code:5``). See
:func:`marker_position`, :func:`render_marker_overlay`, :func:`build_track_truth`.
- **``ground_truth.json``**: offsets, drift, audio pulse (bang) times, events, and stage - **``ground_truth.json``**: offsets, drift, audio pulse (bang) times, events, and stage
corners part of the frozen contract (lanes A and D assert against it). corners part of the frozen contract (lanes A and D assert against it).
FROZEN CONTRACT after foundation. The module is factored so the numeric pieces FROZEN CONTRACT after foundation. The module is factored so the numeric pieces
(audio shifts, poses, PLY round-trip) are unit-testable without invoking ffmpeg. (audio shifts, poses, PLY round-trip, marker geometry) are unit-testable without invoking
ffmpeg.
""" """
from __future__ import annotations from __future__ import annotations
@ -31,8 +37,8 @@ from pathlib import Path
import numpy as np import numpy as np
from festival4d import config, db from festival4d import config, db, tracker_detect
from festival4d.geometry import mat_to_quat from festival4d.geometry import mat_to_quat, quat_to_mat, slerp_pose
log = logging.getLogger("festival4d.synthetic") log = logging.getLogger("festival4d.synthetic")
@ -363,6 +369,209 @@ def read_ply(path: Path) -> tuple[np.ndarray, np.ndarray]:
return points, colors return points, colors
# ---------------------------------------------------------------------------
# Moving markers (phase 6 / M18). Two wearable markers travel known 3D paths across the
# stage; each camera's frame shows them at the pixel its OWN pose projects the 3D point to,
# so the rendered pixels and ``track_truth.json`` are consistent BY CONSTRUCTION — the marker
# pixel is derived from the ground-truth 3D point through the same projection lane H inverts.
#
# Marker A — solid magenta (#ff00ff) disc, color mode, marker_key "hue:150". Slow circle
# (radius 2, y≈1.5) centered on the stage.
# Marker B — blinking bright disc, blink mode, ID 5, marker_key "code:5". A different path;
# the LED follows the frozen OOK protocol (tracker_detect) in GLOBAL time, so
# every camera sees the same on/off state at the same t_global.
# Discs are small (radius ≈2.5% of frame height ⇒ ≪2% frame area) and composited AFTER all
# existing content, so the 189-test floor (offsets/events/poses/SfM) is untouched (pitfall #6).
# ---------------------------------------------------------------------------
MARKER_A_KEY = tracker_detect.color_marker_key(tracker_detect.MAGENTA_OPENCV_HUE) # "hue:150"
MARKER_A_HEX = "#ff00ff"
MARKER_A_BGR = (255, 0, 255) # OpenCV BGR for magenta (drawn AFTER the hue filter)
MARKER_B_ID = 5
MARKER_B_KEY = tracker_detect.code_marker_key(MARKER_B_ID) # "code:5"
MARKER_B_ON_BGR = (255, 255, 255) # bright white while the LED is ON
MARKER_DISC_HEIGHT_FRAC = 0.025 # disc radius / frame height (≈9 px @ 360 -> 0.11% area)
MARKER_A_PERIOD_S = 20.0 # one slow loop across the fixture span
MARKER_B_PERIOD_S = 16.0 # different path + rate so the two never lock together
MARKER_TRUTH_DT_S = 0.05 # track_truth.json sampling step (t_global seconds)
def marker_position(marker_key: str, t_global: float) -> tuple[float, float, float]:
"""Ground-truth 3D position (Three.js scene space) of a marker at ``t_global`` seconds.
Pure and total over all real ``t_global`` (the markers exist for the whole span). This is
THE ground truth: it is what ``track_truth.json`` records and what a correct solver
recovers. FROZEN lane H's acceptance is measured against these paths.
"""
if marker_key == MARKER_A_KEY:
theta = 2.0 * np.pi * t_global / MARKER_A_PERIOD_S
return (2.0 * float(np.cos(theta)), 1.5, 2.0 * float(np.sin(theta)))
if marker_key == MARKER_B_KEY:
phi = 2.0 * np.pi * t_global / MARKER_B_PERIOD_S
return (
2.4 * float(np.sin(phi)),
1.0 + 0.4 * float(np.sin(2.0 * phi)),
-0.8 + 1.2 * float(np.cos(phi)),
)
raise KeyError(f"unknown marker_key {marker_key!r}")
MARKER_KEYS = (MARKER_A_KEY, MARKER_B_KEY)
def _interp_pose(poses: list[dict], t_video: float):
"""Interpolate a COLMAP pose ``(q, t, (fx,fy,cx,cy))`` at ``t_video`` from ``camera_track``
output (ascending by ``t_video_s``). Mirrors ``resolve._pose_at`` / the frontend ``poseAt``
exactly slerp the rotation, lerp the translation via the FROZEN ``geometry.slerp_pose``
so the marker pixels a solver back-projects land on the same 3D point. Clamps (no extrap.).
"""
def tup(p):
return ([p["qw"], p["qx"], p["qy"], p["qz"]],
[p["tx"], p["ty"], p["tz"]],
(p["fx"], p["fy"], p["cx"], p["cy"]))
if t_video <= poses[0]["t_video_s"]:
return tup(poses[0])
if t_video >= poses[-1]["t_video_s"]:
return tup(poses[-1])
lo, hi = 0, len(poses) - 1
while hi - lo > 1:
mid = (lo + hi) // 2
if poses[mid]["t_video_s"] <= t_video:
lo = mid
else:
hi = mid
a, b = poses[lo], poses[hi]
span = b["t_video_s"] - a["t_video_s"]
alpha = (t_video - a["t_video_s"]) / span if span > 1e-9 else 0.0
q, t = slerp_pose(
[a["qw"], a["qx"], a["qy"], a["qz"]], [a["tx"], a["ty"], a["tz"]],
[b["qw"], b["qx"], b["qy"], b["qz"]], [b["tx"], b["ty"], b["tz"]],
alpha,
)
return list(q), list(t), (a["fx"], a["fy"], a["cx"], a["cy"])
def project_point(q, t, intr, X) -> tuple[float, float, float] | None:
"""Project world point ``X`` through a COLMAP world->camera pose (the inverse of
``geometry.ray_from_pixel``). Returns ``(px, py, z_cam)`` in pixels, or ``None`` if the
point is behind the camera. COLMAP camera axes: +x right, +y down, +z forward.
"""
R = quat_to_mat(q) # world -> cam
Xc = R @ np.asarray(X, dtype=np.float64).reshape(3) + np.asarray(t, dtype=np.float64).reshape(3)
z = float(Xc[2])
if z <= 1e-6:
return None
fx, fy, cx, cy = intr
return fx * float(Xc[0]) / z + cx, fy * float(Xc[1]) / z + cy, z
def _disc_radius_px(height: int) -> int:
return max(6, int(round(MARKER_DISC_HEIGHT_FRAC * height)))
def render_marker_overlay(out_dir: Path, poses: list[dict], offset_ms: float,
width: int, height: int, fps: float, duration_s: float) -> int:
"""Write a transparent BGRA PNG per frame with the two markers drawn at their projected
pixels (phase 6 / M18). Composited over one camera's clip by :func:`render_video`.
For frame ``k`` at local ``t_video = k/fps``: convert to ``t_global`` with the FROZEN
``config.t_global_from_video``, look up each marker's 3D truth at ``t_global``, and project
it through this camera's pose interpolated at ``t_video``. Marker A is always drawn; marker
B is drawn only when its LED is ON per the OOK schedule (``tracker_detect.led_on``, in
global time). Off-frame / behind-camera projections are simply not drawn. Returns frame count.
"""
import cv2
out_dir = Path(out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
radius = _disc_radius_px(height)
n = int(round(duration_s * fps))
for k in range(n):
t_video = k / fps
t_global = config.t_global_from_video(t_video, offset_ms, 0.0)
q, t, intr = _interp_pose(poses, t_video)
img = np.zeros((height, width, 4), dtype=np.uint8) # transparent BGRA
_draw_marker_disc(img, q, t, intr, marker_position(MARKER_A_KEY, t_global),
MARKER_A_BGR, radius)
if tracker_detect.led_on(t_global, MARKER_B_ID):
_draw_marker_disc(img, q, t, intr, marker_position(MARKER_B_KEY, t_global),
MARKER_B_ON_BGR, radius)
if not cv2.imwrite(str(out_dir / f"{k:05d}.png"), img):
raise RuntimeError(f"failed to write marker overlay frame: {out_dir}/{k:05d}.png")
return n
def _draw_marker_disc(img, q, t, intr, X, bgr, radius) -> None:
import cv2
proj = project_point(q, t, intr, X)
if proj is None:
return
px, py, _ = proj
h, w = img.shape[:2]
if not (0.0 <= px < w and 0.0 <= py < h):
return # off this camera's frame ⇒ not visible here
cv2.circle(img, (int(round(px)), int(round(py))), radius,
(int(bgr[0]), int(bgr[1]), int(bgr[2]), 255), thickness=-1, lineType=cv2.LINE_8)
def build_track_truth(g_lo: float, g_hi: float, dt: float = MARKER_TRUTH_DT_S) -> dict:
"""The ``track_truth.json`` body: each marker's 3D path sampled over ``[g_lo, g_hi]``.
Shape (spec M18): ``{"markers": [{"marker_key", "points": [{"t_global_s","x","y","z"}]}]}``
in Three.js scene space. ``generated_by`` is added for provenance (extra keys are fine).
"""
n = max(1, int(round((g_hi - g_lo) / dt)) + 1)
markers = []
for key in MARKER_KEYS:
points = []
for i in range(n):
tg = g_lo + i * dt
x, y, z = marker_position(key, tg)
points.append({"t_global_s": round(tg, 6), "x": x, "y": y, "z": z})
markers.append({"marker_key": key, "points": points})
return {"markers": markers, "generated_by": "festival4d.synthetic"}
def render_blink_clip(out_path: Path, marker_id: int = MARKER_B_ID, fps: float = 30.0,
duration_s: float = 3.0, width: int = 320, height: int = 240) -> Path:
"""Render a small clip of a single blinking badge at frame center (phase 6 / M24 helper).
A fixed-position bright disc blinking ``marker_id`` in the frozen OOK protocol on a black
background no camera geometry, so lane K's ``badge_selftest --synthetic`` can round-trip
the ID through ``tracker_detect``'s blink decoder without a full fixture. No audio track.
"""
import cv2
if shutil.which("ffmpeg") is None:
raise RuntimeError("ffmpeg not found on PATH — required to render a blink clip")
out_path = Path(out_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
radius = _disc_radius_px(height)
cx, cy = width // 2, height // 2
n = int(round(duration_s * fps))
with tempfile.TemporaryDirectory() as tmp:
frames_dir = Path(tmp)
for k in range(n):
t = k / fps
img = np.zeros((height, width, 3), dtype=np.uint8) # black background
if tracker_detect.led_on(t, marker_id):
cv2.circle(img, (cx, cy), radius, (255, 255, 255), thickness=-1,
lineType=cv2.LINE_8)
cv2.imwrite(str(frames_dir / f"{k:05d}.png"), img)
cmd = [
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-framerate", str(fps), "-start_number", "0",
"-i", str(frames_dir / "%05d.png"),
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-preset", "veryfast", "-crf", "20",
"-movflags", "+faststart",
str(out_path),
]
subprocess.run(cmd, check=True)
return out_path
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Video rendering (ffmpeg) # Video rendering (ffmpeg)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@ -396,8 +605,15 @@ def _has_drawtext() -> bool:
def render_video(out_path: Path, wav_path: Path, cam_index: int, def render_video(out_path: Path, wav_path: Path, cam_index: int,
width: int, height: int, fps: float, duration_s: float) -> None: width: int, height: int, fps: float, duration_s: float,
"""Render one synthetic clip: a distinct ``testsrc2`` visual muxed with ``wav_path``.""" poses: list[dict] | None = None, offset_ms: float = 0.0) -> None:
"""Render one synthetic clip: a distinct ``testsrc2`` visual muxed with ``wav_path``.
When ``poses`` is given (this camera's ``camera_track``), the two moving markers (M18) are
composited on top via a per-frame RGBA overlay produced by :func:`render_marker_overlay`
(drawn AFTER the base visual + label, so existing behavior/tests are untouched). ``offset_ms``
aligns the marker schedule to global time for this camera.
"""
if shutil.which("ffmpeg") is None: if shutil.which("ffmpeg") is None:
raise RuntimeError("ffmpeg not found on PATH — required for the synthetic fixture") raise RuntimeError("ffmpeg not found on PATH — required for the synthetic fixture")
out_path.parent.mkdir(parents=True, exist_ok=True) out_path.parent.mkdir(parents=True, exist_ok=True)
@ -410,11 +626,7 @@ def render_video(out_path: Path, wav_path: Path, cam_index: int,
"x=24:y=24:fontsize=44:fontcolor=white:box=1:boxcolor=black@0.5") "x=24:y=24:fontsize=44:fontcolor=white:box=1:boxcolor=black@0.5")
src = f"testsrc2=size={width}x{height}:rate={fps}:duration={duration_s}" src = f"testsrc2=size={width}x{height}:rate={fps}:duration={duration_s}"
cmd = [ common_out = [
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", src,
"-i", str(wav_path),
"-vf", vf,
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-preset", "veryfast", "-crf", "28", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-preset", "veryfast", "-crf", "28",
"-profile:v", "baseline", "-level", "3.1", "-profile:v", "baseline", "-level", "3.1",
"-movflags", "+faststart", "-movflags", "+faststart",
@ -422,7 +634,33 @@ def render_video(out_path: Path, wav_path: Path, cam_index: int,
"-shortest", "-shortest",
str(out_path), str(out_path),
] ]
subprocess.run(cmd, check=True)
if poses is None:
cmd = [
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", src,
"-i", str(wav_path),
"-vf", vf,
*common_out,
]
subprocess.run(cmd, check=True)
return
# Markers: pre-render the RGBA overlay frames, then composite them over the base visual.
with tempfile.TemporaryDirectory() as ov:
ov_dir = Path(ov)
render_marker_overlay(ov_dir, poses, offset_ms, width, height, fps, duration_s)
cmd = [
"ffmpeg", "-y", "-hide_banner", "-loglevel", "error",
"-f", "lavfi", "-i", src, # 0: base video
"-i", str(wav_path), # 1: audio
"-framerate", str(fps), "-start_number", "0",
"-i", str(ov_dir / "%05d.png"), # 2: marker overlay
"-filter_complex", f"[0:v]{vf}[bg];[bg][2:v]overlay=0:0:eof_action=pass[vo]",
"-map", "[vo]", "-map", "1:a",
*common_out,
]
subprocess.run(cmd, check=True)
def ffprobe_video(path: Path) -> dict: def ffprobe_video(path: Path) -> dict:
@ -481,11 +719,15 @@ def build(base_dir: Path | str | None = None, duration_s: float | None = None,
filename = f"cam{i}.mp4" filename = f"cam{i}.mp4"
out_path = raw / filename out_path = raw / filename
clip = slice_for_offset(master, sr, offset_ms, duration) clip = slice_for_offset(master, sr, offset_ms, duration)
poses = camera_track(i, duration, fps, w, h)
if run_ffmpeg: if run_ffmpeg:
wav_path = tmp / f"cam{i}.wav" wav_path = tmp / f"cam{i}.wav"
write_wav(wav_path, clip, sr) write_wav(wav_path, clip, sr)
render_video(out_path, wav_path, i, w, h, fps, duration) # Pass this camera's poses + offset so the M18 markers render through the same
# projection lane H inverts (the marker pixel derives from its 3D truth).
render_video(out_path, wav_path, i, w, h, fps, duration,
poses=poses, offset_ms=float(offset_ms))
probed = ffprobe_video(out_path) probed = ffprobe_video(out_path)
vw, vh, vfps, vdur = (probed["width"], probed["height"], vw, vh, vfps, vdur = (probed["width"], probed["height"],
probed["fps"], probed["duration_s"]) probed["fps"], probed["duration_s"])
@ -496,7 +738,7 @@ def build(base_dir: Path | str | None = None, duration_s: float | None = None,
filename=filename, duration_s=vdur, fps=vfps, width=vw, height=vh, filename=filename, duration_s=vdur, fps=vfps, width=vw, height=vh,
offset_ms=float(offset_ms), drift_ppm=0.0, sync_confidence=1.0, offset_ms=float(offset_ms), drift_ppm=0.0, sync_confidence=1.0,
) )
db.set_poses(video.id, camera_track(i, duration, fps, w, h)) db.set_poses(video.id, poses)
videos_gt.append({ videos_gt.append({
"video_id": video.id, "filename": filename, "video_id": video.id, "filename": filename,
"offset_ms": float(offset_ms), "drift_ppm": 0.0, "offset_ms": float(offset_ms), "drift_ppm": 0.0,
@ -537,17 +779,28 @@ def build(base_dir: Path | str | None = None, duration_s: float | None = None,
gt_path = work / "ground_truth.json" gt_path = work / "ground_truth.json"
gt_path.write_text(json.dumps(ground_truth, indent=2)) gt_path.write_text(json.dumps(ground_truth, indent=2))
# Friend-track ground truth (M18): the two markers' 3D paths over the union of the
# cameras' global coverage. Written for both ffmpeg and no-ffmpeg builds — it is pure
# geometry (the same paths the markers are rendered from), so tests can check it cheaply.
offsets_s = [o / 1000.0 for o in config.SYNTH_OFFSETS_MS]
track_truth = build_track_truth(min(offsets_s), max(offsets_s) + duration)
tt_path = work / "track_truth.json"
tt_path.write_text(json.dumps(track_truth, indent=2))
summary = { summary = {
"base_dir": str(base), "base_dir": str(base),
"videos": [v["filename"] for v in videos_gt], "videos": [v["filename"] for v in videos_gt],
"points_ply": str(work / "points.ply"), "points_ply": str(work / "points.ply"),
"point_count": int(len(points)), "point_count": int(len(points)),
"ground_truth": str(gt_path), "ground_truth": str(gt_path),
"track_truth": str(tt_path),
"markers": [m["marker_key"] for m in track_truth["markers"]],
"events": len(SEED_EVENTS), "events": len(SEED_EVENTS),
"anchors": len(STAGE_CORNERS), "anchors": len(STAGE_CORNERS),
} }
log.info("synthetic: done — %d videos, %d points, %d events, %d anchors", log.info("synthetic: done — %d videos, %d points, %d events, %d anchors, %d markers",
len(videos_gt), len(points), len(SEED_EVENTS), len(STAGE_CORNERS)) len(videos_gt), len(points), len(SEED_EVENTS), len(STAGE_CORNERS),
len(track_truth["markers"]))
return summary return summary

View File

@ -0,0 +1,171 @@
"""Marker detection (spec M19, lane H) — STUB with the frozen contracts baked in.
foundation3 ships this file with:
- the **concrete, foundation-owned protocol constants + pure encoders** every side of the
system must agree on (the blink OOK protocol #4, the color-mode marker-key convention #1).
``synthetic.py`` renders the fixture markers with these, ``scripts/badge_selftest.py``
(lane K) imports this decoder path, and ``hardware/badge/protocol.h`` (lane K) duplicates
the constants with a documented copy-check. **One source of truth lives here.**
- a **stub** :func:`run_detect` that raises ``NotImplementedError``. Lane H fills the body
(and adds ``backend/tests/test_tracker_detect.py``); it must not change the constants or
the emitted ``marker_key`` conventions, only implement the detection logic.
Nothing in this module raises at import time ``synthetic.py`` imports the constants and the
``encode_word`` / ``led_on`` helpers, so the import must stay side-effect-free and light
(no OpenCV at module scope; import ``cv2`` inside ``run_detect``).
================================================================================
FROZEN CONTRACTS (plan/30-phase6.md copy, do not reinterpret)
================================================================================
Contract #1 — Marker hues (color mode)
--------------------------------------
Magenta ``#ff00ff`` and cyan ``#00ffff`` are the two supported fixture hues; real shoots may
configure others via ``FESTIVAL4D_MARKER_HUES`` (comma-separated hex). Detection converts to
HSV and gates on **saturation AND value before hue** (pitfall #2 — stage lighting lies about
color). A color blob's ``marker_key`` is ``"hue:<H>"`` where ``<H>`` is the **OpenCV hue**
(0..179) of the configured marker color, i.e. ``round(matplotlib_hue_degrees / 2)``:
magenta ``hue:150``, cyan ``hue:90``. Use :func:`color_marker_key` never format the
string by hand.
Contract #2 — Schema (what the solver, not the detector, writes; here for context)
----------------------------------------------------------------------------------
``tracks(id INTEGER PK, marker_key TEXT NOT NULL, label TEXT, color TEXT, created_at TEXT)``
``track_points(id INTEGER PK, track_id FK, t_global_s FLOAT, x FLOAT, y FLOAT, z FLOAT,
quality FLOAT, views INTEGER)`` points ordered by ``t_global_s``;
``views`` = cameras used (1 ground-plane fallback); coordinates in **Three.js scene space**.
Contract #3 — detections.json (this module's OUTPUT — write via `db`? No: a JSON file)
--------------------------------------------------------------------------------------
Write ``config.DETECTIONS_JSON`` (``data/work/detections.json``) with EXACTLY this shape::
{"videos": {"<video_id>": [{"t_video_s": float, "marker_key": str,
"cx": float, "cy": float, "area": float,
"conf": float 0..1}]},
"generated_by": "festival4d.tracker_detect"}
``cx`` / ``cy`` are **normalized 0..1** (the bbox-annotation convention: ``cx = px / width``,
``cy = py / height``). ``area`` is the blob area in pixels. Detections at ``t_video`` (the
frame's local time) — the solver converts to ``t_global`` with the FROZEN
``config.t_global_from_video`` helper (pitfall #1). No markers / dark / markerless video →
a valid-empty ``{"videos": {}, "generated_by": ...}``, never a crash.
Contract #4 — Blink protocol (OOK, camera-decodable)
----------------------------------------------------
Bit period **133 ms** (4 frames @ 30 fps). One **word** = preamble ``11100`` + 6-bit ID
(MSB first) + even parity = **12 bits**, repeated continuously (~1.6 s/word). ID **63** is
reserved for the sync beacon. Decoding needs **24 fps** video and **1 full word** of
visibility. Decode by **integrating brightness per bit window using the video's real fps**
(``meta.fps``) never by counting frames (pitfall #3). Emit ``marker_key = "code:<id>"``
via :func:`code_marker_key`. The constants below are the byte-identical source for
``hardware/badge/protocol.h`` (lane K).
Contract #5 — Ground plane (solver context)
-------------------------------------------
Synthetic ground is ``y = 0`` (scene space). Real projects estimate it as the
5th-percentile-y horizontal plane of the point cloud. Single-view fallback intersects the
pixel ray with this plane; such points get ``views = 1`` and ``quality 0.5``.
"""
from __future__ import annotations
import logging
import os
log = logging.getLogger("festival4d.tracker_detect")
# ---------------------------------------------------------------------------
# Contract #4 — blink OOK protocol. FROZEN. Byte-identical to hardware/badge/protocol.h.
# ---------------------------------------------------------------------------
BLINK_BIT_PERIOD_S: float = 0.133 # 133 ms bit period (≈4 frames @ 30 fps)
BLINK_PREAMBLE: tuple[int, ...] = (1, 1, 1, 0, 0) # start-of-word marker
BLINK_ID_BITS: int = 6 # payload width (0..63)
BLINK_WORD_BITS: int = len(BLINK_PREAMBLE) + BLINK_ID_BITS + 1 # +1 parity = 12
BLINK_BEACON_ID: int = 63 # reserved: the sync/calibration beacon
BLINK_MIN_FPS: float = 24.0 # decoder needs at least this frame rate
# ---------------------------------------------------------------------------
# Contract #1 — color mode marker hues. OpenCV hue is 0..179 (degrees / 2).
# ---------------------------------------------------------------------------
MAGENTA_OPENCV_HUE: int = 150 # #ff00ff
CYAN_OPENCV_HUE: int = 90 # #00ffff
DEFAULT_MARKER_HUES_HEX: tuple[str, ...] = ("#ff00ff", "#00ffff")
def encode_word(marker_id: int) -> list[int]:
"""The 12-bit OOK word for ``marker_id`` (contract #4). Pure; no I/O.
``preamble(11100) + 6-bit ID MSB-first + even-parity bit``. The parity bit is chosen so
the **whole word has an even number of 1-bits** (parity over preamble+ID). This is the
exact bit sequence a badge emits, ``synthetic.py`` renders, and lane H's decoder inverts.
>>> encode_word(5)
[1, 1, 1, 0, 0, 0, 0, 0, 1, 0, 1, 1]
"""
mid = int(marker_id)
if not 0 <= mid < (1 << BLINK_ID_BITS):
raise ValueError(f"marker_id {marker_id} out of range 0..{(1 << BLINK_ID_BITS) - 1}")
id_bits = [(mid >> (BLINK_ID_BITS - 1 - i)) & 1 for i in range(BLINK_ID_BITS)]
body = list(BLINK_PREAMBLE) + id_bits
parity = sum(body) % 2 # even parity: word weight (incl. parity) is even
return body + [parity]
def led_on(t_seconds: float, marker_id: int, epoch: float = 0.0) -> bool:
"""Whether the badge LED for ``marker_id`` is lit at wall-clock time ``t_seconds``.
The word repeats continuously from ``epoch``; a single physical badge blinks in real
(global) time, so every camera that captures a frame at the same ``t_global`` sees the
same LED state. Used by the synthetic fixture to render marker B faithfully.
"""
bit_index = int((t_seconds - epoch) // BLINK_BIT_PERIOD_S)
word = encode_word(marker_id)
return bool(word[bit_index % BLINK_WORD_BITS])
def color_marker_key(opencv_hue: int) -> str:
"""The ``marker_key`` for a color blob of the given OpenCV hue (contract #1)."""
return f"hue:{int(opencv_hue)}"
def code_marker_key(marker_id: int) -> str:
"""The ``marker_key`` for a decoded blink ID (contract #4)."""
return f"code:{int(marker_id)}"
def marker_hues_hex() -> list[str]:
"""Configured color-mode hues as hex strings (``FESTIVAL4D_MARKER_HUES`` or the default)."""
env = os.environ.get("FESTIVAL4D_MARKER_HUES")
if env:
hues = [h.strip() for h in env.split(",") if h.strip()]
if hues:
return hues
return list(DEFAULT_MARKER_HUES_HEX)
# ---------------------------------------------------------------------------
# Lane H fills these. Signatures FROZEN.
# ---------------------------------------------------------------------------
def run_detect(sample_fps: float = 8.0) -> dict:
"""Detect markers in every video and write ``detections.json`` (spec M19, contract #3).
Lane H implements:
1. Sample frames per video with the ``frames.py`` machinery at ~``sample_fps`` fps
(document the exact rate; ~8 fps comfortably oversamples the 133 ms bit period).
2. **Color mode:** BGRHSV; gate on saturation AND value *before* hue (pitfall #2);
connected components per configured hue (:func:`marker_hues_hex`); blob centroid
normalized ``cx``/``cy``; ``marker_key = color_marker_key(<opencv_hue>)``.
3. **Blink mode:** track bright blobs across the sampled window; integrate brightness
per :data:`BLINK_BIT_PERIOD_S` window using the video's **real fps** (``meta.fps``,
pitfall #3); correlate against :data:`BLINK_PREAMBLE`; check even parity; emit
``marker_key = code_marker_key(<id>)``.
4. Write ``config.DETECTIONS_JSON`` in the contract-#3 shape. No markers / dark /
markerless valid-empty ``{"videos": {}, "generated_by": "festival4d.tracker_detect"}``.
Returns a summary dict (per-video detection counts, decoded blink IDs). Never crash on a
markerless or unreadable video log and continue.
"""
raise NotImplementedError(
"marker detection not implemented yet (lane H / M19) — fill tracker_detect.run_detect"
)

View File

@ -0,0 +1,72 @@
"""Track solving (spec M20, lane H) — STUB with the frozen contract in the docstring.
foundation3 ships the stub; lane H fills :func:`run_solve` (and adds
``backend/tests/test_tracker_solve.py``). This is pure orchestration over the FROZEN
geometry primitives it must NEVER inline pose math (read ``resolve.py`` first; it is the
exact per-timestep pattern), and it writes through the ``db`` track helpers only.
================================================================================
FROZEN CONTRACTS (plan/30-phase6.md copy, do not reinterpret)
================================================================================
Input ``config.DETECTIONS_JSON`` (contract #3, written by ``tracker_detect.run_detect``)::
{"videos": {"<video_id>": [{"t_video_s", "marker_key", "cx", "cy", "area", "conf"}]},
"generated_by": "festival4d.tracker_detect"}
``cx``/``cy`` are normalized 0..1; multiply by the video's width/height for pixel coords.
Output the ``tracks`` / ``track_points`` tables (contract #2), via ``db`` helpers::
tracks(id, marker_key, label, color, created_at)
track_points(id, track_id, t_global_s, x, y, z, quality, views) # ordered by t_global_s
# coordinates in Three.js scene space; views = cameras used (1 ⇒ ground-plane fallback)
Algorithm (spec M20 the parts that bite):
1. Load detections; group by ``marker_key``.
2. **Time alignment is the whole game** (pitfall #1): a detection is found at ``t_video``;
association happens at ``t_global``. Convert with the FROZEN
``config.t_global_from_video(t_video, offset_ms, drift_ppm)`` ONLY never re-derive it.
3. Per time bucket (**0.25 s** of ``t_global``), collect same-``marker_key`` detections
across cameras at matching ``t_global``:
- **2+ views triangulate** through ``geometry`` primitives exactly as ``resolve.py``
does (``geometry.ray_from_pixel`` per view + ``geometry.triangulate_rays``, or a
multi-ray least squares built on them). Record ``views`` and a ``quality`` derived
from the residual/gap.
- **1 view ray ground plane** (contract #5): the synthetic ground is ``y = 0``;
real projects estimate the 5th-percentile-y horizontal plane of the point cloud and
record it in the result dict. Such points get ``views = 1`` and ``quality 0.5``.
4. Smooth each track (moving average or Catmull-Rom your call; **document which**).
5. Solid color for hue markers, palette color for code markers.
6. Write via ``db.add_track`` + ``db.set_track_points``. **Re-solve replaces that marker's
track** (idempotent never stack duplicates on repeated runs).
7. Degradation: no detections zero tracks + a note; never crash.
``POST /api/tracks/solve`` and ``python -m festival4d track`` both run detectsolve
end-to-end once this and ``tracker_detect`` are filled.
"""
from __future__ import annotations
import logging
log = logging.getLogger("festival4d.tracker_solve")
# Poses are keyed every ~0.5 s; bucket association at 0.25 s (spec M20). Kept here so lane H
# and any test reference one number.
BUCKET_S: float = 0.25
GROUND_PLANE_Y: float = 0.0 # synthetic fixture ground (contract #5, scene space)
SINGLE_VIEW_MAX_QUALITY: float = 0.5 # ground-plane fallback quality cap (contract #5)
def run_solve() -> dict:
"""Solve detections into 3D friend tracks (spec M20, see module docstring).
Lane H implements the full detect-output ``tracks``/``track_points`` pipeline described
above. Returns a summary dict (tracks written, per-track point counts + coverage, the
ground-plane used, and any degradation note). Idempotent: re-running replaces each
marker's track rather than stacking duplicates.
"""
raise NotImplementedError(
"track solving not implemented yet (lane H / M20) — fill tracker_solve.run_solve"
)

View File

@ -0,0 +1,26 @@
# Status — foundation3 (Phase 6a / M18)
## Round 6 — 2026-07-17 — STATUS: in_progress
**Directives acknowledged:** round 6 of plan/DIRECTIVES.md. Directive #1: `foundation3` only,
branch `foundation3`, push — do NOT merge. Contracts correctness beats speed; keep 189 green.
**Acceptance checklist** (M18, plan/30-phase6.md §M18):
- [ ] 1. Synthetic fixture: two moving markers rendered via the existing camera projection;
`data/work/track_truth.json` derived through that projection (Three.js scene space)
- [ ] 2. DB: `tracks` + `track_points` (contract #2) + helpers add/get/set_points/patch/delete
- [ ] 3. API: GET /api/tracks, PATCH/DELETE /api/tracks/{id}, POST /api/tracks/solve (degrades),
manifest `has_tracks` + conscious test_api key-set edit + CR
- [ ] 4. CLI: `track [--detect-only|--solve-only]`
- [ ] 5. Backend stubs: tracker_detect.py / tracker_solve.py (NotImplementedError + full contracts)
- [ ] 6. Frontend: friendTracks.js stub wired into main.js loop; #btn-friends + #friends-panel;
state.tracks loaded when manifest.has_tracks
- [ ] 7. Capsule: bake api/tracks (additive) + conscious bundle-structure test edit + CR
- [ ] 8. Tests for every new route/helper; suite ≥ 189 + new; all green
**Done this round:** branch `foundation3` cut from main (HEAD 56e569f). Read all canonical
specs + every file M18 touches. Baseline confirmed pending (running `uv run pytest -q`).
**Blockers / questions for coordinator:** none yet.
**Next:** implement steps 18 in order; re-run full suite after the fixture edit and at the end.