Buckets:
| """cwm3/probes/future_vs_predicted.py — P2 encoding-vs-dynamics separation for the E0_pooled seed-0 world-model head. | |
| One frozen read-out per (event, horizon k) is trained on the ACTUAL future front latent F(t+k) (s12 future_*.npz, future_pca 256; the s5 clip | |
| ending at the last s5 timestamp <= t+k, window [t+k-2, t+k]) -> event state at t+k, then evaluated unchanged on | |
| (a) actual F(t+k) offline reference only (not a model result: it needs the future) | |
| (b) predicted F_hat(t+k) the E0 F head (probes/dump_latents.py), trained with 1 - cos on the same F | |
| (c) copy-current F_cur(t) the current front latent (s5 clip ending at the last ts <= t) projected with the SAME future_pca | |
| All three are L2-normalised before the read-out (the F head is trained with a cosine loss, so its norm is arbitrary). Rows: probe rows with | |
| Fok[k], Vf available, and s5 axis exactly aligned with the frame/20 axis (probes/audit_features.py lag == 0) — see latent_audit.md A.2. | |
| Events at t+k (step 5k-1): road pressure P active / onset, old demand D active / onset, vehicle brake / gas pressed and onsets, hands_on. | |
| Subsets: all rows; NON-overlapping windows only (k >= 3: the window [t+k-2, t+k] shares no frame with the history [.., t]); rows within +-1 s | |
| of an onset of the family. Metric: PR-AUC with device-cluster bootstrap CI; paired Δ (b - c) with the same resamples. Also the cosine | |
| diagnostics cos(F_hat, F) vs cos(F_cur, F) vs cos(F_hat, F_cur) per k. No openpilot DMS field is read. | |
| 2026-09-11 opt-ins (innovation 3 b2; defaults reproduce the original run): --space v2 = the same protocol in the pack-v2 future-PCA space (actual F from | |
| results/pack_v2/future_*.npz on the probe rows, copy-current = the current clip-mean token projected with future_pca_v2.npz — the pack-v2 basis is NOT the v1 one, | |
| cos 0.68 on aligned rows; the projected current clip has cos 0.904 with F_v2(k=1), the v1 analogue 0.897); --fhat-runs <run> ... = predicted latents from | |
| results/probes/fhat/<run>_{regime}_{split}.npz (probes/dump_f_hat.py) instead of latents_*.npz, one CSV row block per run (`run` column); --tag = suffix of the | |
| CSV / cos json / report; --events REGEX / --subsets to restrict the cells. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import os | |
| import re | |
| 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 / "future_vs_predicted.log" | |
| CSV = C.OUT / "future_vs_predicted_latest.csv" | |
| COLS = ["stamp", "regime", "probe", "event", "k", "subset", "n_te", "npos_te", "base_rate", "ap_actual", "ci_actual", "ap_pred", "ci_pred", "ap_copy", "ci_copy", | |
| "d_pred_minus_copy", "d_ci", "frac_gt0", "auroc_actual", "auroc_pred", "auroc_copy", "hp", "run", "space"] | |
| PACK_V2 = C.CWM3 / "results" / "pack_v2" | |
| FHAT = C.OUT / "fhat" | |
| def log(m): | |
| C.log(m, LOG) | |
| def events_at(raw, k): | |
| """event labels/masks at horizon k (step index s = 5k-1) -> dict name -> (y, m, near_onset).""" | |
| s = 5 * k - 1 | |
| M = raw["M"].astype(bool) | |
| out = {} | |
| lo, hi = max(0, s - 5), min(24, s + 5) | |
| def onset_near(Z, Zv, Z0, Z0v): | |
| z = np.concatenate([Z0[:, None], Z], 1).astype(bool) # step -1 = anchor | |
| v = np.concatenate([Z0v[:, None], Zv], 1).astype(bool) | |
| on = z[:, 1:] & ~z[:, :-1] & v[:, 1:] & v[:, :-1] | |
| return on[:, lo:hi + 1].any(1), on | |
| for pref, Z, Zv, Z0, Z0v, mov, fams in (("P", raw["P"], raw["Pvalid"], raw["P0"], raw["P0valid"], raw["PMoving"], C.P_FAMILIES), | |
| ("D", raw["D"], raw["Dvalid"], raw["D0"], raw["D0valid"], raw.get("Moving", raw["PMoving"]), C.D_FAMILIES)): | |
| for j, f in enumerate(fams): | |
| valid = Zv[:, :, j].astype(bool) & mov.astype(bool) & M | |
| near, on = onset_near(Z[:, :, j], Zv[:, :, j], Z0[:, j], Z0v[:, j]) | |
| out[f"{pref}_{f}_active"] = (Z[:, s, j].astype(bool), valid[:, s], near) | |
| out[f"{pref}_{f}_onset"] = (on[:, s], valid[:, s] & (valid[:, s - 1] if s > 0 else Z0v[:, j].astype(bool)), near) | |
| Y, X = raw["Y"].astype(np.float32), raw["X"].astype(np.float32) | |
| for name, c in (("brake", 2), ("gas", 1)): | |
| z = np.concatenate([(X[:, -1:, c] > 0.5), Y[:, :, c] > 0.5], 1) | |
| on = z[:, 1:] & ~z[:, :-1] | |
| out[f"veh_{name}_pressed"] = (z[:, s + 1], M[:, s], on[:, lo:hi + 1].any(1)) | |
| out[f"veh_{name}_onset"] = (on[:, s], M[:, s] & (M[:, s - 1] if s > 0 else np.ones(len(M), bool)), on[:, lo:hi + 1].any(1)) | |
| S = raw["S"].astype(np.float32) | |
| out["drv_hands_on"] = (S[:, s, 0] > 0.5, np.isfinite(S[:, s, 0]) & M[:, s], np.zeros(len(M), bool)) | |
| return out | |
| class Data: | |
| def __init__(self, regime, split, space="v1", fhat_runs=None): | |
| self.raw = C.load_raw(regime, split) | |
| z = np.load(C.feats_path(regime, split), allow_pickle=True) | |
| self.Vf = z["Vf4224"] | |
| rows = C.probe_rows(regime, split, verbose=lambda m: None)["rows"] | |
| if space == "v2": # pack-v2 actual future latent (frame/20 axis, v2 PCA basis) on the probe rows | |
| with np.load(PACK_V2 / f"future_{regime}_{split}.npz") as z2: | |
| self.F, self.Fok = z2["F"][rows].astype(np.float32), z2["ok"][rows].astype(bool) | |
| fp = np.load(PACK_V2 / "future_pca_v2.npz") | |
| else: | |
| self.F, self.Fok = self.raw["F"].astype(np.float32), self.raw["Fok"].astype(bool) | |
| fp = np.load(C.S12R / "future_pca.npz") | |
| self.F_hat = {} # run -> predicted future [n,5,256] (test split only when fhat runs are given) | |
| if fhat_runs: | |
| if split == "test": | |
| for run in fhat_runs: | |
| with np.load(FHAT / f"{run}_{regime}_{split}.npz") as zf: | |
| assert np.array_equal(zf["rows"], rows), f"{run}: fhat rows != probe rows" | |
| self.F_hat[run] = zf["F_hat"].astype(np.float32) | |
| else: | |
| lat = np.load(C.OUT / f"latents_{regime}_{split}.npz") | |
| self.F_hat["latents"] = lat["F_hat"].astype(np.float32) | |
| cur = self.Vf[:, 1408:2816].astype(np.float32) | |
| self.F_cur = (cur - fp["mu"]) @ fp["W"] | |
| lagp = C.OUT / f"s5_lag_{regime}_{split}.npy" | |
| self.lag = np.load(lagp) if lagp.exists() else np.zeros(len(self.raw["t"]), np.float32) | |
| self.aligned = np.isfinite(self.Vf[:, 0]) & (self.lag == 0) | |
| self.n = len(self.raw["t"]) | |
| def rows(self, k): | |
| return self.aligned & self.Fok[:, k - 1] | |
| def unit(X): | |
| return X / (np.linalg.norm(X, axis=1, keepdims=True) + 1e-6) | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--n-boot", type=int, default=300) | |
| ap.add_argument("--probes", default="ridge,mlp64") | |
| ap.add_argument("--report-only", action="store_true") | |
| ap.add_argument("--space", default="v1", choices=["v1", "v2"], help="future-PCA space of the protocol (v2 = pack-v2 F / future_pca_v2; 2026-09-11 opt-in)") | |
| ap.add_argument("--fhat-runs", nargs="*", default=None, help="runs whose F_hat (probes/dump_f_hat.py) play the `predicted` role; default = the seed-0 latents_*.npz") | |
| ap.add_argument("--tag", default="", help="suffix of the CSV / cos json / report (default: the original files)") | |
| ap.add_argument("--events", default="", help="regex on the event names (default all)") | |
| ap.add_argument("--subsets", default="all,nonoverlap,near_onset") | |
| a = ap.parse_args() | |
| global CSV, LOG | |
| if a.tag: | |
| CSV, LOG = C.OUT / f"future_vs_predicted_{a.tag}.csv", C.OUT / f"future_vs_predicted_{a.tag}.log" | |
| if a.report_only: | |
| write_report(a.tag) | |
| return | |
| stamp = C.stamp() | |
| t_all = time.time() | |
| new = not CSV.exists() | |
| f = open(CSV, "a", newline="") | |
| w = csv.DictWriter(f, fieldnames=COLS) | |
| if new: | |
| w.writeheader() | |
| cosrows = [] | |
| for regime in a.regimes.split(","): | |
| D = {s: Data(regime, s, a.space, a.fhat_runs) for s in C.SPLITS} | |
| log(f"{regime}: aligned rows train/val/test {D['train'].aligned.mean():.3f}/{D['val'].aligned.mean():.3f}/{D['test'].aligned.mean():.3f}") | |
| for k in range(1, 6): | |
| ev = {s: events_at(D[s].raw, k) for s in C.SPLITS} | |
| names = [n for n in ev["train"].keys() if not a.events or re.search(a.events, n)] | |
| rows = {s: D[s].rows(k) for s in C.SPLITS} | |
| Fa = {s: unit(D[s].F[rows[s], k - 1]) for s in C.SPLITS} | |
| Fc = unit(D["test"].F_cur[rows["test"]]) | |
| Ft = Fa["test"] | |
| runs = list(D["test"].F_hat) | |
| Fp_all = {run: unit(D["test"].F_hat[run][rows["test"], k - 1]) for run in runs} | |
| for run, Fp in Fp_all.items(): # cosine diagnostics on the test rows, per predicted-latent source | |
| cosrows.append({"regime": regime, "k": k, "n": int(rows["test"].sum()), "run": run, "space": a.space, "cos(F_hat,F)": float((Fp * Ft).sum(1).mean()), "cos(F_cur,F)": float((Fc * Ft).sum(1).mean()), | |
| "cos(F_hat,F_cur)": float((Fp * Fc).sum(1).mean()), "cos(F_hat,F)-cos(F_cur,F) p>0": float(((Fp * Ft).sum(1) > (Fc * Ft).sum(1)).mean())}) | |
| std = C.Std(Fa["train"]) | |
| Ztr, Zva = std(Fa["train"]), std(Fa["val"]) | |
| Zte = {"actual": std(Ft), "copy": std(Fc), **{f"pred:{run}": std(Fp) for run, Fp in Fp_all.items()}} | |
| Y = {s: np.stack([ev[s][n][0] for n in names], 1).astype(np.float32) for s in C.SPLITS} | |
| Mm = {s: np.stack([ev[s][n][1] for n in names], 1) for s in C.SPLITS} | |
| near = {s: np.stack([ev[s][n][2] for n in names], 1) for s in C.SPLITS} | |
| Ytr, Mtr = Y["train"][rows["train"]], Mm["train"][rows["train"]] | |
| Yva, Mva = Y["val"][rows["val"]], Mm["val"][rows["val"]] | |
| Yte, Mte, Nte = Y["test"][rows["test"]], Mm["test"][rows["test"]], near["test"][rows["test"]] | |
| dte = D["test"].raw["dongle"][rows["test"]] | |
| scores = {} | |
| if "ridge" in a.probes: | |
| sva, s_act = C.ridge_fit_predict(Ztr, Ytr, Mtr, Zva, Zte["actual"]) | |
| _, s_copy = C.ridge_fit_predict(Ztr, Ytr, Mtr, Zva[:1], Zte["copy"]) | |
| s_pred = {run: C.ridge_fit_predict(Ztr, Ytr, Mtr, Zva[:1], Zte[f"pred:{run}"])[1] for run in runs} | |
| for j, n in enumerate(names): | |
| if not np.isfinite(sva[j, 0, 0]): | |
| continue | |
| aps = [C.ap_weighted(Yva[Mva[:, j], j], sva[j, x][Mva[:, j]]) if Yva[Mva[:, j], j].sum() > 0 else np.nan for x in range(sva.shape[1])] | |
| if not np.any(np.isfinite(aps)): | |
| continue | |
| x = int(np.nanargmax(aps)) | |
| for run in runs: | |
| scores[("ridge", n, run)] = (s_act[j, x], s_pred[run][j, x], s_copy[j, x], f"alpha={C.RIDGE_ALPHAS[x]:g}") | |
| if "mlp64" in a.probes: | |
| import torch | |
| # one multi-task MLP on actual F; apply to every input at each target's best-val epoch (re-run forward at that epoch) | |
| Zcat = np.concatenate([Zte["actual"], Zte["copy"]] + [Zte[f"pred:{run}"] for run in runs]) | |
| te, va, ep, apv = C.mlp_fit_predict(Ztr, Ytr, Mtr, Zva, Yva, Mva, Zcat, hidden=64, epochs=25, seed=0) | |
| n_te = len(Zte["actual"]) | |
| for j, n in enumerate(names): | |
| if apv[j] < 0: | |
| continue | |
| for r_i, run in enumerate(runs): | |
| scores[("mlp64", n, run)] = (te[j, :n_te], te[j, (2 + r_i) * n_te:(3 + r_i) * n_te], te[j, n_te:2 * n_te], f"epoch={ep[j]}") | |
| for (probe, n, run), (sa, sp, sc, hp) in scores.items(): | |
| j = names.index(n) | |
| for subset in [x for x in ("all", "nonoverlap", "near_onset") if x in a.subsets.split(",")]: | |
| if subset == "nonoverlap" and k < 3: | |
| continue | |
| m = Mte[:, j].copy() | |
| if subset == "near_onset": | |
| m &= Nte[:, j] | |
| if m.sum() < 100 or Yte[m, j].sum() < 10 or (~Yte[m, j].astype(bool)).sum() < 10: | |
| continue | |
| y = Yte[m, j].astype(bool) | |
| ra = C.cluster_boot(y, sa[m], dte[m], n_boot=a.n_boot) | |
| rp = C.cluster_boot(y, sp[m], dte[m], n_boot=a.n_boot, s_b=sc[m]) | |
| rc = C.cluster_boot(y, sc[m], dte[m], n_boot=a.n_boot) | |
| w.writerow({"stamp": stamp, "regime": regime, "probe": probe, "event": n, "k": k, "subset": subset, "n_te": int(m.sum()), "npos_te": int(y.sum()), | |
| "base_rate": round(float(y.mean()), 5), "ap_actual": round(ra["ap"], 5), "ci_actual": f"{ra['ci'][0]:.3f},{ra['ci'][1]:.3f}", | |
| "ap_pred": round(rp["ap"], 5), "ci_pred": f"{rp['ci'][0]:.3f},{rp['ci'][1]:.3f}", "ap_copy": round(rc["ap"], 5), "ci_copy": f"{rc['ci'][0]:.3f},{rc['ci'][1]:.3f}", | |
| "d_pred_minus_copy": round(rp["delta"], 5), "d_ci": f"{rp['delta_ci'][0]:.3f},{rp['delta_ci'][1]:.3f}", "frac_gt0": round(rp["frac_gt0"], 3), | |
| "auroc_actual": round(ra["auroc"], 4), "auroc_pred": round(rp["auroc"], 4), "auroc_copy": round(rc["auroc"], 4), "hp": hp, "run": run, "space": a.space}) | |
| f.flush() | |
| log(f" {regime} k={k}: rows {rows['test'].sum():,} test; cells {len(scores)}; cos(F_hat,F)={cosrows[-1]['cos(F_hat,F)']:.3f} cos(F_cur,F)={cosrows[-1]['cos(F_cur,F)']:.3f}") | |
| del D | |
| f.close() | |
| sfx = f"_{a.tag}" if a.tag else "" | |
| C.write_json(C.OUT / f"future_cos{sfx}_{stamp}.json", cosrows) | |
| C.write_json(C.OUT / f"future_cos{sfx}_latest.json", cosrows) | |
| log(f"done in {time.time() - t_all:.0f}s") | |
| write_report(a.tag) | |
| def write_report(tag=""): | |
| import json | |
| sfx = f"_{tag}" if tag else "" | |
| with open(CSV) as fh: | |
| rows = list(csv.DictReader(fh)) | |
| last = max(r["stamp"] for r in rows) | |
| rows = [r for r in rows if r["stamp"] == last] | |
| for r in rows: | |
| r.setdefault("run", "latents") | |
| cos = json.load(open(C.OUT / f"future_cos{sfx}_latest.json")) | |
| runs = sorted({r["run"] for r in rows}) | |
| space = rows[0].get("space", "v1") if rows else "v1" | |
| L = [f"# Encoding vs dynamics (P2) — actual future latent vs the world model's predicted future vs copy-current" + (f" — {tag} (future-PCA space {space}; predicted = {', '.join(runs)})" if tag else ""), "", | |
| f"Generated {C.stamp()} by `probes/future_vs_predicted.py` (run stamp {last}); raw cells `results/probes/{CSV.name}`. " | |
| "Read-out trained on the ACTUAL future front latent F(t+k) of the TRAIN probe rows (alpha / epoch selected on VAL with actual F), evaluated frozen on TEST rows " | |
| "(disjoint devices) fed with (a) actual F(t+k) — offline reference, needs the future, NOT a model result; (b) E0_pooled seed-0 predicted F_hat(t+k); " | |
| "(c) copy-current: the front latent of the window ending at t in the same PCA space. Inputs L2-normalised. Rows restricted to anchors whose s5 axis is exactly " | |
| "aligned with the frame/20 axis (latent_audit.md A.2) and Fok[k]. PR-AUC with device-cluster bootstrap 95 % CI; Δ = (b) − (c) with shared resamples.", "", | |
| "## 0. Cosine diagnostics on the test rows (is the predicted future closer to the actual future than a copy of the present?)", "", | |
| C.md_table(["regime", "k (s)", "n", "predicted from", "cos(F_hat, F)", "cos(F_cur, F)", "cos(F_hat, F_cur)", "share of rows where F_hat beats copy-current"], | |
| [[c["regime"], c["k"], c["n"], c.get("run", "latents"), f"{c['cos(F_hat,F)']:.3f}", f"{c['cos(F_cur,F)']:.3f}", f"{c['cos(F_hat,F_cur)']:.3f}", f"{c['cos(F_hat,F)-cos(F_cur,F) p>0']:.3f}"] for c in cos]), ""] | |
| events = [] | |
| for r in rows: | |
| if r["event"] not in events: | |
| events.append(r["event"]) | |
| for run in runs: | |
| for regime in ("manual", "engaged"): | |
| for probe in ("ridge", "mlp64"): | |
| for subset in ("all", "nonoverlap", "near_onset"): | |
| sub = [r for r in rows if r["regime"] == regime and r["probe"] == probe and r["subset"] == subset and r["run"] == run] | |
| if not sub: | |
| continue | |
| title = {"all": "all rows", "nonoverlap": "NON-overlapping windows only (k >= 3)", "near_onset": "rows within +-1 s of an onset of the family"}[subset] | |
| L += [f"## {regime} — {probe} — {title}" + (f" — predicted from `{run}`" if len(runs) > 1 or run != "latents" else ""), "", "Cell = AP actual / **predicted** / copy-current; Δ(pred − copy) [CI]; n / pos.", ""] | |
| hdr = ["event"] + [f"k={k}" for k in range(1, 6)] | |
| body = [] | |
| for e in events: | |
| rr = {int(r["k"]): r for r in sub if r["event"] == e} | |
| if not rr: | |
| continue | |
| row = [e] | |
| for k in range(1, 6): | |
| r = rr.get(k) | |
| if r is None: | |
| row.append("–") | |
| continue | |
| d, (lo, hi) = float(r["d_pred_minus_copy"]), map(float, r["d_ci"].split(",")) | |
| mark = "**" if (lo > 0 or hi < 0) else "" | |
| row.append(f"{float(r['ap_actual']):.3f} / **{float(r['ap_pred']):.3f}** / {float(r['ap_copy']):.3f}; Δ {mark}{d:+.3f} [{lo:+.3f}, {hi:+.3f}]{mark}; {r['n_te']}/{r['npos_te']}") | |
| body.append(row) | |
| L += [C.md_table(hdr, body), ""] | |
| # recovered share of the copy-current -> actual gap: (pred - copy) / (actual - copy), state ("active"/"pressed") events, ridge, all rows | |
| L += ["## Summary — share of the gap between copy-current and the actual future that the predicted latent recovers, (P − C) / (A − C), ridge, state events", ""] | |
| for run in runs: | |
| for regime in ("manual", "engaged"): | |
| sub = [r for r in rows if r["regime"] == regime and r["probe"] == "ridge" and r["subset"] == "all" and (r["event"].endswith("_active") or r["event"].endswith("_pressed")) and r["run"] == run] | |
| if not sub: | |
| continue | |
| evs = [] | |
| for r in sub: | |
| if r["event"] not in evs: | |
| evs.append(r["event"]) | |
| body = [] | |
| for e in evs: | |
| row = [e] | |
| for k in range(1, 6): | |
| r = next((x for x in sub if x["event"] == e and int(x["k"]) == k), None) | |
| if r is None: | |
| row.append("–"); continue | |
| A, P_, Cc = float(r["ap_actual"]), float(r["ap_pred"]), float(r["ap_copy"]) | |
| row.append(f"{(P_ - Cc) / (A - Cc):+.2f}" if abs(A - Cc) > 0.01 else "n/a (A≈C)") | |
| body.append(row) | |
| L += [f"### {regime}" + (f" — `{run}`" if len(runs) > 1 or run != "latents" else ""), "", C.md_table(["event", "k=1", "k=2", "k=3", "k=4", "k=5"], body), ""] | |
| L += ["Reading: 0 = the prediction is no better than copying the present; 1 = it reaches the offline actual-future read-out; negative = worse than copying. " | |
| "k ≥ 3 columns are the NON-overlapping windows (the +k window shares no frame with the history).", ""] | |
| L += ["## Interpretation key (applied in docs/latent_quality.md)", "", | |
| C.md_table(["pattern", "reading"], [["actual-future decodes poorly", "representation / pooling / decoder bottleneck — the latent itself does not carry the event"], | |
| ["actual good but predicted poor", "dynamics bottleneck — the information exists in the future latent but the world model does not produce it"], | |
| ["predicted ≈ copy-current", "no learned change — the F head reproduces the present"], | |
| ["predicted > copy-current and close to actual", "the world model anticipates the change"]]), ""] | |
| (C.TABLES / f"latent_future_vs_predicted{sfx}.md").write_text("\n".join(L) + "\n") | |
| log(f"report -> {C.TABLES / f'latent_future_vs_predicted{sfx}.md'}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 20.5 kB
- Xet hash:
- 34aa5ded1b07a2b9af67195720066f5cee92bba0823b0a821a1245f79c9e2723
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.