Spaces:
Running
Running
| """armnet eval — a one-click, no-setup policy eval on real hardware. | |
| Pick a task, paste (or pick) a policy repo id, press Run, and watch the rollouts | |
| on a real SO-101 cell: live video, and per-rollout success auto-scored | |
| by the instrumented BusyBox (no human judging). The run is shared — anyone who | |
| opens this Space follows the same fleet in real time. | |
| The task selector sits above the tabs and drives all three: which dataset the | |
| instructions point at, which policy the Live tab defaults to, and which board | |
| the Leaderboard shows. Jobs route to any free BusyBox cell, so more than one | |
| eval can run at once; the Live tab lets you pick which cell's feed to watch. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import gradio as gr | |
| import config | |
| import leaderboard | |
| from leaderboard import LEADERBOARD | |
| from runner import STATE, display_cell_id, fleet_availability | |
| # The text timer is a lightweight heartbeat so any viewer notices a run another | |
| # viewer started. Video takes a different path: one long-lived generator stream | |
| # per viewer, not one Gradio request per frame. Five FPS is enough to make the | |
| # robot motion readable without making every public viewer encode and receive | |
| # all 20 policy-rate frames. | |
| _TEXT_REFRESH_SECONDS = 2.0 | |
| _VIDEO_STREAM_SECONDS = 0.2 | |
| _VIDEO_IDLE_SLEEP_SECONDS = 0.25 | |
| def _resolve_policy(task_label: str, preset_label: str, custom: str) -> str: | |
| custom = (custom or "").strip() | |
| if custom: | |
| return custom | |
| task = config.task_for(task_label) | |
| return task.presets.get(preset_label, task.reference_policy) | |
| def _session_id(request: gr.Request | None) -> str | None: | |
| return getattr(request, "session_hash", None) if request is not None else None | |
| def _status_markdown(snap: dict) -> str: | |
| lines = [] | |
| task = snap["task"] | |
| policy = snap["policy"] | |
| if not policy: | |
| return "_Nothing running. Pick a policy and press Run._" | |
| lines.append(f"**Policy:** `{policy}`") | |
| cell = display_cell_id(snap.get("cell_id")) | |
| if cell: | |
| lines.append(f"**Cell:** `{cell}`") | |
| if task is not None: | |
| lines.append(f"**Task:** {task.label}") | |
| lines.append(f"**Status:** {snap['status']}") | |
| if snap["job_id"]: | |
| lines.append(f"**Job:** `{snap['job_id']}`") | |
| lines.append( | |
| f"**Progress:** {snap['n_scored']}/{config.N_EPISODES} rollouts scored" | |
| f" · {snap['n_success']} succeeded" | |
| ) | |
| if snap["error"]: | |
| lines.append(f"\n⚠️ **Error:** {snap['error']}") | |
| return " \n".join(lines) | |
| def _rollout_markdown(snap: dict) -> str: | |
| """Render the scored rollouts as a Markdown table. | |
| Markdown (unlike gr.Dataframe here) re-renders reliably on every Timer tick, | |
| so the table grows live as each rollout is scored. | |
| """ | |
| rows = snap["rollouts"] | |
| if not rows: | |
| return "### Rollouts\n\n_No rollouts scored yet._" | |
| lines = ["### Rollouts", "", "| Rollout | Result | Scored by |", "| --- | --- | --- |"] | |
| for r in rows: | |
| result = "✅ success" if r["success"] else "❌ fail" | |
| lines.append(f"| {r['episode']} | {result} | {r['scored_by']} |") | |
| return "\n".join(lines) | |
| def _summary_markdown(snap: dict) -> str: | |
| summary = snap["summary"] | |
| if not summary: | |
| if snap["n_scored"]: | |
| rate = 100.0 * snap["n_success"] / snap["n_scored"] | |
| return f"### Live: {snap['n_success']}/{snap['n_scored']} succeeded ({rate:.0f}%)" | |
| return "" | |
| agg = summary.get("aggregated", {}) if isinstance(summary, dict) else {} | |
| pc = agg.get("pc_success") | |
| parts = [] | |
| if pc is not None: | |
| parts.append(f"### Final: {pc:.0f}% success over {agg.get('n_episodes', '?')} rollouts") | |
| url = summary.get("eval_dataset_url") if isinstance(summary, dict) else None | |
| if url: | |
| parts.append(f"[Recorded eval dataset →]({url})") | |
| return " \n".join(parts) | |
| def _refresh_text(task_label: str, selected_feed: str | None, request: gr.Request): | |
| """Lightweight heartbeat for status shared between every viewer.""" | |
| task = config.task_for(task_label) | |
| session_id = _session_id(request) | |
| if selected_feed: | |
| STATE.set_view(session_id, selected_feed) | |
| snap = STATE.snapshot(session_id=session_id, task_slug=task.slug) | |
| return ( | |
| _status_markdown(snap), | |
| _rollout_markdown(snap), | |
| _summary_markdown(snap), | |
| "\n".join(snap["events"][-30:]), | |
| _run_button_update(snap, task), | |
| _cell_light_html(snap, task), | |
| _queue_markdown(snap), | |
| _feed_selector_update(snap), | |
| _feed_caption(snap), | |
| ) | |
| async def _stream_video(request: gr.Request): | |
| """Yield newest frames over one Gradio stream, capped at five FPS. | |
| ``STATE`` itself is fed continuously by the orchestrator websocket. Reading | |
| only its newest version preserves the old drop-to-stay-live semantics: a | |
| slow browser skips stale frames instead of building latency. | |
| The generator deliberately stays alive while the page is open, including | |
| between evals. That lets a viewer who did not press Run see a shared run | |
| started by someone else without bringing back a polling request per frame. | |
| Gradio cancels it when that browser disconnects. The selected cell is read | |
| from this session's view on every tick so switching feeds does not restart | |
| the stream. | |
| """ | |
| session_id = _session_id(request) | |
| last_version = -1 | |
| last_cell = None | |
| while True: | |
| snap = STATE.snapshot(session_id=session_id) | |
| cell = snap.get("selected_cell") | |
| version = snap["frame_version"] | |
| if cell != last_cell: | |
| last_cell = cell | |
| last_version = -1 | |
| if snap["latest_frame"] is None: | |
| yield None | |
| await asyncio.sleep(_VIDEO_IDLE_SLEEP_SECONDS) | |
| continue | |
| if version == last_version or snap["latest_frame"] is None: | |
| # An async sleep is cancellable when the browser disconnects. The | |
| # old synchronous generator blocked one of Gradio's 40 worker | |
| # threads inside next() whenever the Space was idle. | |
| await asyncio.sleep(_VIDEO_IDLE_SLEEP_SECONDS) | |
| continue | |
| last_version = version | |
| # decode_frame() already resized this once process-wide. Do not make | |
| # every connected viewer repeat the same PIL work. | |
| yield snap["latest_frame"] | |
| await asyncio.sleep(_VIDEO_STREAM_SECONDS) | |
| def _load_leaderboard(task_label: str, sort_choice: str, force: bool = False): | |
| by = "rate" if "Average" in (sort_choice or "") else "lower" | |
| return LEADERBOARD.rows(config.task_for(task_label).slug, by=by, force=force) | |
| def _on_run(task_label: str, preset_label: str, custom: str, request: gr.Request): | |
| task = config.task_for(task_label) | |
| ok, message = STATE.start_run(_resolve_policy(task_label, preset_label, custom), task) | |
| snap = STATE.snapshot(session_id=_session_id(request), task_slug=task.slug) | |
| banner = ("✅ " if ok else "⛔ ") + message | |
| return ( | |
| banner, | |
| _status_markdown(snap), | |
| _run_button_update(snap, task), | |
| _feed_selector_update(snap), | |
| _feed_caption(snap), | |
| _cell_light_html(snap, task), | |
| ) | |
| def _on_feed_change(selected_feed: str, task_label: str, request: gr.Request): | |
| task = config.task_for(task_label) | |
| STATE.set_view(_session_id(request), selected_feed) | |
| snap = STATE.snapshot(session_id=_session_id(request), task_slug=task.slug) | |
| return ( | |
| _status_markdown(snap), | |
| _rollout_markdown(snap), | |
| _summary_markdown(snap), | |
| "\n".join(snap["events"][-30:]), | |
| _feed_caption(snap), | |
| ) | |
| def _on_task_change(task_label: str, sort_choice: str, request: gr.Request): | |
| """Re-point every tab at the newly selected task.""" | |
| task = config.task_for(task_label) | |
| snap = STATE.snapshot(session_id=_session_id(request), task_slug=task.slug) | |
| return ( | |
| _instructions_markdown(task), | |
| gr.update(choices=list(task.presets), value=task.reference_label), | |
| gr.update(placeholder=task.reference_policy, value=""), | |
| _cell_light_html(snap, task), | |
| _run_button_update(snap, task), | |
| _load_leaderboard(task_label, sort_choice), | |
| _task_blurb(task), | |
| _feed_selector_update(snap), | |
| _feed_caption(snap), | |
| ) | |
| def _task_blurb(task: config.Task) -> str: | |
| return f"**{task.label}.** {task.blurb}" | |
| def _instructions_markdown(task: config.Task) -> str: | |
| return ( | |
| "## How to use this Space\n\n" | |
| f"1. **Train a LeRobot-compatible policy** on either the task-specific " | |
| f"[`{task.dataset}`](https://huggingface.co/datasets/{task.dataset}) dataset " | |
| f"or the language-conditioned " | |
| f"[`{config.MULTITASK_DATASET}`]" | |
| f"(https://huggingface.co/datasets/{config.MULTITASK_DATASET}) dataset. " | |
| "Make sure the trained policy is **uploaded to Hugging Face and publicly " | |
| "available**.\n" | |
| "2. **Copy your repo ID** into the text box on the **Live** tab and submit it. " | |
| "Armnet will automatically run and score the policy on a real SO-101 cell. " | |
| "If every cell is busy, the job waits in the shared queue. If more than one " | |
| "eval is running, pick which cell's live feed to watch.\n\n" | |
| "You can run a policy more than once. The more rollouts a policy accumulates, the " | |
| "closer its **lower confidence interval** matches its **average success rate** — " | |
| "see the **Leaderboard** tab.\n\n" | |
| f"Every task has its own dataset and its own board, so pick the one you want at " | |
| "the top of the page before you start." | |
| ) | |
| # Gradio's Dataframe markdown strips <img> sizing attributes, so force a small, | |
| # inline avatar next to the trainer name via CSS. Hugging Face shows these as | |
| # squares, not circles. Also styles the round, blinking cell-availability light. | |
| _CSS = """ | |
| #leaderboard-table img { | |
| width: 20px !important; | |
| height: 20px !important; | |
| min-width: 20px !important; | |
| border-radius: 2px; | |
| display: inline-block; | |
| vertical-align: middle; | |
| margin: 0 4px 0 0 !important; | |
| object-fit: cover; | |
| } | |
| #leaderboard-table .lb-verified { | |
| color: #16a34a; | |
| font-weight: 600; | |
| white-space: nowrap; | |
| cursor: help; | |
| } | |
| .cell-status { | |
| display: flex; | |
| align-items: center; | |
| gap: 8px; | |
| font-weight: 600; | |
| margin: 4px 0 2px 0; | |
| } | |
| .cell-dot { | |
| width: 14px; | |
| height: 14px; | |
| border-radius: 50%; | |
| display: inline-block; | |
| flex: 0 0 auto; | |
| } | |
| .cell-dot--green { background: #22c55e; box-shadow: 0 0 7px 1px #22c55e; animation: cell-blink 1.4s ease-in-out infinite; } | |
| .cell-dot--yellow { background: #eab308; box-shadow: 0 0 7px 1px #eab308; animation: cell-blink 1.0s ease-in-out infinite; } | |
| .cell-dot--red { background: #ef4444; box-shadow: 0 0 7px 1px #ef4444; animation: cell-blink 0.8s ease-in-out infinite; } | |
| .cell-dot--grey { background: #9ca3af; } | |
| @keyframes cell-blink { 0%, 100% { opacity: 1; } 50% { opacity: 0.25; } } | |
| """ | |
| def _cell_light_html(snap: dict, task: config.Task) -> str: | |
| color = snap.get("availability_color") | |
| caption = snap.get("availability_caption") | |
| if not color or not caption: | |
| color, caption = fleet_availability( | |
| snap.get("cell_statuses") or {}, | |
| aggregate=snap.get("cell_status", {}).get(task.slug, "offline"), | |
| ) | |
| return _light_html(color, caption) | |
| def _light_html(color: str, caption: str) -> str: | |
| return ( | |
| '<div class="cell-status">' | |
| f'<span class="cell-dot cell-dot--{color}"></span>' | |
| f"<span>{caption}</span>" | |
| "</div>" | |
| ) | |
| def _feed_caption(snap: dict) -> str: | |
| policy = snap.get("policy") | |
| if not policy: | |
| return "" | |
| cell = display_cell_id(snap.get("cell_id")) | |
| task = snap.get("task") | |
| parts = [f"**Watching:** `{policy}`"] | |
| if cell: | |
| parts.append(f"on `{cell}`") | |
| if task is not None: | |
| parts.append(f"· {task.label}") | |
| return " ".join(parts) | |
| def _feed_selector_update(snap: dict): | |
| feeds = snap.get("live_feeds") or [] | |
| # Offer a choice whenever two evals can be watched — including a run that | |
| # just finished while another cell is still going. | |
| shown = feeds if len(feeds) > 1 else [] | |
| choices = [] | |
| for feed in shown: | |
| cell = display_cell_id(feed["cell_id"]) or "routing" | |
| suffix = "" if feed["active"] else f" ({feed['status']})" | |
| choices.append((f"{feed['policy']} · {cell}{suffix}", feed["cell_id"])) | |
| selected = snap.get("selected_cell") | |
| values = [choice[1] for choice in choices] | |
| value = selected if selected in values else (values[0] if values else None) | |
| return gr.update(choices=choices, value=value, visible=len(choices) > 1) | |
| def _queue_mode(snap: dict, task: config.Task) -> bool: | |
| statuses = snap.get("cell_statuses") or {} | |
| if statuses: | |
| return not any( | |
| status == "available" for status in statuses.values() | |
| ) and any( | |
| status in {"occupied", "maintenance"} for status in statuses.values() | |
| ) | |
| return snap.get("cell_status", {}).get(task.slug, "offline") == "occupied" | |
| def _can_submit(snap: dict, task: config.Task) -> bool: | |
| """Allow immediate runs, or queueing while every matching cell is busy.""" | |
| if snap["queued_eval_total"] >= config.MAX_QUEUED_JOBS: | |
| return False | |
| statuses = snap.get("cell_statuses") or {} | |
| if statuses: | |
| return any( | |
| status in {"available", "occupied", "maintenance"} | |
| for status in statuses.values() | |
| ) | |
| status = snap.get("cell_status", {}).get(task.slug, "offline") | |
| return status in {"available", "occupied"} | |
| def _run_button_update(snap: dict, task: config.Task): | |
| queue_mode = _queue_mode(snap, task) | |
| return gr.update( | |
| value=( | |
| f"Queue {config.N_EPISODES} rollouts" | |
| if queue_mode | |
| else f"Run {config.N_EPISODES} rollouts" | |
| ), | |
| variant="secondary" if queue_mode else "primary", | |
| interactive=_can_submit(snap, task), | |
| ) | |
| def _queue_markdown(snap: dict) -> str: | |
| total = snap["queue_total"] | |
| eval_total = snap["queued_eval_total"] | |
| lines = [ | |
| f"### Shared queue ({total} waiting)", | |
| "", | |
| f"_Armnet eval submissions: {eval_total}/{config.MAX_QUEUED_JOBS}._", | |
| "", | |
| ] | |
| if not snap["queue_items"]: | |
| lines.append("_No jobs waiting._") | |
| else: | |
| lines.extend( | |
| f"{index}. `{label}`" | |
| for index, label in enumerate(snap["queue_items"], start=1) | |
| ) | |
| if eval_total >= config.MAX_QUEUED_JOBS: | |
| lines.extend(["", "_Armnet eval queue full — submissions are temporarily disabled._"]) | |
| return "\n".join(lines) | |
| _DEFAULT_TASK = config.TASKS[config.DEFAULT_TASK_LABEL] | |
| with gr.Blocks(title="armnet eval — evaluate a policy on real robots") as demo: | |
| # Inject CSS via a <style> element (Gradio 6 moved the Blocks `css=` arg to | |
| # launch(), which the Space controls, so this is the portable way). | |
| gr.HTML(f"<style>{_CSS}</style>") | |
| gr.Markdown( | |
| "# armnet eval — evaluate a policy on real robots\n" | |
| "[armnet.dev](https://armnet.dev/)\n\n" | |
| "Pick a task, pick or paste a policy, press **Run**, and watch **" | |
| f"{config.N_EPISODES} rollouts** on a real SO-101 cell. Success is " | |
| "auto-scored by the instrumented BusyBox — no human judging. Anyone here " | |
| "can watch live evals; if more than one cell is running, pick which feed " | |
| "to follow." | |
| ) | |
| # Above the tabs because it scopes all of them: the dataset the instructions | |
| # point at, the policy the Live tab defaults to, and the board shown under | |
| # Leaderboard. | |
| task_sel = gr.Radio( | |
| config.TASK_LABELS, | |
| value=config.DEFAULT_TASK_LABEL, | |
| label="Task", | |
| ) | |
| task_blurb = gr.Markdown(_task_blurb(_DEFAULT_TASK)) | |
| with gr.Tabs(): | |
| with gr.Tab("Instructions"): | |
| instructions = gr.Markdown(_instructions_markdown(_DEFAULT_TASK)) | |
| with gr.Tab("Live"): | |
| if not config.eval_image_configured(): | |
| gr.Markdown( | |
| "> ⚠️ **Not configured.** Set `ARMNET_DEMO_EVAL_IMAGE` (a pre-built " | |
| "lerobot-eval runtime image ref) in the Space settings — see the README." | |
| ) | |
| # Not "cells offline": a page that never gets its first refresh | |
| # would then look exactly like a fleet that is down. | |
| cell_light = gr.HTML(_light_html("grey", "checking cells…")) | |
| with gr.Row(): | |
| preset = gr.Dropdown( | |
| label="Pre-trained policy", | |
| choices=list(_DEFAULT_TASK.presets), | |
| value=_DEFAULT_TASK.reference_label, | |
| ) | |
| custom = gr.Textbox( | |
| label="…or a custom HF repo id", | |
| placeholder=_DEFAULT_TASK.reference_policy, | |
| ) | |
| run_btn = gr.Button( | |
| f"Run {config.N_EPISODES} rollouts", | |
| variant="primary", | |
| interactive=False, | |
| scale=0, | |
| ) | |
| banner = gr.Markdown("") | |
| queue_view = gr.Markdown( | |
| "### Shared queue (0 waiting)\n\n" | |
| f"_Armnet eval submissions: 0/{config.MAX_QUEUED_JOBS}._\n\n" | |
| "_No jobs waiting._" | |
| ) | |
| feed_sel = gr.Radio( | |
| label="Live eval", | |
| choices=[], | |
| visible=False, | |
| ) | |
| feed_caption = gr.Markdown("") | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| video = gr.Image( | |
| label="Live rollout", | |
| height=320, | |
| interactive=False, | |
| streaming=True, | |
| ) | |
| with gr.Column(scale=2): | |
| status = gr.Markdown("") | |
| summary = gr.Markdown("") | |
| rollouts = gr.Markdown("### Rollouts\n\n_No rollouts scored yet._") | |
| events = gr.Textbox(label="Live log", lines=8, interactive=False, autoscroll=True) | |
| with gr.Tab("Leaderboard") as leaderboard_tab: | |
| gr.Markdown( | |
| "Each task has **its own board** — the one shown follows the task " | |
| "selected above. Results are **pooled by policy repo + revision** " | |
| "across every run, and ranked by the **95% lower-bound** success rate " | |
| "by default — so running more rollouts only helps if the policy is " | |
| "genuinely good (spamming low-quality evals can't game it). Switch to " | |
| "*Average success rate* to rank by the raw mean." | |
| ) | |
| with gr.Row(): | |
| lb_sort = gr.Radio( | |
| ["Lower bound (95%)", "Average success rate"], | |
| value="Lower bound (95%)", | |
| label="Sort by", | |
| scale=3, | |
| ) | |
| lb_refresh = gr.Button("Refresh", scale=0) | |
| lb_table = gr.Dataframe( | |
| headers=leaderboard.HEADERS, | |
| datatype=leaderboard.DATATYPES, | |
| interactive=False, | |
| wrap=True, | |
| label="Leaderboard", | |
| elem_id="leaderboard-table", | |
| ) | |
| # Text is cheap enough to poll. Video uses one generator response for the | |
| # lifetime of the page, so increasing its FPS does not multiply HTTP/API | |
| # requests per viewer. | |
| text_timer = gr.Timer(_TEXT_REFRESH_SECONDS) | |
| text_outputs = [ | |
| status, | |
| rollouts, | |
| summary, | |
| events, | |
| run_btn, | |
| cell_light, | |
| queue_view, | |
| feed_sel, | |
| feed_caption, | |
| ] | |
| run_btn.click( | |
| _on_run, | |
| inputs=[task_sel, preset, custom], | |
| outputs=[banner, status, run_btn, feed_sel, feed_caption, cell_light], | |
| ) | |
| # concurrency_limit=None on both: they share one Gradio concurrency group | |
| # (keyed by the function), and its default limit of 1 would put every | |
| # viewer's refresh behind a single worker. It only reads a locked snapshot. | |
| text_timer.tick( | |
| _refresh_text, | |
| inputs=[task_sel, feed_sel], | |
| outputs=text_outputs, | |
| concurrency_limit=None, | |
| ) | |
| feed_sel.change( | |
| _on_feed_change, | |
| inputs=[feed_sel, task_sel], | |
| outputs=[status, rollouts, summary, events, feed_caption], | |
| ) | |
| demo.load( | |
| _refresh_text, | |
| inputs=[task_sel, feed_sel], | |
| outputs=text_outputs, | |
| concurrency_limit=None, | |
| ) | |
| demo.load( | |
| _stream_video, | |
| outputs=video, | |
| show_progress="hidden", | |
| concurrency_limit=None, | |
| stream_every=_VIDEO_STREAM_SECONDS, | |
| api_visibility="private", | |
| ) | |
| # Switching task re-points every tab at once, so a viewer never sees one | |
| # task's dataset next to another's leaderboard. | |
| task_sel.change( | |
| _on_task_change, | |
| inputs=[task_sel, lb_sort], | |
| outputs=[ | |
| instructions, | |
| preset, | |
| custom, | |
| cell_light, | |
| run_btn, | |
| lb_table, | |
| task_blurb, | |
| feed_sel, | |
| feed_caption, | |
| ], | |
| ) | |
| # Leaderboard: load on tab open / sort change / manual refresh (and once on | |
| # page load). Not on a timer — it changes only when a run finishes, so | |
| # constant polling would just risk rate limiting for no benefit. | |
| leaderboard_tab.select(_load_leaderboard, inputs=[task_sel, lb_sort], outputs=[lb_table]) | |
| lb_sort.change(_load_leaderboard, inputs=[task_sel, lb_sort], outputs=[lb_table]) | |
| lb_refresh.click( | |
| lambda t, s: _load_leaderboard(t, s, force=True), | |
| inputs=[task_sel, lb_sort], | |
| outputs=[lb_table], | |
| ) | |
| demo.load(_load_leaderboard, inputs=[task_sel, lb_sort], outputs=[lb_table]) | |
| if __name__ == "__main__": | |
| demo.queue().launch() | |