"""Browser-native replay for the recorded camera-gimbal experiment.
The renderer accepts only completed Python trajectories. Its embedded script
updates SVG geometry, plots trajectory prefixes, and controls playback; it does
not contain plant, estimator, or controller equations.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
import html
import json
import math
from pathlib import Path
import re
import uuid
from typing import Any
RUN_ORDER = ("accelerometer", "gyro", "complementary")
FALLBACK_ID = "fig-gimbal-observation-fallback"
class GimbalReplayError(ValueError):
"""Raised when a replay artifact does not satisfy the renderer contract."""
def _load_source(source: Path | str | Mapping[str, Any]) -> dict[str, Any]:
if isinstance(source, Mapping):
return dict(source)
path = Path(source)
try:
value = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
raise FileNotFoundError(f"gimbal replay artifact is missing: {path}") from None
except json.JSONDecodeError as error:
raise GimbalReplayError(f"invalid gimbal replay JSON: {error}") from error
if not isinstance(value, dict):
raise GimbalReplayError("gimbal replay root must be an object")
return value
def _finite_number(value: Any, field: str) -> float:
if isinstance(value, bool):
raise GimbalReplayError(f"{field} must be numeric")
try:
number = float(value)
except (TypeError, ValueError):
raise GimbalReplayError(f"{field} must be numeric") from None
if not math.isfinite(number):
raise GimbalReplayError(f"{field} must be finite")
return number
def _normalise(source: Path | str | Mapping[str, Any]) -> dict[str, Any]:
artifact = _load_source(source)
if artifact.get("schema_version") != 1:
raise GimbalReplayError("unsupported gimbal replay schema version")
raw_runs = artifact.get("runs")
if not isinstance(raw_runs, Mapping):
raise GimbalReplayError("gimbal replay must contain a runs object")
fields = (
"time_s",
"true_angle_deg",
"estimated_angle_deg",
"true_bias_deg_s",
"estimated_bias_deg_s",
"torque_nm",
"base_acceleration_mps2",
"tap_torque_nm",
"accelerometer_angle_deg",
)
metrics_fields = (
"rms_angle_deg",
"peak_acceleration_window_deg",
"final_abs_angle_deg",
"normalized_torque_effort",
"saturation_fraction",
"regulation_score",
"estimator_rmse_deg",
"final_bias_true_deg_s",
"final_bias_estimate_deg_s",
"peak_torque_nm",
)
runs: dict[str, Any] = {}
frame_count: int | None = None
reference_times: list[float] | None = None
for name in RUN_ORDER:
raw_run = raw_runs.get(name)
if not isinstance(raw_run, Mapping):
raise GimbalReplayError(f"missing run: {name}")
raw_frames = raw_run.get("frames")
if not isinstance(raw_frames, Sequence) or isinstance(raw_frames, (str, bytes)):
raise GimbalReplayError(f"{name}.frames must be an array")
if len(raw_frames) < 2:
raise GimbalReplayError(f"{name}.frames must contain at least two samples")
frames: list[dict[str, float]] = []
for index, raw_frame in enumerate(raw_frames):
if not isinstance(raw_frame, Mapping):
raise GimbalReplayError(f"{name}.frames[{index}] must be an object")
frames.append(
{
field: _finite_number(
raw_frame.get(field), f"{name}.frames[{index}].{field}"
)
for field in fields
}
)
times = [frame["time_s"] for frame in frames]
if any(next_time <= time for time, next_time in zip(times, times[1:])):
raise GimbalReplayError(f"{name} replay times must be strictly increasing")
if frame_count is None:
frame_count = len(frames)
reference_times = times
elif frame_count != len(frames) or any(
not math.isclose(left, right, rel_tol=0.0, abs_tol=1e-9)
for left, right in zip(reference_times or [], times)
):
raise GimbalReplayError("all estimators must share the same replay time grid")
raw_metrics = raw_run.get("metrics")
if not isinstance(raw_metrics, Mapping):
raise GimbalReplayError(f"{name}.metrics must be an object")
metrics = {
field: _finite_number(raw_metrics.get(field), f"{name}.metrics.{field}")
for field in metrics_fields
}
style = raw_run.get("style", {})
if not isinstance(style, Mapping):
style = {}
runs[name] = {
"label": str(raw_run.get("label") or name),
"color": str(style.get("color") or "#1B2430"),
"dash": str(style.get("dash") or "-"),
"frames": frames,
"metrics": metrics,
}
scenario = artifact.get("scenario")
parameters = artifact.get("parameters")
if not isinstance(scenario, Mapping) or not isinstance(parameters, Mapping):
raise GimbalReplayError("gimbal replay must include scenario and parameters")
duration_s = _finite_number(scenario.get("duration_s"), "scenario.duration_s")
if duration_s <= 0.0:
raise GimbalReplayError("scenario.duration_s must be positive")
if not reference_times or not math.isclose(
reference_times[-1], duration_s, rel_tol=0.0, abs_tol=1e-9
):
raise GimbalReplayError("replay must end at scenario.duration_s")
replay = {
"title": str(artifact.get("title") or "Partial observation in camera stabilization"),
"description": str(
artifact.get("description")
or "Recorded state-estimation and feedback trajectories."
),
"fps": max(1.0, _finite_number(artifact.get("playback_fps", 25), "playback_fps")),
"duration_s": duration_s,
"torque_limit_nm": _finite_number(
parameters.get("torque_limit_nm"), "parameters.torque_limit_nm"
),
"tap_start_s": _finite_number(scenario.get("tap_start_s"), "scenario.tap_start_s"),
"tap_end_s": _finite_number(scenario.get("tap_end_s"), "scenario.tap_end_s"),
"acceleration_start_s": _finite_number(
scenario.get("acceleration_start_s"), "scenario.acceleration_start_s"
),
"acceleration_end_s": _finite_number(
scenario.get("acceleration_end_s"), "scenario.acceleration_end_s"
),
"runs": runs,
}
return replay
def _dom_id(prefix: str | None) -> str:
stem = re.sub(r"[^a-zA-Z0-9_-]+", "-", prefix or "gimbal-replay").strip("-")
if not stem:
stem = "gimbal-replay"
return f"{stem}-{uuid.uuid4().hex[:10]}"
def _safe_json(value: Any) -> str:
return (
json.dumps(value, separators=(",", ":"), allow_nan=False)
.replace("<", "\\u003c")
.replace(">", "\\u003e")
.replace("&", "\\u0026")
)
def render_gimbal_replay(
source: Path | str | Mapping[str, Any],
*,
replay_id: str | None = None,
fallback_id: str = FALLBACK_ID,
) -> str:
"""Return a self-contained accessible HTML replay of recorded trajectories."""
replay = _normalise(source)
root = _dom_id(replay_id)
data_id = f"{root}-data"
title_id = f"{root}-title"
description_id = f"{root}-description"
marker_id = f"{root}-arrow"
escaped_fallback = json.dumps(str(fallback_id))
data_json = _safe_json(replay)
card_markup = "".join(
f"""
{html.escape(replay['runs'][name]['label'])}
true angle
+0.0°
estimate
+0.0°
bias estimate
+0.00°/s
torque
+0.000 N·m
"""
for name in RUN_ORDER
)
metric_rows = "".join(
f"""