Skip to content

feat: live control-loop rate readout during inference #85

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ZouzouWP wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: live control-loop rate readout during inference #85

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions frontend/src/lib/inferenceApi.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<FpsTone, string> = {
good: "text-green-400",
warn: "text-amber-400",
bad: "text-red-400",
};

export async function startInference(
baseUrl: string,
fetcher: Fetcher,
Expand Down
129 changes: 128 additions & 1 deletion frontend/src/pages/Inference.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand All @@ -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 (
<svg
viewBox={`0 0 ${W} ${H}`}
preserveAspectRatio="none"
className="w-full h-10"
aria-label="Control-loop rate over the last minute"
>
{target != null && target > 0 && (
<line
x1={0}
y1={y(target)}
x2={W}
y2={y(target)}
stroke="#374151"
strokeWidth={1}
strokeDasharray="3 3"
vectorEffect="non-scaling-stroke"
/>
)}
<polyline
points={points}
fill="none"
stroke="#4ade80"
strokeWidth={1.5}
vectorEffect="non-scaling-stroke"
/>
</svg>
);
};

const FpsTile: React.FC<{
value: string;
label: string;
valueClass?: string;
borderClass?: string;
}> = ({ value, label, valueClass = "text-slate-200", borderClass = "border-gray-700" }) => (
<div className={`flex-1 bg-black/40 border rounded-lg px-3 py-2 ${borderClass}`}>
<div className={`font-mono text-xl leading-none tabular-nums ${valueClass}`}>{value}</div>
<div className="text-[10px] text-gray-500 mt-1">{label}</div>
</div>
);

const Inference: React.FC = () => {
const navigate = useNavigate();
const { baseUrl, fetchWithHeaders } = useApi();
const { toast } = useToast();
const [status, setStatus] = useState<InferenceStatus | null>(null);
const [fpsHistory, setFpsHistory] = useState<number[]>([]);
const [showStopConfirm, setShowStopConfirm] = useState(false);
const navigatedAwayRef = useRef(false);
// Independent flag: we may request a stop (safety net) before the run
Expand All @@ -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",
});
Expand Down Expand Up @@ -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 (
<div className="min-h-screen bg-black text-white flex flex-col p-4 sm:p-6 lg:p-8">
<div className="flex items-center gap-4 mb-8">
Expand Down Expand Up @@ -212,6 +290,55 @@ const Inference: React.FC = () => {
/>
</div>

<div className="mb-6">
<div className="flex items-baseline justify-between mb-2">
<span className="text-[10px] uppercase tracking-widest text-gray-500">
control loop
</span>
{targetFps != null && (
<span className="text-[10px] text-gray-500">target {targetFps} Hz</span>
)}
</div>
{!hasFps ? (
<div className="text-xs text-gray-600 border border-gray-800 rounded-lg px-3 py-4 text-center">
{isSettingUp
? "Waiting for the control loop…"
: "No rate reported yet."}
</div>
) : (
<>
<div className="flex gap-2">
<FpsTile
value={fmtFps(fpsNow)}
label="FPS now"
valueClass={nowTone ? FPS_TONE_TEXT[nowTone] : "text-slate-500"}
borderClass={
nowTone === "bad"
? "border-red-900"
: nowTone === "warn"
? "border-amber-900"
: nowTone === "good"
? "border-green-900"
: "border-gray-700"
}
/>
<FpsTile value={fmtFps(status.fps_avg)} label="average" />
<FpsTile
value={fmtFps(status.fps_min)}
label="worst second"
valueClass={minTone ? FPS_TONE_TEXT[minTone] : "text-slate-200"}
/>
</div>
<FpsSparkline values={fpsHistory} target={targetFps} />
<div className="text-[10px] text-gray-500">
{status.fps_worst_gap_ms != null &&
`longest stall ${Math.round(status.fps_worst_gap_ms)} ms · `}
{status.fps_ticks} loop ticks
</div>
</>
)}
</div>

<div className="text-xs text-slate-500 break-all mb-6">
policy: {status.policy_ref ?? "(unknown)"}
</div>
Expand Down
68 changes: 64 additions & 4 deletions lelab/rollout.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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:
Expand All @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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()}",
Expand Down Expand Up @@ -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,
Expand All @@ -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:
Expand All @@ -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:
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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),
}
Loading