8f2a488c51
Completes M18 on top of the WIP checkpoint (6a703c3). Three lanes (H/J/K) depend only on
what this freezes.
- synthetic.py: base testsrc2 saturation/brightness-muted so its magenta/cyan colour bars
don't swamp lane H's colour detector; the two moving markers stay the only detectable blobs.
- Frontend (step 6): friendTracks.js stub (never-throws, fx.js pattern) wired into main.js;
hidden #btn-friends + #friends-panel; state.tracks loaded when manifest.has_tracks.
- Tests (step 8): test_tracks_foundation.py (20) — DB helpers, /api/tracks routes, solve
degradation, manifest.has_tracks, frozen OOK protocol + encoders, track_truth shape, the
projection round-trip (rendered pixels back-project to 3D truth, worst err <1e-3 → lane H's
median<0.3 reachable by construction), and a real-ffmpeg render test proving marker A is the
dominant magenta blob after base-muting.
- CR-6: the two sanctioned M18 test-lock edits (test_api manifest key-set +has_tracks;
test_capsule bundle paths +api/tracks). Additive; flagged for integration3 ratification.
Suite: 14 pre-existing files sum to 189 (floor held) + 20 new = 209 passed, exit 0.
Frontend `npm run build` clean. Do NOT merge — coordinator reviews + merges (DIRECTIVES R6 #3).
Recovered by the coordinator session after the foundation3 agent's isolation worktree was
removed mid-run, leaving the work uncommitted; every file independently reviewed, full suite +
build re-run green, then committed. No work redone.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
352 lines
16 KiB
Python
352 lines
16 KiB
Python
"""foundation3 (phase 6 / M18) contract tests — the ground three lanes stand on.
|
|
|
|
Covers what foundation3 owns and freezes, NOT the lane implementations (detection = lane H's
|
|
test_tracker_detect.py; solving = test_tracker_solve.py; ribbons = lane J's live-browser
|
|
evidence):
|
|
|
|
A. tracks / track_points DB helpers (contract #2) — round-trips, replace + delete semantics.
|
|
B. /api/tracks routes (contract #2) + POST /api/tracks/solve degradation + manifest.has_tracks.
|
|
C. The frozen OOK blink protocol + color marker_key convention (contracts #1, #4) — pure,
|
|
no OpenCV. These constants are byte-identical to hardware/badge/protocol.h (lane K).
|
|
D. The synthetic fixture (M18 step 1): track_truth.json shape, marker paths, and — the one
|
|
that matters most — a projection ROUND-TRIP proving the rendered marker pixels
|
|
back-project to the ground-truth 3D point through the frozen geometry primitives. If this
|
|
passes, lane H's M20 recovery (median 3D error < 0.3) is achievable by construction.
|
|
|
|
Hermetic: no ffmpeg (geometry only), temp data dir from conftest. DB-touching tests clean up
|
|
after themselves so later modules see the tables as they found them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import shutil
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from festival4d import config, db, synthetic, tracker_detect
|
|
|
|
_HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
|
|
|
|
|
|
# ===========================================================================
|
|
# A. DB helpers (contract #2)
|
|
# ===========================================================================
|
|
def _use_temp_db(tmp_path):
|
|
db.init_engine(tmp_path / "t.db")
|
|
db.reset_db()
|
|
|
|
|
|
def test_add_and_get_tracks_ordered_by_id(tmp_path):
|
|
_use_temp_db(tmp_path)
|
|
a = db.add_track("hue:150", label="Alice", color="#ff00ff")
|
|
b = db.add_track("code:5")
|
|
assert db.has_tracks() is True
|
|
tracks = db.get_tracks()
|
|
assert [t.id for t in tracks] == sorted(t.id for t in tracks) == [a.id, b.id]
|
|
assert tracks[0].marker_key == "hue:150" and tracks[0].label == "Alice"
|
|
assert tracks[1].label is None and tracks[1].color is None # optional fields default None
|
|
assert db.get_track(a.id).marker_key == "hue:150"
|
|
assert db.get_track(999999) is None
|
|
|
|
|
|
def test_has_tracks_false_on_empty(tmp_path):
|
|
_use_temp_db(tmp_path)
|
|
assert db.has_tracks() is False
|
|
assert db.get_tracks() == []
|
|
|
|
|
|
def test_set_track_points_replaces_and_orders(tmp_path):
|
|
_use_temp_db(tmp_path)
|
|
t = db.add_track("code:5")
|
|
# Insert out of order; helper must return them sorted by t_global_s (contract #2).
|
|
n = db.set_track_points(t.id, [
|
|
{"t_global_s": 2.0, "x": 1, "y": 2, "z": 3, "quality": 0.9, "views": 2},
|
|
{"t_global_s": 1.0, "x": 0, "y": 0, "z": 0, "quality": 0.4, "views": 1},
|
|
])
|
|
assert n == 2
|
|
pts = db.get_track_points(t.id)
|
|
assert [p.t_global_s for p in pts] == [1.0, 2.0]
|
|
assert pts[0].views == 1 and pts[0].quality == 0.4
|
|
assert pts[1].views == 2
|
|
# Re-solve replaces (idempotent — never stacks duplicates, per M20 contract).
|
|
db.set_track_points(t.id, [{"t_global_s": 5.0, "x": 9, "y": 9, "z": 9}])
|
|
pts = db.get_track_points(t.id)
|
|
assert len(pts) == 1 and pts[0].t_global_s == 5.0
|
|
assert pts[0].quality is None and pts[0].views is None # omitted optionals -> None
|
|
|
|
|
|
def test_set_track_points_unknown_track_raises(tmp_path):
|
|
_use_temp_db(tmp_path)
|
|
with pytest.raises(KeyError):
|
|
db.set_track_points(424242, [{"t_global_s": 0.0, "x": 0, "y": 0, "z": 0}])
|
|
|
|
|
|
def test_patch_track_partial_and_missing(tmp_path):
|
|
_use_temp_db(tmp_path)
|
|
t = db.add_track("hue:150", label="Old", color="#111111")
|
|
db.patch_track(t.id, label="New") # color untouched
|
|
got = db.get_track(t.id)
|
|
assert got.label == "New" and got.color == "#111111"
|
|
db.patch_track(t.id, color="#00ffff")
|
|
assert db.get_track(t.id).color == "#00ffff"
|
|
with pytest.raises(KeyError):
|
|
db.patch_track(424242, label="x")
|
|
|
|
|
|
def test_delete_track_removes_points(tmp_path):
|
|
_use_temp_db(tmp_path)
|
|
t = db.add_track("code:5")
|
|
db.set_track_points(t.id, [{"t_global_s": 0.0, "x": 0, "y": 0, "z": 0}])
|
|
assert db.delete_track(t.id) is True
|
|
assert db.get_track(t.id) is None
|
|
assert db.get_track_points(t.id) == [] # points cascade (SQLite doesn't by default)
|
|
assert db.delete_track(t.id) is False # already gone
|
|
|
|
|
|
# ===========================================================================
|
|
# B. /api/tracks routes (contract #2)
|
|
# ===========================================================================
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
from fastapi.testclient import TestClient
|
|
|
|
from festival4d import api
|
|
|
|
with TestClient(api.app) as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture()
|
|
def _api_db(client):
|
|
"""Point the API at a clean per-test db and restore the shared engine afterwards."""
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
d = Path(tempfile.mkdtemp(prefix="tracks-api-"))
|
|
db.init_engine(d / "api.db")
|
|
db.reset_db()
|
|
yield
|
|
db.init_engine(config.DB_PATH)
|
|
|
|
|
|
def test_get_tracks_shape(client, _api_db):
|
|
t = db.add_track("hue:150", label="Alice", color="#ff00ff")
|
|
db.set_track_points(t.id, [
|
|
{"t_global_s": 1.0, "x": 0.1, "y": 1.5, "z": 2.0, "quality": 0.9, "views": 2},
|
|
{"t_global_s": 1.25, "x": 0.2, "y": 1.5, "z": 1.9, "quality": 0.5, "views": 1},
|
|
])
|
|
listing = client.get("/api/tracks").json()
|
|
assert len(listing) == 1
|
|
row = listing[0]
|
|
assert set(row) == {"id", "marker_key", "label", "color", "points"}
|
|
assert row["marker_key"] == "hue:150" and row["label"] == "Alice"
|
|
assert [p["t_global_s"] for p in row["points"]] == [1.0, 1.25] # ordered
|
|
assert set(row["points"][0]) == {"t_global_s", "x", "y", "z", "quality", "views"}
|
|
assert row["points"][1]["views"] == 1
|
|
|
|
|
|
def test_patch_and_delete_routes(client, _api_db):
|
|
t = db.add_track("code:5")
|
|
patched = client.patch(f"/api/tracks/{t.id}", json={"label": "Bob", "color": "#00ffff"}).json()
|
|
assert patched["label"] == "Bob" and patched["color"] == "#00ffff"
|
|
# partial patch leaves the other field
|
|
assert client.patch(f"/api/tracks/{t.id}", json={"color": "#fff"}).json()["label"] == "Bob"
|
|
assert client.patch("/api/tracks/999999", json={"label": "x"}).status_code == 404
|
|
assert client.delete(f"/api/tracks/{t.id}").json() == {"deleted": t.id}
|
|
assert client.get("/api/tracks").json() == []
|
|
assert client.delete(f"/api/tracks/{t.id}").status_code == 404
|
|
|
|
|
|
def test_solve_degrades_while_stubbed(client, _api_db):
|
|
# Lane H is a stub (NotImplementedError). The frozen route must NOT 500 — it returns a
|
|
# valid empty result + a note (the house pattern, mirrors POST /api/events/detect).
|
|
resp = client.post("/api/tracks/solve")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["result"] is None
|
|
assert "not implemented" in body["note"].lower()
|
|
assert body["tracks"] == [] # no tracks solved yet
|
|
|
|
|
|
def test_manifest_has_tracks_reflects_db(client, _api_db):
|
|
assert client.get("/api/manifest").json()["has_tracks"] is False
|
|
db.add_track("hue:150")
|
|
assert client.get("/api/manifest").json()["has_tracks"] is True
|
|
|
|
|
|
# ===========================================================================
|
|
# C. Frozen blink OOK protocol + color marker_key (contracts #1, #4)
|
|
# ===========================================================================
|
|
def test_protocol_constants_match_spec():
|
|
assert tracker_detect.BLINK_BIT_PERIOD_S == 0.133
|
|
assert tracker_detect.BLINK_PREAMBLE == (1, 1, 1, 0, 0)
|
|
assert tracker_detect.BLINK_ID_BITS == 6
|
|
assert tracker_detect.BLINK_WORD_BITS == 12 # 5 preamble + 6 id + 1 parity
|
|
assert tracker_detect.BLINK_BEACON_ID == 63
|
|
assert tracker_detect.BLINK_MIN_FPS == 24.0
|
|
assert tracker_detect.MAGENTA_OPENCV_HUE == 150 # #ff00ff, hue 300°/2
|
|
assert tracker_detect.CYAN_OPENCV_HUE == 90 # #00ffff, hue 180°/2
|
|
|
|
|
|
def test_encode_word_id5_exact_and_even_parity():
|
|
word = tracker_detect.encode_word(5)
|
|
assert word == [1, 1, 1, 0, 0, 0, 0, 0, 1, 0, 1, 1] # preamble + 000101 + parity
|
|
assert len(word) == 12
|
|
for mid in range(0, 64):
|
|
w = tracker_detect.encode_word(mid)
|
|
assert tuple(w[:5]) == tracker_detect.BLINK_PREAMBLE
|
|
assert sum(w) % 2 == 0 # even parity over the whole word
|
|
# 6-bit ID MSB-first recovers the value
|
|
recovered = sum(bit << (5 - i) for i, bit in enumerate(w[5:11]))
|
|
assert recovered == mid
|
|
|
|
|
|
def test_encode_word_rejects_out_of_range():
|
|
for bad in (-1, 64, 100):
|
|
with pytest.raises(ValueError):
|
|
tracker_detect.encode_word(bad)
|
|
|
|
|
|
def test_led_on_matches_encoded_word_windows():
|
|
# The LED state at bit-window k equals bit k of the repeating word (global time).
|
|
word = tracker_detect.encode_word(5)
|
|
period = tracker_detect.BLINK_BIT_PERIOD_S
|
|
for k in range(len(word) * 2): # two full words
|
|
t_mid = (k + 0.5) * period # sample mid-window
|
|
assert tracker_detect.led_on(t_mid, 5) == bool(word[k % len(word)])
|
|
|
|
|
|
def test_marker_key_helpers_and_hue_env(monkeypatch):
|
|
assert tracker_detect.color_marker_key(150) == "hue:150"
|
|
assert tracker_detect.code_marker_key(5) == "code:5"
|
|
monkeypatch.delenv("FESTIVAL4D_MARKER_HUES", raising=False)
|
|
assert tracker_detect.marker_hues_hex() == ["#ff00ff", "#00ffff"]
|
|
monkeypatch.setenv("FESTIVAL4D_MARKER_HUES", "#ff0000, #00ff00")
|
|
assert tracker_detect.marker_hues_hex() == ["#ff0000", "#00ff00"]
|
|
|
|
|
|
# ===========================================================================
|
|
# D. Synthetic fixture: track_truth shape, marker paths, projection round-trip
|
|
# ===========================================================================
|
|
def test_track_truth_shape_and_both_markers():
|
|
tt = synthetic.build_track_truth(0.0, 2.0, dt=0.05)
|
|
assert set(tt) >= {"markers"}
|
|
keys = {m["marker_key"] for m in tt["markers"]}
|
|
assert keys == {"hue:150", "code:5"} # marker A (color) + marker B (blink)
|
|
for m in tt["markers"]:
|
|
ts = [p["t_global_s"] for p in m["points"]]
|
|
assert ts == sorted(ts) # monotonically increasing
|
|
assert len(ts) == len(set(ts)) # no dup timesteps
|
|
assert set(m["points"][0]) == {"t_global_s", "x", "y", "z"} # Three.js scene space
|
|
|
|
|
|
def test_marker_paths_and_unknown_key():
|
|
# Marker A: a circle radius 2 at y=1.5 (contract, spec M18).
|
|
for tg in (0.0, 3.7, 11.2):
|
|
x, y, z = synthetic.marker_position("hue:150", tg)
|
|
assert y == pytest.approx(1.5)
|
|
assert math.hypot(x, z) == pytest.approx(2.0, abs=1e-9)
|
|
# Marker B follows a different path (never identical to A) so the two never lock.
|
|
assert synthetic.marker_position("code:5", 3.7) != synthetic.marker_position("hue:150", 3.7)
|
|
with pytest.raises(KeyError):
|
|
synthetic.marker_position("code:99", 0.0)
|
|
|
|
|
|
def test_marker_discs_are_small():
|
|
# Both discs together must be well under 2% of frame area (spec M18: ≤2%, pitfall #6).
|
|
w, h = config.SYNTH_VIDEO_W, config.SYNTH_VIDEO_H
|
|
r = synthetic._disc_radius_px(h)
|
|
two_discs = 2 * math.pi * r * r
|
|
assert two_discs / (w * h) < 0.02
|
|
|
|
|
|
def test_projection_round_trip_recovers_ground_truth():
|
|
"""THE foundation contract: rendered marker pixels back-project to the 3D truth.
|
|
|
|
For each marker at several t_global: project its ground-truth 3D point through every
|
|
camera's interpolated pose (exactly as render_marker_overlay draws the pixel), then treat
|
|
the in-frame pixels as a perfect detector's output and triangulate them back through the
|
|
FROZEN geometry primitives. Recovery must be ~exact on the noise-free fixture — this is
|
|
what makes lane H's M20 (median 3D error < 0.3) reachable by construction.
|
|
"""
|
|
from festival4d import geometry
|
|
|
|
w, h = config.SYNTH_VIDEO_W, config.SYNTH_VIDEO_H
|
|
fps, dur = float(config.SYNTH_FPS), config.SYNTH_DURATION_S
|
|
# Poses per camera, generated the same way build() does (no DB, no ffmpeg needed).
|
|
cam_poses = [synthetic.camera_track(i, dur, fps, w, h)
|
|
for i in range(len(config.SYNTH_OFFSETS_MS))]
|
|
|
|
worst = 0.0
|
|
checked = 0
|
|
for key in synthetic.MARKER_KEYS:
|
|
for t_global in (3.0, 6.5, 10.0, 14.25, 17.5): # inside every camera's valid span
|
|
truth = np.array(synthetic.marker_position(key, t_global))
|
|
rays = []
|
|
for i, offset_ms in enumerate(config.SYNTH_OFFSETS_MS):
|
|
t_video = config.t_video_from_global(t_global, offset_ms, 0.0)
|
|
q, t, intr = synthetic._interp_pose(cam_poses[i], t_video)
|
|
proj = synthetic.project_point(q, t, intr, truth)
|
|
if proj is None:
|
|
continue
|
|
px, py, _ = proj
|
|
if not (0.0 <= px < w and 0.0 <= py < h):
|
|
continue # off this camera's frame
|
|
fx, fy, cx, cy = intr
|
|
rays.append(geometry.ray_from_pixel(q, t, fx, fy, cx, cy, px, py))
|
|
assert len(rays) >= 2, f"{key} @ {t_global}s: need 2+ views, got {len(rays)}"
|
|
point, gap = geometry.triangulate_rays(rays[0][0], rays[0][1], rays[1][0], rays[1][1])
|
|
err = float(np.linalg.norm(point - truth))
|
|
worst = max(worst, err)
|
|
checked += 1
|
|
assert checked >= 8
|
|
# Noise-free synthetic geometry: recovery is essentially exact, far under the 0.3 M20 bar.
|
|
assert worst < 1e-3, f"worst round-trip error {worst:.2e} (>1e-3)"
|
|
|
|
|
|
@pytest.mark.skipif(not _HAS_FFMPEG, reason="ffmpeg/ffprobe required to render the fixture video")
|
|
def test_rendered_marker_a_is_the_dominant_magenta_blob(tmp_path):
|
|
"""The base-muting contract (BASE_SATURATION / BASE_BRIGHTNESS) actually renders a solvable
|
|
fixture: testsrc2's own solid magenta/cyan colour bars must NOT out-mass marker A's disc, or
|
|
lane H's colour detector locks onto a bar and M19/M20 fail. This emulates that detector —
|
|
largest saturated-magenta blob in a real rendered frame — and asserts it lands on the disc.
|
|
Guards against a future un-muting of the base swamping detection.
|
|
"""
|
|
import cv2
|
|
|
|
dur = 2.0
|
|
synthetic.build(base_dir=tmp_path, duration_s=dur, run_ffmpeg=True)
|
|
try:
|
|
w, h = config.SYNTH_VIDEO_W, config.SYNTH_VIDEO_H
|
|
poses = synthetic.camera_track(0, dur, float(config.SYNTH_FPS), w, h)
|
|
t_video = 1.0
|
|
q, t, intr = synthetic._interp_pose(poses, t_video)
|
|
t_global = config.t_global_from_video(t_video, config.SYNTH_OFFSETS_MS[0], 0.0)
|
|
proj = synthetic.project_point(q, t, intr,
|
|
synthetic.marker_position(synthetic.MARKER_A_KEY, t_global))
|
|
assert proj and 0 <= proj[0] < w and 0 <= proj[1] < h, "marker A not in cam0 frame at t=1s"
|
|
|
|
cap = cv2.VideoCapture(str(tmp_path / "raw" / "cam0.mp4"))
|
|
cap.set(cv2.CAP_PROP_POS_MSEC, t_video * 1000.0)
|
|
ok, img = cap.read()
|
|
cap.release()
|
|
assert ok, "could not read cam0 frame"
|
|
|
|
hsv = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
|
|
hh, ss, vv = hsv[:, :, 0].astype(int), hsv[:, :, 1], hsv[:, :, 2]
|
|
# OpenCV-hue magenta ≈150, gated on high saturation AND value (pitfall #2).
|
|
mask = (((hh >= 145) & (hh <= 155)) & (ss > 170) & (vv > 170)).astype(np.uint8)
|
|
n, _lab, stats, cent = cv2.connectedComponentsWithStats(mask, 8)
|
|
assert n > 1, "no saturated-magenta blob at all — marker A not rendered?"
|
|
biggest = 1 + int(np.argmax(stats[1:, cv2.CC_STAT_AREA]))
|
|
cx, cy = cent[biggest]
|
|
dist = math.hypot(cx - proj[0], cy - proj[1])
|
|
assert dist < 10.0, (
|
|
f"dominant magenta blob at ({cx:.0f},{cy:.0f}) is {dist:.0f}px from marker A's "
|
|
f"projected centre ({proj[0]:.0f},{proj[1]:.0f}) — base bars are swamping the disc"
|
|
)
|
|
finally:
|
|
db.init_engine(config.DB_PATH) # build() rebound the engine; restore the shared one
|