Buckets:
| """cwm3/data/build_road_dynamics_targets.py — ROAD-DYNAMICS forecasting targets + CAN road history on the s12 anchor grid (WP4 batch 1, F5b/F5c; | |
| pre-registration docs/decisions.md 2026-09-16T06:46:35Z 'WP4 F+ forecasting upgrade'; plan <local-notes>/plans/elegant-dazzling-stallman.md). | |
| Question answered downstream (eval/road_dynamics_eval.py, results/tables/road_dynamics.md): can road DEMAND be predicted from the road's own | |
| dynamics rather than only recognised from the current signal (kill_report_final K7)? The targets are therefore FUTURE ROAD QUANTITIES — not V-JEPA | |
| features: lead relative distance / relative speed / lead speed and the lane-centre offset at t+0.2 .. t+5.0 s (k = 1..25), with validity masks, and the | |
| derived demand events (ttc < 4 s, lane excursion, lead braking — the P families of data/build_pressure_targets.py, copied bit-for-bit so every arm is | |
| scored on exactly the population of the FINAL model's P chains). | |
| Sources (all through the canspec whitelist; the grid10 C block IS canspec.can_features' output): | |
| s14_corr/cache/grid10/<uid>.npz {t[N], G[N,32]} at 10 Hz — columns read (index): vEgo 12, lane_center_offset 21, lane_width 22, lane_valid 23, | |
| lead_dRel 24, lead_vRel 25, lead_vLead 26, lead_valid 27 (+ badas_risk 28 / ttc 29 / lane_off 30 through build_pressure_targets.route_pressure for the | |
| current margins M0). Never columns 0-11 (DMS block 0-6, pose block 7-11): asserted at import through build_pressure_targets.READ_COLS. | |
| results/targets_pressure/pressure_{regime}_{split}.npz — P / Pvalid / P0 / P0valid / P_first / Moving / Moving0 / M0 / M0valid (row-aligned; uid, t asserted). | |
| results/pack_v2/vf256_{regime}_{split}.npz — the FINAL E base's per-anchor side vector (PCA-256 of the causal full-frame V-JEPA features), for the | |
| F5b side stream only. | |
| Outputs (results/targets_road_dyn/, row order = s12 anchor grid of targets_{regime}_{split}.npz): | |
| road_dyn_{regime}_{split}.npz | |
| H[N,40,8] f16 CAN road history t-7.8 .. t @ 5 Hz (grid samples k-78, k-76, ..., k; the LAST step is the anchor sample), channels HIST_CHANNELS = | |
| lead_dRel, lead_vRel, lead_vLead, lane_center_offset, lane_width, vEgo, lead_valid, lane_valid (canspec sentinels: no lead -> 200/0/0, | |
| no lane -> 0/0; the valid channels disambiguate). Hvalid[N,40] bool in range & vEgo finite | |
| Yf[N,25,4] f32 future lead_dRel, lead_vRel, lead_vLead, lane_center_offset at t+0.2k (grid k+2j). Yvalid[N,25,4] bool lead channels: lead_valid at | |
| t+0.2k & in range & vEgo finite; lane: lane_valid at t+0.2k & in range & vEgo finite. Yw[N,25] f32 future lane_width (valid with lane) | |
| E[N,25,3] E0[N,3] Evalid E0valid E_first[N,3] Moving[N,25] Moving0[N] the derived demand events = targets_pressure families EVENT_FAMILIES (ttc, | |
| lane_excursion, lead_braking) copied; asserted equal to the pack | |
| M0[N,10] f32 M0valid current margins (build_pressure_targets.MARGIN_NAMES) recomputed here and asserted equal to the pack's M0 | |
| ecan[N,40] f32 ecan_ok[N] the 40 RAW anchor features of the F5b side stream (ECAN_NAMES: the 10 current margins M0 + 30 summaries of the 8 s | |
| history: mean / slope / min / max of dRel, vRel, vLead, lane offset and vEgo, lane-width mean, lead / lane valid fractions, number of lead | |
| target switches, minima of ttc / headway / lateral margin / lead acceleration, approaching and excursion fractions) | |
| uid t dongle hist_channels target_channels event_families margin_names ecan_names | |
| ecan_stats.npz TRAIN-only (both regimes pooled, float64) mean / sd of the 40 ecan features over ecan_ok rows — the standardisation of the side streams | |
| vfcan256_{regime}_{split}.npz Z f16 [N,256] = [vf256 PCA components 0..215 (0 where vf256 row_ok is False), 40 standardised ecan features (clip +-8, 0 where | |
| ecan_ok is False)], row_ok = vf256 row_ok | ecan_ok, n — the SAME file contract as results/pack_v2/vf256_* (Z / row_ok / n), so the data | |
| layer loads it as a side stream (`side: vfcan256`, d_side 256 unchanged: no model code changes). The 40 lowest-variance PCA components | |
| (sd 0.40-0.43 vs 16 for the first) are dropped to make room; documented as part of the variant definition. | |
| can256_{regime}_{split}.npz Z f16 [N,256] = [216 zeros, the 40 standardised ecan features], row_ok = ecan_ok — the SENSOR-ONLY control side stream | |
| (E0 `pooled_P_canonly`: `tokens: none` + `side: can256`). | |
| Strict-future contract (tests/test_road_dynamics_targets.py): perturbing grid samples AFTER the anchor changes only Yf / Yw / E, never H / M0 / ecan. | |
| Run: cd cwm3 && python data/build_road_dynamics_targets.py [--regimes manual,engaged] [--splits train,val,test] [--force] | |
| No openpilot DMS field is read; grid10 columns 0-11 are never read (FORBIDDEN block guarded by build_pressure_targets). | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import sys | |
| import time | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "8") | |
| import numpy as np # noqa: E402 | |
| ROOT = Path(__file__).resolve().parents[1] | |
| EXPLORE = ROOT.parent | |
| sys.path.insert(0, str(ROOT / "common")) | |
| sys.path.insert(0, str(ROOT / "data")) | |
| from runmeta import write_run_json, sha256_file # noqa: E402 | |
| import build_pressure_targets as bp # noqa: E402 (the grid10 reader with the column firewall; route_pressure = the P / M0 definitions) | |
| OUT = ROOT / "results" / "targets_road_dyn" | |
| PRESS = ROOT / "results" / "targets_pressure" | |
| PACK_V2 = ROOT / "results" / "pack_v2" | |
| # ---- 2026-09-17 WP5 BATON-holdout (coordinator decision docs/decisions.md 2026-09-17T15:51:24Z): ADDITIVE path override. `--holdout` (root from env | |
| # CWM3_HOLDOUT_ROOT, default results/holdout) reads pressure / pack_v2 from the holdout artefacts and writes ONLY into <root>/targets_road_dyn/; the s12 anchor | |
| # grid (bp.S12/targets_{r}_holdout.npz) and the grid10 cache (bp.GRID) are read from their usual places, where the holdout lane wrote the 60 routes. The side | |
| # streams are standardised with the FROZEN TRAIN statistics results/targets_road_dyn/ecan_stats.npz — LOADED (sha256 pinned below, values asserted equal to the | |
| # F5b build's run json), never recomputed, never rewritten. Default behaviour (no flag) is unchanged. | |
| HOLDOUT_ROOT = ROOT / "results" / "holdout" | |
| FROZEN_ECAN_STATS = ROOT / "results" / "targets_road_dyn" / "ecan_stats.npz" | |
| FROZEN_ECAN_STATS_SHA256 = "5aa420636042233306a7d65111d775a4da254d7e0f801babcb11dd0d68ffdd53" # file of 2026-09-16T07:06Z (F5b build; every F5b E0 / joint run used side streams standardised with it) | |
| FROZEN_ECAN_STATS_RUNJSON = ROOT / "results" / "targets_road_dyn" / "run_build_road_dynamics_targets_20260916T070733Z.json" # records the same mu / sd / n (extra.ecan_stats) | |
| def configure_holdout(root=None): | |
| """point OUT / PRESS / PACK_V2 at the holdout artefacts; returns the paths. Nothing else changes (bp.S12, bp.GRID untouched).""" | |
| global OUT, PRESS, PACK_V2 | |
| root = Path(root or os.environ.get("CWM3_HOLDOUT_ROOT") or HOLDOUT_ROOT) | |
| OUT, PRESS, PACK_V2 = root / "targets_road_dyn", root / "targets_pressure", root / "pack_v2" | |
| return dict(OUT=OUT, PRESS=PRESS, PACK_V2=PACK_V2) | |
| def load_frozen_ecan_stats(path=FROZEN_ECAN_STATS, expected_sha256=FROZEN_ECAN_STATS_SHA256, runjson=FROZEN_ECAN_STATS_RUNJSON): | |
| """-> (mu float64[40], sd float64[40], n) of the FROZEN TRAIN statistics; refuses a file whose sha256 differs from the pinned F5b value and checks the | |
| stored mu / sd / n against the F5b build's run json (float32 storage of the recorded float64 values). Read-only.""" | |
| path = Path(path) | |
| sha = sha256_file(path) | |
| assert sha == expected_sha256, f"{path}: sha256 {sha} != frozen F5b ecan_stats {expected_sha256} — refusing (the side streams must use the statistics the ECAN runs were trained with)" | |
| with np.load(path, allow_pickle=True) as z: | |
| mu32, sd32, n = z["mu"].astype(np.float32), z["sd"].astype(np.float32), int(z["n"]) | |
| names = [str(x) for x in z["names"]] | |
| assert names == ECAN_NAMES and mu32.shape == sd32.shape == (N_ECAN,), (names[:3], mu32.shape) | |
| if runjson is not None and Path(runjson).exists(): | |
| rj = json.loads(Path(runjson).read_text()) | |
| rec = (rj.get("extra") or rj).get("ecan_stats", {}) | |
| assert int(rec.get("n", -1)) == n and np.array_equal(mu32, np.array(rec["mu"], np.float32)) and np.array_equal(sd32, np.array(rec["sd"], np.float32)), \ | |
| f"{path}: stored statistics differ from the F5b build record {runjson}" | |
| return mu32.astype(np.float64), sd32.astype(np.float64), n | |
| FS = 10.0 | |
| H_STEPS, STEP_GRID = 25, 2 # 25 future steps of 0.2 s | |
| HIST_STEPS, HIST_STRIDE = 40, 2 # 40 history steps of 0.2 s (8 s @ 5 Hz), last = anchor | |
| HIST_WIN = 80 # 8 s @ 10 Hz for the summaries (samples k-79 .. k) | |
| HIST_CHANNELS = ["lead_dRel", "lead_vRel", "lead_vLead", "lane_center_offset", "lane_width", "vEgo", "lead_valid", "lane_valid"] | |
| TARGET_CHANNELS = ["lead_dRel", "lead_vRel", "lead_vLead", "lane_center_offset"] | |
| EVENT_FAMILIES = ["ttc", "lane_excursion", "lead_braking"] | |
| EVENT_COLS = [bp.FAMILIES.index(f) for f in EVENT_FAMILIES] | |
| N_PCA_KEEP, N_ECAN = 216, 40 | |
| ECAN_NAMES = [f"m0_{n}" for n in bp.MARGIN_NAMES] + \ | |
| [f"h8_{c}_{s}" for c in ("dRel", "vRel", "vLead", "laneoff") for s in ("mean", "slope", "min", "max")] + ["h8_lanew_mean"] + \ | |
| [f"h8_vEgo_{s}" for s in ("mean", "slope", "min", "max")] + \ | |
| ["h8_lead_frac", "h8_lane_frac", "h8_n_lead_switch", "h8_ttc_min", "h8_headway_min", "h8_latmargin_min", "h8_aLead_min", "h8_frac_approach", "h8_frac_excursion"] | |
| assert len(ECAN_NAMES) == N_ECAN, len(ECAN_NAMES) | |
| # the grid columns this module reads directly (all in build_pressure_targets.READ_COLS -> asserted >= 12 there) | |
| GRID_READ = ["vEgo", "lane_center_offset", "lane_width", "lane_valid", "lead_dRel", "lead_vRel", "lead_vLead", "lead_valid"] | |
| for _c in GRID_READ: | |
| assert _c in bp.READ_COLS and bp.CI[_c] >= bp.N_FORBIDDEN, _c | |
| # ----------------------------------------------------------------------------- per route | |
| def route_dynamics(G): | |
| """per-sample road arrays of one route: the pressure builder's margins / events (route_pressure) + the raw canspec road channels.""" | |
| rd = bp.route_pressure(G) | |
| raw = np.stack([bp.col(G, c) for c in HIST_CHANNELS]).astype(np.float32) # [8,T] | |
| raw = np.nan_to_num(raw, nan=0.0) | |
| lead_valid = bp.col(G, "lead_valid") > 0.5 | |
| dRel = bp.col(G, "lead_dRel") | |
| drop1 = np.zeros(rd["T"], np.float64) | |
| drop1[1:] = np.where(lead_valid[1:] & lead_valid[:-1], dRel[:-1] - dRel[1:], 0.0) | |
| gap_n = int(round(bp.THR["cut_in_gap_s"] * FS)) | |
| cs_inv = np.cumsum(np.r_[0, (~lead_valid).astype(np.int64)]) | |
| prev_all_invalid = np.zeros(rd["T"], bool) | |
| prev_all_invalid[gap_n:] = (cs_inv[gap_n:gap_n + rd["T"] - gap_n] - cs_inv[:rd["T"] - gap_n]) == gap_n | |
| switch = ((drop1 > bp.THR["same_target_jump_m"]) | (lead_valid & prev_all_invalid)).astype(np.float32) # lead target switch / new lead events | |
| vfin = np.isfinite(bp.col(G, "vEgo")) | |
| return dict(T=rd["T"], M=rd["M"], Mvalid=rd["Mvalid"], mv=rd["mv"], thr=rd["thr"], radar_present=rd["radar_present"], lane_cov=rd["lane_cov"], | |
| badas_cov=rd["badas_cov"], n_moving=rd["n_moving"], raw=raw, switch=switch, vfin=vfin) | |
| def _masked_stats(x, w, dt, sent_mean, sent_min, sent_max): | |
| """x, w [n,W] (w = validity 0/1), dt [W] sample offsets (s) -> mean, OLS slope (per s), min, max over valid samples; sentinels where < 1 (< 3 for the slope).""" | |
| n_ok = w.sum(1) | |
| sw = np.maximum(n_ok, 1.0) | |
| mean = np.where(n_ok > 0, (x * w).sum(1) / sw, sent_mean) | |
| tbar = (w * dt[None, :]).sum(1) / sw | |
| dtc = dt[None, :] - tbar[:, None] | |
| num = (w * dtc * (x - mean[:, None])).sum(1) | |
| den = (w * dtc ** 2).sum(1) | |
| slope = np.where((n_ok >= 3) & (den > 1e-9), num / np.maximum(den, 1e-9), 0.0) | |
| big = 1e9 | |
| mn = np.where(n_ok > 0, np.where(w > 0, x, big).min(1), sent_min) | |
| mx = np.where(n_ok > 0, np.where(w > 0, x, -big).max(1), sent_max) | |
| return mean, slope, mn, mx | |
| def gather_dynamics(rd, k): | |
| """anchors k (grid sample index) -> history H / Hvalid, future Yf / Yvalid / Yw, current margins M0 / M0valid and the 40 raw ecan features.""" | |
| T = rd["T"] | |
| n = len(k) | |
| raw, M, Mv, vfin = rd["raw"], rd["M"], rd["Mvalid"], rd["vfin"] | |
| k0ok = (k >= 0) & (k < T) | |
| k0c = np.clip(k, 0, T - 1) | |
| # --- history 8 s @ 5 Hz (last step = anchor) | |
| hoff = -(HIST_STEPS - 1 - np.arange(HIST_STEPS)) * HIST_STRIDE # -78 .. 0 | |
| hi = k[:, None] + hoff[None, :] | |
| hin = (hi >= 0) & (hi < T) | |
| hic = np.clip(hi, 0, T - 1) | |
| H = raw[:, hic].transpose(1, 2, 0) * hin[:, :, None] # [n,40,8] | |
| Hvalid = hin & vfin[hic] | |
| # --- future 25 steps | |
| foff = (np.arange(1, H_STEPS + 1) * STEP_GRID)[None, :] | |
| fi = k[:, None] + foff | |
| fin = (fi >= 0) & (fi < T) | |
| fic = np.clip(fi, 0, T - 1) | |
| Yf = raw[:4][:, fic].transpose(1, 2, 0).astype(np.float32) # [n,25,4] | |
| lead_f = (raw[6][fic] > 0.5) & fin & vfin[fic] | |
| lane_f = (raw[7][fic] > 0.5) & fin & vfin[fic] | |
| Yvalid = np.stack([lead_f, lead_f, lead_f, lane_f], -1) | |
| Yw = raw[4][fic].astype(np.float32) * lane_f | |
| # --- current margins (the pack's M0 definition, recomputed) | |
| M0 = (M[:, k0c].T * k0ok[:, None]).astype(np.float32) | |
| M0valid = Mv[:, k0c].T & k0ok[:, None] | |
| # --- 8 s summaries @ 10 Hz (samples k-79 .. k) | |
| woff = np.arange(-HIST_WIN + 1, 1) | |
| wi = k[:, None] + woff[None, :] | |
| win = (wi >= 0) & (wi < T) | |
| wic = np.clip(wi, 0, T - 1) | |
| dt = woff / FS | |
| wv = (win & vfin[wic]).astype(np.float64) | |
| lead_w = wv * (raw[6][wic] > 0.5) | |
| lane_w_ok = wv * (raw[7][wic] > 0.5) * (raw[4][wic] >= bp.THR["lane_w_min"]) * (raw[4][wic] <= bp.THR["lane_w_max"]) | |
| dRel, vRel, vLead, loff, lw, v = (raw[c][wic].astype(np.float64) for c in range(6)) | |
| feats = [M0[:, j] for j in range(M0.shape[1])] | |
| feats += list(_masked_stats(dRel, lead_w, dt, bp.THR["drel_none"], bp.THR["drel_none"], bp.THR["drel_none"])) | |
| feats += list(_masked_stats(vRel, lead_w, dt, 0.0, 0.0, 0.0)) | |
| feats += list(_masked_stats(vLead, lead_w, dt, 0.0, 0.0, 0.0)) | |
| feats += list(_masked_stats(loff, lane_w_ok, dt, 0.0, 0.0, 0.0)) | |
| feats.append(_masked_stats(lw, lane_w_ok, dt, 0.0, 0.0, 0.0)[0]) | |
| feats += list(_masked_stats(v, wv, dt, 0.0, 0.0, 0.0)) | |
| nw = np.maximum(wv.sum(1), 1.0) | |
| feats.append(lead_w.sum(1) / nw) | |
| feats.append((wv * (raw[7][wic] > 0.5)).sum(1) / nw) | |
| feats.append((wv * rd["switch"][wic]).sum(1)) | |
| approach = lead_w * (vRel < -0.1) | |
| ttc = np.where(approach > 0, dRel / np.maximum(-vRel, 0.1), bp.THR["ttc_none"]) | |
| feats.append(np.minimum(np.where(wv > 0, ttc, bp.THR["ttc_none"]).min(1), bp.THR["ttc_none"])) | |
| headway = np.where(lead_w > 0, dRel / np.maximum(v, bp.THR["vego_floor"]), bp.THR["headway_none"]) | |
| feats.append(np.minimum(np.where(wv > 0, headway, bp.THR["headway_none"]).min(1), bp.THR["headway_none"])) | |
| lat = np.where(lane_w_ok > 0, lw / 2.0 - np.abs(loff), bp.THR["lat_none"]) | |
| feats.append(np.minimum(lat.min(1), bp.THR["lat_none"])) | |
| aL = M[4][wic].astype(np.float64) * (Mv[4][wic] & (win)) # same-target backward 1 s lead acceleration (0 where undefined) | |
| feats.append(aL.min(1)) | |
| feats.append(approach.sum(1) / np.maximum(lead_w.sum(1), 1.0)) | |
| feats.append((lane_w_ok * (lat < bp.THR["veh_half_w"])).sum(1) / np.maximum(lane_w_ok.sum(1), 1.0)) | |
| ecan = np.stack(feats, 1).astype(np.float32) | |
| assert ecan.shape == (n, N_ECAN), ecan.shape | |
| ecan_ok = k0ok & vfin[k0c] | |
| ecan[~ecan_ok] = 0.0 | |
| return dict(H=H.astype(np.float16), Hvalid=Hvalid, Yf=Yf, Yvalid=Yvalid, Yw=Yw, M0=M0, M0valid=M0valid, ecan=ecan, ecan_ok=ecan_ok) | |
| # ----------------------------------------------------------------------------- split | |
| def build_split(regime, split, cache, thr_rows): | |
| d, wa = bp.build_split(regime, split, cache, thr_rows, route_fn=route_dynamics, gather_fn=gather_dynamics) | |
| with np.load(PRESS / f"pressure_{regime}_{split}.npz", allow_pickle=True) as z: | |
| assert np.array_equal(z["uid"], d["uid"]) and np.allclose(z["t"], d["t"]), "pressure pack rows != anchor rows" | |
| assert [str(m) for m in z["margin_names"]] == bp.MARGIN_NAMES and [str(f) for f in z["families"]] == bp.FAMILIES | |
| pm0, pm0v = z["M0"], z["M0valid"] | |
| dm = np.abs(pm0 - d["M0"])[pm0v & d["M0valid"]] | |
| assert np.array_equal(pm0v, d["M0valid"]) and (dm.max() if dm.size else 0.0) < 1e-3, f"M0 recompute differs from the pack ({dm.max()})" | |
| d["E"], d["Evalid"] = z["P"][:, :, EVENT_COLS], z["Pvalid"][:, :, EVENT_COLS] | |
| d["E0"], d["E0valid"], d["E_first"] = z["P0"][:, EVENT_COLS], z["P0valid"][:, EVENT_COLS], z["P_first"][:, EVENT_COLS] | |
| d["Moving"], d["Moving0"] = z["Moving"], z["Moving0"] | |
| d.update(hist_channels=np.array(HIST_CHANNELS), target_channels=np.array(TARGET_CHANNELS), event_families=np.array(EVENT_FAMILIES), | |
| margin_names=np.array(bp.MARGIN_NAMES), ecan_names=np.array(ECAN_NAMES)) | |
| return d, wa | |
| def summary(d): | |
| Yv = d["Yvalid"] | |
| out = {"N": int(len(d["t"])), "hist_valid_frac": float(d["Hvalid"].mean()), "ecan_ok_frac": float(d["ecan_ok"].mean())} | |
| for j, c in enumerate(TARGET_CHANNELS): | |
| v = Yv[:, :, j] | |
| out[f"{c}_valid_frac"] = float(v.mean()) | |
| out[f"{c}_anchor_and_k15_valid_frac"] = float((Yv[:, 14, j] & (d["M0valid"][:, 8] if j < 3 else True) & ((d["M0"][:, 8] > 0.5) if j < 3 else (d["M0"][:, 9] > 0.5))).mean()) | |
| for j, f in enumerate(EVENT_FAMILIES): | |
| sel = (d["E_first"][:, j] != -2) & d["Evalid"][:, :15, j].all(1) | |
| y = (d["E_first"][sel, j] >= 0) & (d["E_first"][sel, j] < 15) | |
| out[f"{f}_onset3s_base_rate"] = float(y.mean()) if sel.any() else float("nan") | |
| out[f"{f}_onset3s_n"] = int(sel.sum()) | |
| return out | |
| def standardise_and_write_side(regimes, splits, force, frozen_stats=None): | |
| """ecan TRAIN stats (both regimes pooled) -> ecan_stats.npz; vfcan256_* / can256_* side streams for every (regime, split). | |
| frozen_stats=(mu, sd, n) (2026-09-17 holdout mode): use these statistics instead — nothing is computed from `splits` and NO ecan_stats.npz is written.""" | |
| extra_meta = {} | |
| if frozen_stats is None: | |
| sm, sq, cnt = np.zeros(N_ECAN), np.zeros(N_ECAN), 0 | |
| for r in regimes: | |
| with np.load(OUT / f"road_dyn_{r}_train.npz") as z: | |
| e, ok = z["ecan"].astype(np.float64), z["ecan_ok"] | |
| sm += e[ok].sum(0) | |
| sq += (e[ok] ** 2).sum(0) | |
| cnt += int(ok.sum()) | |
| mu = sm / max(cnt, 1) | |
| sd = np.sqrt(np.maximum(sq / max(cnt, 1) - mu ** 2, 0.0)) + 1e-6 | |
| np.savez(OUT / "ecan_stats.npz", mu=mu.astype(np.float32), sd=sd.astype(np.float32), n=np.int64(cnt), names=np.array(ECAN_NAMES), | |
| rule="float64 mean / sd over ecan_ok TRAIN rows, both regimes pooled (train/batch.compute_stats v2 rule)") | |
| else: | |
| mu, sd, cnt = frozen_stats | |
| mu, sd = np.asarray(mu, np.float64), np.asarray(sd, np.float64) | |
| assert mu.shape == sd.shape == (N_ECAN,) | |
| extra_meta = dict(ecan_stats_source=str(FROZEN_ECAN_STATS), ecan_stats_sha256=FROZEN_ECAN_STATS_SHA256, ecan_stats_n=np.int64(cnt), frozen_stats=np.bool_(True)) | |
| files = [] | |
| for r in regimes: | |
| for s in splits: | |
| with np.load(OUT / f"road_dyn_{r}_{s}.npz") as z: | |
| e, ok, uid, t = z["ecan"].astype(np.float32), z["ecan_ok"], z["uid"], z["t"] | |
| es = np.clip((e - mu.astype(np.float32)) / sd.astype(np.float32), -8.0, 8.0) * ok[:, None] | |
| vf = PACK_V2 / f"vf256_{r}_{s}.npz" | |
| with np.load(vf) as z: | |
| Z, rok = z["Z"].astype(np.float32), z["row_ok"].astype(bool) | |
| assert int(z["n"]) == len(t) and Z.shape == (len(t), 256), (vf, Z.shape, len(t)) | |
| Zc = np.concatenate([Z[:, :N_PCA_KEEP] * rok[:, None], es], 1).astype(np.float16) | |
| meta = dict(n=np.int64(len(t)), stream="vfcan256", roi="fullframe_causal", version=np.int64(1), source=str(vf), source_sha256=sha256_file(vf), | |
| composition=f"vf256 PCA components 0..{N_PCA_KEEP - 1} (0 where vf256 row_ok False) + {N_ECAN} standardised ecan features (train stats ecan_stats.npz, clip +-8, 0 where ecan_ok False)", | |
| ecan_names=np.array(ECAN_NAMES), **extra_meta) | |
| p1 = OUT / f"vfcan256_{r}_{s}.npz" | |
| p2 = OUT / f"can256_{r}_{s}.npz" | |
| for p in (p1, p2): | |
| if p.exists() and not force: | |
| raise FileExistsError(f"{p} exists (append-only results; pass --force)") | |
| np.savez_compressed(p1, Z=Zc, row_ok=rok | ok, **meta) | |
| Zs = np.concatenate([np.zeros((len(t), N_PCA_KEEP), np.float32), es], 1).astype(np.float16) | |
| meta.update(stream="can256", composition=f"{N_PCA_KEEP} zeros + {N_ECAN} standardised ecan features (sensor-only control; no visual information)", roi="none") | |
| np.savez_compressed(p2, Z=Zs, row_ok=ok, **meta) | |
| files += [str(p1), str(p2)] | |
| print(f"[side] {r}/{s} vfcan256 row_ok={float((rok | ok).mean()):.4f} can256 row_ok={float(ok.mean()):.4f}", flush=True) | |
| return files, {"mu": mu.tolist(), "sd": sd.tolist(), "n": cnt} | |
| def write_scaled_side(regimes, splits, scale, suffix, force): | |
| """POST-HOC sensitivity side stream vfcan256<suffix> (2026-09-16, after the first E0 results): the SAME composition as vfcan256 with the 40 standardised CAN | |
| features multiplied by `scale`. Reason: the E branch applies LayerNorm over the 256-d side vector; unit-sd CAN features sit ~6x lower after that LN in | |
| vfcan256 (mixed with PCA components of sd 0.4-16) than in can256 (216 zeros) — the scale arm separates 'vision hurts' from 'CAN under-scaled'.""" | |
| with np.load(OUT / "ecan_stats.npz") as z: | |
| mu, sd = z["mu"].astype(np.float32), z["sd"].astype(np.float32) | |
| files = [] | |
| for r in regimes: | |
| for s in splits: | |
| with np.load(OUT / f"road_dyn_{r}_{s}.npz") as z: | |
| e, ok, t = z["ecan"].astype(np.float32), z["ecan_ok"], z["t"] | |
| es = np.clip((e - mu) / sd, -8.0, 8.0) * ok[:, None] * float(scale) | |
| vf = PACK_V2 / f"vf256_{r}_{s}.npz" | |
| with np.load(vf) as z: | |
| Z, rok = z["Z"].astype(np.float32), z["row_ok"].astype(bool) | |
| assert int(z["n"]) == len(t) and Z.shape == (len(t), 256) | |
| p = OUT / f"vfcan256{suffix}_{r}_{s}.npz" | |
| if p.exists() and not force: | |
| raise FileExistsError(f"{p} exists (append-only results; pass --force)") | |
| np.savez_compressed(p, Z=np.concatenate([Z[:, :N_PCA_KEEP] * rok[:, None], es], 1).astype(np.float16), row_ok=rok | ok, n=np.int64(len(t)), stream=f"vfcan256{suffix}", | |
| roi="fullframe_causal", version=np.int64(1), source=str(vf), source_sha256=sha256_file(vf), scale_can=np.float32(scale), | |
| composition=f"vf256 PCA components 0..{N_PCA_KEEP - 1} + {N_ECAN} standardised ecan features x {scale} (post-hoc scale sensitivity)", ecan_names=np.array(ECAN_NAMES)) | |
| files.append(str(p)) | |
| print(f"[side] {r}/{s} vfcan256{suffix} (CAN x{scale}) row_ok={float((rok | ok).mean()):.4f}", flush=True) | |
| return files | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--splits", default="train,val,test") | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--no-side", action="store_true", help="skip the vfcan256 / can256 side streams") | |
| ap.add_argument("--scaled-side-only", type=float, default=0.0, help="POST-HOC: write only vfcan256s_* with the CAN block x this factor (needs the road_dyn packs + ecan_stats)") | |
| ap.add_argument("--holdout", action="store_true", help="2026-09-17 WP5: build road_dyn + vfcan256 / can256 for the BATON-holdout split from <CWM3_HOLDOUT_ROOT|results/holdout>/{targets_pressure,pack_v2} " | |
| "into <root>/targets_road_dyn/, standardised with the FROZEN TRAIN results/targets_road_dyn/ecan_stats.npz (loaded + sha256-asserted, never recomputed / rewritten); implies --splits holdout") | |
| a = ap.parse_args() | |
| frozen = None | |
| if a.holdout: | |
| assert not a.scaled_side_only, "--holdout and --scaled-side-only are exclusive" | |
| paths = configure_holdout() | |
| if a.splits == "train,val,test": | |
| a.splits = "holdout" | |
| assert a.splits == "holdout", f"--holdout builds the holdout split only (got --splits {a.splits})" | |
| frozen = load_frozen_ecan_stats() | |
| print(f"[holdout] paths {({k: str(v) for k, v in paths.items()})}; frozen ecan_stats {FROZEN_ECAN_STATS} sha256 {FROZEN_ECAN_STATS_SHA256[:16]} n={frozen[2]:,}", flush=True) | |
| if a.scaled_side_only: | |
| t0 = time.time() | |
| files = write_scaled_side(a.regimes.split(","), a.splits.split(","), a.scaled_side_only, "s", a.force) | |
| write_run_json(OUT, "build_road_dynamics_scaled_side", [Path(__file__)], [OUT / "ecan_stats.npz"], 0, t0, | |
| extra=dict(files_written=files, scale_can=a.scaled_side_only, dms_fields_used=[]), input_listing=False) | |
| print("SCALED_SIDE_DONE", flush=True) | |
| return | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| t_start = time.time() | |
| regimes, splits = a.regimes.split(","), a.splits.split(",") | |
| cache, thr_rows, stats, files = {}, [], {}, [] | |
| for r in regimes: | |
| for s in splits: | |
| p = OUT / f"road_dyn_{r}_{s}.npz" | |
| if p.exists() and not a.force: | |
| raise FileExistsError(f"{p} exists (append-only results; pass --force)") | |
| d, wa = build_split(r, s, cache, thr_rows) | |
| np.savez_compressed(p, **d) | |
| files.append(str(p)) | |
| stats[f"{r}_{s}"] = dict(align=wa, **summary(d)) | |
| print(f"[{r}/{s}] N={len(d['t']):,} align={wa:.1e} -> {p.name} " + " ".join(f"{k}={v:.4f}" for k, v in stats[f"{r}_{s}"].items() if isinstance(v, float) and k != "align"), flush=True) | |
| del d | |
| side_files, ecan_stats = ([], {}) if a.no_side else standardise_and_write_side(regimes, splits, a.force, frozen_stats=frozen) | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| (OUT / f"build_stats_{stamp}.json").write_text(json.dumps(stats, indent=1)) | |
| (OUT / "build_stats_latest.json").write_text(json.dumps(stats, indent=1)) | |
| inputs = [bp.S12 / f"targets_{r}_{s}.npz" for r in regimes for s in splits] + [PRESS / f"pressure_{r}_{s}.npz" for r in regimes for s in splits] + [bp.GRID_SRC] | |
| write_run_json(OUT, "build_road_dynamics_targets", [Path(__file__), Path(bp.__file__)], inputs, 0, t_start, | |
| extra=dict(files_written=files + side_files, hist_channels=HIST_CHANNELS, target_channels=TARGET_CHANNELS, event_families=EVENT_FAMILIES, | |
| ecan_names=ECAN_NAMES, ecan_stats=ecan_stats, n_pca_keep=N_PCA_KEEP, grid_columns_read=[(c, bp.CI[c]) for c in bp.READ_COLS], | |
| grid_columns_never_read=[(bp.COLS[i], i) for i in range(bp.N_FORBIDDEN)], n_routes=len(cache), dms_fields_used=[], thresholds=bp.THR, | |
| holdout=bool(a.holdout), frozen_ecan_stats=(dict(path=str(FROZEN_ECAN_STATS), sha256=FROZEN_ECAN_STATS_SHA256, n=frozen[2]) if a.holdout else None), | |
| out_dir=str(OUT), press_dir=str(PRESS), pack_v2_dir=str(PACK_V2))) | |
| print("ROAD_DYN_DONE", f"{time.time() - t_start:.0f}s", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 28 kB
- Xet hash:
- 079a3be08e98cc1dfc090be81cac52b5c920bde8a4b8e03bd4a57ebf3a44cbce
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.