6a703c314d
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>
810 lines
34 KiB
Python
810 lines
34 KiB
Python
"""Synthetic fixture generator (spec M0) — the universal test bed.
|
|
|
|
``python -m festival4d synthetic`` builds a complete fake project so every later layer is
|
|
testable without real footage:
|
|
|
|
- **3 fake videos** (``cam0/1/2.mp4``, 640x360, 30 fps, 20 s), each a distinct ``testsrc2``
|
|
visual but sharing the *same* synthesized music-like audio, shifted by known offsets
|
|
(0, +1370, -842 ms). ``cam0`` is the reference (offset 0).
|
|
- **Fake poses**: 3 camera trajectories on arcs around a "stage" box at the origin,
|
|
looking at the stage, written into ``camera_poses`` (COLMAP world->camera convention).
|
|
- **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.
|
|
- **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
|
|
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
|
|
(audio shifts, poses, PLY round-trip, marker geometry) are unit-testable without invoking
|
|
ffmpeg.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import wave
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
|
|
from festival4d import config, db, tracker_detect
|
|
from festival4d.geometry import mat_to_quat, quat_to_mat, slerp_pose
|
|
|
|
log = logging.getLogger("festival4d.synthetic")
|
|
|
|
# Audio timeline: the master signal covers global time [G_START, G_END] so that even the
|
|
# negatively-offset video (started before global 0) samples a valid window.
|
|
G_START = -2.0
|
|
G_END = 24.0
|
|
BEAT_INTERVAL_S = 0.5 # regular beat clicks
|
|
BANG_TIMES_S = (3.0, 10.0, 17.0) # loud transients (M7 candidates); well separated
|
|
TONE_FREQS_HZ = (220.0, 330.0)
|
|
|
|
# Camera intrinsics (pinhole, after undistortion) for the synthetic cameras.
|
|
SYNTH_HFOV_DEG = 60.0
|
|
|
|
# Stage geometry (world coords, +y up, +z toward the cameras/audience).
|
|
STAGE_HALF_W = 3.0 # x extent
|
|
STAGE_HALF_D = 2.0 # z extent
|
|
STAGE_HEIGHT = 1.0 # y top
|
|
STAGE_TARGET = (0.0, 0.5, 0.0) # cameras aim here
|
|
|
|
STAGE_CORNERS = {
|
|
"Stage FL": (-STAGE_HALF_W, STAGE_HEIGHT, STAGE_HALF_D),
|
|
"Stage FR": (STAGE_HALF_W, STAGE_HEIGHT, STAGE_HALF_D),
|
|
"Stage BL": (-STAGE_HALF_W, STAGE_HEIGHT, -STAGE_HALF_D),
|
|
"Stage BR": (STAGE_HALF_W, STAGE_HEIGHT, -STAGE_HALF_D),
|
|
}
|
|
|
|
# Seeded timeline events (pre-classified) so lane C can render markers immediately.
|
|
SEED_EVENTS = [
|
|
(3.0, "bass_drop", 0.94, "Bass drops and the crowd erupts as the beat hits."),
|
|
(6.0, "crowd_wave", 0.72, "A wave ripples through the audience."),
|
|
(8.0, "quiet_moment", 0.61, "A brief hush before the next build."),
|
|
(10.0, "pyro", 0.88, "Pyro columns fire across the front of the stage."),
|
|
(13.0, "light_show", 0.90, "Sweeping beams sync to the breakdown."),
|
|
(17.0, "confetti", 0.81, "Confetti cannons burst over the crowd."),
|
|
(19.0, "artist_moment", 0.77, "The artist steps to the edge for a solo."),
|
|
]
|
|
|
|
# Fonts to try for the "CAM N" label (optional nicety; skipped if none exist).
|
|
_FONT_CANDIDATES = (
|
|
"/System/Library/Fonts/Supplemental/Arial.ttf",
|
|
"/System/Library/Fonts/Supplemental/Helvetica.ttc",
|
|
"/Library/Fonts/Arial.ttf",
|
|
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
|
|
)
|
|
|
|
_PLY_DTYPE = np.dtype(
|
|
[("x", "<f4"), ("y", "<f4"), ("z", "<f4"),
|
|
("red", "u1"), ("green", "u1"), ("blue", "u1")]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Audio
|
|
# ---------------------------------------------------------------------------
|
|
def synth_master_audio(sr: int = config.SYNTH_AUDIO_SR) -> tuple[np.ndarray, dict]:
|
|
"""Synthesize the master music-like signal over global time [G_START, G_END].
|
|
|
|
Returns the float32 signal (peak-normalized to ~0.95) and a ground-truth dict with the
|
|
sample rate, the window start, and the beat/bang times (global seconds).
|
|
"""
|
|
n = int(round((G_END - G_START) * sr))
|
|
t = G_START + np.arange(n) / sr
|
|
sig = np.zeros(n, dtype=np.float64)
|
|
|
|
# Low tone bed (kept quiet so it never dominates the transients GCC-PHAT keys on).
|
|
for f in TONE_FREQS_HZ:
|
|
sig += 0.05 * np.sin(2.0 * np.pi * f * t)
|
|
|
|
# Regular beat clicks: short 2 kHz bursts with a fast exponential decay -> sharp,
|
|
# broadband-ish transients that give a clean cross-correlation peak.
|
|
beat_times = np.arange(np.ceil(G_START / BEAT_INTERVAL_S) * BEAT_INTERVAL_S,
|
|
G_END, BEAT_INTERVAL_S)
|
|
for bt in beat_times:
|
|
_add_transient(sig, t, sr, center=bt, amp=0.25, freq=2000.0, decay_s=0.03)
|
|
|
|
# Loud "bangs" (bass drops / pyro): louder, longer, low-frequency thump + noise burst.
|
|
rng = np.random.default_rng(7)
|
|
for bang in BANG_TIMES_S:
|
|
_add_transient(sig, t, sr, center=bang, amp=0.9, freq=60.0, decay_s=0.18)
|
|
_add_noise_burst(sig, t, sr, center=bang, amp=0.5, decay_s=0.12, rng=rng)
|
|
|
|
# Faint noise floor for realism (SNR stays high so sync remains exact).
|
|
sig += 0.004 * rng.standard_normal(n)
|
|
|
|
peak = float(np.max(np.abs(sig)))
|
|
if peak > 0:
|
|
sig *= 0.95 / peak
|
|
|
|
ground_truth = {
|
|
"sample_rate": sr,
|
|
"g_start_s": G_START,
|
|
"g_end_s": G_END,
|
|
"beat_interval_s": BEAT_INTERVAL_S,
|
|
"bang_times_global_s": list(BANG_TIMES_S),
|
|
"pulse_times_global_s": list(BANG_TIMES_S), # alias lane D matches candidates against
|
|
}
|
|
return sig.astype(np.float32), ground_truth
|
|
|
|
|
|
def _add_transient(sig, t, sr, center, amp, freq, decay_s):
|
|
"""Add an exponentially decaying sinusoid burst centered at global time ``center``."""
|
|
length = int(decay_s * 5 * sr)
|
|
start = int(round((center - G_START) * sr))
|
|
i0 = max(0, start)
|
|
i1 = min(len(sig), start + length)
|
|
if i1 <= i0:
|
|
return
|
|
local = (np.arange(i0, i1) - start) / sr
|
|
env = np.exp(-local / decay_s)
|
|
sig[i0:i1] += amp * env * np.sin(2.0 * np.pi * freq * local)
|
|
|
|
|
|
def _add_noise_burst(sig, t, sr, center, amp, decay_s, rng):
|
|
length = int(decay_s * 5 * sr)
|
|
start = int(round((center - G_START) * sr))
|
|
i0 = max(0, start)
|
|
i1 = min(len(sig), start + length)
|
|
if i1 <= i0:
|
|
return
|
|
local = (np.arange(i0, i1) - start) / sr
|
|
env = np.exp(-local / decay_s)
|
|
sig[i0:i1] += amp * env * rng.standard_normal(i1 - i0)
|
|
|
|
|
|
def slice_for_offset(master: np.ndarray, sr: int, offset_ms: float,
|
|
duration_s: float) -> np.ndarray:
|
|
"""Extract a video's audio window from the master (spec M0 known-offset shift).
|
|
|
|
A video with ``offset_ms`` has local time ``tau`` mapping to global ``offset_ms/1000 +
|
|
tau`` (drift 0), so its audio is the master window starting at that global time.
|
|
"""
|
|
start_global = offset_ms / 1000.0
|
|
start = int(round((start_global - G_START) * sr))
|
|
length = int(round(duration_s * sr))
|
|
if start < 0 or start + length > len(master):
|
|
raise ValueError(f"offset {offset_ms} ms window falls outside the master signal")
|
|
return master[start:start + length]
|
|
|
|
|
|
def write_wav(path: Path, signal: np.ndarray, sr: int) -> None:
|
|
"""Write a mono 16-bit PCM WAV (stdlib only)."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
clipped = np.clip(signal, -1.0, 1.0)
|
|
pcm = (clipped * 32767.0).astype("<i2")
|
|
with wave.open(str(path), "wb") as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(sr)
|
|
w.writeframes(pcm.tobytes())
|
|
|
|
|
|
def read_wav(path: Path) -> tuple[np.ndarray, int]:
|
|
"""Read a mono 16-bit PCM WAV back to float [-1, 1] (used by tests)."""
|
|
with wave.open(str(path), "rb") as w:
|
|
sr = w.getframerate()
|
|
frames = w.readframes(w.getnframes())
|
|
pcm = np.frombuffer(frames, dtype="<i2").astype(np.float64) / 32767.0
|
|
return pcm, sr
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Geometry: poses + point cloud
|
|
# ---------------------------------------------------------------------------
|
|
def intrinsics(width: int, height: int) -> dict:
|
|
fx = (width / 2.0) / np.tan(np.radians(SYNTH_HFOV_DEG) / 2.0)
|
|
return {"fx": float(fx), "fy": float(fx), "cx": width / 2.0, "cy": height / 2.0}
|
|
|
|
|
|
def look_at_colmap(center, target, world_up=(0.0, 1.0, 0.0)) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Build a COLMAP world->camera pose (q=[w,x,y,z], t) for a camera at ``center``
|
|
looking at ``target`` with the given world up. Camera axes: +x right, +y down, +z fwd.
|
|
"""
|
|
C = np.asarray(center, dtype=np.float64)
|
|
T = np.asarray(target, dtype=np.float64)
|
|
up = np.asarray(world_up, dtype=np.float64)
|
|
|
|
z_c = T - C
|
|
z_c /= np.linalg.norm(z_c)
|
|
if abs(np.dot(up, z_c)) > 0.999: # looking near-vertical: pick a safe up
|
|
up = np.array([0.0, 0.0, 1.0])
|
|
up_c = up - np.dot(up, z_c) * z_c # image up (perp to forward)
|
|
up_c /= np.linalg.norm(up_c)
|
|
x_c = np.cross(z_c, up_c) # right
|
|
y_c = -up_c # down
|
|
|
|
R_c2w = np.column_stack([x_c, y_c, z_c]) # camera->world
|
|
R_w2c = R_c2w.T # world->camera (COLMAP R)
|
|
t = -R_w2c @ C
|
|
q = mat_to_quat(R_w2c)
|
|
return q, t
|
|
|
|
|
|
def camera_track(cam_index: int, duration_s: float, fps: float,
|
|
width: int, height: int) -> list[dict]:
|
|
"""One camera's trajectory: a gentle arc in front of the stage, ~2 poses/second."""
|
|
base_az = np.radians([-35.0, 0.0, 35.0][cam_index])
|
|
base_radius = 9.0
|
|
base_height = [1.6, 2.4, 2.0][cam_index]
|
|
sweep = np.radians(12.0)
|
|
intr = intrinsics(width, height)
|
|
|
|
poses = []
|
|
step = 0.5 # seconds -> ~2 poses/second
|
|
n = int(round(duration_s / step)) + 1
|
|
for k in range(n):
|
|
t_video = min(k * step, duration_s)
|
|
phase = 2.0 * np.pi * (t_video / duration_s)
|
|
az = base_az + sweep * np.sin(phase)
|
|
radius = base_radius + 0.5 * np.sin(phase + cam_index)
|
|
height_y = base_height + 0.3 * np.sin(phase * 2.0)
|
|
center = (radius * np.sin(az), height_y, radius * np.cos(az))
|
|
q, t = look_at_colmap(center, STAGE_TARGET)
|
|
poses.append({
|
|
"frame_idx": int(round(t_video * fps)),
|
|
"t_video_s": float(t_video),
|
|
"qw": float(q[0]), "qx": float(q[1]), "qy": float(q[2]), "qz": float(q[3]),
|
|
"tx": float(t[0]), "ty": float(t[1]), "tz": float(t[2]),
|
|
**intr,
|
|
"registered": True,
|
|
})
|
|
return poses
|
|
|
|
|
|
def generate_point_cloud(seed: int = 42) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Random points on the stage box + ground plane (with the named corners included).
|
|
|
|
Returns ``(points Nx3 float32, colors Nx3 uint8)``.
|
|
"""
|
|
rng = np.random.default_rng(seed)
|
|
pts: list[np.ndarray] = []
|
|
cols: list[np.ndarray] = []
|
|
|
|
hw, hd, h = STAGE_HALF_W, STAGE_HALF_D, STAGE_HEIGHT
|
|
stage_color = np.array([210, 90, 60]) # warm stage
|
|
ground_color = np.array([70, 80, 70]) # dim ground
|
|
|
|
# Stage box: sample points on all 6 faces.
|
|
n_face = 260
|
|
faces = [
|
|
# (fixed axis, value, u range, v range) -> build points
|
|
("x", -hw), ("x", hw), ("z", -hd), ("z", hd), ("y", 0.0), ("y", h),
|
|
]
|
|
for axis, val in faces:
|
|
u = rng.uniform(-hw, hw, n_face)
|
|
v = rng.uniform(0.0, h, n_face) if axis in ("x", "z") else rng.uniform(-hd, hd, n_face)
|
|
w = rng.uniform(-hd, hd, n_face)
|
|
if axis == "x":
|
|
p = np.column_stack([np.full(n_face, val), v, w])
|
|
elif axis == "z":
|
|
p = np.column_stack([u, v, np.full(n_face, val)])
|
|
else: # y
|
|
p = np.column_stack([u, np.full(n_face, val), w])
|
|
pts.append(p)
|
|
jitter = rng.integers(-15, 16, size=(n_face, 3))
|
|
cols.append(np.clip(stage_color + jitter, 0, 255))
|
|
|
|
# Ground plane: scattered points on y=0 out to a radius, excluding the stage footprint.
|
|
n_ground = 1600
|
|
gx = rng.uniform(-12.0, 12.0, n_ground)
|
|
gz = rng.uniform(-12.0, 12.0, n_ground)
|
|
on_stage = (np.abs(gx) < hw) & (np.abs(gz) < hd)
|
|
gx, gz = gx[~on_stage], gz[~on_stage]
|
|
gp = np.column_stack([gx, np.zeros_like(gx), gz])
|
|
pts.append(gp)
|
|
jitter = rng.integers(-12, 13, size=(len(gx), 3))
|
|
cols.append(np.clip(ground_color + jitter, 0, 255))
|
|
|
|
# Include the exact named stage corners so the M8 nearest-point fallback can hit them.
|
|
corners = np.array(list(STAGE_CORNERS.values()))
|
|
pts.append(corners)
|
|
cols.append(np.tile(np.array([255, 230, 120]), (len(corners), 1)))
|
|
|
|
points = np.vstack(pts).astype(np.float32)
|
|
colors = np.vstack(cols).astype(np.uint8)
|
|
return points, colors
|
|
|
|
|
|
def write_ply(path: Path, points: np.ndarray, colors: np.ndarray) -> None:
|
|
"""Write a binary-little-endian PLY (x,y,z float + red,green,blue uchar). FROZEN format."""
|
|
points = np.asarray(points, dtype=np.float32).reshape(-1, 3)
|
|
colors = np.asarray(colors, dtype=np.uint8).reshape(-1, 3)
|
|
if len(points) != len(colors):
|
|
raise ValueError("points and colors length mismatch")
|
|
n = len(points)
|
|
arr = np.empty(n, dtype=_PLY_DTYPE)
|
|
arr["x"], arr["y"], arr["z"] = points[:, 0], points[:, 1], points[:, 2]
|
|
arr["red"], arr["green"], arr["blue"] = colors[:, 0], colors[:, 1], colors[:, 2]
|
|
assert arr.dtype.itemsize == 15 # no padding — 3*4 + 3*1
|
|
|
|
header = (
|
|
"ply\n"
|
|
"format binary_little_endian 1.0\n"
|
|
"comment Festival 4D point cloud\n"
|
|
f"element vertex {n}\n"
|
|
"property float x\n"
|
|
"property float y\n"
|
|
"property float z\n"
|
|
"property uchar red\n"
|
|
"property uchar green\n"
|
|
"property uchar blue\n"
|
|
"end_header\n"
|
|
)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, "wb") as f:
|
|
f.write(header.encode("ascii"))
|
|
f.write(arr.tobytes())
|
|
|
|
|
|
def read_ply(path: Path) -> tuple[np.ndarray, np.ndarray]:
|
|
"""Read a PLY written by :func:`write_ply` (used by tests). Returns (points, colors)."""
|
|
with open(path, "rb") as f:
|
|
line = f.readline()
|
|
if line.strip() != b"ply":
|
|
raise ValueError("not a PLY file")
|
|
n = 0
|
|
while True:
|
|
line = f.readline()
|
|
if not line:
|
|
raise ValueError("unexpected EOF in PLY header")
|
|
if line.startswith(b"element vertex"):
|
|
n = int(line.split()[-1])
|
|
if line.strip() == b"end_header":
|
|
break
|
|
arr = np.frombuffer(f.read(n * _PLY_DTYPE.itemsize), dtype=_PLY_DTYPE, count=n)
|
|
points = np.column_stack([arr["x"], arr["y"], arr["z"]]).astype(np.float32)
|
|
colors = np.column_stack([arr["red"], arr["green"], arr["blue"]]).astype(np.uint8)
|
|
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)
|
|
# ---------------------------------------------------------------------------
|
|
def _find_font() -> str | None:
|
|
for candidate in _FONT_CANDIDATES:
|
|
if Path(candidate).exists():
|
|
return candidate
|
|
return None
|
|
|
|
|
|
_DRAWTEXT_AVAILABLE: bool | None = None
|
|
|
|
|
|
def _has_drawtext() -> bool:
|
|
"""Whether this ffmpeg build includes the ``drawtext`` filter (needs libfreetype)."""
|
|
global _DRAWTEXT_AVAILABLE
|
|
if _DRAWTEXT_AVAILABLE is None:
|
|
try:
|
|
out = subprocess.run(
|
|
["ffmpeg", "-hide_banner", "-filters"],
|
|
check=True, capture_output=True, text=True,
|
|
).stdout
|
|
_DRAWTEXT_AVAILABLE = any(
|
|
line.split()[1] == "drawtext"
|
|
for line in out.splitlines()
|
|
if len(line.split()) > 1
|
|
)
|
|
except Exception:
|
|
_DRAWTEXT_AVAILABLE = False
|
|
return _DRAWTEXT_AVAILABLE
|
|
|
|
|
|
def render_video(out_path: Path, wav_path: Path, cam_index: int,
|
|
width: int, height: int, fps: float, duration_s: float,
|
|
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:
|
|
raise RuntimeError("ffmpeg not found on PATH — required for the synthetic fixture")
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
hue = cam_index * 47
|
|
vf = f"hue=h={hue}"
|
|
font = _find_font()
|
|
if font and _has_drawtext():
|
|
vf += (f",drawtext=fontfile='{font}':text='CAM {cam_index}':"
|
|
"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}"
|
|
common_out = [
|
|
"-c:v", "libx264", "-pix_fmt", "yuv420p", "-preset", "veryfast", "-crf", "28",
|
|
"-profile:v", "baseline", "-level", "3.1",
|
|
"-movflags", "+faststart",
|
|
"-c:a", "aac", "-b:a", "128k",
|
|
"-shortest",
|
|
str(out_path),
|
|
]
|
|
|
|
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:
|
|
"""Probe width/height/fps/duration of a video with ffprobe."""
|
|
out = subprocess.run(
|
|
["ffprobe", "-v", "error", "-select_streams", "v:0",
|
|
"-show_entries", "stream=width,height,avg_frame_rate:format=duration",
|
|
"-of", "json", str(path)],
|
|
check=True, capture_output=True, text=True,
|
|
).stdout
|
|
data = json.loads(out)
|
|
stream = data["streams"][0]
|
|
num, den = stream["avg_frame_rate"].split("/")
|
|
fps = float(num) / float(den) if float(den) else 0.0
|
|
return {
|
|
"width": int(stream["width"]),
|
|
"height": int(stream["height"]),
|
|
"fps": fps,
|
|
"duration_s": float(data["format"]["duration"]),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Orchestration
|
|
# ---------------------------------------------------------------------------
|
|
def build(base_dir: Path | str | None = None, duration_s: float | None = None,
|
|
run_ffmpeg: bool = True) -> dict:
|
|
"""Generate the full synthetic project. Returns a summary dict.
|
|
|
|
``base_dir`` defaults to ``config.DATA_DIR``. ``duration_s`` defaults to the fixture's
|
|
20 s (tests may pass a shorter value). ``run_ffmpeg=False`` skips video rendering (for
|
|
fast, hermetic tests of the DB/pose/PLY path).
|
|
"""
|
|
base = Path(base_dir) if base_dir is not None else config.DATA_DIR
|
|
duration = float(duration_s) if duration_s is not None else config.SYNTH_DURATION_S
|
|
raw = base / "raw"
|
|
work = base / "work"
|
|
raw.mkdir(parents=True, exist_ok=True)
|
|
work.mkdir(parents=True, exist_ok=True)
|
|
|
|
sr = config.SYNTH_AUDIO_SR
|
|
w, h, fps = config.SYNTH_VIDEO_W, config.SYNTH_VIDEO_H, config.SYNTH_FPS
|
|
|
|
log.info("synthetic: building project at %s (duration=%.1fs, ffmpeg=%s)",
|
|
base, duration, run_ffmpeg)
|
|
|
|
db.init_engine(base / "project.db")
|
|
db.reset_db()
|
|
|
|
master, audio_gt = synth_master_audio(sr)
|
|
|
|
videos_gt = []
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp = Path(tmp)
|
|
for i, offset_ms in enumerate(config.SYNTH_OFFSETS_MS):
|
|
filename = f"cam{i}.mp4"
|
|
out_path = raw / filename
|
|
clip = slice_for_offset(master, sr, offset_ms, duration)
|
|
poses = camera_track(i, duration, fps, w, h)
|
|
|
|
if run_ffmpeg:
|
|
wav_path = tmp / f"cam{i}.wav"
|
|
write_wav(wav_path, clip, sr)
|
|
# 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)
|
|
vw, vh, vfps, vdur = (probed["width"], probed["height"],
|
|
probed["fps"], probed["duration_s"])
|
|
else:
|
|
vw, vh, vfps, vdur = w, h, float(fps), duration
|
|
|
|
video = db.add_video(
|
|
filename=filename, duration_s=vdur, fps=vfps, width=vw, height=vh,
|
|
offset_ms=float(offset_ms), drift_ppm=0.0, sync_confidence=1.0,
|
|
)
|
|
db.set_poses(video.id, poses)
|
|
videos_gt.append({
|
|
"video_id": video.id, "filename": filename,
|
|
"offset_ms": float(offset_ms), "drift_ppm": 0.0,
|
|
"is_reference": i == 0,
|
|
})
|
|
log.info("synthetic: cam%d offset=%+.0fms -> %s", i, offset_ms, out_path.name)
|
|
|
|
# Point cloud
|
|
points, colors = generate_point_cloud()
|
|
write_ply(work / "points.ply", points, colors)
|
|
|
|
# Anchors (stage corners)
|
|
corner_colors = {"Stage FL": "#ff5533", "Stage FR": "#ffaa33",
|
|
"Stage BL": "#33aaff", "Stage BR": "#aa55ff"}
|
|
for label, (x, y, z) in STAGE_CORNERS.items():
|
|
db.add_anchor(label, x, y, z, corner_colors.get(label))
|
|
|
|
# Seeded events
|
|
for t_global, event_type, conf, desc in SEED_EVENTS:
|
|
db.add_event(t_global_s=t_global, event_type=event_type, source="ai",
|
|
duration_s=1.0, confidence=conf, description=desc)
|
|
|
|
# Ground truth JSON
|
|
ground_truth = {
|
|
"videos": videos_gt,
|
|
"reference_filename": "cam0.mp4",
|
|
"audio": audio_gt,
|
|
"events": [
|
|
{"t_global_s": t, "event_type": et, "confidence": c, "description": d}
|
|
for (t, et, c, d) in SEED_EVENTS
|
|
],
|
|
"scene": {
|
|
"stage_corners": {k: list(v) for k, v in STAGE_CORNERS.items()},
|
|
"stage_target": list(STAGE_TARGET),
|
|
"point_count": int(len(points)),
|
|
},
|
|
}
|
|
gt_path = work / "ground_truth.json"
|
|
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 = {
|
|
"base_dir": str(base),
|
|
"videos": [v["filename"] for v in videos_gt],
|
|
"points_ply": str(work / "points.ply"),
|
|
"point_count": int(len(points)),
|
|
"ground_truth": str(gt_path),
|
|
"track_truth": str(tt_path),
|
|
"markers": [m["marker_key"] for m in track_truth["markers"]],
|
|
"events": len(SEED_EVENTS),
|
|
"anchors": len(STAGE_CORNERS),
|
|
}
|
|
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(track_truth["markers"]))
|
|
return summary
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
logging.basicConfig(level=logging.INFO)
|
|
print(json.dumps(build(), indent=2))
|