Buckets:
| #!/usr/bin/env python3 | |
| """cwm3/eval/report_can_arms.py — results/tables/can_arms.md for the OPTIONAL BATON-side CAN arms on the driver chains (plan Part A §A.5; pre-registration | |
| docs/decisions.md 2026-09-15T01:48:39Z). | |
| Reads ONLY: results/paired_compare.csv (rows written by the unchanged eval/paired_compare_joint.py, tags canU_s<k>_<nocan|cangate>_vs_u0base; latest row per | |
| tag / regime / cell / metric), the arm runs' run json + history.jsonl + train.log (provenance, wall time), the BEST checkpoint of the cangate runs (the learned | |
| a, b) and the `gate2` diagnostic of their prediction packs. Verdict per cell = cwm2/eval/kill_checks.beats over the 3 seeds (hazard NLL: Δ sign flipped, lower | |
| is better) — the same rule and rendering as the DMS layer-1 driver-onset table. Nothing is recomputed; the table is (re)written in full, never edited by hand. | |
| python eval/report_can_arms.py [--out results/tables/can_arms.md] [--seeds 0 1 2] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| CWM2 = CWM3.parent / "cwm2" | |
| sys.path.insert(0, str(CWM2 / "eval")) | |
| import kill_checks as KC # noqa: E402 | |
| CSV = CWM3 / "results" / "paired_compare.csv" | |
| RUNS = CWM3 / "results" / "runs" | |
| PREREG_UTC = "2026-09-15T01:48:39Z" | |
| ARMS = {"nocan": ("U0_nocan_s{k}_kcv2_clip8_ln", "kin_can_noK_clip8_ln_nocan", "the 5 CAN columns of Kx zeroed in train/batch.prep_U (recipe u_nocan)"), | |
| "cangate": ("U0_cangate_s{k}_kcv2_clip8_ln", "kin_can_noK_clip8_ln_cangate", "CAN release gate λ' = clamp(λ·2g), g = σ(a·act + b) (recipe u_can_gate; models/branches.CanReleaseGate)")} | |
| BASE = "U0_kin_can_noK_s{k}_kcv2_clip8_ln" | |
| EVENTS = ("hand_leave", "hand_return", "head_turn_onset") | |
| CELLS = [(f"drv_onset|{e}|3s", "pr_auc") for e in EVENTS] + [(f"hazard_nll|{e}", "nll") for e in EVENTS] | |
| REGIMES = ("manual", "engaged") | |
| def latest_rows(tags): | |
| df = pd.read_csv(CSV, low_memory=False) | |
| df = df[df.tag.isin(tags)].copy() | |
| if not len(df): | |
| return df | |
| df["seed"] = df.tag.str.extract(r"_s(\d+)_")[0].astype(int) | |
| return df.sort_values("run_stamp").groupby(["tag", "regime", "split", "cell", "metric"], as_index=False).tail(1) | |
| def run_info(d): | |
| """provenance of one run dir from its run json / history / train.log (None when the run is not finished).""" | |
| js = sorted(d.glob("run_train_*_latest.json")) | |
| if not js or not (d / "PRED_DONE").exists(): | |
| return None | |
| r = json.loads(js[-1].read_text()) | |
| hist = r.get("history", []) | |
| t_first = hist[0]["t_utc"] if hist else "" | |
| t_last = hist[-1]["t_utc"] if hist else "" | |
| train_s = None | |
| if hist: | |
| t0 = time.mktime(time.strptime(r["t_start_utc"], "%Y-%m-%dT%H:%M:%SZ")) | |
| t1 = time.mktime(time.strptime(t_last, "%Y-%m-%dT%H:%M:%SZ")) | |
| train_s = t1 - t0 | |
| info = {"run": d.name, "variant": r.get("variant"), "t_start_utc": r.get("t_start_utc"), "t_end_utc": r.get("t_end_utc"), "wall_s": r.get("wall_s"), | |
| "train_phase_s": train_s, "epochs_run": r.get("epochs_run"), "best_epoch": r.get("best_epoch"), "val_objective": r.get("val_objective"), | |
| "params": (r.get("params") or {}).get("total"), "code_sha256": r.get("code_sha256", "")[:12], "command": r.get("command", ""), | |
| "kclip": r.get("kclip"), "stats_version": r.get("stats_version"), "u_nocan": r.get("u_nocan"), "u_can_gate": r.get("u_can_gate"), | |
| "t_first_epoch": t_first, "t_last_epoch": t_last, "after_prereg": bool(r.get("t_start_utc", "") > PREREG_UTC)} | |
| if r.get("u_can_gate"): | |
| try: | |
| import torch | |
| best = (d / "BEST").read_text().strip() | |
| sd = torch.load(d / best, map_location="cpu", weights_only=False)["model"] | |
| info["gate_a"], info["gate_b"] = float(sd["can_gate.a"]), float(sd["can_gate.b"]) | |
| except Exception as ex: # noqa: BLE001 | |
| info["gate_a"] = info["gate_b"] = None | |
| info["gate_err"] = str(ex)[:80] | |
| g = [] | |
| for reg in REGIMES: | |
| p = d / f"pred_{reg}_test.npz" | |
| if p.exists(): | |
| with np.load(p) as z: | |
| if "gate2" in z.files: | |
| v = z["gate2"][z["valid"].astype(bool)] | |
| g.append(v[np.isfinite(v)]) | |
| if g: | |
| v = np.concatenate(g).astype(np.float64) | |
| info["gate2_test"] = {"mean": float(v.mean()), "p05": float(np.percentile(v, 5)), "p50": float(np.percentile(v, 50)), "p95": float(np.percentile(v, 95)), | |
| "min": float(v.min()), "max": float(v.max()), "n": int(len(v))} | |
| return info | |
| def fmt(x, nd=4): | |
| return "—" if x is None or (isinstance(x, float) and not np.isfinite(x)) else f"{x:.{nd}f}" | |
| def render_arm(arm, seeds, df): | |
| run_t, variant, desc = ARMS[arm] | |
| tags = [f"canU_s{s}_{arm}_vs_u0base" for s in seeds] | |
| lines = [f"## {arm} — A = `{run_t.replace('{k}', '<k>')}` ({variant}: {desc}), B = `{BASE.replace('{k}', '<k>')}` (the frozen survivor base)", ""] | |
| g_all = df[df.tag.isin(tags)] if len(df) else df | |
| if not len(g_all): | |
| lines += [f"no rows yet (tags `{tags}`)", ""] | |
| return lines, {} | |
| lines += ["| regime | cell | metric | " + " | ".join(f"A s{s}" for s in seeds) + " | " + " | ".join(f"B s{s}" for s in seeds) + " | " | |
| + " | ".join(f"Δ s{s} [95 % CI]" for s in seeds) + " | mean Δ ± sd (n, pos) | verdict |", | |
| "|---|---|---|" + "---|" * (3 * len(seeds)) + "---|---|"] | |
| verdicts = {} | |
| for regime in REGIMES: | |
| for cell, metric in CELLS: | |
| g = g_all[(g_all.regime == regime) & (g_all.cell == cell) & (g_all.metric == metric)].sort_values("seed") | |
| if not len(g): | |
| verdicts[(regime, cell)] = "MISSING" | |
| lines.append(f"| {regime} | {cell.replace('|', '/')} | {metric} | " + " | ".join(["—"] * (3 * len(seeds))) + " | — | MISSING |") | |
| continue | |
| lower = cell.startswith("hazard_nll") | |
| pr = pd.DataFrame({"seed": g.seed, "delta": g.delta.astype(float) * (-1 if lower else 1), | |
| "delta_ci_lo": (g.delta_ci_lo if not lower else -g.delta_ci_hi).astype(float), | |
| "delta_ci_hi": (g.delta_ci_hi if not lower else -g.delta_ci_lo).astype(float), "status": "secondary"}) | |
| res = KC.beats(pr, pd.DataFrame({"value": g.value_a.astype(float)}), pd.DataFrame({"value": g.value_b.astype(float)}), min_seeds=3) | |
| by = {int(s): r for s, r in zip(g.seed, g.itertuples())} | |
| A = [f"{by[s].value_a:.4f}" if s in by else "—" for s in seeds] | |
| B = [f"{by[s].value_b:.4f}" if s in by else "—" for s in seeds] | |
| D = [f"{by[s].delta:+.4f} [{by[s].delta_ci_lo:+.4f}, {by[s].delta_ci_hi:+.4f}]" if s in by else "—" for s in seeds] | |
| d = g.delta.astype(float).to_numpy() | |
| sd = d.std(ddof=1) if len(d) > 1 else 0.0 | |
| verdicts[(regime, cell)] = res["verdict"] | |
| lines.append(f"| {regime} | {cell.replace('|', '/')} | {metric}{' (lower better)' if lower else ''} | " + " | ".join(A) + " | " + " | ".join(B) + " | " | |
| + " | ".join(D) + f" | {d.mean():+.4f} ± {sd:.4f} (n={len(d)}, pos={int(g.n_pos.iloc[0])}) | {res['verdict']} |") | |
| lines.append("") | |
| return lines, verdicts | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--out", default=str(CWM3 / "results" / "tables" / "can_arms.md")) | |
| ap.add_argument("--seeds", nargs="+", type=int, default=[0, 1, 2]) | |
| a = ap.parse_args() | |
| seeds = tuple(a.seeds) | |
| stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| tags = [f"canU_s{s}_{arm}_vs_u0base" for arm in ARMS for s in seeds] | |
| df = latest_rows(tags) if CSV.exists() else pd.DataFrame() | |
| L = [f"# CAN arms on the driver chains — OPTIONAL BATON-side ablations (plan Part A §A.5) — {stamp}", "", | |
| f"Pre-registration: `docs/decisions.md` row {PREREG_UTC} (arms, formula, run names, comparison argv, reading rules). NOT part of the AIDE K-v2 programme; " | |
| "no paper change without owner review; every pre-registered cell is reported.", "", | |
| "Evaluator: `eval/paired_compare_joint.py` (unchanged; `--kin-targets causal_routes --kin-version v2`, split test, n_boot 1000 device-cluster bootstrap, identical rows " | |
| "= rows both layouts scored), argv exactly the B1 pairing of `eval/run_joint_evals_final_k0.sh` (also the DMS layer-1 `fit_u_base.pair_argv`). Verdict per cell = " | |
| "`cwm2/eval/kill_checks.beats` over the 3 seeds (≥ 2/3 seed CIs excluding 0, same sign, |mean Δ| > seed sd → BEATS / WORSE; all CIs contain 0 and every |Δ| < 0.01 → " | |
| "NOT_WORSE; otherwise INCONCLUSIVE). Hazard NLL: lower is better, Δ = A − B is shown as computed and its sign is flipped before the verdict. Every number is read from " | |
| f"`results/paired_compare.csv` (tags `canU_s<k>_<arm>_vs_u0base`); n = paired rows, pos = positives (onset cells) / rows with an onset (NLL cells).", "", | |
| "Reading rules (pre-registered): **nocan** WORSE on a cell = the parent's CAN input contributes to that chain; NOT_WORSE = dispensable for that cell. **cangate** BEATS = the " | |
| "heuristic helps that cell; WORSE = it hurts; INCONCLUSIVE = no evidence either way.", ""] | |
| infos = {arm: [run_info(RUNS / ARMS[arm][0].format(k=s)) for s in seeds] for arm in ARMS} | |
| # ---- 0. runs | |
| L += ["## 0. Runs (provenance)", "", | |
| "| run | variant | t_start_utc (> pre-reg?) | wall s (invocation) | train phase s | epochs / best | val objective | params | code sha | gate (a, b) | 2g on test rows (mean, p05 / p50 / p95, n) |", | |
| "|---|---|---|---|---|---|---|---|---|---|---|"] | |
| for arm in ARMS: | |
| for s, inf in zip(seeds, infos[arm]): | |
| if inf is None: | |
| L.append(f"| {ARMS[arm][0].format(k=s)} | {ARMS[arm][1]} | not finished | | | | | | | | |") | |
| continue | |
| gate = f"({inf['gate_a']:+.4f}, {inf['gate_b']:+.4f})" if inf.get("gate_a") is not None else ("—" if not inf.get("u_can_gate") else f"n/a ({inf.get('gate_err', '')})") | |
| g2 = inf.get("gate2_test") | |
| g2s = f"{g2['mean']:.4f}, {g2['p05']:.4f} / {g2['p50']:.4f} / {g2['p95']:.4f}, n={g2['n']:,}" if g2 else "—" | |
| L.append(f"| {inf['run']} | {inf['variant']} | {inf['t_start_utc']} ({'yes' if inf['after_prereg'] else 'NO'}) | {fmt(inf['wall_s'], 1)} | {fmt(inf['train_phase_s'], 0)} | " | |
| f"{inf['epochs_run']} / {inf['best_epoch']} | {fmt(inf['val_objective'], 5)} | {inf['params']:,} | {inf['code_sha256']} | {gate} | {g2s} |") | |
| base_infos = [run_info(RUNS / BASE.format(k=s)) for s in seeds] | |
| for s, inf in zip(seeds, base_infos): | |
| if inf: | |
| L.append(f"| {inf['run']} (B) | {inf['variant']} | {inf['t_start_utc']} (frozen base) | {fmt(inf['wall_s'], 1)} | {fmt(inf['train_phase_s'], 0)} | {inf['epochs_run']} / {inf['best_epoch']} | " | |
| f"{fmt(inf['val_objective'], 5)} | {inf['params']:,} | {inf['code_sha256']} | — | — |") | |
| L.append("") | |
| # ---- 1/2. arms | |
| all_verdicts = {} | |
| for i, arm in enumerate(ARMS, 1): | |
| lines, v = render_arm(arm, seeds, df) | |
| lines[0] = f"## {i}. " + lines[0][3:] | |
| L += lines | |
| all_verdicts[arm] = v | |
| # ---- 3. summary | |
| L += ["## 3. Verdict summary (12 pre-registered cells per arm)", "", "| arm | BEATS | WORSE | NOT_WORSE | INCONCLUSIVE | other | cells WORSE | cells BEATS |", "|---|---|---|---|---|---|---|---|"] | |
| for arm, v in all_verdicts.items(): | |
| cnt = {k: sum(1 for x in v.values() if x == k) for k in ("BEATS", "WORSE", "NOT_WORSE", "INCONCLUSIVE")} | |
| other = len(v) - sum(cnt.values()) | |
| worse = ", ".join(f"{r}/{c.replace('|', '/')}" for (r, c), x in v.items() if x == "WORSE") or "—" | |
| beats = ", ".join(f"{r}/{c.replace('|', '/')}" for (r, c), x in v.items() if x == "BEATS") or "—" | |
| L.append(f"| {arm} | {cnt['BEATS']} | {cnt['WORSE']} | {cnt['NOT_WORSE']} | {cnt['INCONCLUSIVE']} | {other} | {worse} | {beats} |") | |
| L.append("") | |
| # ---- 4. GPU minutes | |
| tot = sum((inf or {}).get("wall_s") or 0 for arm in ARMS for inf in infos[arm]) | |
| tr = sum((inf or {}).get("train_phase_s") or 0 for arm in ARMS for inf in infos[arm]) | |
| L += ["## 4. GPU minutes", "", | |
| f"Six U0 runs: {tot / 60:.1f} min of invocation wall time in total ({tr / 60:.1f} min in the training phases; each run's parent loads ~9.5 GiB of host arrays before the first " | |
| "batch). The GPU was shared with the DMS extraction job (~9.4 GB) and other trainers throughout (launch gate: free VRAM ≥ 10 GB, MemAvailable ≥ 28 GB, trainers < 4; " | |
| "≤ 2 of these runs concurrently) — wall times are upper bounds on the GPU cost. Pairings are CPU.", ""] | |
| # ---- 5. provenance | |
| L += ["## 5. Provenance", "", | |
| f"- Table written by `eval/report_can_arms.py` at {stamp} from `results/paired_compare.csv` ({len(df)} latest rows across the {len(tags)} tags) and the run dirs above.", | |
| f"- Training argv (each run's run json `command`): `{(infos['nocan'][0] or infos['cangate'][0] or {}).get('command', 'n/a')}` (seed / variant / --out per run).", | |
| "- Pairing argv: `eval/paired_compare_joint.py --regime <manual|engaged> --a results/runs/<run> --b results/runs/U0_kin_can_noK_s<k>_kcv2_clip8_ln --tag canU_s<k>_<arm>_vs_u0base " | |
| "--kin-targets causal_routes --kin-version v2` (split test, n_boot 1000).", | |
| "- Code: `train/stages.yaml` (variants kin_can_noK_clip8_ln_{nocan,cangate}), `train/batch.py` (prep_U u_nocan; loss_U 'HG'), `models/branches.py` (CanReleaseGate, DriverBranch can_gate), " | |
| "`models/heads.py` (gate_hazards / hazards_gated / chain_bce_h / driver_loss_gated), `train/train_cwm3.py` (model_kwargs can_gate; predict_split gated hazards + gate2), " | |
| "`tests/test_can_gate.py`, `train/run_can_arms.sh`. Existing runs, tables and the behaviour of every existing variant are untouched (additive, default-off flags).", | |
| "- Launcher pids: `results/can_arms/pids.txt`; per-run train logs and pairing logs under `results/can_arms/`.", ""] | |
| Path(a.out).write_text("\n".join(L) + "\n") | |
| print("\n".join(L)) | |
| print(f"\nwritten {a.out}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 14.6 kB
- Xet hash:
- 2b96198d35fc785ed44f5324eccd158c071e95672c86352733c96042c28a2c41
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.