diff --git a/frontend/src/lib/inferenceApi.ts b/frontend/src/lib/inferenceApi.ts index 3258c989..5628c40e 100644 --- a/frontend/src/lib/inferenceApi.ts +++ b/frontend/src/lib/inferenceApi.ts @@ -26,8 +26,40 @@ export interface InferenceStatus { log_path: string | null; exited?: boolean; exit_code?: number | null; + /** Control-loop telemetry, measured in the rollout subprocess. */ + target_fps: number | null; + /** Rate over the last second — null while setting up or once the run ends. */ + fps_now: number | null; + fps_avg: number | null; + /** Slowest one-second window of the run. */ + fps_min: number | null; + /** Longest single gap between two control-loop ticks. */ + fps_worst_gap_ms: number | null; + fps_ticks: number; + fps_stale: boolean; } +export type FpsTone = "good" | "warn" | "bad"; + +/** + * How healthy a measured rate is against the target. The thresholds are + * deliberately forgiving: a rollout that holds 26 of its 30 Hz is fine, one + * that halves is not. + */ +export function fpsTone(fps: number, target: number | null): FpsTone { + if (!target || target <= 0) return "good"; + const ratio = fps / target; + if (ratio >= 0.85) return "good"; + if (ratio >= 0.55) return "warn"; + return "bad"; +} + +export const FPS_TONE_TEXT: Record = { + good: "text-green-400", + warn: "text-amber-400", + bad: "text-red-400", +}; + export async function startInference( baseUrl: string, fetcher: Fetcher, diff --git a/frontend/src/pages/Inference.tsx b/frontend/src/pages/Inference.tsx index 5e8faa6d..9a8ff868 100644 --- a/frontend/src/pages/Inference.tsx +++ b/frontend/src/pages/Inference.tsx @@ -16,12 +16,16 @@ import { AlertDialogTitle, } from "@/components/ui/alert-dialog"; import { + FPS_TONE_TEXT, InferenceStatus, + fpsTone, getInferenceStatus, stopInference, } from "@/lib/inferenceApi"; const POLL_MS = 1000; +// One sample per poll, so this is the last minute of control-loop rate. +const FPS_HISTORY_LEN = 60; function formatTime(seconds: number): string { const s = Math.max(0, Math.floor(seconds)); @@ -30,11 +34,67 @@ function formatTime(seconds: number): string { return `${String(mins).padStart(2, "0")}:${String(secs).padStart(2, "0")}`; } +/** Rate history over a fixed 60-sample axis: the curve grows in from the + * left, then slides once the window is full. The dashed line is the target. */ +const FpsSparkline: React.FC<{ values: number[]; target: number | null }> = ({ + values, + target, +}) => { + const W = 200; + const H = 40; + if (values.length < 2) return null; + const ceiling = Math.max(target ?? 0, ...values) * 1.1 || 1; + const y = (v: number) => H - (v / ceiling) * H; + const step = W / (FPS_HISTORY_LEN - 1); + const points = values.map((v, i) => `${(i * step).toFixed(1)},${y(v).toFixed(1)}`).join(" "); + return ( + + {target != null && target > 0 && ( + + )} + + + ); +}; + +const FpsTile: React.FC<{ + value: string; + label: string; + valueClass?: string; + borderClass?: string; +}> = ({ value, label, valueClass = "text-slate-200", borderClass = "border-gray-700" }) => ( +
+
{value}
+
{label}
+
+); + const Inference: React.FC = () => { const navigate = useNavigate(); const { baseUrl, fetchWithHeaders } = useApi(); const { toast } = useToast(); const [status, setStatus] = useState(null); + const [fpsHistory, setFpsHistory] = useState([]); const [showStopConfirm, setShowStopConfirm] = useState(false); const navigatedAwayRef = useRef(false); // Independent flag: we may request a stop (safety net) before the run @@ -56,15 +116,26 @@ const Inference: React.FC = () => { const next = await getInferenceStatus(baseUrl, fetchWithHeaders); if (cancelled) return; setStatus(next); + if (next.fps_now != null && !next.fps_stale) { + const sample = next.fps_now; + setFpsHistory((h) => [...h, sample].slice(-FPS_HISTORY_LEN)); + } // Auto-bounce home once the run is done. if (!next.inference_active && !navigatedAwayRef.current) { navigatedAwayRef.current = true; if (next.exited) { + // The average is the one number worth carrying off the page: it + // says whether the policy actually ran at the rate it was trained at. + const rate = + next.fps_avg != null + ? ` Averaged ${next.fps_avg.toFixed(1)} Hz` + + (next.target_fps ? ` of ${next.target_fps} Hz.` : ".") + : ""; toast({ title: "Inference finished", description: next.exit_code === 0 - ? "Run completed." + ? `Run completed.${rate}` : `Exit code ${next.exit_code}. See ${next.log_path}.`, variant: next.exit_code === 0 ? "default" : "destructive", }); @@ -152,6 +223,13 @@ const Inference: React.FC = () => { : "FINISHED"; const timerSeconds = isRunning ? rolloutElapsed : setupElapsed; + const targetFps = status.target_fps; + const fpsNow = status.fps_now; + const hasFps = fpsNow != null || status.fps_avg != null; + const nowTone = fpsNow != null ? fpsTone(fpsNow, targetFps) : null; + const minTone = status.fps_min != null ? fpsTone(status.fps_min, targetFps) : null; + const fmtFps = (v: number | null) => (v != null ? v.toFixed(1) : "—"); + return (
@@ -212,6 +290,55 @@ const Inference: React.FC = () => { />
+
+
+ + control loop + + {targetFps != null && ( + target {targetFps} Hz + )} +
+ {!hasFps ? ( +
+ {isSettingUp + ? "Waiting for the control loop…" + : "No rate reported yet."} +
+ ) : ( + <> +
+ + + +
+ +
+ {status.fps_worst_gap_ms != null && + `longest stall ${Math.round(status.fps_worst_gap_ms)} ms · `} + {status.fps_ticks} loop ticks +
+ + )} +
+
policy: {status.policy_ref ?? "(unknown)"}
diff --git a/lelab/rollout.py b/lelab/rollout.py index cfff79a8..23b939bb 100644 --- a/lelab/rollout.py +++ b/lelab/rollout.py @@ -55,6 +55,12 @@ class InferenceRequest(BaseModel): _inference_started_at: float | None = None _inference_rollout_started_at: float | None = None _inference_meta: dict[str, Any] = {} +# Latest control-loop rate sample emitted by lelab.rollout_metrics, and the +# target the rollout was configured with. Both are written by the stdout pump +# thread and read by the status route; plain assignments of immutable values, +# same pattern as _inference_rollout_started_at above. +_inference_fps: dict[str, Any] | None = None +_inference_target_fps: float | None = None # Guards mutations to the globals above; held only for the short critical # sections in start/stop/status. _state_lock = threading.Lock() @@ -65,12 +71,23 @@ class InferenceRequest(BaseModel): # UI can present a "rollout time" separate from the multi-second policy # load + bus connect + camera connect setup overhead. _ROLLOUT_START_MARKER = "Rollout setup complete" +# One per second from lelab.rollout_metrics' rate meter. +_FPS_RE = re.compile( + r"\[LELAB_FPS\] now=([\d.]+) avg=([\d.]+) min=([\d.]+) gap=([\d.]+) n=(\d+)" +) +# lerobot announces the configured target once: "Robot: so101_follower | FPS: 30 | ...". +# Read it rather than assuming RolloutConfig's 30.0 default, so the UI stays +# honest if the rollout config ever changes. +_TARGET_FPS_RE = re.compile(r"\bFPS:\s*([\d.]+)") +# A sample older than this means the rate meter stopped reporting (run ending, +# subprocess wedged) — the UI must not keep showing a stale number as live. +_FPS_STALE_AFTER_S = 3.0 def _pump_stdout(proc: subprocess.Popen, log_handle) -> None: - """Tee the subprocess's stdout to the log file and watch for the - rollout-start marker.""" - global _inference_rollout_started_at + """Tee the subprocess's stdout to the log file, watch for the rollout-start + marker, and pick up the control-loop rate samples.""" + global _inference_rollout_started_at, _inference_fps, _inference_target_fps try: for raw in iter(proc.stdout.readline, b""): try: @@ -88,6 +105,17 @@ def _pump_stdout(proc: subprocess.Popen, log_handle) -> None: "Inference rollout main loop started after %.1fs of setup", _inference_rollout_started_at - (_inference_started_at or _inference_rollout_started_at), ) + elif (m := _FPS_RE.search(line)) is not None: + _inference_fps = { + "now": float(m.group(1)), + "avg": float(m.group(2)), + "min": float(m.group(3)), + "worst_gap_ms": float(m.group(4)), + "ticks": int(m.group(5)), + "at": time.time(), + } + elif _inference_target_fps is None and (m := _TARGET_FPS_RE.search(line)) is not None: + _inference_target_fps = float(m.group(1)) except Exception as exc: logger.exception("Inference stdout pump failed: %s", exc) finally: @@ -235,11 +263,29 @@ def _classify_outcome(rc: int | None, rollout_started: bool, error_text: str | N return "failed" +def _fps_payload(sample: dict[str, Any] | None, target: float | None, live: bool) -> dict[str, Any]: + """Flatten the last rate sample into the status response. + + `live` is False once the run is over: the averages stay meaningful as a + summary, but there is no current rate any more.""" + stale = not live or sample is None or (time.time() - sample["at"]) > _FPS_STALE_AFTER_S + return { + "target_fps": target, + "fps_now": None if (sample is None or stale) else sample["now"], + "fps_avg": sample["avg"] if sample else None, + "fps_min": sample["min"] if sample else None, + "fps_worst_gap_ms": sample["worst_gap_ms"] if sample else None, + "fps_ticks": sample["ticks"] if sample else 0, + "fps_stale": stale, + } + + def handle_start_inference(request: InferenceRequest) -> dict[str, Any]: """Start a one-shot rollout subprocess. Returns a dict — the route layer turns it into a JSON response or HTTPException as appropriate.""" global inference_active, _inference_proc, _inference_started_at global _inference_rollout_started_at, _inference_meta + global _inference_fps, _inference_target_fps # Mutex with teleop and recording: all three drive the same serial bus. from . import record as _record, teleoperate as _teleoperate @@ -277,7 +323,9 @@ def handle_start_inference(request: InferenceRequest) -> dict[str, Any]: cmd = [ sys.executable, "-m", - "lerobot.scripts.lerobot_rollout", + # Thin wrapper around lerobot.scripts.lerobot_rollout that installs + # the control-loop rate meter before handing over to it. + "lelab.rollout_metrics", "--strategy.type=base", f"--policy.path={policy_path}", f"--policy.device={_detect_device()}", @@ -334,6 +382,8 @@ def handle_start_inference(request: InferenceRequest) -> dict[str, Any]: _inference_proc = proc _inference_started_at = time.time() _inference_rollout_started_at = None + _inference_fps = None + _inference_target_fps = None _inference_meta = { "policy_ref": request.policy_ref, "duration_s": request.duration_s, @@ -346,6 +396,7 @@ def handle_start_inference(request: InferenceRequest) -> dict[str, Any]: def handle_stop_inference() -> dict[str, Any]: global inference_active, _inference_proc, _inference_started_at global _inference_rollout_started_at, _inference_meta + global _inference_fps, _inference_target_fps with _state_lock: if not inference_active or _inference_proc is None: @@ -369,12 +420,15 @@ def handle_stop_inference() -> dict[str, Any]: _inference_started_at = None _inference_rollout_started_at = None _inference_meta = {} + _inference_fps = None + _inference_target_fps = None return {"success": True, "message": "Inference stopped"} def handle_inference_status() -> dict[str, Any]: global inference_active, _inference_proc, _inference_started_at global _inference_rollout_started_at, _inference_meta + global _inference_fps, _inference_target_fps # Finalise state lazily if the subprocess died on its own. with _state_lock: @@ -385,11 +439,15 @@ def handle_inference_status() -> dict[str, Any]: finished_meta = _inference_meta finished_started = _inference_started_at finished_rollout_started = _inference_rollout_started_at + finished_fps = _inference_fps + finished_target_fps = _inference_target_fps inference_active = False _inference_proc = None _inference_started_at = None _inference_rollout_started_at = None _inference_meta = {} + _inference_fps = None + _inference_target_fps = None # On failure, surface the real error from the log so the UI doesn't # have to send the user digging through the cache. error = _extract_error_from_log(finished_meta.get("log_path")) if rc else None @@ -408,6 +466,7 @@ def handle_inference_status() -> dict[str, Any]: "rollout_started_at": finished_rollout_started, "rollout_elapsed_s": 0, "elapsed_s": 0, + **_fps_payload(finished_fps, finished_target_fps, live=False), } elapsed = (time.time() - _inference_started_at) if _inference_started_at else 0 rollout_elapsed = time.time() - _inference_rollout_started_at if _inference_rollout_started_at else 0 @@ -420,4 +479,5 @@ def handle_inference_status() -> dict[str, Any]: "duration_s": _inference_meta.get("duration_s"), "policy_ref": _inference_meta.get("policy_ref"), "log_path": _inference_meta.get("log_path"), + **_fps_payload(_inference_fps, _inference_target_fps, live=inference_active), } diff --git a/lelab/rollout_metrics.py b/lelab/rollout_metrics.py new file mode 100644 index 00000000..9b66abd8 --- /dev/null +++ b/lelab/rollout_metrics.py @@ -0,0 +1,146 @@ +# Copyright 2025 The HuggingFace Inc. team. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""lerobot_rollout wrapper that measures the real control-loop rate. + +lerobot logs a warning when the control loop misses its target rate, but only +per missed iteration and only into the inference log. A run that quietly drops +from 30 Hz to 10 Hz therefore looks, in the UI, exactly like one that holds — +which matters, because a policy behaves differently when it is stepped at a +fraction of the rate it was trained at. + +Rather than patch lerobot, this wrapper counts ``ThreadSafeRobot.get_observation`` +calls: the rollout control loop makes exactly one per iteration, so counting +them over a window *is* the loop rate. One ``[LELAB_FPS]`` line is printed per +second; ``lelab/rollout.py``'s stdout pump parses it and exposes the numbers on +``/inference-status``. + +Metering must never break the rollout: every failure path is swallowed, and a +run whose meter fails to install simply reports no rate. +""" + +from __future__ import annotations + +import threading +import time + +_RATE_WINDOW_S = 1.0 # matches the UI's 1 Hz status poll + + +class _RateMeter: + """Counts control-loop ticks and the worst gap between two of them. + + A window is timed from its own first to its own last tick, not by the + reporter's wall clock: a window the loop only half-filled (it just started, + or the run just ended) then reports the rate it actually saw instead of a + phantom slowdown.""" + + def __init__(self) -> None: + self.ticks = 0 + self.max_gap = 0.0 + self._first: float | None = None + self._newest: float | None = None + # Kept across drains so a gap straddling two windows is still measured. + self._last: float | None = None + self._lock = threading.Lock() + + def tick(self, now: float) -> None: + with self._lock: + if self._last is not None: + gap = now - self._last + if gap > self.max_gap: + self.max_gap = gap + self._last = now + if self._first is None: + self._first = now + self._newest = now + self.ticks += 1 + + def drain(self) -> tuple[int, float, float]: + """(ticks, seconds spanned by those ticks, worst gap).""" + with self._lock: + span = (self._newest - self._first) if self.ticks >= 2 else 0.0 + out = (self.ticks, span, self.max_gap) + self.ticks = 0 + self.max_gap = 0.0 + self._first = None + self._newest = None + return out + + +def _install_rate_meter() -> None: + """Report the real control-loop rate once a second on stdout.""" + try: + from lerobot.rollout.robot_wrapper import ThreadSafeRobot + + meter = _RateMeter() + original = ThreadSafeRobot.get_observation + + def get_observation(self, *args, **kwargs): + try: + return original(self, *args, **kwargs) + finally: + meter.tick(time.perf_counter()) + + ThreadSafeRobot.get_observation = get_observation + + def report_loop() -> None: + total_ticks, intervals, measured = 0, 0, 0.0 + worst_fps: float | None = None + worst_gap = 0.0 + started = False + while True: + time.sleep(_RATE_WINDOW_S) + ticks, span, max_gap = meter.drain() + # Under two ticks means the control loop isn't running: setup, + # or the run has ended. Nothing to report, and staying quiet is + # what lets the server flag the sample stale. + if ticks < 2 or span <= 0: + continue + # Setup takes one observation of its own and the first live + # window straddles warmup; both would smear the average, so + # discard the first window that looks like a running loop. + if not started: + started = True + continue + total_ticks += ticks + intervals += ticks - 1 + measured += span + fps_now = (ticks - 1) / span + worst_fps = fps_now if worst_fps is None else min(worst_fps, fps_now) + worst_gap = max(worst_gap, max_gap) + # print(), not logger: this must reach stdout whatever lerobot's + # init_logging() decided, and the server tees stdout to the log + # file anyway, so the run stays diagnosable after the fact. + print( + f"[LELAB_FPS] now={fps_now:.1f} avg={intervals / measured:.1f} " + f"min={worst_fps:.1f} gap={worst_gap * 1000:.0f} n={total_ticks}", + flush=True, + ) + + threading.Thread(target=report_loop, name="lelab-rate-meter", daemon=True).start() + except Exception: + # Telemetry is best-effort; inference must run regardless. + pass + + +def main() -> None: + _install_rate_meter() + from lerobot.scripts.lerobot_rollout import main as rollout_main + + rollout_main() + + +if __name__ == "__main__": + main() diff --git a/tests/test_rollout.py b/tests/test_rollout.py index c0bb1f3d..5027ff9b 100644 --- a/tests/test_rollout.py +++ b/tests/test_rollout.py @@ -35,6 +35,8 @@ def _reset_rollout_globals(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(rollout, "_inference_started_at", None) monkeypatch.setattr(rollout, "_inference_rollout_started_at", None) monkeypatch.setattr(rollout, "_inference_meta", {}) + monkeypatch.setattr(rollout, "_inference_fps", None) + monkeypatch.setattr(rollout, "_inference_target_fps", None) def test_inference_request_rejects_missing_required_fields() -> None: @@ -219,6 +221,69 @@ def test_handle_inference_status_when_idle_returns_dict_with_expected_keys() -> assert result["inference_active"] is False for key in ("started_at", "rollout_started_at", "elapsed_s", "rollout_elapsed_s"): assert key in result + for key in ("target_fps", "fps_now", "fps_avg", "fps_min", "fps_stale"): + assert key in result + + +def test_fps_regex_parses_a_rate_meter_line() -> None: + """The line format is the contract between lelab.rollout_metrics (which + prints it in the subprocess) and the stdout pump that reads it.""" + from lelab.rollout import _FPS_RE + + m = _FPS_RE.search("[LELAB_FPS] now=28.6 avg=27.9 min=11.4 gap=452 n=1742") + assert m is not None + assert m.groups() == ("28.6", "27.9", "11.4", "452", "1742") + + +def test_target_fps_regex_reads_lerobots_banner_and_ignores_our_own_line() -> None: + from lelab.rollout import _TARGET_FPS_RE + + banner = "INFO Robot: so101_follower | FPS: 30 | Duration: 60s" + assert _TARGET_FPS_RE.search(banner).group(1) == "30" + assert _TARGET_FPS_RE.search("[LELAB_FPS] now=28.6 avg=27.9 min=11.4 gap=4 n=17") is None + + +def test_fps_payload_without_a_sample_reports_stale() -> None: + from lelab.rollout import _fps_payload + + payload = _fps_payload(None, None, live=True) + assert payload["fps_stale"] is True + assert payload["fps_now"] is None + assert payload["fps_ticks"] == 0 + + +def test_fps_payload_drops_a_sample_older_than_the_stale_window() -> None: + """A wedged subprocess stops emitting; the UI must not keep showing its + last rate as if it were live.""" + import time + + from lelab.rollout import _FPS_STALE_AFTER_S, _fps_payload + + sample = { + "now": 28.6, "avg": 27.9, "min": 11.4, "worst_gap_ms": 452.0, + "ticks": 1742, "at": time.time() - (_FPS_STALE_AFTER_S + 1), + } + payload = _fps_payload(sample, 30.0, live=True) + assert payload["fps_stale"] is True + assert payload["fps_now"] is None + # The run summary stays available even once the live rate is gone. + assert payload["fps_avg"] == 27.9 + + +def test_fps_payload_keeps_the_summary_but_drops_the_live_rate_once_finished() -> None: + import time + + from lelab.rollout import _fps_payload + + sample = { + "now": 28.6, "avg": 27.9, "min": 11.4, "worst_gap_ms": 452.0, + "ticks": 1742, "at": time.time(), + } + payload = _fps_payload(sample, 30.0, live=False) + assert payload["fps_now"] is None + assert payload["fps_avg"] == 27.9 + assert payload["fps_min"] == 11.4 + assert payload["target_fps"] == 30.0 def _stub_request():