Buckets:
| """cwm3/eval/verify_identity_final.py — roadmap §三 4 numeric identity check of the assembled `indep` joint run against its three base runs (CPU, read-only). | |
| python eval/verify_identity_final.py --run I1_indep_s0_final --bases V0_can4s_s0_stats2 U0_kin_can_noK_s0_kcv2_clip8_ln E0_pooled_P_s0_v2 [--out results/tables/identity_final.md] | |
| Checks per regime on the TEST split (val listed too): | |
| (a) quick metrics: every `<event>:onset_prauc@3s` of the joint's metrics_quick_*.json equals the emitting base's value to 4 decimals (V: gas / brake / | |
| steer_phase / resp; U: hand_return / hand_leave / head_turn_onset / head_return; E: the six D chains and the six `P_<chain>` road-pressure chains); | |
| (b) layouts: h_on[:, :, e] of the joint layout vs the base layout for every event the base emits — max |Δ| must be 0 with identical NaN masks (same | |
| f16 hazards); onset_p = 1 - prod(1 - h) is recomputed by each writer before the f16 cast (write_pred_layout.py in float32, write_pred_layout_joint.py | |
| in float64 with a [1e-6, 1-1e-6] clip), so it may differ by ONE f16 ulp (2^-11 for values in [0.5, 1)): tolerance F16_ULP, reported descriptively; | |
| (c) P onset: the joint layout's `p_onset_p` vs the E base layout's — max |Δ| must be 0; the chain names are compared modulo the `P_` prefix | |
| (the base writer names the columns `P_<chain>`, the joint writer `<chain>`; values are what the warning head consumes). | |
| Result: markdown table (+ PASS / FAIL) and a dict {ok, problems, rows}. No openpilot DMS field is read. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| RUNS = CWM3 / "results" / "runs" | |
| EVENTS_OF = {"V": ("gas", "brake", "steer_phase", "resp"), "U": ("hand_return", "hand_leave", "head_turn_onset", "head_return"), | |
| "E": ("steer_rate", "decel", "ttc", "ttc_drop", "lane", "badas", "P_ttc", "P_ttc_drop", "P_cut_in", "P_lead_braking", "P_lane_excursion", "P_badas")} | |
| GAZE_EVENTS = ("gaze_off_on", "gaze_off_off") # gaze U bases (2026-09-12 appendix variant): checked when the U base's quick metrics carry them; layout block gaze_off_h_on / h_off / onset_p | |
| F16_ULP = 2.0 ** -11 # one float16 ulp for values in [0.5, 1): the only admissible onset_p difference (different float precision of the two writers, hazards identical) | |
| LAYOUT_EVENTS_OF = {"V": ("gas_on", "brake_on", "steer_phase_on", "resp_on"), "U": ("hand_return", "hand_leave", "head_turn_onset", "head_return"), | |
| "E": tuple(f"D_{f}" for f in ("steer_rate", "decel", "ttc", "ttc_drop", "lane", "badas"))} | |
| def _meta(p): | |
| return json.loads(str(np.load(p, allow_pickle=True)["meta"])) | |
| def check(run, bases, out=None, splits=("test", "val"), regimes=("manual", "engaged")): | |
| rd = RUNS / run | |
| rows, problems = [], [] | |
| bc = json.loads((rd / "base_ckpts.json").read_text())["base_ckpts"] | |
| for b, d in bases.items(): | |
| if bc.get(b, "").split(":")[0] != d: | |
| problems.append(f"{run} base {b} is {bc.get(b)} not {d}") | |
| for regime in regimes: | |
| for split in splits: | |
| qj = json.loads((rd / f"metrics_quick_{regime}_{split}.json").read_text()) | |
| lj = np.load(rd / f"pred_layout_{regime}_{split}.npz", allow_pickle=True) | |
| evj = [str(x) for x in lj["event_names"]] | |
| for b, d in bases.items(): | |
| bd = RUNS / d | |
| qb = json.loads((bd / f"metrics_quick_{regime}_{split}.json").read_text()) | |
| evs = EVENTS_OF[b] + (GAZE_EVENTS if b == "U" and any(f"{e}:onset_prauc@3s" in qb for e in GAZE_EVENTS) else ()) | |
| for ev in evs: | |
| k = f"{ev}:onset_prauc@3s" | |
| if k in qj and k in qb: | |
| dq = abs(round(float(qj[k]), 4) - round(float(qb[k]), 4)) | |
| rows.append((regime, split, b, f"quick {k}", f"{float(qj[k]):.4f}", f"{float(qb[k]):.4f}", f"{dq:.4f}", "OK" if dq == 0 else "DIFF")) | |
| if dq != 0: | |
| problems.append(f"{regime}/{split} {k}: joint {qj[k]:.6f} vs {d} {qb[k]:.6f}") | |
| elif split == "test": | |
| problems.append(f"{regime}/{split} {k} missing in {'joint' if k not in qj else d}") | |
| lb = np.load(bd / f"pred_layout_{regime}_{split}.npz", allow_pickle=True) | |
| evb = [str(x) for x in lb["event_names"]] | |
| for ev in LAYOUT_EVENTS_OF[b]: | |
| if ev not in evj or ev not in evb: | |
| problems.append(f"{regime}/{split} layout event {ev} missing") | |
| continue | |
| hj, hb = lj["h_on"][:, :, evj.index(ev)].astype(np.float32), lb["h_on"][:, :, evb.index(ev)].astype(np.float32) | |
| both = np.isfinite(hj) & np.isfinite(hb) | |
| nan_mismatch = int((np.isfinite(hj) != np.isfinite(hb)).sum()) | |
| dmax = float(np.abs(hj[both] - hb[both]).max()) if both.any() else float("nan") | |
| oj, ob = lj["onset_p"][:, :, evj.index(ev)].astype(np.float32), lb["onset_p"][:, :, evb.index(ev)].astype(np.float32) | |
| bo = np.isfinite(oj) & np.isfinite(ob) | |
| omax = float(np.abs(oj[bo] - ob[bo]).max()) if bo.any() else float("nan") | |
| ok_ev = dmax == 0 and nan_mismatch == 0 and omax <= F16_ULP | |
| rows.append((regime, split, b, f"layout h_on {ev}", f"n={int(both.any(1).sum())}", f"nan-mask mismatches {nan_mismatch}", f"max|Δ| {dmax:.3g}; onset_p max|Δ| {omax:.3g} ({'≤ 1 f16 ulp' if omax <= F16_ULP else '> 1 f16 ulp'})", "OK" if ok_ev else "DIFF")) | |
| if not ok_ev: | |
| problems.append(f"{regime}/{split} h_on {ev}: max|Δ| {dmax:.3g}, onset_p {omax:.3g}, nan mismatches {nan_mismatch}") | |
| if b == "U" and "gaze_off_h_on" in lb.files: | |
| # gaze U bases: both writers compute the block with the same float64 arithmetic from the same f16 hazards -> exact equality required everywhere | |
| if "gaze_off_h_on" not in lj.files: | |
| problems.append(f"{regime}/{split}: joint layout has no gaze_off_h_on block (the U base emits the gaze chain)") | |
| else: | |
| for key in ("gaze_off_h_on", "gaze_off_h_off", "gaze_off_onset_p", "gaze_off_pi"): | |
| hj, hb = lj[key].astype(np.float32), lb[key].astype(np.float32) | |
| both = np.isfinite(hj) & np.isfinite(hb) | |
| nan_mismatch = int((np.isfinite(hj) != np.isfinite(hb)).sum()) if hj.shape == hb.shape else -1 | |
| dmax = float(np.abs(hj[both] - hb[both]).max()) if hj.shape == hb.shape and both.any() else float("nan") | |
| ok_g = hj.shape == hb.shape and dmax == 0 and nan_mismatch == 0 | |
| rows.append((regime, split, "U", f"layout {key} (gaze chain)", f"n={int(both.reshape(len(both), -1).any(1).sum())}", f"nan-mask mismatches {nan_mismatch}", f"max|Δ| {dmax:.3g}", "OK" if ok_g else "DIFF")) | |
| if not ok_g: | |
| problems.append(f"{regime}/{split} {key}: max|Δ| {dmax:.3g}, nan mismatches {nan_mismatch}, shapes {hj.shape} vs {hb.shape}") | |
| if b == "E" and "p_onset_p" in lb.files: | |
| if "p_onset_p" not in lj.files: | |
| problems.append(f"{regime}/{split}: joint layout has no p_onset_p") | |
| continue | |
| strip = lambda n: str(n)[2:] if str(n).startswith("P_") else str(n) # noqa: E731 (base writer: P_<chain>; joint writer: <chain>) | |
| pn_j, pn_b = [strip(x) for x in lj["p_names"]], [strip(x) for x in lb["p_names"]] | |
| pj, pb = lj["p_onset_p"].astype(np.float32), lb["p_onset_p"].astype(np.float32) | |
| bp = np.isfinite(pj) & np.isfinite(pb) | |
| pmax = float(np.abs(pj[bp] - pb[bp]).max()) if bp.any() else float("nan") | |
| ok = pn_j == pn_b and pmax == 0 and pj.shape == pb.shape | |
| rows.append((regime, split, "E", "layout p_onset_p (6 P chains x 4 taus)", f"chain names equal (modulo P_ prefix) {pn_j == pn_b}", f"shape {pj.shape}", f"max|Δ| {pmax:.3g}", "OK" if ok else "DIFF")) | |
| if not ok: | |
| problems.append(f"{regime}/{split} p_onset_p: names {pn_j == pn_b}, max|Δ| {pmax:.3g}") | |
| ok = not problems | |
| lines = [f"# Identity check of `{run}` against its bases (eval/verify_identity_final.py; roadmap §三 4)", "", | |
| f"Bases: " + ", ".join(f"{b} = `{d}` (BEST {bc.get(b, '?').split(':')[-1]})" for b, d in bases.items()), "", | |
| "Rule: quick onset PR-AUC@3s equal to 4 decimals for every event a base emits; layout `h_on` max |Δ| = 0 with identical NaN masks (onset_p, recomputed by each " | |
| "writer before the f16 cast — float32 in write_pred_layout.py, float64 in write_pred_layout_joint.py — may differ by one f16 ulp = 2^-11 and is reported); " | |
| "the joint's `p_onset_p` identical to the E base's (P head copied at init; column names compared modulo the `P_` prefix of the base writer); " | |
| "gaze U bases: `gaze_off_on` / `gaze_off_off` quick metrics to 4 decimals and the layout block gaze_off_h_on / h_off / onset_p / pi exactly equal.", "", | |
| "| regime | split | base | check | joint | base | Δ | status |", "|---|---|---|---|---|---|---|---|"] | |
| lines += [f"| {r} | {s} | {b} | {c} | {a} | {bb} | {d} | {st} |" for r, s, b, c, a, bb, d, st in rows] | |
| lines += ["", f"## Verdict: **{'PASS' if ok else 'FAIL'}**" + ("" if ok else " — " + "; ".join(problems[:20]))] | |
| md = "\n".join(lines) + "\n" | |
| if out: | |
| Path(out).write_text(md) | |
| return {"ok": ok, "problems": problems, "rows": rows, "md": md} | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--run", required=True) | |
| ap.add_argument("--bases", nargs=3, required=True, metavar=("V_DIR", "U_DIR", "E_DIR")) | |
| ap.add_argument("--out", default=str(CWM3 / "results" / "tables" / "identity_final.md")) | |
| a = ap.parse_args() | |
| res = check(a.run, dict(zip(("V", "U", "E"), a.bases)), a.out) | |
| print(res["md"]) | |
| print(f"written {a.out}") | |
| sys.exit(0 if res["ok"] else 1) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 10.5 kB
- Xet hash:
- dd27eb00ee301357abc01238ada85199ba87be31af982fad8c899108b34663b0
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.