"""Report pages for jobs (simulation runs and the jobs built on them), and their signed links."""
from __future__ import annotations
import hashlib
import hmac
import html
import json
import secrets
import uuid
from .sim_backend import TERMINAL
from .storage import Storage
# Official embedding: @foxglove/embed (vendored in api/static) drives an embed.foxglove.dev iframe.
# The iframe fetches the MCAP itself, so the R2 bucket must allow CORS from https://embed.foxglove.dev.
_FOXGLOVE_SCRIPT = """
"""
_FALLBACK_SECRET = secrets.token_hex(32)
def report_token(secret: str, run_id: str) -> str:
key = (secret or _FALLBACK_SECRET).encode()
return hmac.new(key, run_id.encode(), hashlib.sha256).hexdigest()[:32]
def check_token(secret: str, run_id: str, token: str | None) -> bool:
return bool(token) and hmac.compare_digest(report_token(secret, run_id), token)
def new_run_id() -> str:
return uuid.uuid4().hex[:16]
# --------------------------------------------------------------------------- report page
_PAGE = """
RosDiff {title_short} {run_id}
{refresh}
{title} {run_id}
{meta}
{status_label} {status_note}
{body}
"""
def _fmt(v) -> str:
if v is None:
return "–"
if isinstance(v, bool):
return "yes" if v else "no"
return html.escape(str(v))
def render_report(
run: dict,
storage: Storage | None,
foxglove_org: str | None = None,
children: list[dict] | None = None,
link=None,
) -> str:
"""HTML report for any job. `children`: jobs made from this one; `link(job_id)`: its signed report URL."""
from .jobs import KINDS
status = run["status"]
kind = run.get("kind") or "sim"
running = status not in TERMINAL
notes = {
"SUBMITTING": "Submitting to the GPU queue…",
"IN_QUEUE": "Waiting for a GPU worker (a cold start can take a minute).",
"IN_PROGRESS": "Simulating and rendering…",
"COMPLETED": "",
"FAILED": "The run failed.",
"TIMED_OUT": "The run hit the execution time limit.",
"CANCELLED": "The run was cancelled.",
}
body = []
output = run.get("output") or {}
summary = output.get("summary") or {}
files = output.get("files") or {}
if run.get("error"):
body.append(f'
Error
{html.escape(run["error"])}
')
if kind != "sim":
body.extend(_kind_body(kind, run, storage))
if kind == "sim" and status == "COMPLETED" and summary:
cells = [
("Simulated", f"{summary.get('simulated_seconds')} s"),
("Robot fell", summary.get("fell")),
("Fell at", f"{summary['fell_at']} s" if summary.get("fell_at") is not None else None),
(
"Root height min / final",
f"{summary.get('root_min_height')} / {summary.get('root_final_height')} m"
if "root_min_height" in summary
else None,
),
("Unstable at", summary.get("unstable_at")),
("Max contacts", summary.get("max_contacts")),
("Real-time factor", summary.get("realtime_factor")),
("GPU time", f"{run['execution_ms'] / 1000:.1f} s" if run.get("execution_ms") else None),
]
grid = "".join(f"
{html.escape(k)}{_fmt(v)}
" for k, v in cells)
warn = summary.get("warnings") or []
extra = f'
MuJoCo warnings: {html.escape(", ".join(warn))}
' if warn else ""
body.append(f'
{grid}
{extra}
')
if kind == "sim" and status == "COMPLETED" and storage is not None:
videos = [(files[k], mime) for k, mime in (("video", "video/mp4"), ("video_webm", "video/webm")) if k in files]
if videos:
sources = "".join(f'' for key, mime in videos)
body.append(
f'
')
if "mcap" in files:
mcap_url = storage.url(files["mcap"])
body.append(
'
3D replay (Foxglove) '
f'— topics /scene, /tf, /state · download '
".mcap (opens in any Foxglove app)"
f''
'
Loading the Foxglove viewer… it asks you to sign in to a '
"(free) Foxglove account the first time.
" + _FOXGLOVE_SCRIPT
)
links = [f'{k}' for k in ("scene", "summary") if k in files]
if links:
body.append(f'
Files: {" · ".join(links)} (links valid for 7 days)
')
if children:
rows = []
for c in children:
label = KINDS[c["kind"]].label if c.get("kind") in KINDS else (c.get("kind") or "unknown")
href = link(c["id"]) if link else None
name = f'{html.escape(label)}' if href else html.escape(label)
verdict = _child_note(c)
rows.append(
f'
{name} {html.escape(c["id"])} · '
f"{html.escape((c.get('status') or 'UNKNOWN').replace('_', ' ').title())}{verdict}
"
)
body.append('
Follow-up jobs' + "".join(rows) + "
")
if kind == "sim":
meta = (
f"{_fmt(run.get('robot') or 'no robot')} · {run.get('duration_s') or 0:g} s · controller: "
f"{_fmt(run.get('controller'))} · started {_fmt(run['created_at'][:19])} UTC"
)
else:
parent = f"from run {html.escape(run['parent_id'])} · " if run.get("parent_id") else ""
meta = f"{parent}started {_fmt(run['created_at'][:19])} UTC"
if run.get("cost_usd") is not None:
meta += f" · est. cost ${run['cost_usd']:.3f}"
title = KINDS[kind].label if kind in KINDS else kind
return _PAGE.format(
run_id=html.escape(run["id"]),
title=html.escape(title),
title_short=html.escape(kind),
meta=meta,
refresh='' if running else "",
status=html.escape(status),
status_label=html.escape(status.replace("_", " ").title()),
status_note=html.escape(notes.get(status, "")) + (" This page refreshes every 10 s." if running else ""),
body="\n".join(body),
)
# --------------------------------------------------------------------------- other job kinds
# kind -> function(job, storage) -> list of HTML cards; registered by the modules that add each kind.
KIND_BODIES: dict = {}
def _kind_body(kind: str, job: dict, storage: Storage | None) -> list[str]:
if job["status"] != "COMPLETED":
return []
if kind in KIND_BODIES:
return KIND_BODIES[kind](job, storage)
output = job.get("output") or {}
cards = []
summary = output.get("summary") or {}
if summary:
cells = "".join(
f"
{html.escape(str(k))}{_fmt(v)}
"
for k, v in summary.items()
if not isinstance(v, (dict, list))
)
cards.append(f'
{cells}
')
cards.extend(file_links(output.get("files") or {}, storage))
return cards
def file_links(files: dict, storage: Storage | None) -> list[str]:
if not files or storage is None:
return []
links = [f'{html.escape(name)}' for name, key in files.items()]
return [f'
Files: {" · ".join(links)} (links valid for 7 days)
']
def _child_note(job: dict) -> str:
"""One-line result shown next to a follow-up job on its parent's page."""
note = ((job.get("output") or {}).get("summary") or {}).get("headline")
return f" — {html.escape(str(note))}" if note else ""
def _badge(ok: bool | None, yes: str, no: str) -> str:
if ok is None:
return 'not run'
return f'{html.escape(yes if ok else no)}'
def _list(items, limit: int = 30) -> str:
items = list(items or [])
if not items:
return 'none'
more = f" … and {len(items) - limit} more" if len(items) > limit else ""
return "" + ", ".join(html.escape(str(i)) for i in items[:limit]) + "" + more
def _ros_check_body(job: dict, storage: Storage | None) -> list[str]:
output = job.get("output") or {}
r = output.get("summary") or {}
files = output.get("files") or {}
b, la, te, li = r.get("build") or {}, r.get("launch"), r.get("test"), r.get("lint") or {}
cards = [f'
{html.escape(r.get("headline", ""))}
']
cells = [
("ROS 2", r.get("distro")),
("Packages", ", ".join(r.get("packages") or [])),
("colcon build", _badge(b.get("ok"), "builds", "fails")),
("Launch (10 s)", _badge(la.get("ok") if la else None, "runs", "fails")),
(
"Tests",
_badge(te.get("ok"), f"{te.get('tests')} passed", f"{te.get('failures', 0) + te.get('errors', 0)} failing")
if te and te.get("ran")
else 'none in package',
),
("Lint issues", li.get("total", 0)),
]
grid = "".join(
f"
{html.escape(k)}{v if isinstance(v, str) and v.startswith('<') else _fmt(v)}
"
for k, v in cells
)
cards.append(f'
{grid}
')
problems = []
if b.get("errors"):
problems.append("Build errors
" + html.escape("\n".join(b["errors"])) + "
")
missing = (b.get("missing_dependencies") or []) + ((la or {}).get("missing_dependencies") or [])
if missing:
problems.append(
f"
Missing dependencies: {_list(dict.fromkeys(missing))} — not installed in the sandbox, or "
"not declared in package.xml.
"
)
if la and la.get("crashed"):
problems.append(f"
Crashed: {_list(la['crashed'])}
")
if la and la.get("errors"):
problems.append("Runtime errors
" + html.escape("\n".join(la["errors"])) + "
")
if li.get("issues"):
issues = ", ".join(f"{k}: {v}" for k, v in li["issues"].items() if v)
if issues:
problems.append(f'
Lint: {html.escape(issues)} (see lint.log)
')
if problems:
cards.append('
' + "".join(problems) + "
")
if la:
cards.append(
f'
While running
Started: {_list(la.get("started"))}
'
f"
Nodes: {_list(la.get('nodes'))}
Topics: {_list(la.get('topics'))}
"
)
if storage is not None and "bag.mcap" in files:
url = storage.url(files["bag.mcap"])
cards.append(
'
Recorded topics (Foxglove)— every topic during the '
f'launch, recorded with ros2 bag · download .mcap'
f''
'
Loading the Foxglove viewer…
' + _FOXGLOVE_SCRIPT
)
cards.extend(file_links({k: v for k, v in files.items() if k != "bag.mcap"}, storage))
return cards
KIND_BODIES["ros_check"] = _ros_check_body
def _sweep_body(job: dict, storage: Storage | None) -> list[str]:
output = job.get("output") or {}
s = output.get("summary") or {}
files = output.get("files") or {}
cards = [f'
{html.escape(s.get("headline", ""))}
']
r = s.get("ranges") or {}
cells = [
("Worlds", s.get("worlds")),
("Engine", s.get("engine")),
("Simulated each", f"{s.get('simulated_seconds_each')} s"),
("Controller", s.get("controller")),
("Fall rate", f"{100 * s['fall_rate']:.0f}%" if "fall_rate" in s else None),
("Largest push survived", f"{s['max_push_n_survived']} N" if "max_push_n_survived" in s else None),
("Friction scale", "–".join(map(str, r.get("friction_scale", [])))),
("Object mass scale", "–".join(map(str, r.get("object_mass_scale", [])))),
(
"Push",
f"0–{(r.get('push_newtons') or [0, 0])[1]} N for {r.get('push_duration_s')} s at {r.get('push_at_s')} s",
),
("Compute time", f"{s.get('wall_seconds')} s"),
]
grid = "".join(f"
{html.escape(k)}{_fmt(v)}
" for k, v in cells)
note = f'
{html.escape(s["engine_note"])}
' if s.get("engine_note") else ""
cards.append(f'
{grid}
{note}
')
worlds = []
if storage is not None and "sweep.json" in files:
try:
worlds = json.loads(storage.read_text(files["sweep.json"]) or "{}").get("worlds", [])
except (ValueError, AttributeError):
worlds = []
if worlds and "fall_rate" in s:
cards.append(
'
Push vs. friction— each dot is one world; '
"red fell, green stayed up" + _scatter(worlds) + "
"
)
cards.extend(file_links(files, storage))
return cards
def _scatter(worlds: list[dict], w: int = 640, h: int = 280) -> str:
pad = 40
pushes = [x["push_n"] for x in worlds] or [1]
fr = [x["friction"] for x in worlds] or [1]
px_max, f_min, f_max = max(max(pushes), 1e-6), min(fr), max(max(fr), min(fr) + 1e-6)
dots = []
for x in worlds:
cx = pad + (w - 2 * pad) * x["push_n"] / px_max
cy = h - pad - (h - 2 * pad) * (x["friction"] - f_min) / (f_max - f_min)
color = "var(--bad)" if x.get("fell") else "var(--ok)"
tip = f"push {x['push_n']} N, friction x{x['friction']}, mass x{x['object_mass']}"
dots.append(
f'{html.escape(tip)}'
)
axes = (
f''
f''
f''
f"push (N), 0–{px_max:g}"
f'friction ×{f_min:.2f}–{f_max:.2f}'
)
return f''
KIND_BODIES["sweep"] = _sweep_body
def _review_body(job: dict, storage: Storage | None) -> list[str]:
r = (job.get("output") or {}).get("summary") or {}
cards = [
f'
An automatic first pass by NVIDIA Cosmos Reason, to point reviewers at runs worth a '
"look. It can be wrong: the video and the numbers on the run's own report are what count.
"
)
return cards
KIND_BODIES["review"] = _review_body
def _export_body(job: dict, storage: Storage | None) -> list[str]:
output = job.get("output") or {}
r = output.get("summary") or {}
files = output.get("files") or {}
link = ""
if r.get("hub_url"):
link = f' · open on Hugging Face'
cards = [f'
A replay: MuJoCo simulated the run, Isaac Sim draws the same motion with RTX '
"lighting (no second physics engine), so the two videos show the same thing.