Buckets:
| """cwm3/probes/probe_layers.py — P1 compression-stage probes + P3 conditional-gain probes -> results/tables/latent_probes.md. | |
| Same rows (results/probes/rows_*), same split protocol (fit on TRAIN, select hyper-parameters on VAL = a disjoint device set, report on | |
| TEST with a device-cluster bootstrap CI), matched probe capacity for every representation: | |
| ridge : closed-form L2 linear read-out (labels centred), lambda = alpha * trace(G)/d, alpha selected on val AP per target | |
| mlp64 : one multi-task 2-layer MLP (d -> 64 -> all targets, masked BCE, positive re-weighting), per-target best-val epoch | |
| Representations (frozen unless stated): raw V-JEPA2 clip-mean 1408 (window ending at floor(2t)/2), region 16x1408 (mean; flattened PCA-256 | |
| fitted on the train region rows), pooled 4224 [mean,last,trend], PCA-256 (vf256), road token bank (16x4x64 flattened / slot-mean / last slot), | |
| bank pooled256 (last / 16 slots), cabin Vc (last 1408 / 4224 / vc256), crop256 (route-median ROI, NON-causal, diagnostic), the TEMPORAL trunk | |
| latents s_t of the seed-0 base runs (probes/dump_latents.py), the E0 predicted future F_hat, and hand-crafted kin+CAN summaries (P3 base). | |
| Targets (probes/common.build_targets): current state at t (hands_on, head_turned, P0 x 6, vEgo bins, gas, brake) and onsets within 3 s | |
| (hand_leave / hand_return / head_turn, P x 6, D x 6, brake_on / gas_on); identity = which dongle (within-split, route-grouped holdout). | |
| Metrics: PR-AUC (AP) with device-cluster bootstrap 95 % CI, AUROC. No openpilot DMS field is read. | |
| python probes/probe_layers.py [--regimes manual,engaged] [--reps ...] [--n-boot 300] [--report-only] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import json | |
| import os | |
| import sys | |
| import time | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| import numpy as np # noqa: E402 | |
| sys.path.insert(0, str(Path(__file__).resolve().parent)) | |
| import common as C # noqa: E402 | |
| LOG = C.OUT / "probe_layers.log" | |
| CSV = C.OUT / "probes_latest.csv" | |
| COLS = ["stamp", "regime", "rows", "rep", "d", "probe", "target", "kind", "family", "hp", "val_ap", "test_ap", "ci_lo", "ci_hi", "auroc", | |
| "n_te", "npos_te", "base_rate_te", "n_tr", "npos_tr"] | |
| IDCOLS = ["stamp", "regime", "split", "rep", "d", "probe", "n_train", "n_test", "n_dongles", "top1", "top1_chance_majority", "top5"] | |
| PAIRCOLS = ["stamp", "regime", "rows", "probe", "target", "kind", "family", "rep_a", "rep_b", "ap_a", "ap_b", "delta", "d_lo", "d_hi", "frac_gt0", "n_te", "npos_te"] | |
| FROZEN_ROAD = ["raw1408", "Vf4224", "vf256", "bank_pooled_last", "bank_pooled16_flat", "bank_tok_last", "bank_tok16_mean", "bank_tok16_flat"] | |
| FROZEN_CABIN = ["Vc_last1408", "Vc4224", "vc256", "crop256"] | |
| TEMPORAL = ["s_V4", "s_V8", "s_U", "s_E", "s_E_full", "F_hat"] | |
| HANDCRAFT = ["kincan"] | |
| COMBOS = ["kincan+vf256", "kincan+s_E_full", "kincan+bank_tok16_mean", "kincan+vc256", "kincan+crop256", "kincan+s_U", "s_U+vf256", "s_U+vc256"] | |
| REGION = ["region_mean", "region_pca256"] | |
| ALL_REPS = FROZEN_ROAD + FROZEN_CABIN + TEMPORAL + HANDCRAFT + COMBOS | |
| def log(m): | |
| C.log(m, LOG) | |
| # ----------------------------------------------------------------------------- features | |
| def kincan_features(raw, st): | |
| """hand-crafted kp-derived kinematics + CAN summary (~80 dims), standardised with TRAIN statistics `st` (kmu/ksd/xmu/xsd).""" | |
| Kh = raw["Kh"].astype(np.float32) # [n,40,11] NaN = invalid | |
| valid = np.isfinite(Kh[:, :, 10]) & np.isfinite(Kh[:, :, 0]) | |
| Z = np.clip((Kh - st["kmu"]) / st["ksd"], -5, 5) | |
| Zn = np.where(valid[:, :, None], Z, np.nan) | |
| with np.errstate(all="ignore"): | |
| last = np.nan_to_num(Zn[:, -1]) | |
| m1 = np.nan_to_num(np.nanmean(Zn[:, -10:], 1)) | |
| m4 = np.nan_to_num(np.nanmean(Zn, 1)) | |
| s4 = np.nan_to_num(np.nanstd(Zn, 1)) | |
| d1 = np.nan_to_num(Zn[:, -1] - np.nanmean(Zn[:, -11:-9], 1)) | |
| vfrac = valid.mean(1, keepdims=True) | |
| X = raw["X"].astype(np.float32) | |
| Xs = np.clip((X - st["xmu"]) / st["xsd"], -8, 8) | |
| xl, xm, xs, xd = Xs[:, -1], Xs.mean(1), Xs.std(1), Xs[:, -1] - Xs[:, -11] | |
| S0 = np.nan_to_num(raw["S0"].astype(np.float32), nan=0.5) | |
| z0 = np.stack([S0[:, 0], S0[:, 1], (X[:, -1, 1] > 0.5), (X[:, -1, 2] > 0.5), np.isfinite(raw["S0"][:, 0])], 1).astype(np.float32) | |
| return np.concatenate([last, m1, m4, s4, d1, vfrac, xl, xm, xs, xd, z0], 1) | |
| class Reps: | |
| """Lazy builders of the representation matrices for one (regime, split).""" | |
| def __init__(self, regime, split, st): | |
| self.raw = C.load_raw(regime, split) | |
| self.fe = np.load(C.feats_path(regime, split), allow_pickle=True) # NpzFile (stored members): read on access -> bounded RSS | |
| self.lat = dict(np.load(C.OUT / f"latents_{regime}_{split}.npz")) | |
| self.st = st | |
| self.n = len(self.raw["t"]) | |
| self._cache = {} | |
| def common_ok(self): | |
| fe, raw, lat = self.fe, self.raw, self.lat | |
| ok = np.isfinite(fe["raw1408"][:, 0]) & np.isfinite(fe["Vf4224"][:, 0]) & np.isfinite(fe["Vc4224"][:, 0]) & np.isfinite(fe["crop4224"][:, 0]) | |
| ok &= raw["vf256_ok"].astype(bool) & raw["vc256_ok"].astype(bool) & raw["crop256_ok"].astype(bool) & fe["tok_ok"][:, -1] | |
| return ok # s_V8 (8 s CAN, x8_ok: 43 % of manual test rows) is probed on its own rows, tagged x8_rows | |
| def region_ok(self): | |
| ok = np.zeros(self.n, bool) | |
| ok[np.flatnonzero(self.fe["region_rows"])[self.fe["region_filled"]]] = True | |
| return ok | |
| def get(self, name): | |
| if name in self._cache: | |
| return self._cache[name] | |
| fe, raw, lat = self.fe, self.raw, self.lat | |
| n = self.n | |
| if "+" in name: | |
| X = np.concatenate([self.get(p) for p in name.split("+")], 1) | |
| elif name == "raw1408": | |
| X = fe["raw1408"].astype(np.float32) | |
| elif name == "Vf4224": | |
| X = fe["Vf4224"].astype(np.float32) | |
| elif name == "vf256": | |
| X = raw["vf256"].astype(np.float32) | |
| elif name == "bank_pooled_last": | |
| X = fe["pooled16"][:, -1].astype(np.float32) | |
| elif name == "bank_pooled16_flat": | |
| X = fe["pooled16"].reshape(n, -1).astype(np.float32) | |
| elif name == "bank_tok_last": | |
| X = fe["tok16"][:, -1].reshape(n, -1).astype(np.float32) | |
| elif name == "bank_tok16_mean": | |
| ok = fe["tok_ok"].astype(np.float32) | |
| X = ((fe["tok16"].astype(np.float32) * ok[:, :, None, None]).sum(1) / np.maximum(ok.sum(1), 1)[:, None, None]).reshape(n, -1) | |
| elif name == "bank_tok16_flat": | |
| X = fe["tok16"].reshape(n, -1).astype(np.float32) | |
| elif name == "Vc_last1408": | |
| X = fe["Vc4224"][:, 1408:2816].astype(np.float32) | |
| elif name == "Vc4224": | |
| X = fe["Vc4224"].astype(np.float32) | |
| elif name == "vc256": | |
| X = raw["vc256"].astype(np.float32) | |
| elif name == "crop256": | |
| X = raw["crop256"].astype(np.float32) | |
| elif name in ("s_V4", "s_V8", "s_U", "s_E"): | |
| X = lat[name].astype(np.float32) | |
| elif name == "s_E_full": | |
| X = np.concatenate([lat["s_E"], lat["s_Evf"]], 1).astype(np.float32) | |
| elif name == "F_hat": | |
| X = lat["F_hat"].reshape(n, -1).astype(np.float32) | |
| elif name == "kincan": | |
| X = kincan_features(raw, self.st) | |
| elif name == "region_mean": | |
| X = np.full((n, 1408), np.nan, np.float32) | |
| reg = np.load(str(fe["region_file"]), mmap_mode="r") | |
| rr = np.flatnonzero(fe["region_rows"]) | |
| for i in range(0, len(rr), 2000): | |
| blk = np.asarray(reg[i:i + 2000]).astype(np.float32) | |
| X[rr[i:i + 2000]] = blk.mean(1) | |
| X[~self.region_ok()] = np.nan | |
| elif name == "region_flat": | |
| reg = np.load(str(fe["region_file"]), mmap_mode="r") | |
| rr = np.flatnonzero(fe["region_rows"]) | |
| return reg.reshape(len(rr), -1), rr # f16 memmap view [n_reg, 22528]; rows = rr (read in chunks by the callers) | |
| else: | |
| raise KeyError(name) | |
| if X.shape[1] <= 128: # only tiny matrices are cached (memory discipline on the shared host) | |
| self._cache[name] = X | |
| return X | |
| def region_pca(reps_tr, reps_va, reps_te, k=256, device=None): | |
| """flattened 16x1408 -> PCA-k fitted on the TRAIN region rows (torch low-rank SVD) -> matrices [n, k] (NaN outside the region subset).""" | |
| import torch | |
| dev = device or C.torch_device() | |
| Xtr, rr_tr = reps_tr.get("region_flat") | |
| ok_tr = reps_tr.fe["region_filled"] | |
| Xt = torch.cat([torch.as_tensor(np.asarray(Xtr[i:i + 2000]), device=dev).float() for i in range(0, len(Xtr), 2000)])[torch.as_tensor(ok_tr, device=dev)] | |
| mu = Xt.mean(0) | |
| Xt -= mu | |
| torch.manual_seed(0) | |
| U, S, V = torch.svd_lowrank(Xt, q=k + 32, niter=4) | |
| W = V[:, :k].contiguous() | |
| tot = float((Xt ** 2).sum() / len(Xt)) | |
| evr = float((S[:k] ** 2).sum() / len(Xt) / tot) | |
| out = [] | |
| for reps in (reps_tr, reps_va, reps_te): | |
| X, rr = reps.get("region_flat") | |
| Z = np.full((reps.n, k), np.nan, np.float32) | |
| for i in range(0, len(rr), 2000): | |
| blk = torch.as_tensor(np.asarray(X[i:i + 2000]), device=dev).float() - mu | |
| Z[rr[i:i + 2000]] = (blk @ W).cpu().numpy() | |
| Z[~reps.region_ok()] = np.nan | |
| out.append(Z) | |
| del Xt | |
| return out, {"k": k, "evr_k": evr, "n_fit": int(ok_tr.sum())} | |
| # ----------------------------------------------------------------------------- one representation | |
| def score_path(stamp, regime, tag, name): | |
| d = C.OUT / "scores" / stamp | |
| d.mkdir(parents=True, exist_ok=True) | |
| return d / f"{regime}_{tag}_{name}.npz" | |
| def run_rep(name, regime, R, T, Yc, Mc, rows_ok, dongle_te, n_boot, stamp, probes, writer, pair_store, hp_tag=None, done_cells=()): | |
| import gc | |
| tag = "common" if hp_tag is None else hp_tag | |
| sp = score_path(stamp, regime, tag, name) | |
| if (regime, tag, name) in done_cells and sp.exists(): | |
| z = np.load(sp, allow_pickle=True) | |
| for key, arr in zip(z["keys"], z["scores"]): | |
| pr, tn = str(key).split("|") | |
| pair_store[(name, pr, tn)] = (arr, z["mask"][:, list(T["train"].keys()).index(tn)]) | |
| log(f" {regime} {name} ({tag}): cells exist for stamp {stamp} — skipped (scores reloaded)") | |
| return None | |
| Xtr, Xva, Xte = (R[s].get(name) for s in ("train", "val", "test")) | |
| ok = {s: rows_ok[s] & np.isfinite(X[:, 0]) for s, X in zip(("train", "val", "test"), (Xtr, Xva, Xte))} | |
| std = C.Std(Xtr[ok["train"]]) | |
| Ztr, Zva, Zte = std(Xtr[ok["train"]]), std(Xva[ok["val"]]), std(Xte[ok["test"]]) | |
| d = Ztr.shape[1] | |
| names = list(T["train"].keys()) | |
| Ytr, Mtr = Yc["train"][ok["train"]], Mc["train"][ok["train"]] | |
| Yva, Mva = Yc["val"][ok["val"]], Mc["val"][ok["val"]] | |
| Yte, Mte = Yc["test"][ok["test"]], Mc["test"][ok["test"]] | |
| dte = dongle_te[ok["test"]] | |
| t0 = time.time() | |
| out_scores = {} | |
| if "ridge" in probes: | |
| sva, ste = C.ridge_fit_predict(Ztr, Ytr, Mtr, Zva, Zte) | |
| for j, tn in enumerate(names): | |
| if not np.isfinite(sva[j, 0, 0]): | |
| continue | |
| aps = [C.ap_weighted(Yva[Mva[:, j], j], sva[j, a][Mva[:, j]]) if Mva[:, j].sum() > 20 and Yva[Mva[:, j], j].sum() > 0 else np.nan for a in range(sva.shape[1])] | |
| a_best = int(np.nanargmax(aps)) if np.any(np.isfinite(aps)) else 0 | |
| out_scores[("ridge", tn)] = (ste[j, a_best], f"alpha={C.RIDGE_ALPHAS[a_best]:g}", aps[a_best]) | |
| if "mlp64" in probes: | |
| te, va, ep, apv = C.mlp_fit_predict(Ztr, Ytr, Mtr, Zva, Yva, Mva, Zte, hidden=64, epochs=25, seed=0) | |
| for j, tn in enumerate(names): | |
| if apv[j] < 0: | |
| continue | |
| out_scores[("mlp64", tn)] = (te[j], f"epoch={ep[j]}", apv[j]) | |
| n_cells = 0 | |
| keys, arrs = [], [] | |
| for (probe, tn), (s_te, hp, apv) in out_scores.items(): | |
| j = names.index(tn) | |
| m = Mte[:, j] | |
| if m.sum() < 100 or Yte[m, j].sum() < 10: | |
| continue | |
| keys.append(f"{probe}|{tn}"); arrs.append(s_te.astype(np.float32)) | |
| r = C.cluster_boot(Yte[m, j], s_te[m], dte[m], n_boot=n_boot, seed=0) | |
| writer.writerow({"stamp": stamp, "regime": regime, "rows": "common" if hp_tag is None else hp_tag, "rep": name, "d": d, "probe": probe, "target": tn, | |
| "kind": T["train"][tn]["kind"], "family": T["train"][tn]["family"], "hp": hp, "val_ap": round(float(apv), 5), "test_ap": round(r["ap"], 5), | |
| "ci_lo": round(r["ci"][0], 5), "ci_hi": round(r["ci"][1], 5), "auroc": round(r["auroc"], 5), "n_te": r["n"], "npos_te": r["n_pos"], | |
| "base_rate_te": round(float(Yte[m, j].mean()), 5), "n_tr": int(Mtr[:, j].sum()), "npos_tr": int(Ytr[Mtr[:, j], j].sum())}) | |
| pair_store[(name, probe, tn)] = (s_te, m) | |
| n_cells += 1 | |
| if arrs: | |
| np.savez(sp, keys=np.array(keys), scores=np.stack(arrs), mask=Mte, ok_test=ok["test"], n_te=np.array(len(Zte))) | |
| log(f" {regime} {name} d={d} n_tr={len(Ztr):,} n_te={len(Zte):,} cells={n_cells} ({time.time() - t0:.0f}s)") | |
| del Xtr, Xva, Xte, Ztr, Zva, Zte | |
| gc.collect() | |
| return ok["test"] | |
| def identity_probe(name, regime, R, split, stamp, writer, probes): | |
| """which dongle: within one split, route-grouped holdout (alternate routes per dongle; single-route dongles split at the time median).""" | |
| import torch | |
| reps = R[split] | |
| X = reps.get(name) | |
| if isinstance(X, tuple): | |
| return | |
| ok = np.isfinite(X[:, 0]) & (reps.common_ok() if not name.startswith("region") else reps.region_ok()) | |
| uid, dong, t = reps.raw["uid"][ok], reps.raw["dongle"][ok], reps.raw["t"][ok] | |
| X = X[ok] | |
| hold = np.zeros(len(X), bool) | |
| for dn in np.unique(dong): | |
| rows = np.flatnonzero(dong == dn) | |
| routes = np.unique(uid[rows]) | |
| if len(routes) >= 2: | |
| hb = set(routes[1::2].tolist()) | |
| hold[rows] = np.isin(uid[rows], list(hb)) | |
| else: | |
| hold[rows] = t[rows] > np.median(t[rows]) | |
| codes, y = np.unique(dong, return_inverse=True) | |
| K = len(codes) | |
| tr, te = ~hold, hold | |
| # only dongles present on both sides | |
| keep = np.isin(y, np.intersect1d(np.unique(y[tr]), np.unique(y[te]))) | |
| tr, te = tr & keep, te & keep | |
| std = C.Std(X[tr]) | |
| Ztr, Zte = std(X[tr]), std(X[te]) | |
| Yoh = np.eye(K, dtype=np.float32)[y[tr]] | |
| maj = float(np.bincount(y[te], minlength=K).max() / te.sum()) | |
| if "ridge" in probes: | |
| sva, ste = C.ridge_fit_predict(Ztr, Yoh, np.ones_like(Yoh, bool), Zte[:1], Zte, alphas=(1e-1,)) | |
| S = ste[:, 0, :].T # [n_te, K] | |
| top1 = float((S.argmax(1) == y[te]).mean()) | |
| top5 = float(np.mean([y[te][i] in np.argsort(-S[i])[:5] for i in range(len(S))])) | |
| writer.writerow({"stamp": stamp, "regime": regime, "split": split, "rep": name, "d": Ztr.shape[1], "probe": "ridge_ovr", "n_train": int(tr.sum()), "n_test": int(te.sum()), | |
| "n_dongles": int(K), "top1": round(top1, 4), "top1_chance_majority": round(maj, 4), "top5": round(top5, 4)}) | |
| if "mlp64" in probes: | |
| dev = C.torch_device() | |
| torch.manual_seed(0) | |
| net = torch.nn.Sequential(torch.nn.Linear(Ztr.shape[1], 64), torch.nn.GELU(), torch.nn.Linear(64, K)).to(dev) | |
| opt = torch.optim.AdamW(net.parameters(), lr=1e-3, weight_decay=1e-4) | |
| Xt, yt = torch.as_tensor(Ztr, device=dev), torch.as_tensor(y[tr], device=dev) | |
| for ep in range(20): | |
| perm = torch.randperm(len(Xt), device=dev) | |
| for i in range(0, len(Xt), 1024): | |
| idx = perm[i:i + 1024] | |
| loss = torch.nn.functional.cross_entropy(net(Xt[idx]), yt[idx]) | |
| opt.zero_grad(); loss.backward(); opt.step() | |
| with torch.no_grad(): | |
| S = torch.cat([net(torch.as_tensor(Zte[i:i + 8192], device=dev)) for i in range(0, len(Zte), 8192)]).cpu().numpy() | |
| top1 = float((S.argmax(1) == y[te]).mean()) | |
| top5 = float(np.mean([y[te][i] in np.argsort(-S[i])[:5] for i in range(len(S))])) | |
| writer.writerow({"stamp": stamp, "regime": regime, "split": split, "rep": name, "d": Ztr.shape[1], "probe": "mlp64", "n_train": int(tr.sum()), "n_test": int(te.sum()), | |
| "n_dongles": int(K), "top1": round(top1, 4), "top1_chance_majority": round(maj, 4), "top5": round(top5, 4)}) | |
| log(f" identity {regime}/{split} {name}: top1 written") | |
| # ----------------------------------------------------------------------------- main | |
| def done_cells_for(stamp): | |
| if not CSV.exists(): | |
| return set() | |
| with open(CSV) as fh: | |
| return {(r["regime"], r["rows"], r["rep"]) for r in csv.DictReader(fh) if r["stamp"] == stamp} | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--reps", default="all") | |
| ap.add_argument("--probes", default="ridge,mlp64") | |
| ap.add_argument("--n-boot", type=int, default=300) | |
| ap.add_argument("--stage", default="reps", choices=["reps", "identity", "mode", "report"], help="reps = representations + region + P3 pairs per regime") | |
| ap.add_argument("--stamp", default="", help="reuse a run stamp: cells already in the CSV for that stamp are skipped, their saved scores reused") | |
| ap.add_argument("--no-region", action="store_true") | |
| ap.add_argument("--report-only", action="store_true") | |
| a = ap.parse_args() | |
| if a.report_only or a.stage == "report": | |
| write_report() | |
| return | |
| import torch | |
| stamp = a.stamp or C.stamp() | |
| t_all = time.time() | |
| probes = a.probes.split(",") | |
| reps_wanted = ALL_REPS if a.reps == "all" else a.reps.split(",") | |
| st = {k: np.asarray(v, np.float32) for k, v in torch.load(C.RUNS / "U0_kin_can_noK_s0" / next(iter((C.RUNS / "U0_kin_can_noK_s0" / "BEST").read_text().split())), | |
| map_location="cpu", weights_only=False)["stats"].items()} | |
| log(f"start stamp={stamp} stage={a.stage} device={C.torch_device()} regimes={a.regimes} probes={probes}") | |
| new = not CSV.exists() | |
| f = open(CSV, "a", newline="") | |
| w = csv.DictWriter(f, fieldnames=COLS) | |
| if new: | |
| w.writeheader() | |
| fid = open(C.OUT / "identity_latest.csv", "a", newline="") | |
| wid = csv.DictWriter(fid, fieldnames=IDCOLS) | |
| if fid.tell() == 0: | |
| wid.writeheader() | |
| fpair = open(C.OUT / "pairs_latest.csv", "a", newline="") | |
| wpair = csv.DictWriter(fpair, fieldnames=PAIRCOLS) | |
| if fpair.tell() == 0: | |
| wpair.writeheader() | |
| done_cells = done_cells_for(stamp) | |
| if a.stage == "reps": | |
| for regime in a.regimes.split(","): | |
| R = {s: Reps(regime, s, st) for s in C.SPLITS} | |
| T = {s: C.build_targets(R[s].raw) for s in C.SPLITS} | |
| names = list(T["train"].keys()) | |
| Yc = {s: np.stack([T[s][k]["y"] for k in names], 1).astype(np.float32) for s in C.SPLITS} | |
| Mc = {s: np.stack([T[s][k]["m"] for k in names], 1) for s in C.SPLITS} | |
| rows_ok = {s: R[s].common_ok() for s in C.SPLITS} | |
| log(f"{regime}: rows common ok train/val/test = {rows_ok['train'].mean():.3f}/{rows_ok['val'].mean():.3f}/{rows_ok['test'].mean():.3f}; targets={len(names)}") | |
| C.write_json(C.OUT / f"targets_{regime}_test.json", C.target_table(T["test"], R["test"].raw["dongle"])) | |
| pair_store = {} | |
| dongle_te = R["test"].raw["dongle"] | |
| for name in reps_wanted: | |
| if name == "s_V8": | |
| rows_x8 = {s: rows_ok[s] & np.isfinite(R[s].lat["s_V8"][:, 0]) for s in C.SPLITS} | |
| run_rep("s_V8", regime, R, T, Yc, Mc, rows_x8, dongle_te, a.n_boot, stamp, probes, w, {}, hp_tag="x8_rows", done_cells=done_cells) | |
| run_rep("s_V4", regime, R, T, Yc, Mc, rows_x8, dongle_te, a.n_boot, stamp, probes, w, {}, hp_tag="x8_rows", done_cells=done_cells) | |
| f.flush() | |
| continue | |
| run_rep(name, regime, R, T, Yc, Mc, rows_ok, dongle_te, a.n_boot, stamp, probes, w, pair_store, done_cells=done_cells) | |
| f.flush() | |
| if not a.no_region: | |
| need = [n for n in ["region_mean", "region_pca256", "raw1408", "vf256", "bank_tok_last", "kincan"] if (regime, "region_subset", n) not in done_cells] | |
| if need: | |
| (Ztr, Zva, Zte), info = region_pca(R["train"], R["val"], R["test"]) | |
| for s_, Z in zip(C.SPLITS, (Ztr, Zva, Zte)): | |
| R[s_]._cache["region_pca256"] = Z | |
| log(f"{regime}: region PCA-256 fitted on {info['n_fit']} train rows, EVR {info['evr_k']:.4f}") | |
| C.write_json(C.OUT / f"region_pca_{regime}.json", info) | |
| rows_reg = {s_: rows_ok[s_] & R[s_].region_ok() for s_ in C.SPLITS} | |
| for name in ["region_mean", "region_pca256", "raw1408", "vf256", "bank_tok_last", "kincan"]: | |
| run_rep(name, regime, R, T, Yc, Mc, rows_reg, dongle_te, a.n_boot, stamp, probes, w, {}, hp_tag="region_subset", done_cells=done_cells) | |
| f.flush() | |
| for s_ in C.SPLITS: | |
| R[s_]._cache.pop("region_pca256", None) | |
| # P3 paired deltas vs the kin+CAN base on identical rows (scores from this run or reloaded from disk) | |
| for probe in probes: | |
| for tn in names: | |
| if T["train"][tn]["kind"] != "fut": | |
| continue | |
| base = pair_store.get(("kincan", probe, tn)) | |
| if base is None: | |
| continue | |
| sb, mb = base | |
| for name in COMBOS + ["s_U", "vf256", "vc256", "s_E_full"]: | |
| got = pair_store.get((name, probe, tn)) | |
| if got is None: | |
| continue | |
| sa, ma = got | |
| m = ma & mb | |
| y = Yc["test"][rows_ok["test"]][m, names.index(tn)].astype(bool) | |
| if y.sum() < 10: | |
| continue | |
| r = C.cluster_boot(y, sa[m], dongle_te[rows_ok["test"]][m], n_boot=a.n_boot, seed=0, s_b=sb[m]) | |
| wpair.writerow({"stamp": stamp, "regime": regime, "rows": "common", "probe": probe, "target": tn, "kind": "fut", "family": T["train"][tn]["family"], | |
| "rep_a": name, "rep_b": "kincan", "ap_a": round(r["ap"], 5), "ap_b": round(r["ap_b"], 5), "delta": round(r["delta"], 5), | |
| "d_lo": round(r["delta_ci"][0], 5), "d_hi": round(r["delta_ci"][1], 5), "frac_gt0": round(r["frac_gt0"], 3), "n_te": r["n"], "npos_te": r["n_pos"]}) | |
| fpair.flush() | |
| log(f"{regime}: P3 pairs written") | |
| del R, pair_store, Yc, Mc | |
| if a.stage == "identity": | |
| for regime in a.regimes.split(","): | |
| R = {s: Reps(regime, s, st) for s in ("train", "test")} | |
| for split in ("train", "test"): | |
| for name in ["raw1408", "vf256", "bank_tok16_mean", "Vc_last1408", "vc256", "crop256", "s_V4", "s_U", "s_E_full", "kincan"]: | |
| identity_probe(name, regime, R, split, stamp, wid, probes) | |
| fid.flush() | |
| del R | |
| if a.stage == "mode": | |
| Rb = {s: [Reps(r, s, st) for r in C.REGIMES] for s in C.SPLITS} | |
| for name in ["raw1408", "vf256", "bank_tok16_mean", "vc256", "s_E_full", "s_U", "kincan"]: | |
| Xs = {s: np.concatenate([Rb[s][i].get(name) for i in range(2)]) for s in C.SPLITS} | |
| y = {s: np.concatenate([np.full(Rb[s][i].n, i, np.float32) for i in range(2)]) for s in C.SPLITS} | |
| ok = {s: np.concatenate([Rb[s][i].common_ok() for i in range(2)]) & np.isfinite(Xs[s][:, 0]) for s in C.SPLITS} | |
| dte = np.concatenate([Rb["test"][i].raw["dongle"] for i in range(2)]) | |
| std = C.Std(Xs["train"][ok["train"]]) | |
| Z = {s: std(Xs[s][ok[s]]) for s in C.SPLITS} | |
| del Xs | |
| for probe in probes: | |
| if probe == "ridge": | |
| sva, ste = C.ridge_fit_predict(Z["train"], y["train"][ok["train"], None], np.ones((ok["train"].sum(), 1), bool), Z["val"], Z["test"]) | |
| aps = [C.ap_weighted(y["val"][ok["val"]] > 0.5, sva[0, a_]) for a_ in range(sva.shape[1])] | |
| a_ = int(np.nanargmax(aps)); s_te, hp, apv = ste[0, a_], f"alpha={C.RIDGE_ALPHAS[a_]:g}", aps[a_] | |
| else: | |
| te, va, ep, apv = C.mlp_fit_predict(Z["train"], y["train"][ok["train"], None], np.ones((ok["train"].sum(), 1), bool), Z["val"], y["val"][ok["val"], None], | |
| np.ones((ok["val"].sum(), 1), bool), Z["test"], epochs=15) | |
| s_te, hp, apv = te[0], f"epoch={ep[0]}", apv[0] | |
| yt = y["test"][ok["test"]] > 0.5 | |
| r = C.cluster_boot(yt, s_te, dte[ok["test"]], n_boot=a.n_boot) | |
| w.writerow({"stamp": stamp, "regime": "both", "rows": "common", "rep": name, "d": Z["train"].shape[1], "probe": probe, "target": "cur_mode", "kind": "cur", "family": "regime", | |
| "hp": hp, "val_ap": round(float(apv), 5), "test_ap": round(r["ap"], 5), "ci_lo": round(r["ci"][0], 5), "ci_hi": round(r["ci"][1], 5), "auroc": round(r["auroc"], 5), | |
| "n_te": r["n"], "npos_te": r["n_pos"], "base_rate_te": round(float(yt.mean()), 5), "n_tr": int(ok["train"].sum()), "npos_tr": int(y["train"][ok["train"]].sum())}) | |
| f.flush() | |
| log(f" both cur_mode {name} done") | |
| f.close(); fid.close(); fpair.close() | |
| log(f"stage {a.stage} done in {time.time() - t_all:.0f}s") | |
| # ----------------------------------------------------------------------------- report | |
| def _load(path): | |
| with open(path) as f: | |
| return list(csv.DictReader(f)) | |
| def write_report(): | |
| rows = _load(CSV) | |
| last = max(r["stamp"] for r in rows) | |
| rows = [r for r in rows if r["stamp"] == last] or rows | |
| ids = [r for r in _load(C.OUT / "identity_latest.csv")] if (C.OUT / "identity_latest.csv").exists() else [] | |
| pairs = [r for r in _load(C.OUT / "pairs_latest.csv")] if (C.OUT / "pairs_latest.csv").exists() else [] | |
| if ids: | |
| li = max(r["stamp"] for r in ids); ids = [r for r in ids if r["stamp"] == li] | |
| if pairs: | |
| lp = max(r["stamp"] for r in pairs); pairs = [r for r in pairs if r["stamp"] == lp] | |
| cell = {(r["regime"], r["rows"], r["rep"], r["probe"], r["target"]): r for r in rows} | |
| reps_order = [r for r in ALL_REPS + REGION if any(k[2] == r for k in cell)] | |
| targets = [] | |
| for r in rows: | |
| if r["target"] not in targets: | |
| targets.append(r["target"]) | |
| L = ["# Latent probes (P1 / P3) — what each representation retains, on identical rows", "", | |
| f"Generated {C.stamp()} by `probes/probe_layers.py` (run stamp {last}); raw cells `results/probes/probes_latest.csv`, paired deltas `pairs_latest.csv`, " | |
| "identity `identity_latest.csv`. Protocol: fit on the TRAIN probe rows (50k per regime), hyper-parameters selected on VAL (15k, disjoint devices), " | |
| "reported on TEST (40k, disjoint devices) as PR-AUC with a device-cluster bootstrap 95 % CI (n_boot 300). `ridge` = closed-form L2 linear read-out " | |
| "(alpha in {1e-3..10} x trace/d, per target on val); `mlp64` = one multi-task 2-layer MLP (64 hidden, masked BCE, per-target best-val epoch). " | |
| "Rows = the common rows where every non-region representation is available; `region_subset` rows additionally carry the raw 16x1408 tensor.", ""] | |
| L += ["Representations: `raw1408` frontfeat clip-mean token of the window ending at floor(2t)/2; `region_mean` mean of the 16 region tokens (= raw1408 up to fp16); " | |
| "`region_pca256` flattened 16x1408 -> PCA-256 (train fit); `Vf4224` s5 pooled [mean,last,trend]; `vf256` its PCA-256 (the E-branch input); `bank_pooled_last/16_flat` " | |
| "PCA-256 of the clip-mean token at the last slot / all 16 slots; `bank_tok_last/16_mean/16_flat` the 4x64 region-token bank (last slot / slot-mean / 16 slots); " | |
| "`Vc_last1408`, `Vc4224`, `vc256` cabin full-frame (causal); `crop256` cabin route-median ROI (NON-causal, diagnostic); `s_V4/s_V8` TCN trunk of V0 can4s/can8s; " | |
| "`s_U` GRU trunk of U0_kin_can_noK; `s_E` causal-Transformer read-out of E0_pooled, `s_E_full` = [s_E, Linear(vf256)]; `F_hat` the E0 predicted future latent (+1..+5 s); " | |
| "`kincan` hand-crafted kp-kinematics + CAN summary (P3 base).", ""] | |
| for regime in ("manual", "engaged"): | |
| for rows_tag in ("common", "region_subset"): | |
| sub = [k for k in cell if k[0] == regime and k[1] == rows_tag] | |
| if not sub: | |
| continue | |
| reps_here = [r for r in reps_order if any(k[2] == r for k in sub)] | |
| for probe in ("ridge", "mlp64"): | |
| if not any(k[3] == probe for k in sub): | |
| continue | |
| L += [f"## {regime} — {probe} — rows: {rows_tag}", ""] | |
| hdr = ["target", "kind", "n / pos (test)"] + reps_here | |
| body = [] | |
| for tn in targets: | |
| any_ = [cell.get((regime, rows_tag, rp, probe, tn)) for rp in reps_here] | |
| if not any(any_): | |
| continue | |
| ref = next(x for x in any_ if x) | |
| best = max([float(x["test_ap"]) for x in any_ if x]) | |
| row = [tn, ref["kind"], f"{ref['n_te']} / {ref['npos_te']} (base {float(ref['base_rate_te']):.3f})"] | |
| for x in any_: | |
| if x is None: | |
| row.append("–") | |
| else: | |
| v = float(x["test_ap"]) | |
| s = f"{v:.3f} [{float(x['ci_lo']):.3f}, {float(x['ci_hi']):.3f}]" | |
| row.append(f"**{s}**" if v == best else s) | |
| body.append(row) | |
| L += [C.md_table(hdr, body), ""] | |
| both = [r for r in rows if r["regime"] == "both"] | |
| if both: | |
| L += ["## Regime (mode) probe — regimes pooled, train -> test", "", C.md_table(["rep", "probe", "test AP [CI]", "AUROC", "n / pos"], | |
| [[r["rep"], r["probe"], f"{float(r['test_ap']):.3f} [{float(r['ci_lo']):.3f}, {float(r['ci_hi']):.3f}]", f"{float(r['auroc']):.3f}", f"{r['n_te']} / {r['npos_te']}"] for r in both]), ""] | |
| if ids: | |
| L += ["## Identity leakage — which dongle (within-split, route-grouped holdout; top-1 accuracy vs the majority-class rate)", "", | |
| C.md_table(["regime", "split", "rep", "d", "probe", "n train / test", "dongles", "top-1", "majority", "top-5"], | |
| [[r["regime"], r["split"], r["rep"], r["d"], r["probe"], f"{r['n_train']} / {r['n_test']}", r["n_dongles"], f"{float(r['top1']):.3f}", f"{float(r['top1_chance_majority']):.3f}", f"{float(r['top5']):.3f}"] for r in ids]), ""] | |
| if pairs: | |
| L += ["## P3 — conditional gain over the kin+CAN summary on identical test rows (Δ PR-AUC, device-cluster bootstrap 95 % CI; P(Δ>0) = share of resamples)", ""] | |
| for regime in ("manual", "engaged"): | |
| for probe in ("ridge", "mlp64"): | |
| sub = [r for r in pairs if r["regime"] == regime and r["probe"] == probe] | |
| if not sub: | |
| continue | |
| reps_a = [] | |
| for r in sub: | |
| if r["rep_a"] not in reps_a: | |
| reps_a.append(r["rep_a"]) | |
| L += [f"### {regime} — {probe}", ""] | |
| hdr = ["target", "kincan AP", "n / pos"] + [f"Δ {ra}" for ra in reps_a] | |
| body = [] | |
| for tn in targets: | |
| rr = [r for r in sub if r["target"] == tn] | |
| if not rr: | |
| continue | |
| row = [tn, f"{float(rr[0]['ap_b']):.3f}", f"{rr[0]['n_te']} / {rr[0]['npos_te']}"] | |
| for ra in reps_a: | |
| x = next((r for r in rr if r["rep_a"] == ra), None) | |
| if x is None: | |
| row.append("–") | |
| else: | |
| d, lo, hi = float(x["delta"]), float(x["d_lo"]), float(x["d_hi"]) | |
| s = f"{d:+.3f} [{lo:+.3f}, {hi:+.3f}]" | |
| row.append(f"**{s}**" if (lo > 0 or hi < 0) else s) | |
| body.append(row) | |
| L += [C.md_table(hdr, body), ""] | |
| (C.TABLES / "latent_probes.md").write_text("\n".join(L) + "\n") | |
| log(f"report -> {C.TABLES / 'latent_probes.md'}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 33.1 kB
- Xet hash:
- 3bec79412e564c6397f73b49e18c8b09e85ca30eb030a05822042451a7201a3d
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.