From 18adda76b5143d993922210539dbb98a5cd9a4c2 Mon Sep 17 00:00:00 2001 From: Daniel Sanchez Date: Thu, 3 Sep 2026 14:25:09 +0900 Subject: [PATCH 1/4] feat(eval): ego-dimension acceptance gates Pre-deployment checks for a failure open-loop metrics do not detect: the model plans a standstill on geometry it should drive through, at some values of the ego_shape input. When a corpus mixes platforms whose speeds also differ, the token becomes the cheapest predictor of speed and gets read as a speed prior rather than as geometry; the resulting model can score normally on displacement error while being undeployable. The traffic-light gate is the one to run per checkpoint: grading scenes that span a light cycle with a single threshold marks red frames as stalls AND scores a model that runs reds as healthy, so scenes are grouped by the state their own route lanes carry. Amber is reported, not gated. The remaining tools locate the confound in a corpus, separate token-driven stops from real ones, and identify which input group holds the plan. Every entry point refuses a training milestone that still carries EMA weights, matching the convention in the recovery evals. --- rlvr/autoresearch/ego_shape_diag/README.md | 103 ++++++++++++ rlvr/autoresearch/ego_shape_diag/__init__.py | 0 rlvr/autoresearch/ego_shape_diag/_common.py | 98 +++++++++++ .../ego_shape_diag/check_ego_shape_gate.py | 91 ++++++++++ .../ego_shape_diag/check_route_ab.py | 132 +++++++++++++++ .../ego_shape_diag/check_tl_gate.py | 153 +++++++++++++++++ .../ego_shape_diag/check_token_ablation.py | 120 +++++++++++++ .../ego_shape_diag/check_training_data.py | 157 ++++++++++++++++++ rlvr/test_ego_shape_gates.py | 97 +++++++++++ 9 files changed, 951 insertions(+) create mode 100644 rlvr/autoresearch/ego_shape_diag/README.md create mode 100644 rlvr/autoresearch/ego_shape_diag/__init__.py create mode 100644 rlvr/autoresearch/ego_shape_diag/_common.py create mode 100644 rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py create mode 100644 rlvr/autoresearch/ego_shape_diag/check_route_ab.py create mode 100644 rlvr/autoresearch/ego_shape_diag/check_tl_gate.py create mode 100644 rlvr/autoresearch/ego_shape_diag/check_token_ablation.py create mode 100644 rlvr/autoresearch/ego_shape_diag/check_training_data.py create mode 100644 rlvr/test_ego_shape_gates.py diff --git a/rlvr/autoresearch/ego_shape_diag/README.md b/rlvr/autoresearch/ego_shape_diag/README.md new file mode 100644 index 000000000..2a036ab1f --- /dev/null +++ b/rlvr/autoresearch/ego_shape_diag/README.md @@ -0,0 +1,103 @@ +# Ego-dimension diagnostics + +Pre-deployment checks for a failure that open-loop metrics do not detect: the model +plans a **standstill** on geometry it should drive through, at some values of the +`ego_shape` (vehicle-dimension) input. + +`ego_shape` is supposed to describe geometry — it should change the shape of a +manoeuvre, never whether the vehicle moves at all. But if a training corpus mixes +vehicle platforms and those platforms also differ in how fast they drive, the token +becomes the cheapest predictor of speed available to the model, and it gets read as +"which speed distribution do I imitate". The resulting model can score normally on +displacement error while being undeployable, because the collapse is confined to the +geometry and dimensions where it plans to stop. + +These tools measure that directly, in seconds, without a closed-loop simulation. Run +them from the repo root, and point `--model` at a **deployable** checkpoint: a training +milestone carries both raw optimizer iterates and the EMA copy, and loading the raw ones +gives a plausible but wrong verdict. Every script here refuses such a checkpoint and +prints the conversion. + +## 1. `check_tl_gate.py` — the acceptance gate, traffic-light stratified + +The one to run on every checkpoint when the scenes carry traffic-light state. + +``` +python -m rlvr.autoresearch.ego_shape_diag.check_tl_gate \ + --model --scenes \ + --shape [--goal X,Y] +``` + +A stop is only a failure if the light says go. Grading scenes that span a light cycle +with a single threshold gets this backwards twice over: it marks the red frames as +stalls, and it scores a model that *runs* red lights as healthy. So scenes are grouped +by the traffic-light state their own route lanes carry, and the gate asks for +`mean span >= --min_green_m` on green and `<= --max_red_m` on red. Amber is reported +but not gated — easing and proceeding are both defensible, and the choice is a +behaviour difference between checkpoints rather than a pass or a fail. + +If no scene carries a signalled route lane the tool fails rather than reporting a +vacuous pass. + +## 2. `check_ego_shape_gate.py` — the acceptance gate, dimension sweep + +For scenes without traffic-light state, or to locate the cliff. + +``` +python -m rlvr.autoresearch.ego_shape_diag.check_ego_shape_gate \ + --model --scenes [--wheelbases 2.75,3.5,4.0,4.5,4.76] +``` + +Sweeps the wheelbase with the other dimensions fixed and reports the plan span at each, +plus the spread. A healthy model is flat across the sweep. A cliff — normal spans at one +end, near-zero at the other — means the token is acting as a speed prior, and the spread +is the size of the effect. + +## 3. `check_training_data.py` — find the confound in the corpus + +``` +python -m rlvr.autoresearch.ego_shape_diag.check_training_data \ + --scenes [--wheelbase_split 4.0] [--sample 800] +``` + +Splits a scene set by wheelbase and compares travel over the prediction horizon between +the two classes. Two red flags: only a couple of distinct `ego_shape` values (a binary +switch is the easiest thing for a model to key on), and a large gap in median travel +between the classes. Healthy corpora have overlapping travel distributions. This runs on +data alone, so it can be checked before committing to a training run. + +## 4. `check_route_ab.py` — pathological stops versus real ones + +``` +python -m rlvr.autoresearch.ego_shape_diag.check_route_ab \ + --model --scenes --shape_a W,L,W --shape_b W,L,W +``` + +Scores every scene along a route at two dimension settings. A stop where one setting +stands still and the other drives is attributable to the token; a stop where both agree +is a real one. Report the fractions, not individual frames — and cross-check the +pathological timestamps against traffic-light state before concluding anything, since +stops on red are legitimate however the settings differ. + +## 5. `check_token_ablation.py` — which input is holding the plan + +``` +python -m rlvr.autoresearch.ego_shape_diag.check_token_ablation \ + --model --scene --shape W,L,W [--split_line_strings] +``` + +Zeroes each input group in turn; a group whose removal releases the plan is what the +model is reacting to. `--split_line_strings` separates stop lines from road borders and +also pushes the borders outward, which distinguishes "cannot fit" from "will not cross": +if dropping borders releases the plan but moving them away does not, it is the latter. + +## Scenes + +These tools need scenes at the geometry where the model misbehaves — typically converted +from a recorded run, with the same converter the runtime uses so the tensors match. + +Two conversion details worth knowing, both handled here: + +* `ego_agent_past` may arrive as float64; the encoder needs float32. +* a converter may rewrite `goal_pose` onto the ego when the recorded run ends stopped, + which makes the scene meaningless. `--goal X,Y` restores the true ego-frame goal. diff --git a/rlvr/autoresearch/ego_shape_diag/__init__.py b/rlvr/autoresearch/ego_shape_diag/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/rlvr/autoresearch/ego_shape_diag/_common.py b/rlvr/autoresearch/ego_shape_diag/_common.py new file mode 100644 index 000000000..ddf1d8ae2 --- /dev/null +++ b/rlvr/autoresearch/ego_shape_diag/_common.py @@ -0,0 +1,98 @@ +"""Shared helpers for the ego-dimension diagnostics. Run all scripts from the repo root.""" + +from __future__ import annotations + +import glob +import os + +import numpy as np +import torch + +from preference_optimization.utils import load_npz_data +from rlvr.autoresearch.tools.eval_det_avoidance import load_model + + +def load_deployable_model(model_path: str, device): + """Load a checkpoint, refusing one whose deployable copy would be skipped. + + ``load_model`` takes ``ckpt["model"]``, and a training milestone holds BOTH that + (the raw optimizer iterates) and ``ema_state_dict`` (the deployable copy). Loading + the raw weights yields a plausible-looking but wrong verdict with no warning, so + fail loudly and say how to convert instead. + """ + ckpt = torch.load(model_path, map_location="cpu", weights_only=False) + if isinstance(ckpt, dict) and "ema_state_dict" in ckpt: + raise SystemExit( + f"{model_path} is a training milestone carrying both raw and EMA weights; " + "load_model would silently use the RAW iterates. Convert it first:\n" + ' ck = torch.load(src, map_location="cpu", weights_only=False)\n' + ' state = {k.replace("module.", ""): v for k, v in ck["ema_state_dict"].items()}\n' + ' torch.save({"model": state}, dst) # copy args.json next to it' + ) + return load_model(model_path, device) + + +def find_scenes(path: str) -> list[str]: + """Accept a directory (searched recursively), a glob, or a single .npz.""" + if os.path.isdir(path): + out = sorted(glob.glob(os.path.join(path, "**", "*.npz"), recursive=True)) + else: + out = sorted(glob.glob(path)) + if not out: + raise SystemExit(f"no .npz found under {path!r}") + return out + + +def to_f32(d: dict) -> dict: + """Some converters emit float64 for ego_agent_past; the encoder requires float32.""" + return { + k: (v.float() if torch.is_tensor(v) and v.dtype == torch.float64 else v) + for k, v in d.items() + } + + +def load_scene(path: str, device, goal: tuple[float, float] | None = None) -> dict: + """Load one scene, optionally overriding the ego-frame goal. + + The override exists because a converter may rewrite ``goal_pose`` onto the ego when + the recorded run ends stopped; passing the true goal restores the intended scene. + """ + d = to_f32(load_npz_data(path, device)) + if goal is not None: + g = d["goal_pose"].clone().reshape(-1) + g[0], g[1] = goal + if g.numel() >= 4: # [x, y, cos, sin] after heading_to_cos_sin + g[2], g[3] = 1.0, 0.0 + d["goal_pose"] = g.reshape(d["goal_pose"].shape) + return d + + +def set_shape(d: dict, wheelbase: float, length: float, width: float) -> dict: + """Return a copy of the scene with ``ego_shape`` replaced.""" + out = dict(d) + s = d["ego_shape"].clone().reshape(-1) + s[0], s[1], s[2] = wheelbase, length, width + out["ego_shape"] = s.reshape(d["ego_shape"].shape) + return out + + +def plan_span(model, margs, d: dict, device) -> float: + """Straight-line distance from the first to the last predicted ego waypoint. + + This is a coarse "is the model planning to move at all" measure, not a path length: + at cruise it lands in the tens of metres, and a value near zero means the model is + planning to stand still. + """ + from rlvr.autoresearch.tools.eval_det_avoidance import det_inference_batched + + out = det_inference_batched(model, margs, [d], device) + t = out[0] if not isinstance(out, tuple) else out[0][0] + t = np.squeeze(t.detach().cpu().numpy()) + return float(np.hypot(t[-1, 0] - t[0, 0], t[-1, 1] - t[0, 1])) + + +def parse_shape(text: str) -> tuple[float, float, float]: + parts = [float(x) for x in text.split(",")] + if len(parts) != 3: + raise SystemExit(f"expected wheelbase,length,width (got {text!r})") + return parts[0], parts[1], parts[2] diff --git a/rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py b/rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py new file mode 100644 index 000000000..de69547cd --- /dev/null +++ b/rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""ACCEPTANCE GATE: does the planned distance depend on the ego-dimension token? It must not. + +``ego_shape`` describes the vehicle's geometry. A healthy model plans the same distance +whatever it says — geometry decides how a manoeuvre is shaped, not whether the vehicle +moves. When a training set contains only a couple of distinct ``ego_shape`` values and +those groups differ in how fast they drive, the token becomes the cheapest available +predictor of speed, and the model learns to read it as "which speed distribution do I +imitate". The failure then appears only at one end of the range: at some wheelbase the +plan collapses to a standstill on geometry the model otherwise handles. + +Sweep the token and compare. A flat sweep passes; a cliff means the token is acting as a +speed prior. Pair this with ``check_training_data`` (the confound in the data) and +``check_tl_gate`` (whether the standstill is actually correct at that moment). + + python -m rlvr.autoresearch.ego_shape_diag.check_ego_shape_gate \ + --model --scenes \ + [--goal X,Y] [--wheelbases 2.75,3.5,4.0,4.5,4.76] [--min_span 15] [--limit 5] + +Seconds of compute. Open-loop trajectory error does not detect this failure. +""" + +from __future__ import annotations + +import argparse + +import torch + +from rlvr.autoresearch.ego_shape_diag._common import ( + find_scenes, + load_deployable_model, + load_scene, + plan_span, + set_shape, +) + + +def sweep_scene(model, margs, d: dict, wheelbases, length, width, device) -> list[float]: + """Plan span at each wheelbase, holding the other dimensions fixed.""" + return [plan_span(model, margs, set_shape(d, wb, length, width), device) for wb in wheelbases] + + +def report_scene(path: str, wheelbases: list[float], sweep: list[float], min_span: float) -> bool: + """Print one scene's sweep; returns True when it fails the gate.""" + bad = [s for s in sweep if s < min_span] + print(path.split("/")[-1]) + print(" " + " ".join(f"{w}:{s:.1f}" for w, s in zip(wheelbases, sweep))) + print( + f" spread across wheelbase: {max(sweep) - min(sweep):6.2f} m" + f" {'FAIL' if bad else 'pass'}" + ) + return bool(bad) + + +def _parse_args() -> argparse.Namespace: + ap = argparse.ArgumentParser() + ap.add_argument("--model", required=True, help="deployable checkpoint (args.json alongside)") + ap.add_argument("--scenes", required=True, help="scene dir, glob, or single .npz") + ap.add_argument("--goal", default=None, help="x,y ego-frame goal override") + ap.add_argument("--wheelbases", default="2.75,3.5,4.0,4.5,4.76") + ap.add_argument("--length", type=float, default=4.34, help="held fixed across the sweep") + ap.add_argument("--width", type=float, default=1.84, help="held fixed across the sweep") + ap.add_argument("--min_span", type=float, default=15.0, help="gate threshold in metres") + ap.add_argument("--limit", type=int, default=5, help="scenes to test") + return ap.parse_args() + + +def main() -> int: + a = _parse_args() + goal = tuple(float(x) for x in a.goal.split(",")) if a.goal else None + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + model, margs = load_deployable_model(a.model, device) + wheelbases = [float(x) for x in a.wheelbases.split(",")] + scenes = find_scenes(a.scenes)[: a.limit] + + print(f"model : {a.model}") + print(f"scenes : {len(scenes)} gate: span >= {a.min_span} m at every wheelbase\n") + failures = 0 + for path in scenes: + d = load_scene(path, device, goal) + sweep = sweep_scene(model, margs, d, wheelbases, a.length, a.width, device) + failures += report_scene(path, wheelbases, sweep, a.min_span) + print( + f"\n{'FAIL' if failures else 'PASS'}: {failures}/{len(scenes)} scenes below {a.min_span} m" + ) + print("A large spread means the ego-dimension token is acting as a speed prior.") + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/rlvr/autoresearch/ego_shape_diag/check_route_ab.py b/rlvr/autoresearch/ego_shape_diag/check_route_ab.py new file mode 100644 index 000000000..7c2ce68d8 --- /dev/null +++ b/rlvr/autoresearch/ego_shape_diag/check_route_ab.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +"""Route-wide A/B between two ego-dimension settings: which stops are pathological? + +Along a converted route, score every scene twice — once at each of two ``ego_shape`` +settings — and classify the stops: + + PATHOLOGICAL : span_a < --stop_thresh AND span_b >= --ok_thresh + (only the dimension token separates driving from standing still) + LEGITIMATE : both spans < --stop_thresh + (the settings agree there is a reason to stop) + +The counts matter more than any single frame: a handful of pathological stops is noise, +while a steady fraction across a route means the token is driving the decision. Use +``check_tl_gate`` on the same scenes to confirm whether the stops line up with red +signals, which is the usual innocent explanation. + + python -m rlvr.autoresearch.ego_shape_diag.check_route_ab \ + --model --scenes \ + --shape_a W,L,W --shape_b W,L,W [--goal X,Y] +""" + +from __future__ import annotations + +import argparse +import re + +import numpy as np +import torch + +from rlvr.autoresearch.ego_shape_diag._common import ( + find_scenes, + load_deployable_model, + load_scene, + parse_shape, + plan_span, + set_shape, +) + +_FRAME_RE = re.compile(r"(\d+)\.npz") + + +def score_route(model, margs, scenes, device, goal, shape_a, shape_b) -> list[tuple]: + """(frame index, span at shape A, span at shape B) for every scene on the route.""" + rows: list[tuple] = [] + for path in scenes: + found = _FRAME_RE.findall(path) + idx = int(found[-1]) if found else len(rows) + d = load_scene(path, device, goal) + rows.append( + ( + idx, + plan_span(model, margs, set_shape(d, *shape_a), device), + plan_span(model, margs, set_shape(d, *shape_b), device), + ) + ) + return rows + + +def classify(rows: list[tuple], stop_thresh: float, ok_thresh: float) -> tuple[list, list]: + """Split the stops into (pathological, legitimate).""" + patho = [r for r in rows if r[1] < stop_thresh and r[2] >= ok_thresh] + legit = [r for r in rows if r[1] < stop_thresh and r[2] < stop_thresh] + return patho, legit + + +def _report_spans(rows: list[tuple], stop_thresh: float) -> None: + print(f"scenes: {len(rows)}") + for name, index in (("A", 1), ("B", 2)): + spans = np.array([r[index] for r in rows]) + print( + f"shape {name} span: mean={spans.mean():6.2f} p50={np.percentile(spans, 50):6.2f} " + f"frac<{stop_thresh:g}m={(spans < stop_thresh).mean() * 100:5.1f}%" + ) + + +def _report_pathological(patho: list[tuple], dt: float) -> None: + if not patho: + return + print("\npathological frames: t(s) shape A shape B") + for idx, span_a, span_b in patho: + print(f" {idx * dt:8.1f} {span_a:7.2f} {span_b:7.2f}") + print("\nCross-check these timestamps against the traffic-light state; stops on red") + print("are legitimate however the two settings differ.") + + +def report(rows: list[tuple], stop_thresh: float, ok_thresh: float, dt: float) -> None: + patho, legit = classify(rows, stop_thresh, ok_thresh) + _report_spans(rows, stop_thresh) + print( + f"\nPATHOLOGICAL (A<{stop_thresh:g} and B>={ok_thresh:g}): " + f"{len(patho)}/{len(rows)} = {len(patho) / len(rows) * 100:.1f}% of route" + ) + print( + f"LEGITIMATE (both <{stop_thresh:g}, they agree): " + f"{len(legit)}/{len(rows)} = {len(legit) / len(rows) * 100:.1f}%" + ) + _report_pathological(patho, dt) + + +def _parse_args() -> argparse.Namespace: + ap = argparse.ArgumentParser() + ap.add_argument("--model", required=True, help="deployable checkpoint (args.json alongside)") + ap.add_argument("--scenes", required=True, help="dir of NPZs converted from a full run") + ap.add_argument("--shape_a", required=True, help="wheelbase,length,width (the suspect one)") + ap.add_argument("--shape_b", required=True, help="wheelbase,length,width (the reference)") + ap.add_argument("--goal", default=None, help="x,y ego-frame goal override") + ap.add_argument("--stop_thresh", type=float, default=5.0) + ap.add_argument("--ok_thresh", type=float, default=15.0) + ap.add_argument("--dt", type=float, default=0.1, help="seconds per frame index unit") + return ap.parse_args() + + +def main() -> int: + a = _parse_args() + goal = tuple(float(x) for x in a.goal.split(",")) if a.goal else None + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + model, margs = load_deployable_model(a.model, device) + rows = score_route( + model, + margs, + find_scenes(a.scenes), + device, + goal, + parse_shape(a.shape_a), + parse_shape(a.shape_b), + ) + report(rows, a.stop_thresh, a.ok_thresh, a.dt) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py new file mode 100644 index 000000000..306da65ba --- /dev/null +++ b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py @@ -0,0 +1,153 @@ +#!/usr/bin/env python3 +"""ACCEPTANCE GATE, traffic-light stratified: does the model go on green and hold on red? + +A stop is only a failure if the light says go. Scenes captured across a light cycle +therefore cannot be graded with a single plan-span threshold: an unstratified gate marks +the red frames as stalls and — worse — scores a model that *runs* red lights as healthy. +Group the scenes by the traffic-light state their own route lanes carry, then require: + + green scenes -> mean plan span >= --min_green_m (the model commits to go) + red scenes -> mean plan span <= --max_red_m (the model holds) + +Amber is reported but not gated: easing and proceeding are both defensible, so it is a +behaviour difference between checkpoints rather than a pass or a fail. + + python -m rlvr.autoresearch.ego_shape_diag.check_tl_gate \ + --model --scenes \ + [--goal X,Y] [--shape wheelbase,length,width] \ + [--min_green_m 15] [--max_red_m 2] + +Seconds of compute and no closed-loop simulation. Worth running on every checkpoint, +because open-loop trajectory error does not detect this failure: a model can hold a +standstill plan at signalised geometry while scoring normally on displacement metrics. +""" + +from __future__ import annotations + +import argparse + +import numpy as np +import torch +from diffusion_planner.dimensions import ( + TRAFFIC_LIGHT_GREEN, + TRAFFIC_LIGHT_NO_TRAFFIC_LIGHT, + TRAFFIC_LIGHT_RED, + TRAFFIC_LIGHT_YELLOW, +) + +from rlvr.autoresearch.ego_shape_diag._common import ( + find_scenes, + load_deployable_model, + load_scene, + parse_shape, + plan_span, + set_shape, +) + +_TL_SLICE = slice(TRAFFIC_LIGHT_GREEN, TRAFFIC_LIGHT_NO_TRAFFIC_LIGHT + 1) +_CLASSES = { + TRAFFIC_LIGHT_GREEN - TRAFFIC_LIGHT_GREEN: "green", + TRAFFIC_LIGHT_YELLOW - TRAFFIC_LIGHT_GREEN: "amber", + TRAFFIC_LIGHT_RED - TRAFFIC_LIGHT_GREEN: "red", +} +_ORDER = ("green", "amber", "red", "none") + + +def route_tl_class(d: dict) -> str: + """The traffic-light state carried by the scene's own route lanes. + + Red wins over amber and amber over green when several appear: a route whose lanes + include a red signal is a hold situation whatever the other lanes show. Scenes with + no signalled route lane are reported as ``none`` and excluded from the gate. + """ + rl = d["route_lanes"].detach().cpu().numpy() + if rl.ndim > 3: + rl = rl.reshape(-1, rl.shape[-2], rl.shape[-1]) + valid = np.abs(rl).sum(axis=(1, 2)) > 0 + if not valid.any(): + return "none" + onehot = rl[valid][:, 0, _TL_SLICE] + hot = onehot.max(axis=1) > 0 + if not hot.any(): + return "none" + present = {_CLASSES.get(int(c)) for c in onehot[hot].argmax(axis=1)} + for state in ("red", "amber", "green"): + if state in present: + return state + return "none" + + +def measure(model, margs, scenes, device, goal, shape) -> dict[str, list[float]]: + """Plan span for every scene, bucketed by its route traffic-light state.""" + spans: dict[str, list[float]] = {k: [] for k in _ORDER} + for path in scenes: + d = load_scene(path, device, goal) + spans[route_tl_class(d)].append( + plan_span(model, margs, set_shape(d, *shape) if shape else d, device) + ) + return spans + + +def verdict(spans: dict[str, list[float]], min_green_m: float, max_red_m: float) -> list[str]: + """Gate failures, empty when the checkpoint passes.""" + if not spans["green"] and not spans["red"]: + raise SystemExit( + "no signalled route lanes in any scene — this gate needs scenes whose route " + "carries a traffic-light state; refusing to report a vacuous pass" + ) + failures = [] + if spans["green"]: + green = float(np.mean(spans["green"])) + if green < min_green_m: + failures.append(f"green {green:.2f} m < {min_green_m} m (does not commit to go)") + if spans["red"]: + red = float(np.mean(spans["red"])) + if red > max_red_m: + failures.append(f"red {red:.2f} m > {max_red_m} m (runs the light)") + return failures + + +def report(spans: dict[str, list[float]], model_path: str, n_scenes: int, shape) -> None: + print(f"model : {model_path}") + print(f"scenes : {n_scenes} " + " ".join(f"{k} {len(spans[k])}" for k in _ORDER)) + print(f"shape : {shape if shape else 'as recorded'}") + for k in _ORDER: + if spans[k]: + print(f" {k:<6} n={len(spans[k]):<3} mean span {float(np.mean(spans[k])):6.2f} m") + + +def _parse_args() -> argparse.Namespace: + ap = argparse.ArgumentParser() + ap.add_argument("--model", required=True, help="deployable checkpoint (args.json alongside)") + ap.add_argument("--scenes", required=True, help="scene dir, glob, or single .npz") + ap.add_argument("--goal", default=None, help="x,y ego-frame goal override") + ap.add_argument("--shape", default=None, help="wheelbase,length,width to evaluate at") + ap.add_argument("--min_green_m", type=float, default=15.0) + ap.add_argument("--max_red_m", type=float, default=2.0) + ap.add_argument("--limit", type=int, default=0, help="0 = every scene") + return ap.parse_args() + + +def main() -> int: + a = _parse_args() + goal = tuple(float(x) for x in a.goal.split(",")) if a.goal else None + shape = parse_shape(a.shape) if a.shape else None + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + model, margs = load_deployable_model(a.model, device) + scenes = find_scenes(a.scenes) + if a.limit: + scenes = scenes[: a.limit] + + spans = measure(model, margs, scenes, device, goal, shape) + report(spans, a.model, len(scenes), shape) + + failures = verdict(spans, a.min_green_m, a.max_red_m) + print() + for f in failures: + print(f"FAIL: {f}") + print("PASS" if not failures else "FAIL") + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/rlvr/autoresearch/ego_shape_diag/check_token_ablation.py b/rlvr/autoresearch/ego_shape_diag/check_token_ablation.py new file mode 100644 index 000000000..8a6fe4a8a --- /dev/null +++ b/rlvr/autoresearch/ego_shape_diag/check_token_ablation.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +"""Which input group is holding the plan back? Zero one at a time and re-measure. + +When a model plans a standstill on geometry it should drive through, this says which +part of the input it is reacting to. Zero each input group in turn and watch the plan +span: a group whose removal releases the plan is the one the model is responding to. + +``--split_line_strings`` goes further within that group, separating stop lines from road +borders and additionally pushing the borders outward. The distinction matters: if +dropping the borders releases the plan but moving them away does not, the model is not +failing to fit through a gap — it is holding at a boundary it has learned to respect. + + python -m rlvr.autoresearch.ego_shape_diag.check_token_ablation \ + --model --scene --shape W,L,W \ + [--goal X,Y] [--split_line_strings] + +Ablation shows what the model attends to, not what is wrong with the data; confirm any +finding against the corpus with ``check_training_data``. +""" + +from __future__ import annotations + +import argparse + +import torch + +from rlvr.autoresearch.ego_shape_diag._common import ( + find_scenes, + load_deployable_model, + load_scene, + parse_shape, + plan_span, + set_shape, +) + +GROUPS = [ + "neighbor_agents_past", + "static_objects", + "line_strings", + "polygons", + "lanes", + "route_lanes", + "ego_agent_past", +] +_STOP_LINE_CHANNEL = 2 +_ROAD_BORDER_CHANNEL = 3 + + +def ablate_groups(model, margs, base: dict, ref: float, device) -> None: + for key in GROUPS: + if key not in base: + continue + d = dict(base) + d[key] = torch.zeros_like(base[key]) + span = plan_span(model, margs, d, device) + tag = " <-- RELEASES" if span > max(ref * 3, 10) else "" + print(f" {'zero ' + key:32s} {span:7.2f} m{tag}") + + +def split_line_strings(model, margs, base: dict, device) -> None: + """Separate stop lines from road borders, then test clearance by moving borders.""" + strings = base["line_strings"] + is_stop = strings[..., _STOP_LINE_CHANNEL].abs().sum(-1) > 0 + is_border = strings[..., _ROAD_BORDER_CHANNEL].abs().sum(-1) > 0 + print( + f"\n line_strings slots: stop_line={int(is_stop.sum())} road_border={int(is_border.sum())}" + ) + for name, mask in (("stop_line", is_stop), ("road_border", is_border)): + d = dict(base) + d["line_strings"] = strings * (~mask).float().unsqueeze(-1).unsqueeze(-1) + print(f" {'drop ' + name + ' only':32s} {plan_span(model, margs, d, device):7.2f} m") + for push in (1.0, 4.0, 8.0): + d = dict(base) + moved = strings.clone() + xy = moved[..., :2] + radius = xy.norm(dim=-1, keepdim=True).clamp(min=1e-3) + selected = is_border.unsqueeze(-1).unsqueeze(-1).float() + occupied = (xy.abs().sum(-1, keepdim=True) > 0).float() + moved[..., :2] = xy + (xy / radius) * push * selected * occupied + d["line_strings"] = moved + print( + f" {'push road borders +%.0f m' % push:32s} " + f"{plan_span(model, margs, d, device):7.2f} m" + ) + print("\n Borders pushed away but the plan unchanged => not a fit/clearance problem.") + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--model", required=True, help="deployable checkpoint (args.json alongside)") + ap.add_argument("--scene", required=True, help="a single .npz (or dir; the first is used)") + ap.add_argument( + "--shape", required=True, help="ego_shape to hold fixed: wheelbase,length,width" + ) + ap.add_argument("--goal", default=None, help="x,y ego-frame goal override") + ap.add_argument( + "--split_line_strings", + action="store_true", + help="also separate stop lines from road borders within that group", + ) + a = ap.parse_args() + + goal = tuple(float(x) for x in a.goal.split(",")) if a.goal else None + device = torch.device("cuda" if torch.cuda.is_available() else "cpu") + model, margs = load_deployable_model(a.model, device) + scene = find_scenes(a.scene)[0] + base = set_shape(load_scene(scene, device, goal), *parse_shape(a.shape)) + + ref = plan_span(model, margs, base, device) + print(f"scene: {scene.split('/')[-1]}") + print(f"ego_shape held at {a.shape}\n") + print(f" {'reference (untouched)':32s} {ref:7.2f} m") + ablate_groups(model, margs, base, ref, device) + if a.split_line_strings and "line_strings" in base: + split_line_strings(model, margs, base, device) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/rlvr/autoresearch/ego_shape_diag/check_training_data.py b/rlvr/autoresearch/ego_shape_diag/check_training_data.py new file mode 100644 index 000000000..cb08f1ef2 --- /dev/null +++ b/rlvr/autoresearch/ego_shape_diag/check_training_data.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +"""Audit a training scene set for a correlation between ego dimensions and speed. + +If a corpus mixes vehicle platforms, ``ego_shape`` can end up being the only input that +separates them — and if those groups also differ in how far they travel, the token stops +meaning geometry and starts meaning "which speed distribution do I imitate". That is a +property of the data, visible before any training run. + +Reports per dimension class: + * how many distinct ``ego_shape`` values exist (two is a binary switch: highest risk) + * travel over the prediction horizon, the quantity that must NOT be class-dependent + * how often the ego is already stopped, and how often the target is stationary + * goal-distance distribution, as a sanity check on the conversion + + python -m rlvr.autoresearch.ego_shape_diag.check_training_data \ + --scenes [--wheelbase_split 4.0] [--sample 800] + +Healthy result: the travel distributions overlap. If one class is markedly slower, either +decorrelate the token (jitter, dropout, rebalance) or drop the off-platform data — and +check the resulting model with ``check_ego_shape_gate``. +""" + +from __future__ import annotations + +import argparse +import collections +import glob +import json +import os +import random + +import numpy as np + + +def load_list(path: str) -> list[str]: + """Accept a directory, a glob, or a dataset-list JSON (list, or dict of lists).""" + if os.path.isdir(path): + return sorted(glob.glob(os.path.join(path, "**", "*.npz"), recursive=True)) + if path.endswith(".json"): + data = json.loads(open(path).read()) + if isinstance(data, dict): + for key in ("path_list", "scenes", "npz_list"): + if key in data: + return list(data[key]) + return list(next(iter(data.values()))) + return list(data) + return sorted(glob.glob(path)) + + +def collect(paths: list[str], split: float) -> tuple[collections.Counter, dict]: + """Tally distinct ego_shape values and per-class travel / goal / stopped stats.""" + shapes: collections.Counter = collections.Counter() + per: dict = collections.defaultdict(lambda: {"travel": [], "goal": [], "stopped": 0, "n": 0}) + for path in paths: + try: + z = np.load(path) + except Exception: + continue + if "ego_shape" not in z: + continue + shape = tuple(np.round(z["ego_shape"], 4)) + shapes[shape] += 1 + entry = per[f"wheelbase > {split:g}" if shape[0] > split else f"wheelbase <= {split:g}"] + entry["n"] += 1 + if "ego_agent_future" in z: + future = z["ego_agent_future"][:, :2] + entry["travel"].append(float(np.linalg.norm(future[-1] - future[0]))) + if "goal_pose" in z: + g = z["goal_pose"] + entry["goal"].append(float(np.hypot(g[0], g[1]))) + if "ego_current_state" in z: + state = z["ego_current_state"] + if len(state) > 5 and float(np.hypot(state[4], state[5])) < 0.5: + entry["stopped"] += 1 + return shapes, per + + +def report_classes(per: dict) -> None: + for name, entry in per.items(): + travel = np.array(entry["travel"]) + goal = np.array(entry["goal"]) + print(f"\n {name} n={entry['n']}") + if len(travel): + print( + f" horizon travel : p10={np.percentile(travel, 10):6.2f} " + f"p50={np.percentile(travel, 50):6.2f} p90={np.percentile(travel, 90):6.2f} " + f"mean={travel.mean():6.2f} m" + ) + print(f" stationary targets (<2 m): {(travel < 2).mean() * 100:5.1f}%") + print( + f" ego stopped now (<0.5 m/s): {entry['stopped'] / max(entry['n'], 1) * 100:5.1f}%" + ) + if len(goal): + print( + f" goal distance: p5={np.percentile(goal, 5):7.1f} " + f"p50={np.percentile(goal, 50):7.1f} p95={np.percentile(goal, 95):7.1f} m" + f" frac<20m={(goal < 20).mean() * 100:4.1f}%" + ) + + +def report_gap(per: dict) -> None: + """The headline comparison, only meaningful with exactly two classes.""" + if len(per) != 2: + return + (name_a, entry_a), (name_b, entry_b) = list(per.items()) + travel_a, travel_b = np.array(entry_a["travel"]), np.array(entry_b["travel"]) + if not len(travel_a) or not len(travel_b): + return + med_a, med_b = np.percentile(travel_a, 50), np.percentile(travel_b, 50) + slower = name_a if med_a < med_b else name_b + print( + f"\n>>> median horizon travel: {name_a}={med_a:.2f} m vs {name_b}={med_b:.2f} m" + f" ({abs(med_a - med_b) / max(med_a, med_b) * 100:.0f}% gap)" + ) + print(f">>> '{slower}' is the SLOWER class. With two ego_shape values the model can use") + print(">>> the token to select that slower distribution. Decorrelate it.") + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--scenes", required=True, help="dir, glob, or dataset-list json") + ap.add_argument( + "--wheelbase_split", + type=float, + default=4.0, + help="split scenes into two classes at this wheelbase, in metres", + ) + ap.add_argument("--sample", type=int, default=800) + ap.add_argument("--seed", type=int, default=0) + a = ap.parse_args() + + paths = load_list(a.scenes) + random.seed(a.seed) + if len(paths) > a.sample: + paths = random.sample(paths, a.sample) + print(f"sampled {len(paths)} scenes\n") + + shapes, per = collect(paths, a.wheelbase_split) + if not shapes: + raise SystemExit( + f"no scene under {a.scenes!r} carries ego_shape — wrong dataset root or a " + "converter that does not emit it; refusing to report an empty audit" + ) + total = sum(shapes.values()) + binary = " <-- BINARY SWITCH, highest risk" if len(shapes) == 2 else "" + print(f"distinct ego_shape values: {len(shapes)}{binary}") + for shape, count in shapes.most_common(10): + print(f" {shape} -> {count:5d} ({count / total * 100:5.1f}%)") + + print("\nper class:") + report_classes(per) + report_gap(per) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/rlvr/test_ego_shape_gates.py b/rlvr/test_ego_shape_gates.py new file mode 100644 index 000000000..820834034 --- /dev/null +++ b/rlvr/test_ego_shape_gates.py @@ -0,0 +1,97 @@ +"""Tests for the ego-dimension diagnostics. + +The gate logic is tested without a model: traffic-light classification reads the scene, +and the verdict is a pure function of the measured spans. +""" + +import numpy as np +import pytest +import torch +from diffusion_planner.dimensions import ( + TRAFFIC_LIGHT_GREEN, + TRAFFIC_LIGHT_RED, + TRAFFIC_LIGHT_YELLOW, +) + +from rlvr.autoresearch.ego_shape_diag import _common +from rlvr.autoresearch.ego_shape_diag.check_tl_gate import route_tl_class, verdict + + +def _scene(*onehot_indices: int, valid: bool = True) -> dict: + """One scene whose route lanes carry the given traffic-light one-hots.""" + route = np.zeros((max(len(onehot_indices), 1), 20, 33), dtype=np.float32) + for lane, index in enumerate(onehot_indices): + route[lane, :, 0] = 1.0 # mark the lane valid + route[lane, 0, index] = 1.0 + if not valid: + route[:] = 0.0 + return {"route_lanes": torch.from_numpy(route)} + + +def test_route_tl_class_reads_each_state(): + assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN)) == "green" + assert route_tl_class(_scene(TRAFFIC_LIGHT_YELLOW)) == "amber" + assert route_tl_class(_scene(TRAFFIC_LIGHT_RED)) == "red" + + +def test_route_tl_class_unsignalled_is_none(): + # a valid lane carrying no traffic-light one-hot, and an all-zero (absent) route + assert route_tl_class(_scene()) == "none" + assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, valid=False)) == "none" + + +def test_route_tl_class_red_outranks_green(): + """A route that includes a red signal is a hold, whatever the other lanes show.""" + assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, TRAFFIC_LIGHT_RED)) == "red" + assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, TRAFFIC_LIGHT_YELLOW)) == "amber" + + +def test_verdict_passes_when_it_goes_on_green_and_holds_on_red(): + spans = {"green": [16.0, 18.0], "amber": [0.1], "red": [0.04], "none": []} + assert verdict(spans, min_green_m=15.0, max_red_m=2.0) == [] + + +def test_verdict_catches_a_model_that_will_not_go(): + spans = {"green": [1.0, 2.0], "amber": [], "red": [0.04], "none": []} + (failure,) = verdict(spans, min_green_m=15.0, max_red_m=2.0) + assert "does not commit to go" in failure + + +def test_verdict_catches_a_model_that_runs_the_light(): + """The failure an unstratified span gate scores as healthy.""" + spans = {"green": [20.0], "amber": [], "red": [19.0], "none": []} + (failure,) = verdict(spans, min_green_m=15.0, max_red_m=2.0) + assert "runs the light" in failure + + +def test_verdict_refuses_a_vacuous_pass(): + """No signalled scene means the gate proved nothing; it must not report PASS.""" + with pytest.raises(SystemExit, match="no signalled route lanes"): + verdict({"green": [], "amber": [1.0], "red": [], "none": [5.0]}, 15.0, 2.0) + + +def test_amber_is_reported_but_never_gated(): + holds = {"green": [16.0], "amber": [0.1], "red": [0.04], "none": []} + proceeds = {"green": [16.0], "amber": [17.0], "red": [0.04], "none": []} + assert verdict(holds, 15.0, 2.0) == verdict(proceeds, 15.0, 2.0) == [] + + +def test_load_deployable_model_refuses_a_training_milestone(tmp_path): + """A milestone holds raw AND EMA weights; load_model would silently take the raw.""" + milestone = tmp_path / "milestone.pth" + torch.save({"model": {"w": torch.zeros(1)}, "ema_state_dict": {"w": torch.ones(1)}}, milestone) + with pytest.raises(SystemExit, match="training milestone"): + _common.load_deployable_model(str(milestone), torch.device("cpu")) + + +def test_set_shape_replaces_dimensions_without_touching_the_original(): + scene = {"ego_shape": torch.tensor([2.75, 4.34, 1.84])} + out = _common.set_shape(scene, 4.76, 7.24, 2.43) + assert torch.allclose(out["ego_shape"], torch.tensor([4.76, 7.24, 2.43])) + assert torch.allclose(scene["ego_shape"], torch.tensor([2.75, 4.34, 1.84])) + + +def test_parse_shape_rejects_a_wrong_length(): + assert _common.parse_shape("4.76,7.24,2.43") == (4.76, 7.24, 2.43) + with pytest.raises(SystemExit, match="wheelbase,length,width"): + _common.parse_shape("4.76,7.24") From a497bc3881b36e7fadc0770284e03d0f7197b20d Mon Sep 17 00:00:00 2001 From: Daniel Sanchez Date: Fri, 4 Sep 2026 14:59:20 +0900 Subject: [PATCH 2/4] fix(eval): require both gate halves; narrow to the traffic-light gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review P1: verdict() only refused a scene set when BOTH buckets were empty, so a green-only set reported PASS while the red half went untested — and a green-only set is exactly what cannot catch a checkpoint that runs red lights. Both halves are now required; --allow_one_sided grades what is present and names the untested half in the result, which is then not a full acceptance. Four regression tests cover green-only, red-only, the override, and that the override relaxes coverage rather than thresholds. The four ego-dimension diagnostics are dropped from this PR. They were written to root-cause one specific corpus problem and belong with that campaign's assets, not in shared tooling where nothing else exercises them. Review P2 (sweep spread not part of that gate's verdict) is valid and is fixed there rather than here. --- rlvr/autoresearch/ego_shape_diag/README.md | 123 +++++--------- .../ego_shape_diag/check_ego_shape_gate.py | 91 ---------- .../ego_shape_diag/check_route_ab.py | 132 --------------- .../ego_shape_diag/check_tl_gate.py | 48 +++++- .../ego_shape_diag/check_token_ablation.py | 120 ------------- .../ego_shape_diag/check_training_data.py | 157 ------------------ ...est_ego_shape_gates.py => test_tl_gate.py} | 40 ++++- 7 files changed, 118 insertions(+), 593 deletions(-) delete mode 100644 rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py delete mode 100644 rlvr/autoresearch/ego_shape_diag/check_route_ab.py delete mode 100644 rlvr/autoresearch/ego_shape_diag/check_token_ablation.py delete mode 100644 rlvr/autoresearch/ego_shape_diag/check_training_data.py rename rlvr/{test_ego_shape_gates.py => test_tl_gate.py} (70%) diff --git a/rlvr/autoresearch/ego_shape_diag/README.md b/rlvr/autoresearch/ego_shape_diag/README.md index 2a036ab1f..c8f9963fe 100644 --- a/rlvr/autoresearch/ego_shape_diag/README.md +++ b/rlvr/autoresearch/ego_shape_diag/README.md @@ -1,103 +1,60 @@ -# Ego-dimension diagnostics +# Traffic-light acceptance gate -Pre-deployment checks for a failure that open-loop metrics do not detect: the model -plans a **standstill** on geometry it should drive through, at some values of the -`ego_shape` (vehicle-dimension) input. - -`ego_shape` is supposed to describe geometry — it should change the shape of a -manoeuvre, never whether the vehicle moves at all. But if a training corpus mixes -vehicle platforms and those platforms also differ in how fast they drive, the token -becomes the cheapest predictor of speed available to the model, and it gets read as -"which speed distribution do I imitate". The resulting model can score normally on -displacement error while being undeployable, because the collapse is confined to the -geometry and dimensions where it plans to stop. - -These tools measure that directly, in seconds, without a closed-loop simulation. Run -them from the repo root, and point `--model` at a **deployable** checkpoint: a training -milestone carries both raw optimizer iterates and the EMA copy, and loading the raw ones -gives a plausible but wrong verdict. Every script here refuses such a checkpoint and -prints the conversion. - -## 1. `check_tl_gate.py` — the acceptance gate, traffic-light stratified - -The one to run on every checkpoint when the scenes carry traffic-light state. +A pre-deployment check on a planner checkpoint: **does it plan to go on green and hold on +red?** Seconds of compute, no closed-loop simulation. ``` python -m rlvr.autoresearch.ego_shape_diag.check_tl_gate \ --model --scenes \ - --shape [--goal X,Y] + [--shape wheelbase,length,width] [--goal X,Y] ``` -A stop is only a failure if the light says go. Grading scenes that span a light cycle -with a single threshold gets this backwards twice over: it marks the red frames as -stalls, and it scores a model that *runs* red lights as healthy. So scenes are grouped -by the traffic-light state their own route lanes carry, and the gate asks for -`mean span >= --min_green_m` on green and `<= --max_red_m` on red. Amber is reported -but not gated — easing and proceeding are both defensible, and the choice is a -behaviour difference between checkpoints rather than a pass or a fail. +## Why a stratified gate -If no scene carries a signalled route lane the tool fails rather than reporting a -vacuous pass. +Open-loop trajectory error does not detect either failure this gate is for. A model can +hold a standstill plan at signalised geometry and still score normally on displacement +metrics; a model that drives through stop lines scores normally too, because the recorded +future it is compared against is a car that stopped for perfectly ordinary reasons. -## 2. `check_ego_shape_gate.py` — the acceptance gate, dimension sweep +Grading such scenes with a single plan-span threshold is worse than useless. Scenes taken +across a light cycle contain frames where standing still is the *correct* answer, so a +one-threshold gate marks those as stalls — and, in the same breath, scores a model that +runs reds as healthy, since running a red produces exactly the long confident plan the +threshold rewards. -For scenes without traffic-light state, or to locate the cliff. +So scenes are grouped by the traffic-light state their own route lanes carry, and each +group is held to its own standard: -``` -python -m rlvr.autoresearch.ego_shape_diag.check_ego_shape_gate \ - --model --scenes [--wheelbases 2.75,3.5,4.0,4.5,4.76] -``` +| group | requirement | catches | +|---|---|---| +| green | mean plan span ≥ `--min_green_m` | a checkpoint that will not commit to go | +| red | mean plan span ≤ `--max_red_m` | a checkpoint that drives through the stop | +| amber | reported, never graded | — | -Sweeps the wheelbase with the other dimensions fixed and reports the plan span at each, -plus the spread. A healthy model is flat across the sweep. A cliff — normal spans at one -end, near-zero at the other — means the token is acting as a speed prior, and the spread -is the size of the effect. +Amber is deliberately not a criterion: easing and proceeding are both defensible, so it +is a behaviour difference between checkpoints rather than a pass or a fail. It is printed +because that difference is worth seeing. -## 3. `check_training_data.py` — find the confound in the corpus +## What makes a result trustworthy -``` -python -m rlvr.autoresearch.ego_shape_diag.check_training_data \ - --scenes [--wheelbase_split 4.0] [--sample 800] -``` - -Splits a scene set by wheelbase and compares travel over the prediction horizon between -the two classes. Two red flags: only a couple of distinct `ego_shape` values (a binary -switch is the easiest thing for a model to key on), and a large gap in median travel -between the classes. Healthy corpora have overlapping travel distributions. This runs on -data alone, so it can be checked before committing to a training run. - -## 4. `check_route_ab.py` — pathological stops versus real ones - -``` -python -m rlvr.autoresearch.ego_shape_diag.check_route_ab \ - --model --scenes --shape_a W,L,W --shape_b W,L,W -``` - -Scores every scene along a route at two dimension settings. A stop where one setting -stands still and the other drives is attributable to the token; a stop where both agree -is a real one. Report the fractions, not individual frames — and cross-check the -pathological timestamps against traffic-light state before concluding anything, since -stops on red are legitimate however the settings differ. - -## 5. `check_token_ablation.py` — which input is holding the plan - -``` -python -m rlvr.autoresearch.ego_shape_diag.check_token_ablation \ - --model --scene --shape W,L,W [--split_line_strings] -``` - -Zeroes each input group in turn; a group whose removal releases the plan is what the -model is reacting to. `--split_line_strings` separates stop lines from road borders and -also pushes the borders outward, which distinguishes "cannot fit" from "will not cross": -if dropping borders releases the plan but moving them away does not, it is the latter. +* **Both halves must be exercised.** They catch opposite failures, so a set carrying only + one leaves the other undetectable while still printing PASS. A green-only set cannot + notice a red-runner. The gate refuses such a set; `--allow_one_sided` overrides that + and names the untested half in the result, which is then not a full acceptance. +* **Point `--model` at a deployable checkpoint.** A training milestone holds both the raw + optimizer iterates and the EMA copy, and the loader takes the raw ones — a plausible + but wrong verdict. The gate refuses a milestone and prints the conversion. +* **Scenes need a signalled route.** If no scene carries a traffic-light state the gate + fails rather than reporting a vacuous pass. ## Scenes -These tools need scenes at the geometry where the model misbehaves — typically converted -from a recorded run, with the same converter the runtime uses so the tensors match. - -Two conversion details worth knowing, both handled here: +Use scenes at the geometry you care about, converted with the same converter the runtime +uses so the tensors match. Two conversion details, both handled here: * `ego_agent_past` may arrive as float64; the encoder needs float32. * a converter may rewrite `goal_pose` onto the ego when the recorded run ends stopped, - which makes the scene meaningless. `--goal X,Y` restores the true ego-frame goal. + which makes the scene meaningless — `--goal X,Y` restores the true ego-frame goal. + +`--shape` overrides the recorded vehicle dimensions, which is how to check a checkpoint at +the geometry it will actually be deployed on rather than the one it was recorded with. diff --git a/rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py b/rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py deleted file mode 100644 index de69547cd..000000000 --- a/rlvr/autoresearch/ego_shape_diag/check_ego_shape_gate.py +++ /dev/null @@ -1,91 +0,0 @@ -#!/usr/bin/env python3 -"""ACCEPTANCE GATE: does the planned distance depend on the ego-dimension token? It must not. - -``ego_shape`` describes the vehicle's geometry. A healthy model plans the same distance -whatever it says — geometry decides how a manoeuvre is shaped, not whether the vehicle -moves. When a training set contains only a couple of distinct ``ego_shape`` values and -those groups differ in how fast they drive, the token becomes the cheapest available -predictor of speed, and the model learns to read it as "which speed distribution do I -imitate". The failure then appears only at one end of the range: at some wheelbase the -plan collapses to a standstill on geometry the model otherwise handles. - -Sweep the token and compare. A flat sweep passes; a cliff means the token is acting as a -speed prior. Pair this with ``check_training_data`` (the confound in the data) and -``check_tl_gate`` (whether the standstill is actually correct at that moment). - - python -m rlvr.autoresearch.ego_shape_diag.check_ego_shape_gate \ - --model --scenes \ - [--goal X,Y] [--wheelbases 2.75,3.5,4.0,4.5,4.76] [--min_span 15] [--limit 5] - -Seconds of compute. Open-loop trajectory error does not detect this failure. -""" - -from __future__ import annotations - -import argparse - -import torch - -from rlvr.autoresearch.ego_shape_diag._common import ( - find_scenes, - load_deployable_model, - load_scene, - plan_span, - set_shape, -) - - -def sweep_scene(model, margs, d: dict, wheelbases, length, width, device) -> list[float]: - """Plan span at each wheelbase, holding the other dimensions fixed.""" - return [plan_span(model, margs, set_shape(d, wb, length, width), device) for wb in wheelbases] - - -def report_scene(path: str, wheelbases: list[float], sweep: list[float], min_span: float) -> bool: - """Print one scene's sweep; returns True when it fails the gate.""" - bad = [s for s in sweep if s < min_span] - print(path.split("/")[-1]) - print(" " + " ".join(f"{w}:{s:.1f}" for w, s in zip(wheelbases, sweep))) - print( - f" spread across wheelbase: {max(sweep) - min(sweep):6.2f} m" - f" {'FAIL' if bad else 'pass'}" - ) - return bool(bad) - - -def _parse_args() -> argparse.Namespace: - ap = argparse.ArgumentParser() - ap.add_argument("--model", required=True, help="deployable checkpoint (args.json alongside)") - ap.add_argument("--scenes", required=True, help="scene dir, glob, or single .npz") - ap.add_argument("--goal", default=None, help="x,y ego-frame goal override") - ap.add_argument("--wheelbases", default="2.75,3.5,4.0,4.5,4.76") - ap.add_argument("--length", type=float, default=4.34, help="held fixed across the sweep") - ap.add_argument("--width", type=float, default=1.84, help="held fixed across the sweep") - ap.add_argument("--min_span", type=float, default=15.0, help="gate threshold in metres") - ap.add_argument("--limit", type=int, default=5, help="scenes to test") - return ap.parse_args() - - -def main() -> int: - a = _parse_args() - goal = tuple(float(x) for x in a.goal.split(",")) if a.goal else None - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - model, margs = load_deployable_model(a.model, device) - wheelbases = [float(x) for x in a.wheelbases.split(",")] - scenes = find_scenes(a.scenes)[: a.limit] - - print(f"model : {a.model}") - print(f"scenes : {len(scenes)} gate: span >= {a.min_span} m at every wheelbase\n") - failures = 0 - for path in scenes: - d = load_scene(path, device, goal) - sweep = sweep_scene(model, margs, d, wheelbases, a.length, a.width, device) - failures += report_scene(path, wheelbases, sweep, a.min_span) - print( - f"\n{'FAIL' if failures else 'PASS'}: {failures}/{len(scenes)} scenes below {a.min_span} m" - ) - print("A large spread means the ego-dimension token is acting as a speed prior.") - return 1 if failures else 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/rlvr/autoresearch/ego_shape_diag/check_route_ab.py b/rlvr/autoresearch/ego_shape_diag/check_route_ab.py deleted file mode 100644 index 7c2ce68d8..000000000 --- a/rlvr/autoresearch/ego_shape_diag/check_route_ab.py +++ /dev/null @@ -1,132 +0,0 @@ -#!/usr/bin/env python3 -"""Route-wide A/B between two ego-dimension settings: which stops are pathological? - -Along a converted route, score every scene twice — once at each of two ``ego_shape`` -settings — and classify the stops: - - PATHOLOGICAL : span_a < --stop_thresh AND span_b >= --ok_thresh - (only the dimension token separates driving from standing still) - LEGITIMATE : both spans < --stop_thresh - (the settings agree there is a reason to stop) - -The counts matter more than any single frame: a handful of pathological stops is noise, -while a steady fraction across a route means the token is driving the decision. Use -``check_tl_gate`` on the same scenes to confirm whether the stops line up with red -signals, which is the usual innocent explanation. - - python -m rlvr.autoresearch.ego_shape_diag.check_route_ab \ - --model --scenes \ - --shape_a W,L,W --shape_b W,L,W [--goal X,Y] -""" - -from __future__ import annotations - -import argparse -import re - -import numpy as np -import torch - -from rlvr.autoresearch.ego_shape_diag._common import ( - find_scenes, - load_deployable_model, - load_scene, - parse_shape, - plan_span, - set_shape, -) - -_FRAME_RE = re.compile(r"(\d+)\.npz") - - -def score_route(model, margs, scenes, device, goal, shape_a, shape_b) -> list[tuple]: - """(frame index, span at shape A, span at shape B) for every scene on the route.""" - rows: list[tuple] = [] - for path in scenes: - found = _FRAME_RE.findall(path) - idx = int(found[-1]) if found else len(rows) - d = load_scene(path, device, goal) - rows.append( - ( - idx, - plan_span(model, margs, set_shape(d, *shape_a), device), - plan_span(model, margs, set_shape(d, *shape_b), device), - ) - ) - return rows - - -def classify(rows: list[tuple], stop_thresh: float, ok_thresh: float) -> tuple[list, list]: - """Split the stops into (pathological, legitimate).""" - patho = [r for r in rows if r[1] < stop_thresh and r[2] >= ok_thresh] - legit = [r for r in rows if r[1] < stop_thresh and r[2] < stop_thresh] - return patho, legit - - -def _report_spans(rows: list[tuple], stop_thresh: float) -> None: - print(f"scenes: {len(rows)}") - for name, index in (("A", 1), ("B", 2)): - spans = np.array([r[index] for r in rows]) - print( - f"shape {name} span: mean={spans.mean():6.2f} p50={np.percentile(spans, 50):6.2f} " - f"frac<{stop_thresh:g}m={(spans < stop_thresh).mean() * 100:5.1f}%" - ) - - -def _report_pathological(patho: list[tuple], dt: float) -> None: - if not patho: - return - print("\npathological frames: t(s) shape A shape B") - for idx, span_a, span_b in patho: - print(f" {idx * dt:8.1f} {span_a:7.2f} {span_b:7.2f}") - print("\nCross-check these timestamps against the traffic-light state; stops on red") - print("are legitimate however the two settings differ.") - - -def report(rows: list[tuple], stop_thresh: float, ok_thresh: float, dt: float) -> None: - patho, legit = classify(rows, stop_thresh, ok_thresh) - _report_spans(rows, stop_thresh) - print( - f"\nPATHOLOGICAL (A<{stop_thresh:g} and B>={ok_thresh:g}): " - f"{len(patho)}/{len(rows)} = {len(patho) / len(rows) * 100:.1f}% of route" - ) - print( - f"LEGITIMATE (both <{stop_thresh:g}, they agree): " - f"{len(legit)}/{len(rows)} = {len(legit) / len(rows) * 100:.1f}%" - ) - _report_pathological(patho, dt) - - -def _parse_args() -> argparse.Namespace: - ap = argparse.ArgumentParser() - ap.add_argument("--model", required=True, help="deployable checkpoint (args.json alongside)") - ap.add_argument("--scenes", required=True, help="dir of NPZs converted from a full run") - ap.add_argument("--shape_a", required=True, help="wheelbase,length,width (the suspect one)") - ap.add_argument("--shape_b", required=True, help="wheelbase,length,width (the reference)") - ap.add_argument("--goal", default=None, help="x,y ego-frame goal override") - ap.add_argument("--stop_thresh", type=float, default=5.0) - ap.add_argument("--ok_thresh", type=float, default=15.0) - ap.add_argument("--dt", type=float, default=0.1, help="seconds per frame index unit") - return ap.parse_args() - - -def main() -> int: - a = _parse_args() - goal = tuple(float(x) for x in a.goal.split(",")) if a.goal else None - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - model, margs = load_deployable_model(a.model, device) - rows = score_route( - model, - margs, - find_scenes(a.scenes), - device, - goal, - parse_shape(a.shape_a), - parse_shape(a.shape_b), - ) - report(rows, a.stop_thresh, a.ok_thresh, a.dt) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py index 306da65ba..dacb543bf 100644 --- a/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py +++ b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py @@ -88,12 +88,33 @@ def measure(model, margs, scenes, device, goal, shape) -> dict[str, list[float]] return spans -def verdict(spans: dict[str, list[float]], min_green_m: float, max_red_m: float) -> list[str]: - """Gate failures, empty when the checkpoint passes.""" - if not spans["green"] and not spans["red"]: +def untested_halves(spans: dict[str, list[float]]) -> list[str]: + """The gate halves this scene set cannot exercise, in report order.""" + return [state for state in ("green", "red") if not spans[state]] + + +def verdict( + spans: dict[str, list[float]], + min_green_m: float, + max_red_m: float, + allow_one_sided: bool = False, +) -> list[str]: + """Gate failures, empty when the checkpoint passes. + + Both halves have to be exercised for a PASS to mean anything. They catch opposite + failures — the green half catches a checkpoint that will not move, the red half one + that drives through a stop — so a set carrying only one of them leaves the other + failure undetectable while still printing PASS. A green-only set in particular cannot + notice a model that runs red lights, which is the failure an unstratified span gate + already scores as healthy. + """ + missing = untested_halves(spans) + if missing and not allow_one_sided: raise SystemExit( - "no signalled route lanes in any scene — this gate needs scenes whose route " - "carries a traffic-light state; refusing to report a vacuous pass" + f"the scene set carries no {' and no '.join(missing)} scenes, so the " + f"{'/'.join(missing)} half of this gate would never be tested — yet a PASS would " + "claim both. Supply scenes spanning a light cycle, or pass --allow_one_sided to " + "grade only the half that is present." ) failures = [] if spans["green"]: @@ -125,6 +146,12 @@ def _parse_args() -> argparse.Namespace: ap.add_argument("--min_green_m", type=float, default=15.0) ap.add_argument("--max_red_m", type=float, default=2.0) ap.add_argument("--limit", type=int, default=0, help="0 = every scene") + ap.add_argument( + "--allow_one_sided", + action="store_true", + help="grade even when the scenes cover only green or only red; the untested half " + "is named in the result, which is then not a full acceptance", + ) return ap.parse_args() @@ -141,11 +168,18 @@ def main() -> int: spans = measure(model, margs, scenes, device, goal, shape) report(spans, a.model, len(scenes), shape) - failures = verdict(spans, a.min_green_m, a.max_red_m) + failures = verdict(spans, a.min_green_m, a.max_red_m, a.allow_one_sided) print() for f in failures: print(f"FAIL: {f}") - print("PASS" if not failures else "FAIL") + missing = untested_halves(spans) + if failures: + print("FAIL") + elif missing: + # Never let a half that was never exercised read as an acceptance. + print(f"PASS ({'/'.join(missing)} half NOT tested — not a full acceptance)") + else: + print("PASS") return 1 if failures else 0 diff --git a/rlvr/autoresearch/ego_shape_diag/check_token_ablation.py b/rlvr/autoresearch/ego_shape_diag/check_token_ablation.py deleted file mode 100644 index 8a6fe4a8a..000000000 --- a/rlvr/autoresearch/ego_shape_diag/check_token_ablation.py +++ /dev/null @@ -1,120 +0,0 @@ -#!/usr/bin/env python3 -"""Which input group is holding the plan back? Zero one at a time and re-measure. - -When a model plans a standstill on geometry it should drive through, this says which -part of the input it is reacting to. Zero each input group in turn and watch the plan -span: a group whose removal releases the plan is the one the model is responding to. - -``--split_line_strings`` goes further within that group, separating stop lines from road -borders and additionally pushing the borders outward. The distinction matters: if -dropping the borders releases the plan but moving them away does not, the model is not -failing to fit through a gap — it is holding at a boundary it has learned to respect. - - python -m rlvr.autoresearch.ego_shape_diag.check_token_ablation \ - --model --scene --shape W,L,W \ - [--goal X,Y] [--split_line_strings] - -Ablation shows what the model attends to, not what is wrong with the data; confirm any -finding against the corpus with ``check_training_data``. -""" - -from __future__ import annotations - -import argparse - -import torch - -from rlvr.autoresearch.ego_shape_diag._common import ( - find_scenes, - load_deployable_model, - load_scene, - parse_shape, - plan_span, - set_shape, -) - -GROUPS = [ - "neighbor_agents_past", - "static_objects", - "line_strings", - "polygons", - "lanes", - "route_lanes", - "ego_agent_past", -] -_STOP_LINE_CHANNEL = 2 -_ROAD_BORDER_CHANNEL = 3 - - -def ablate_groups(model, margs, base: dict, ref: float, device) -> None: - for key in GROUPS: - if key not in base: - continue - d = dict(base) - d[key] = torch.zeros_like(base[key]) - span = plan_span(model, margs, d, device) - tag = " <-- RELEASES" if span > max(ref * 3, 10) else "" - print(f" {'zero ' + key:32s} {span:7.2f} m{tag}") - - -def split_line_strings(model, margs, base: dict, device) -> None: - """Separate stop lines from road borders, then test clearance by moving borders.""" - strings = base["line_strings"] - is_stop = strings[..., _STOP_LINE_CHANNEL].abs().sum(-1) > 0 - is_border = strings[..., _ROAD_BORDER_CHANNEL].abs().sum(-1) > 0 - print( - f"\n line_strings slots: stop_line={int(is_stop.sum())} road_border={int(is_border.sum())}" - ) - for name, mask in (("stop_line", is_stop), ("road_border", is_border)): - d = dict(base) - d["line_strings"] = strings * (~mask).float().unsqueeze(-1).unsqueeze(-1) - print(f" {'drop ' + name + ' only':32s} {plan_span(model, margs, d, device):7.2f} m") - for push in (1.0, 4.0, 8.0): - d = dict(base) - moved = strings.clone() - xy = moved[..., :2] - radius = xy.norm(dim=-1, keepdim=True).clamp(min=1e-3) - selected = is_border.unsqueeze(-1).unsqueeze(-1).float() - occupied = (xy.abs().sum(-1, keepdim=True) > 0).float() - moved[..., :2] = xy + (xy / radius) * push * selected * occupied - d["line_strings"] = moved - print( - f" {'push road borders +%.0f m' % push:32s} " - f"{plan_span(model, margs, d, device):7.2f} m" - ) - print("\n Borders pushed away but the plan unchanged => not a fit/clearance problem.") - - -def main() -> int: - ap = argparse.ArgumentParser() - ap.add_argument("--model", required=True, help="deployable checkpoint (args.json alongside)") - ap.add_argument("--scene", required=True, help="a single .npz (or dir; the first is used)") - ap.add_argument( - "--shape", required=True, help="ego_shape to hold fixed: wheelbase,length,width" - ) - ap.add_argument("--goal", default=None, help="x,y ego-frame goal override") - ap.add_argument( - "--split_line_strings", - action="store_true", - help="also separate stop lines from road borders within that group", - ) - a = ap.parse_args() - - goal = tuple(float(x) for x in a.goal.split(",")) if a.goal else None - device = torch.device("cuda" if torch.cuda.is_available() else "cpu") - model, margs = load_deployable_model(a.model, device) - scene = find_scenes(a.scene)[0] - base = set_shape(load_scene(scene, device, goal), *parse_shape(a.shape)) - - ref = plan_span(model, margs, base, device) - print(f"scene: {scene.split('/')[-1]}") - print(f"ego_shape held at {a.shape}\n") - print(f" {'reference (untouched)':32s} {ref:7.2f} m") - ablate_groups(model, margs, base, ref, device) - if a.split_line_strings and "line_strings" in base: - split_line_strings(model, margs, base, device) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/rlvr/autoresearch/ego_shape_diag/check_training_data.py b/rlvr/autoresearch/ego_shape_diag/check_training_data.py deleted file mode 100644 index cb08f1ef2..000000000 --- a/rlvr/autoresearch/ego_shape_diag/check_training_data.py +++ /dev/null @@ -1,157 +0,0 @@ -#!/usr/bin/env python3 -"""Audit a training scene set for a correlation between ego dimensions and speed. - -If a corpus mixes vehicle platforms, ``ego_shape`` can end up being the only input that -separates them — and if those groups also differ in how far they travel, the token stops -meaning geometry and starts meaning "which speed distribution do I imitate". That is a -property of the data, visible before any training run. - -Reports per dimension class: - * how many distinct ``ego_shape`` values exist (two is a binary switch: highest risk) - * travel over the prediction horizon, the quantity that must NOT be class-dependent - * how often the ego is already stopped, and how often the target is stationary - * goal-distance distribution, as a sanity check on the conversion - - python -m rlvr.autoresearch.ego_shape_diag.check_training_data \ - --scenes [--wheelbase_split 4.0] [--sample 800] - -Healthy result: the travel distributions overlap. If one class is markedly slower, either -decorrelate the token (jitter, dropout, rebalance) or drop the off-platform data — and -check the resulting model with ``check_ego_shape_gate``. -""" - -from __future__ import annotations - -import argparse -import collections -import glob -import json -import os -import random - -import numpy as np - - -def load_list(path: str) -> list[str]: - """Accept a directory, a glob, or a dataset-list JSON (list, or dict of lists).""" - if os.path.isdir(path): - return sorted(glob.glob(os.path.join(path, "**", "*.npz"), recursive=True)) - if path.endswith(".json"): - data = json.loads(open(path).read()) - if isinstance(data, dict): - for key in ("path_list", "scenes", "npz_list"): - if key in data: - return list(data[key]) - return list(next(iter(data.values()))) - return list(data) - return sorted(glob.glob(path)) - - -def collect(paths: list[str], split: float) -> tuple[collections.Counter, dict]: - """Tally distinct ego_shape values and per-class travel / goal / stopped stats.""" - shapes: collections.Counter = collections.Counter() - per: dict = collections.defaultdict(lambda: {"travel": [], "goal": [], "stopped": 0, "n": 0}) - for path in paths: - try: - z = np.load(path) - except Exception: - continue - if "ego_shape" not in z: - continue - shape = tuple(np.round(z["ego_shape"], 4)) - shapes[shape] += 1 - entry = per[f"wheelbase > {split:g}" if shape[0] > split else f"wheelbase <= {split:g}"] - entry["n"] += 1 - if "ego_agent_future" in z: - future = z["ego_agent_future"][:, :2] - entry["travel"].append(float(np.linalg.norm(future[-1] - future[0]))) - if "goal_pose" in z: - g = z["goal_pose"] - entry["goal"].append(float(np.hypot(g[0], g[1]))) - if "ego_current_state" in z: - state = z["ego_current_state"] - if len(state) > 5 and float(np.hypot(state[4], state[5])) < 0.5: - entry["stopped"] += 1 - return shapes, per - - -def report_classes(per: dict) -> None: - for name, entry in per.items(): - travel = np.array(entry["travel"]) - goal = np.array(entry["goal"]) - print(f"\n {name} n={entry['n']}") - if len(travel): - print( - f" horizon travel : p10={np.percentile(travel, 10):6.2f} " - f"p50={np.percentile(travel, 50):6.2f} p90={np.percentile(travel, 90):6.2f} " - f"mean={travel.mean():6.2f} m" - ) - print(f" stationary targets (<2 m): {(travel < 2).mean() * 100:5.1f}%") - print( - f" ego stopped now (<0.5 m/s): {entry['stopped'] / max(entry['n'], 1) * 100:5.1f}%" - ) - if len(goal): - print( - f" goal distance: p5={np.percentile(goal, 5):7.1f} " - f"p50={np.percentile(goal, 50):7.1f} p95={np.percentile(goal, 95):7.1f} m" - f" frac<20m={(goal < 20).mean() * 100:4.1f}%" - ) - - -def report_gap(per: dict) -> None: - """The headline comparison, only meaningful with exactly two classes.""" - if len(per) != 2: - return - (name_a, entry_a), (name_b, entry_b) = list(per.items()) - travel_a, travel_b = np.array(entry_a["travel"]), np.array(entry_b["travel"]) - if not len(travel_a) or not len(travel_b): - return - med_a, med_b = np.percentile(travel_a, 50), np.percentile(travel_b, 50) - slower = name_a if med_a < med_b else name_b - print( - f"\n>>> median horizon travel: {name_a}={med_a:.2f} m vs {name_b}={med_b:.2f} m" - f" ({abs(med_a - med_b) / max(med_a, med_b) * 100:.0f}% gap)" - ) - print(f">>> '{slower}' is the SLOWER class. With two ego_shape values the model can use") - print(">>> the token to select that slower distribution. Decorrelate it.") - - -def main() -> int: - ap = argparse.ArgumentParser() - ap.add_argument("--scenes", required=True, help="dir, glob, or dataset-list json") - ap.add_argument( - "--wheelbase_split", - type=float, - default=4.0, - help="split scenes into two classes at this wheelbase, in metres", - ) - ap.add_argument("--sample", type=int, default=800) - ap.add_argument("--seed", type=int, default=0) - a = ap.parse_args() - - paths = load_list(a.scenes) - random.seed(a.seed) - if len(paths) > a.sample: - paths = random.sample(paths, a.sample) - print(f"sampled {len(paths)} scenes\n") - - shapes, per = collect(paths, a.wheelbase_split) - if not shapes: - raise SystemExit( - f"no scene under {a.scenes!r} carries ego_shape — wrong dataset root or a " - "converter that does not emit it; refusing to report an empty audit" - ) - total = sum(shapes.values()) - binary = " <-- BINARY SWITCH, highest risk" if len(shapes) == 2 else "" - print(f"distinct ego_shape values: {len(shapes)}{binary}") - for shape, count in shapes.most_common(10): - print(f" {shape} -> {count:5d} ({count / total * 100:5.1f}%)") - - print("\nper class:") - report_classes(per) - report_gap(per) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/rlvr/test_ego_shape_gates.py b/rlvr/test_tl_gate.py similarity index 70% rename from rlvr/test_ego_shape_gates.py rename to rlvr/test_tl_gate.py index 820834034..35cd7e2ff 100644 --- a/rlvr/test_ego_shape_gates.py +++ b/rlvr/test_tl_gate.py @@ -1,4 +1,4 @@ -"""Tests for the ego-dimension diagnostics. +"""Tests for the traffic-light acceptance gate. The gate logic is tested without a model: traffic-light classification reads the scene, and the verdict is a pure function of the measured spans. @@ -14,7 +14,11 @@ ) from rlvr.autoresearch.ego_shape_diag import _common -from rlvr.autoresearch.ego_shape_diag.check_tl_gate import route_tl_class, verdict +from rlvr.autoresearch.ego_shape_diag.check_tl_gate import ( + route_tl_class, + untested_halves, + verdict, +) def _scene(*onehot_indices: int, valid: bool = True) -> dict: @@ -66,10 +70,40 @@ def test_verdict_catches_a_model_that_runs_the_light(): def test_verdict_refuses_a_vacuous_pass(): """No signalled scene means the gate proved nothing; it must not report PASS.""" - with pytest.raises(SystemExit, match="no signalled route lanes"): + with pytest.raises(SystemExit, match="no green and no red scenes"): verdict({"green": [], "amber": [1.0], "red": [], "none": [5.0]}, 15.0, 2.0) +def test_verdict_refuses_a_green_only_set(): + """The case that matters: a green-only set cannot detect a red-runner.""" + green_only = {"green": [16.0], "amber": [], "red": [], "none": []} + with pytest.raises(SystemExit, match="no red scenes"): + verdict(green_only, 15.0, 2.0) + # the override grades what is there, and the untested half is named for the caller + assert verdict(green_only, 15.0, 2.0, allow_one_sided=True) == [] + assert untested_halves(green_only) == ["red"] + + +def test_verdict_refuses_a_red_only_set(): + """The mirror case: a red-only set cannot detect a checkpoint that will not move.""" + red_only = {"green": [], "amber": [], "red": [0.04], "none": []} + with pytest.raises(SystemExit, match="no green scenes"): + verdict(red_only, 15.0, 2.0) + assert verdict(red_only, 15.0, 2.0, allow_one_sided=True) == [] + assert untested_halves(red_only) == ["green"] + + +def test_one_sided_override_still_fails_a_bad_checkpoint(): + """The override relaxes coverage, never the thresholds.""" + runs_reds = {"green": [], "amber": [], "red": [19.0], "none": []} + (failure,) = verdict(runs_reds, 15.0, 2.0, allow_one_sided=True) + assert "runs the light" in failure + + +def test_untested_halves_empty_when_both_covered(): + assert untested_halves({"green": [16.0], "amber": [], "red": [0.04], "none": []}) == [] + + def test_amber_is_reported_but_never_gated(): holds = {"green": [16.0], "amber": [0.1], "red": [0.04], "none": []} proceeds = {"green": [16.0], "amber": [17.0], "red": [0.04], "none": []} From 14bd6228d05ee3d6e70d15fb6dbaddde4de294a8 Mon Sep 17 00:00:00 2001 From: Daniel Sanchez Date: Fri, 4 Sep 2026 15:38:26 +0900 Subject: [PATCH 3/4] fix(eval): do not infer the ego's signal from any red route lane Review P1: route_lanes is not one signal. At an intersection it can carry a perpendicular approach that is correctly red while the ego's own approach is green, so red-wins precedence put ordinary green scenes in the red bucket, where a normal plan is then reported as running a light the ego never faced. Resolving which signal governs the ego needs heading alignment and stop-line proximity, which already exists in the C++ frame filters (detect_red_light_run). A second, uncross-checked Python copy of that geometry is how the two drift apart, so this takes the other option offered in review: a route whose signalled lanes disagree is classified ambiguous and excluded from the verdict, alongside scenes with no signalled lane. Both counts are printed so the excluded set is visible rather than silent. The test that asserted the old precedence encoded the bug; it now asserts exclusion, and that agreeing lanes still classify normally. Bucketing reads only route_lanes, so this is model-independent: on the 30-scene reference set it yields 0 ambiguous and the split is unchanged at 23 green / 5 amber / 2 red. --- .../ego_shape_diag/check_tl_gate.py | 40 ++++++++++++++----- rlvr/test_tl_gate.py | 14 +++++-- 2 files changed, 39 insertions(+), 15 deletions(-) diff --git a/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py index dacb543bf..0f37e607a 100644 --- a/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py +++ b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py @@ -50,15 +50,25 @@ TRAFFIC_LIGHT_YELLOW - TRAFFIC_LIGHT_GREEN: "amber", TRAFFIC_LIGHT_RED - TRAFFIC_LIGHT_GREEN: "red", } -_ORDER = ("green", "amber", "red", "none") +_ORDER = ("green", "amber", "red", "ambiguous", "none") def route_tl_class(d: dict) -> str: - """The traffic-light state carried by the scene's own route lanes. - - Red wins over amber and amber over green when several appear: a route whose lanes - include a red signal is a hold situation whatever the other lanes show. Scenes with - no signalled route lane are reported as ``none`` and excluded from the gate. + """The traffic-light state governing the ego, or ``ambiguous`` / ``none``. + + A route tensor does not carry one signal. At an intersection ``route_lanes`` can + include a perpendicular approach whose light is correctly red while the ego's own + approach is green, so treating "any red lane" as red mislabels ordinary green scenes + — and this gate would then fail a perfectly normal plan for running a light the ego + never faced. + + Deciding which signal governs the ego needs heading alignment and stop-line + proximity. That logic already exists in the C++ frame filters + (``detect_red_light_run``) and is deliberately not reimplemented here, because a + second, uncross-checked copy of the geometry is how the two drift apart. A route + whose signalled lanes disagree is therefore reported as ``ambiguous`` and left out of + the verdict: fewer graded scenes, but no invented answer. A route with no signalled + lane at all is ``none``, likewise excluded. """ rl = d["route_lanes"].detach().cpu().numpy() if rl.ndim > 3: @@ -71,10 +81,12 @@ def route_tl_class(d: dict) -> str: if not hot.any(): return "none" present = {_CLASSES.get(int(c)) for c in onehot[hot].argmax(axis=1)} - for state in ("red", "amber", "green"): - if state in present: - return state - return "none" + present.discard(None) # one-hot slots outside green/amber/red carry no state + if not present: + return "none" + if len(present) > 1: + return "ambiguous" + return present.pop() def measure(model, margs, scenes, device, goal, shape) -> dict[str, list[float]]: @@ -134,7 +146,13 @@ def report(spans: dict[str, list[float]], model_path: str, n_scenes: int, shape) print(f"shape : {shape if shape else 'as recorded'}") for k in _ORDER: if spans[k]: - print(f" {k:<6} n={len(spans[k]):<3} mean span {float(np.mean(spans[k])):6.2f} m") + print(f" {k:<9} n={len(spans[k]):<3} mean span {float(np.mean(spans[k])):6.2f} m") + excluded = len(spans["ambiguous"]) + len(spans["none"]) + if excluded: + print( + f" ({excluded} scene(s) not graded: ambiguous = route lanes disagree, " + "none = no signalled lane)" + ) def _parse_args() -> argparse.Namespace: diff --git a/rlvr/test_tl_gate.py b/rlvr/test_tl_gate.py index 35cd7e2ff..b157c4bd1 100644 --- a/rlvr/test_tl_gate.py +++ b/rlvr/test_tl_gate.py @@ -44,10 +44,16 @@ def test_route_tl_class_unsignalled_is_none(): assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, valid=False)) == "none" -def test_route_tl_class_red_outranks_green(): - """A route that includes a red signal is a hold, whatever the other lanes show.""" - assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, TRAFFIC_LIGHT_RED)) == "red" - assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, TRAFFIC_LIGHT_YELLOW)) == "amber" +def test_route_tl_class_mixed_states_are_ambiguous(): + """route_lanes can hold a perpendicular red while the ego's own approach is green. + + Picking either one would be a guess, and guessing "red" makes this gate fail a normal + plan for running a light the ego never faced. Such scenes are excluded instead. + """ + assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, TRAFFIC_LIGHT_RED)) == "ambiguous" + assert route_tl_class(_scene(TRAFFIC_LIGHT_GREEN, TRAFFIC_LIGHT_YELLOW)) == "ambiguous" + # agreeing lanes are not ambiguous + assert route_tl_class(_scene(TRAFFIC_LIGHT_RED, TRAFFIC_LIGHT_RED)) == "red" def test_verdict_passes_when_it_goes_on_green_and_holds_on_red(): From 1564553637522ff37c02d99912b7a6205d7456c1 Mon Sep 17 00:00:00 2001 From: Daniel Sanchez Date: Wed, 9 Sep 2026 09:02:23 +0900 Subject: [PATCH 4/4] fix(eval): report the spread behind each class mean, not the mean alone A mean is not enough to read this gate. Two cases it hides: A class whose scenes all return the same span is one situation sampled many times, not many independent samples -- its n overstates the evidence. And a class that splits into 'drives' and 'stands still' can average out above the threshold while a chunk of its scenes are collapsed: 18 scenes at 20 m and 5 at 1 m means 15.87 m, which clears a 15 m floor with 5 failures inside it. Both are visible in min/median/max and invisible in the mean, so print all four, and flag a class whose spread is under a centimetre. Compared with a tolerance rather than for equality: repeated measurements of one situation still differ in the last decimals, so an equality test never fires on real data. Verdicts are unchanged -- the gate still grades on the mean, so existing results reproduce exactly. This only makes the evidence behind them legible. --- .../ego_shape_diag/check_tl_gate.py | 28 +++++++++++++++++-- rlvr/test_tl_gate.py | 26 +++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py index 0f37e607a..1317c4183 100644 --- a/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py +++ b/rlvr/autoresearch/ego_shape_diag/check_tl_gate.py @@ -51,6 +51,8 @@ TRAFFIC_LIGHT_RED - TRAFFIC_LIGHT_GREEN: "red", } _ORDER = ("green", "amber", "red", "ambiguous", "none") +# Two spans closer than this are the same plan, not two measurements worth averaging. +_SAME_PLAN_M = 0.01 def route_tl_class(d: dict) -> str: @@ -141,12 +143,34 @@ def verdict( def report(spans: dict[str, list[float]], model_path: str, n_scenes: int, shape) -> None: + """Per class: the mean the gate grades on, and the spread behind it. + + The mean alone is not enough to read a result. A class whose scenes all return the + same span is one situation sampled many times, not many independent samples, and its + n overstates the evidence; a class that splits into "drives" and "stands still" can + average out above the threshold while half its scenes fail. Both are visible in + min/median/max and invisible in the mean, so print all four. + """ print(f"model : {model_path}") print(f"scenes : {n_scenes} " + " ".join(f"{k} {len(spans[k])}" for k in _ORDER)) print(f"shape : {shape if shape else 'as recorded'}") for k in _ORDER: - if spans[k]: - print(f" {k:<9} n={len(spans[k]):<3} mean span {float(np.mean(spans[k])):6.2f} m") + v = sorted(spans[k]) + if not v: + continue + # Compare with a tolerance, not for equality: scenes that are the same situation + # still differ in the last few decimals, and a spread under a centimetre is not a + # difference in behaviour at these thresholds. + degenerate = len(v) > 1 and (v[-1] - v[0]) < _SAME_PLAN_M + note = ( + f" (spread < {_SAME_PLAN_M * 100:.0f} cm - one situation, not {len(v)} samples)" + if degenerate + else "" + ) + print( + f" {k:<9} n={len(v):<3} mean {float(np.mean(v)):6.2f} m" + f" median {float(np.median(v)):6.2f} min {v[0]:6.2f} max {v[-1]:6.2f}{note}" + ) excluded = len(spans["ambiguous"]) + len(spans["none"]) if excluded: print( diff --git a/rlvr/test_tl_gate.py b/rlvr/test_tl_gate.py index b157c4bd1..29bc43969 100644 --- a/rlvr/test_tl_gate.py +++ b/rlvr/test_tl_gate.py @@ -135,3 +135,29 @@ def test_parse_shape_rejects_a_wrong_length(): assert _common.parse_shape("4.76,7.24,2.43") == (4.76, 7.24, 2.43) with pytest.raises(SystemExit, match="wheelbase,length,width"): _common.parse_shape("4.76,7.24") + + +def test_report_flags_a_degenerate_class(capsys): + """All-identical spans mean one situation sampled n times, not n samples.""" + from rlvr.autoresearch.ego_shape_diag.check_tl_gate import report + + # near-identical, as real runs of one situation are - not bit-equal + green = [16.8064 + i * 1e-6 for i in range(23)] + spans = {"green": green, "amber": [], "red": [0.04, 0.04], "none": [], "ambiguous": []} + report(spans, "ckpt.pth", 25, None) + out = capsys.readouterr().out + assert "one situation" in out + assert "median" in out and "min" in out and "max" in out + + +def test_report_exposes_a_split_the_mean_would_hide(capsys): + """18 scenes drive, 5 stand still: the mean clears 15 m while 5 scenes are collapsed.""" + from rlvr.autoresearch.ego_shape_diag.check_tl_gate import report + + spans = {"green": [20.0] * 18 + [1.0] * 5, "amber": [], "red": [], "none": [], "ambiguous": []} + report(spans, "ckpt.pth", 23, None) + out = capsys.readouterr().out + # the mean clears the 15 m threshold, but min exposes the collapsed scenes + assert "mean 15.87" in " ".join(out.split(" ")) or "15.87" in out + assert "min 1.00" in out + assert "all identical" not in out # and this class is not degenerate