Buckets:
| """cwm3/data/build_regionpca256.py — `regionpca256` road stream on the frame/20 axis (pack v2): the frontfeat REGION tensor | |
| (16 regions × 1408, s12_wm/cache/frontfeat/<uid>.npz, clip END times t on the 0.5 s frame/20 grid — extract_crop.py uses round(t·20) | |
| frames, so no re-indexing is needed) of the causal window ending at t, flattened to 22,528-d and reduced by a PCA-256 fitted on | |
| MANUAL TRAIN rows only (≤ 200k-row seeded subsample, IncrementalPCA). Per anchor two vectors → [N,512]: | |
| last = PCA(flattened region of the last clip with END in [t−2, t]) (window rule of the v1 pooled streams: ≥ 3 clips in [t−2, t]) | |
| mean = mean of the PCA vectors of the clips with END in [t−8, t] (≤ 8 s history) | |
| Files results/pack_v2/regionpca256_{regime}_{split}.npz {vector f16[N,512], ok bool[N], axis, version=2, pca_file, pca_sha256, targets, | |
| n_last_window, n_hist} (row order = s12 targets = vf256), results/pack_v2/pca_region_256.npz {mu, W[22528,256], evr, n_fit}. | |
| Probe (probes protocol, CPU): ridge (probes.common.ridge_fit_predict, alpha ∈ {1e-3..10}·trace/d selected on VAL AP) on the probe rows | |
| (results/probes/rows_*), targets cur_P0_lane_excursion / cur_P0_ttc (raw_*.npz P0/P0valid), identical rows for regionpca256 and | |
| vf256 (pack_v2) → AP on val (selected alpha) and test (+ device-cluster bootstrap CI); results/tables/regionpca256_probe.md. | |
| Motivation: docs/latent_quality.md §2/§8 — the flattened spatial PCA-256 keeps lane_excursion (0.352 / 0.558 manual / engaged test) | |
| that the pooled vf256 loses (0.136 / 0.204); target here ≥ those vs vf256's 0.160 / 0.218 on the common rows.""" | |
| import argparse, json, sys, time | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1]; EXPLORE = ROOT.parent | |
| sys.path.insert(0, str(ROOT / "common")); sys.path.insert(0, str(ROOT / "probes")) | |
| from runmeta import write_run_json, sha256_file | |
| S12R = EXPLORE / "s12_wm" / "results"; FF = EXPLORE / "s12_wm" / "cache" / "frontfeat"; OUT = ROOT / "results" / "pack_v2"; TABLES = ROOT / "results" / "tables"; PROBES = ROOT / "results" / "probes" | |
| REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test"); D_REG, K = 16 * 1408, 256; P_FAMILIES = ["ttc", "ttc_drop", "cut_in", "lead_braking", "lane_excursion", "badas"] | |
| def drop_cache(path): | |
| """evict a file from the page cache after use (the host's low-memory guard reacts to free memory, not to available).""" | |
| import os | |
| try: | |
| fd = os.open(str(path), os.O_RDONLY); os.posix_fadvise(fd, 0, 0, os.POSIX_FADV_DONTNEED); os.close(fd) | |
| except Exception: pass | |
| def iter_member(path, key, chunk=1024): | |
| """stream a .npy member of an .npz (compressed or not) in row chunks — never materialises the whole array.""" | |
| import zipfile | |
| from numpy.lib import format as npf | |
| with zipfile.ZipFile(path) as zf, zf.open(key + ".npy") as fh: | |
| version = npf.read_magic(fh); shape, fortran, dtype = npf._read_array_header(fh, version); assert not fortran | |
| rowsz = int(np.prod(shape[1:], dtype=np.int64)) * dtype.itemsize | |
| for i in range(0, shape[0], chunk): | |
| k = min(chunk, shape[0] - i); buf = fh.read(k * rowsz); assert len(buf) == k * rowsz | |
| yield i, np.frombuffer(buf, dtype).reshape((k,) + tuple(shape[1:])) | |
| def windows(tf, t): | |
| lo2 = np.searchsorted(tf, t - 2.0 - 1e-6); hi = np.searchsorted(tf, t + 1e-6, side="right"); lo8 = np.searchsorted(tf, t - 8.0 - 1e-6) | |
| ok = (hi - lo2) >= 3 | |
| return lo2, lo8, hi, ok | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--phase", default="all", help="fit|pack|probe|all"); ap.add_argument("--n-fit", type=int, default=200_000); ap.add_argument("--seed", type=int, default=0); ap.add_argument("--limit-routes", type=int, default=0) | |
| a = ap.parse_args(); t_start = time.time(); OUT.mkdir(parents=True, exist_ok=True); TABLES.mkdir(parents=True, exist_ok=True) | |
| logf = open(OUT / "build_regionpca256.log", "a") | |
| def log(m): print(m, flush=True); logf.write(m + "\n"); logf.flush() | |
| tg = {(r, s): np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True) for r in REGIMES for s in SPLITS} | |
| def route_groups(d): | |
| uid = d["uid"]; order = np.argsort(uid, kind="stable"); b = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1]) | |
| return [(uid[order[x]], order[x:y]) for x, y in zip(b[:-1], b[1:])] | |
| report = json.load(open(OUT / "regionpca256_report.json")) if (OUT / "regionpca256_report.json").exists() else {} | |
| pca_p = OUT / "pca_region_256.npz" | |
| # ------------------------------------------------------------------ fit (manual train, last-window flattened region) | |
| if a.phase in ("fit", "all") and not pca_p.exists(): | |
| from sklearn.decomposition import IncrementalPCA | |
| d = tg[("manual", "train")]; t = d["t"].astype(np.float64); groups = route_groups(d) | |
| if a.limit_routes: groups = groups[:a.limit_routes] | |
| ok_all = np.zeros(len(t), bool); last_idx = np.full(len(t), -1, np.int64) | |
| for u, rows in groups: | |
| p = FF / f"{u}.npz" | |
| if not p.exists(): continue | |
| tf = np.load(p)["t"].astype(np.float64); lo2, lo8, hi, ok = windows(tf, t[rows]); ok_all[rows] = ok; last_idx[rows[ok]] = hi[ok] - 1 | |
| rng = np.random.default_rng(a.seed); pos = np.flatnonzero(ok_all); pick = np.zeros(len(t), bool); pick[pos[rng.choice(len(pos), min(a.n_fit, len(pos)), replace=False)]] = True | |
| ipca = IncrementalPCA(n_components=K); buf = []; nbuf = 0; nfit = 0; t0 = time.time() | |
| for u, rows in groups: | |
| rows = rows[pick[rows]] | |
| if not len(rows): continue | |
| idx = np.unique(last_idx[rows]); X = np.zeros((len(idx), D_REG), np.float32) | |
| for c0, blk in iter_member(FF / f"{u}.npz", "region", 1024): | |
| sel = (idx >= c0) & (idx < c0 + len(blk)); X[sel] = blk[idx[sel] - c0].reshape(-1, D_REG).astype(np.float32) | |
| # one vector per (row) — rows sharing a clip contribute once per row (as the probe's row-level fit) | |
| cnt = np.bincount(np.searchsorted(idx, last_idx[rows]), minlength=len(idx)); X = np.repeat(X, cnt, axis=0) | |
| buf.append(X); nbuf += len(X); drop_cache(FF / f"{u}.npz") | |
| if nbuf >= 4096: Xb = np.concatenate(buf); ipca.partial_fit(Xb); nfit += len(Xb); buf, nbuf = [], 0; log(f" [fit] {nfit} rows {time.time()-t0:.0f}s") | |
| if nbuf >= K: Xb = np.concatenate(buf); ipca.partial_fit(Xb); nfit += len(Xb) | |
| mu = ipca.mean_.astype(np.float32); W = ipca.components_.T.astype(np.float32); evr = ipca.explained_variance_ratio_.astype(np.float32) | |
| np.savez(pca_p, mu=mu, W=W, evr=evr, n_fit=np.array(nfit), seed=np.array(a.seed), d_in=np.array(D_REG), k=np.array(K), method=np.array("sklearn IncrementalPCA(256), batches >= 4096 rows, manual TRAIN row_ok rows (last-window flattened 16x1408 region)"), axis=np.array("frame/20 (frontfeat native)")) | |
| report["pca"] = dict(n_fit=int(nfit), evr256=float(evr.sum()), evr_first8=[float(v) for v in evr[:8]], d_in=D_REG, k=K); log(f"[fit] n={nfit} evr(256)={float(evr.sum()):.4f} top8={np.round(evr[:8], 3).tolist()} {time.time()-t0:.0f}s") | |
| (OUT / "regionpca256_report.json").write_text(json.dumps(report, indent=1)) | |
| # ------------------------------------------------------------------ pack | |
| if a.phase in ("pack", "all"): | |
| z = np.load(pca_p); mu, W = z["mu"].astype(np.float32), z["W"].astype(np.float32); report.setdefault("coverage", {}) | |
| for (r, s), d in tg.items(): | |
| out = OUT / f"regionpca256_{r}_{s}.npz" | |
| if out.exists() and f"{r}_{s}" in report["coverage"]: log(f"[pack] {out.name} exists — skipped"); continue | |
| t = d["t"].astype(np.float64); N = len(t); V = np.zeros((N, 2 * K), np.float16); ok_all = np.zeros(N, bool); nl = np.zeros(N, np.int16); nh = np.zeros(N, np.int16); t0 = time.time(); groups = route_groups(d) | |
| if a.limit_routes: groups = groups[:a.limit_routes] | |
| for u, rows in groups: | |
| p = FF / f"{u}.npz" | |
| if not p.exists(): continue | |
| tf = np.load(p)["t"].astype(np.float64); lo2, lo8, hi, ok = windows(tf, t[rows]) | |
| if not ok.any(): continue | |
| proj = np.zeros((len(tf), K), np.float32) | |
| for c0, blk in iter_member(p, "region", 1024): proj[c0:c0 + len(blk)] = (blk.reshape(-1, D_REG).astype(np.float32) - mu) @ W # streamed: <= 1024 x 22528 f32 (~92 MB) at a time | |
| cs = np.concatenate([np.zeros((1, K), np.float64), np.cumsum(proj.astype(np.float64), 0)], 0) | |
| rr = rows[ok]; h = hi[ok]; l8 = lo8[ok] | |
| V[rr, :K] = proj[h - 1].astype(np.float16); V[rr, K:] = ((cs[h] - cs[l8]) / (h - l8)[:, None]).astype(np.float16) | |
| ok_all[rr] = True; nl[rr] = (hi - lo2)[ok]; nh[rr] = (h - l8); drop_cache(p) | |
| np.savez(out, vector=V, ok=ok_all, n_last_window=nl, n_hist=nh, axis=np.array("frame/20 (frontfeat native t = clip END)"), version=np.array(2), pca_file=np.array(pca_p.name), pca_sha256=np.array(sha256_file(pca_p)), | |
| targets=np.array(str(S12R / f"targets_{r}_{s}.npz")), layout=np.array("[:256] last = PCA(flattened 16x1408 region of the last clip with END in [t-2,t]); [256:] mean of the PCA vectors of clips with END in [t-8,t]"), window_rule=np.array(">= 3 clips with END in [t-2, t] (v1 pooled-stream rule)")) | |
| v1 = np.load(OUT / f"vf256_{r}_{s}.npz")["row_ok"] | |
| report["coverage"][f"{r}_{s}"] = dict(N=int(N), ok=int(ok_all.sum()), ok_frac=float(ok_all.mean()), vf256_v2_ok=int(v1.sum()), both_ok=int((ok_all & v1).sum()), n_hist_mean=float(nh[ok_all].mean()) if ok_all.any() else None) | |
| log(f"[pack] {r}/{s}: ok {ok_all.sum()}/{N} ({ok_all.mean():.4f}); vf256 ok {v1.sum()}; hist clips mean {report['coverage'][f'{r}_{s}']['n_hist_mean']:.1f} {time.time()-t0:.0f}s") | |
| (OUT / "regionpca256_report.json").write_text(json.dumps(report, indent=1)) | |
| # ------------------------------------------------------------------ probe (CPU ridge, probe rows, identical rows for both reps) | |
| if a.phase in ("probe", "all"): | |
| import torch; from sklearn.metrics import average_precision_score as AP | |
| from common import ridge_fit_predict, RIDGE_ALPHAS # probes/common.py | |
| dev = torch.device("cpu"); torch.set_num_threads(6); report["probe"] = {} | |
| for r in REGIMES: | |
| R = {s: np.load(PROBES / f"rows_{r}_{s}.npz")["rows"].astype(np.int64) for s in SPLITS}; RAW = {s: np.load(PROBES / f"raw_{r}_{s}.npz", allow_pickle=True) for s in SPLITS} | |
| reps = {"regionpca256": {s: np.load(OUT / f"regionpca256_{r}_{s}.npz") for s in SPLITS}, "vf256_v2": {s: np.load(OUT / f"vf256_{r}_{s}.npz") for s in SPLITS}} | |
| okm = {s: reps["regionpca256"][s]["ok"][R[s]] & reps["vf256_v2"][s]["row_ok"][R[s]] for s in SPLITS} | |
| X = {n: {s: (reps[n][s]["vector"] if n == "regionpca256" else reps[n][s]["Z"])[R[s]][okm[s]].astype(np.float32) for s in SPLITS} for n in reps} | |
| fam = {"lane_excursion": P_FAMILIES.index("lane_excursion"), "ttc": P_FAMILIES.index("ttc")} | |
| Y = {s: np.stack([RAW[s]["P0"][:, j] for j in fam.values()], 1).astype(np.float32)[okm[s]] for s in SPLITS}; M = {s: np.stack([RAW[s]["P0valid"][:, j] for j in fam.values()], 1).astype(bool)[okm[s]] for s in SPLITS} | |
| dong = {s: RAW[s]["dongle"][okm[s]] for s in SPLITS}; rep_r = dict(rows={s: int(okm[s].sum()) for s in SPLITS}, targets={}) | |
| for n in reps: | |
| mu_, sd_ = X[n]["train"].mean(0), X[n]["train"].std(0) + 1e-6; Xs = {s: (X[n][s] - mu_) / sd_ for s in SPLITS} | |
| sva, ste = ridge_fit_predict(Xs["train"], Y["train"], M["train"], Xs["val"], Xs["test"], alphas=RIDGE_ALPHAS, device=dev) | |
| for j, tn in enumerate(fam): | |
| mv, mt = M["val"][:, j], M["test"][:, j] | |
| aps_v = [AP(Y["val"][mv, j], sva[j, a_][mv]) if mv.sum() and Y["val"][mv, j].sum() else np.nan for a_ in range(len(RIDGE_ALPHAS))] | |
| a_best = int(np.nanargmax(aps_v)); ap_te = float(AP(Y["test"][mt, j], ste[j, a_best][mt])) if mt.sum() and Y["test"][mt, j].sum() else None | |
| rng = np.random.default_rng(0); dv = dong["test"][mt]; yv = Y["test"][mt, j]; sv = ste[j, a_best][mt]; ud = np.unique(dv); boots = [] | |
| for _ in range(200): | |
| pick = rng.choice(ud, len(ud), replace=True); idx = np.concatenate([np.flatnonzero(dv == u_) for u_ in pick]) | |
| if yv[idx].sum() > 0: boots.append(AP(yv[idx], sv[idx])) | |
| rep_r["targets"].setdefault(f"cur_P0_{tn}", {})[n] = dict(val_ap=float(aps_v[a_best]), alpha=RIDGE_ALPHAS[a_best], test_ap=ap_te, test_ci95=[float(np.percentile(boots, 2.5)), float(np.percentile(boots, 97.5))] if boots else None, | |
| n_val=int(mv.sum()), pos_val=int(Y["val"][mv, j].sum()), n_test=int(mt.sum()), pos_test=int(Y["test"][mt, j].sum()), d=int(X[n]["train"].shape[1])) | |
| log(f"[probe] {r} {n}: " + " ".join(f"{tn} val {rep_r['targets'][f'cur_P0_{tn}'][n]['val_ap']:.3f} test {rep_r['targets'][f'cur_P0_{tn}'][n]['test_ap']:.3f}" for tn in fam)) | |
| report["probe"][r] = rep_r; (OUT / "regionpca256_report.json").write_text(json.dumps(report, indent=1)) | |
| ref = {"manual": {"lane_excursion": (0.352, 0.136, 0.160), "ttc": (0.286, 0.313, 0.293)}, "engaged": {"lane_excursion": (0.558, 0.204, 0.218), "ttc": (None, None, 0.267)}} | |
| L = ["# regionpca256 probe — flattened 16×1408 region PCA-256 (last ⊕ 8 s mean, 512-d) vs vf256 (pack v2), identical probe rows", "", f"Generated {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())} by cwm3/data/build_regionpca256.py; ridge = probes.common.ridge_fit_predict (alpha ∈ {{1e-3..10}}·trace/d chosen on VAL AP), PR-AUC; test CI = device-cluster bootstrap (200). Reference = docs/latent_quality.md §2 (region_subset test rows: region_pca256 / vf256) and the common-rows vf256.", "", | |
| "| regime | target | rows val / test (pos) | regionpca256 val AP | regionpca256 test AP [CI] | vf256_v2 val AP | vf256_v2 test AP [CI] | ref region_pca256 test | ref vf256 (subset / common) |", "|---|---|---|---|---|---|---|---|---|"] | |
| for r in REGIMES: | |
| for tn in ("lane_excursion", "ttc"): | |
| q = report["probe"][r]["targets"][f"cur_P0_{tn}"]; g, v = q["regionpca256"], q["vf256_v2"]; rf = ref[r][tn] | |
| L.append(f"| {r} | cur_P0_{tn} | {g['n_val']:,} / {g['n_test']:,} ({g['pos_val']} / {g['pos_test']}) | {g['val_ap']:.3f} | {g['test_ap']:.3f} [{g['test_ci95'][0]:.3f}, {g['test_ci95'][1]:.3f}] | {v['val_ap']:.3f} | {v['test_ap']:.3f} [{v['test_ci95'][0]:.3f}, {v['test_ci95'][1]:.3f}] | {rf[0]} | {rf[1]} / {rf[2]} |") | |
| L += ["", f"PCA: {json.dumps(report.get('pca'))}", "", "Coverage: " + json.dumps(report.get("coverage"))] | |
| (TABLES / "regionpca256_probe.md").write_text("\n".join(L) + "\n") | |
| write_run_json(OUT, "build_regionpca256", [Path(__file__)], [S12R / f"targets_{r}_{s}.npz" for r in REGIMES for s in SPLITS], a.seed, t_start, extra=dict(summary={k: v for k, v in report.items() if k != "probe"}, probe=report.get("probe")), input_listing=False) | |
| log("REGIONPCA256_DONE") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 15.5 kB
- Xet hash:
- 4fbbf8a431dedb36b17bf49ee6e04c4fd2a51bd9df3c9fcc49ec03e4164dac65
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.