Buckets:
| """cwm3/data/build_pack_v2.py — pooled V-JEPA2 streams (Vf/Vc → vf256/vc256) and future-latent targets F rebuilt on the | |
| frame/20 axis as pack VERSION 2 (results/pack_v2/), leaving v1 (results/pack, s12_wm/results/{vis,future}_*) untouched. | |
| FINDING (probes/audit_features.py, audit A): the s5 dense library (s5_features/s5_extract_v4.py) labels each clip with a | |
| CONTAINER-fps time: `fps = vr.get_avg_fps() or 20.0`, `ts = arange(2.0, n_frames/fps, 0.5)`, frames at round((ts+rel)·fps). | |
| CAN, pose and the frontfeat token bank are on frame/20. When avg fps != 20 (road camera: 442/487 clip-map leaves) the s5 label | |
| drifts from the true frame/20 time by up to ~18 s over a route (lag > 0 = stale but causal). | |
| FIX: per route and camera, the container fps used by s5 is recovered EXACTLY (cv2 CAP_PROP_FPS == decord get_avg_fps() == | |
| ffprobe avg_frame_rate, verified 2026-09-10 on both cameras) — locally with cv2, remotely with ffprobe over the SSH ControlMaster, | |
| cross-checked against cwm2/private/clip_map_full.json cab/road fps where available — and every clip is re-indexed: | |
| frame_end = round(ts · fps_container) (the END frame s5 actually read, rel = 0), t_true = frame_end / 20. | |
| Then the v1 rules are re-applied on t_true with the v1 code paths reproduced bit-for-bit: | |
| * pooled Vf / Vc (s7_visual_fusion/featpool.FeatStore.pool): clips with END in [t-2, t] (searchsorted with 1e-6), >= 3 clips, | |
| [mean, last, last-first] of the float32 clip features -> f16 (4224); row_ok = both cameras pooled (materialize_s12.py); | |
| * PCA-256 per stream re-fitted on MANUAL TRAIN only (200k seeded rows, exact covariance eigh, as cwm3/data/pca_streams.py) | |
| -> results/pack_v2/pca_{Vf,Vc}_256.npz ; packs results/pack_v2/{vf256,vc256}_{regime}_{split}.npz with the v1 keys | |
| {Z f16[N,256], row_ok, pca_file, pca_sha256, source, targets, n, stream, roi} (+ axis, fps_table); | |
| * F(t+k), k=1..5 (s12_wm/materialize_oracle_future.mat_future): s5 front feature at the last t_true <= t+k with | |
| |t_true-(t+k)| <= 0.6, projected by a future PCA re-fitted on TRAIN routes only (manual ∪ engaged train uids; v1 fitted on | |
| all routes) -> results/pack_v2/future_{regime}_{split}.npz {F f16[N,5,256], ok bool[N,5]} (v1 lives in | |
| s12_wm/results/future_{regime}_{split}.npz and is read by cwm3/data/cwm3_dataset.py:126-128 as F / Fok). | |
| Row order = the s12 anchor grid (uid/t asserted equal to targets_{regime}_{split}.npz = v1 order). | |
| VERIFICATION (results/tables/pack_v2_alignment.md, results/pack_v2/alignment.json): per split/regime and camera the analytic lag | |
| of the clip each anchor uses, lag = s5 label − true time (v1) vs 0 by construction (v2), the content staleness t − t_true(last | |
| clip) (v1 can be negative = future content), |lag| shares (<= 0.25 s, >= 1 s, >= 3 s, max), routes changed, fraction of anchors | |
| whose v2 vector equals v1 (v2 pooled vector projected with the v1 PCA basis == v1 pack, byte-identical for fps == 20 routes), | |
| F: fraction of v1 F(t+k) targets whose content END is <= t ("the past") and fraction of identical clip picks; plus the | |
| audit-style EMPIRICAL check on the probe rows (results/probes/rows_*, feats_* raw1408 = frontfeat bank last slot on frame/20): | |
| best-match lag of the v2-labelled s5 clips. Acceptance: |lag| <= 0.25 s for >= 99.5 % of resolvable anchors. | |
| CPU only (OMP_NUM_THREADS <= 8).""" | |
| import argparse, json, os, subprocess, sys, time, fractions | |
| from collections import defaultdict | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "common")) | |
| from runmeta import write_run_json, sha256_file | |
| EXPLORE = ROOT.parent; S12R = EXPLORE / "s12_wm" / "results" | |
| S5_DIR = {"front": EXPLORE / "s5_features" / "vjepa2g384" / "front", "cabin": EXPLORE / "s5_features" / "vjepa2g384" / "cabin"} | |
| CAM_FILE = {"front": "qcamera.mp4", "cabin": "dcamera.mp4"}; STREAM_CAM = {"Vf": "front", "Vc": "cabin"}; TAG = {"Vf": "vf256", "Vc": "vc256"} | |
| OUT = ROOT / "results" / "pack_v2"; TABLES = ROOT / "results" / "tables"; PACK1 = ROOT / "results" / "pack"; PCA1 = ROOT / "results" / "pca"; PROBES = ROOT / "results" / "probes" | |
| LEAVES = ROOT / "results" / "baton_leaves.json"; CLIP_MAP = EXPLORE / "cwm2" / "private" / "clip_map_full.json"; LROOT = Path("/home/USER/BATON") | |
| REMOTE = "USER@REMOTE_HOST"; RROOT = "/data/backup/BATON"; SSH = ["ssh", "-S", "/tmp/baton_bak_ctl", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"] | |
| REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test"); D_IN, K, TAUS = 4224, 256, (1.0, 2.0, 3.0, 4.0, 5.0) | |
| def leaf_of(uid): p = uid.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:]) | |
| def s12_uids(): | |
| return sorted(set().union(*[set(np.unique(np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True)["uid"]).tolist()) for r in REGIMES for s in SPLITS])) | |
| def train_uids(): | |
| return set().union(*[set(np.unique(np.load(S12R / f"targets_{r}_train.npz", allow_pickle=True)["uid"]).tolist()) for r in REGIMES]) | |
| # ------------------------------------------------------------------------------------------------ phase A: container fps table | |
| def fps_table(uids, log): | |
| p = OUT / "container_fps.json" | |
| tab = json.load(open(p)) if p.exists() else {} | |
| lv = json.load(open(LEAVES))["leaves"]; cm = json.load(open(CLIP_MAP)); cmf = {} | |
| for v in cm.values(): cmf[v["uid"]] = {"front": (v["road"]["fps"], v["road"]["n_frames"]), "cabin": (v["cab"]["fps"], v["cab"]["n_frames"])} | |
| import cv2 | |
| n_cv = n_ff = n_cm = 0; dif = [] | |
| for u in uids: | |
| if u in tab and all(c in tab[u] for c in CAM_FILE): continue # recorded (fps may be None = video absent) | |
| leaf = leaf_of(u); info = lv.get(leaf) or {}; rec = tab.get(u, {}) | |
| for cam, fn in CAM_FILE.items(): | |
| if cam in rec and rec[cam].get("fps"): continue | |
| fps = nfr = None; src = None | |
| if info.get("local_status") == "local_identical_size" and (LROOT / leaf / fn).exists(): | |
| cap = cv2.VideoCapture(str(LROOT / leaf / fn)); fps, nfr = float(cap.get(cv2.CAP_PROP_FPS)), int(cap.get(cv2.CAP_PROP_FRAME_COUNT)); cap.release(); src = "cv2_local"; n_cv += 1 | |
| elif info: | |
| r = subprocess.run(SSH + [REMOTE, f"ffprobe -v error -select_streams v:0 -show_entries stream=avg_frame_rate,nb_frames -of json '{RROOT}/{leaf}/{fn}'"], capture_output=True, text=True, timeout=300) | |
| try: | |
| j = json.loads(r.stdout)["streams"][0]; fps = float(fractions.Fraction(j["avg_frame_rate"])); nfr = int(j.get("nb_frames", 0)); src = "ffprobe_remote"; n_ff += 1 | |
| except Exception as e: log(f" ffprobe failed {u} {cam}: {r.stderr[:120]} {e}") | |
| if fps is None and u in cmf: fps, nfr = cmf[u][cam]; src = "clip_map"; n_cm += 1 | |
| if fps is not None and u in cmf and src != "clip_map": dif.append(abs(fps - cmf[u][cam][0])) | |
| rec[cam] = {"fps": fps, "n_frames": nfr, "source": src} | |
| tab[u] = rec | |
| if (n_cv + n_ff) % 100 == 0: p.write_text(json.dumps(tab, indent=1)) | |
| p.write_text(json.dumps(tab, indent=1)) | |
| miss = [u for u in uids if any(tab[u][c]["fps"] is None for c in CAM_FILE)] | |
| st = dict(n_routes=len(uids), probed_cv2=n_cv, probed_ffprobe_remote=n_ff, from_clip_map=n_cm, n_missing_fps=len(miss), clip_map_crosscheck_n=len(dif), clip_map_crosscheck_max_abs_dfps=(max(dif) if dif else None), | |
| fps_eq_20={c: int(sum(1 for u in uids if tab[u][c]["fps"] == 20.0)) for c in CAM_FILE}, fps_ne_20={c: int(sum(1 for u in uids if tab[u][c]["fps"] not in (None, 20.0))) for c in CAM_FILE}) | |
| log(f"[fps] {st}"); return tab, st | |
| # ------------------------------------------------------------------------------------------------ phase B/C helpers | |
| class S5: | |
| """s5 route with re-indexed true times; LRU.""" | |
| def __init__(self, fps_tab): self.tab = fps_tab; self.cache = {} | |
| def get(self, uid, cam): | |
| key = (uid, cam) | |
| if key not in self.cache: | |
| if len(self.cache) > 12: self.cache.pop(next(iter(self.cache))) | |
| p = S5_DIR[cam] / f"{uid}.npz"; fps = self.tab.get(uid, {}).get(cam, {}).get("fps") | |
| if not p.exists() or fps is None: self.cache[key] = None | |
| else: | |
| z = np.load(p); ts = z["timestamps"].astype(np.float64); F = z["features"] | |
| nfr = self.tab[uid][cam].get("n_frames") or 0 | |
| fe = np.round(ts * fps).astype(np.int64) | |
| if nfr: fe = np.clip(fe, 0, nfr - 1) | |
| self.cache[key] = dict(ts=ts, tt=fe / 20.0, F=F, fps=fps) | |
| return self.cache[key] | |
| def pool_rows(tt, F, t): | |
| """v1 pooling rule on axis tt for anchors t -> (P f16[n,4224], ok[n], last_idx[n]) (exact featpool arithmetic).""" | |
| lo = np.searchsorted(tt, t - 2.0 - 1e-6); hi = np.searchsorted(tt, t + 1e-6, side="right"); ok = (hi - lo) >= 3 | |
| P = np.zeros((len(t), 3 * F.shape[1]), np.float16) # 4224 for the 1408-d s5 features | |
| for i in np.flatnonzero(ok): | |
| W = F[lo[i]:hi[i]].astype(np.float32); P[i] = np.concatenate([W.mean(0), W[-1], W[-1] - W[0]]).astype(np.float16) | |
| return P, ok, hi - 1 | |
| def fit_pca(X64_sum, X64_gram, n): | |
| mu = X64_sum / n; cov = X64_gram / n - np.outer(mu, mu); evals, evecs = np.linalg.eigh(cov); o = np.argsort(evals)[::-1]; evals, evecs = evals[o], evecs[:, o] | |
| return mu.astype(np.float32), evecs[:, :K].astype(np.float32), (evals[:K] / evals.sum()).astype(np.float32) | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--phase", default="all", help="fps|streams|future|verify|all"); ap.add_argument("--n-fit", type=int, default=200_000); ap.add_argument("--seed", 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_pack_v2.log", "a") | |
| def log(m): print(m, flush=True); logf.write(m + "\n"); logf.flush() | |
| uids = s12_uids(); tr_uids = train_uids() | |
| tab, fps_st = fps_table(uids, log) | |
| if a.phase == "fps": return | |
| store = S5(tab) | |
| report = json.load(open(OUT / "alignment.json")) if (OUT / "alignment.json").exists() else dict(streams={}, future={}, empirical={}) | |
| report["fps"] = fps_st if fps_st.get("probed_cv2") else report.get("fps", fps_st) | |
| tg = {(r, s): np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True) for r in REGIMES for s in SPLITS} | |
| # ---------------------------------------------------------------- streams: pooled on t_true, PCA fit on manual train, packs, lag stats | |
| if a.phase in ("streams", "all"): | |
| pca1 = {st: np.load(PCA1 / f"pca_{st}_256.npz") for st in STREAM_CAM} | |
| pca2 = {} | |
| if all((OUT / f"pca_{st}_256.npz").exists() for st in STREAM_CAM): | |
| for st in STREAM_CAM: | |
| z = np.load(OUT / f"pca_{st}_256.npz"); pca2[st] = (z["mu"], z["W"], z["evr"], OUT / f"pca_{st}_256.npz") | |
| log("[fit v2] reusing existing pca_{Vf,Vc}_256.npz") | |
| # pass 1: row_ok_v2 + fit sample for manual/train | |
| d = tg[("manual", "train")]; uid, t = d["uid"], d["t"].astype(np.float64); N = len(t) | |
| order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1]) | |
| ok2 = np.zeros(N, bool) | |
| for a_, b_ in zip(bounds[:-1], bounds[1:]): | |
| rows = order[a_:b_]; u = uid[rows[0]]; oks = [] | |
| for cam in CAM_FILE: | |
| g = store.get(u, cam) | |
| if g is None: oks.append(np.zeros(len(rows), bool)); continue | |
| lo = np.searchsorted(g["tt"], t[rows] - 2.0 - 1e-6); hi = np.searchsorted(g["tt"], t[rows] + 1e-6, side="right"); oks.append((hi - lo) >= 3) | |
| ok2[rows] = oks[0] & oks[1] | |
| rng = np.random.default_rng(a.seed); pos = np.flatnonzero(ok2); pick = np.zeros(N, bool); pick[pos[rng.choice(len(pos), min(a.n_fit, len(pos)), replace=False)]] = True | |
| S_ = {st: np.zeros(D_IN, np.float64) for st in STREAM_CAM}; G_ = {st: np.zeros((D_IN, D_IN), np.float64) for st in STREAM_CAM}; nfit = 0 | |
| for a_, b_ in zip(bounds[:-1], bounds[1:]): | |
| if pca2: break | |
| rows = order[a_:b_]; rows = rows[pick[rows]] | |
| if not len(rows): continue | |
| u = uid[rows[0]] | |
| for st, cam in STREAM_CAM.items(): | |
| g = store.get(u, cam) | |
| for c0 in range(0, len(rows), 8192): | |
| P, ok, _ = pool_rows(g["tt"], g["F"], t[rows[c0:c0 + 8192]]); x = P[ok].astype(np.float64); S_[st] += x.sum(0); G_[st] += x.T @ x; del P, x | |
| nfit += len(rows) | |
| for st in STREAM_CAM: | |
| if st in pca2: continue | |
| mu, W, evr = fit_pca(S_[st], G_[st], nfit); pth = OUT / f"pca_{st}_256.npz" | |
| np.savez(pth, mu=mu, W=W, evr=evr, n_fit=np.array(nfit), seed=np.array(a.seed), stream=np.array(st), axis=np.array("frame/20 (v2: s5 clips re-indexed by container fps)"), created_utc=np.array(time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))) | |
| pca2[st] = (mu, W, evr, pth); log(f"[fit v2] {st}: n={nfit} evr(256)={float(evr.sum()):.4f} top8={np.round(evr[:8], 3).tolist()} (v1 evr {float(pca1[st]['evr'].sum()):.4f})") | |
| report["streams"][f"pca_{st}"] = dict(n_fit=int(nfit), evr256_v2=float(evr.sum()), evr256_v1=float(pca1[st]["evr"].sum()), evr_first8_v2=[float(v) for v in evr[:8]]) | |
| # pass 2: packs + lag statistics for every split | |
| for (r, s), d in tg.items(): | |
| if f"{r}_{s}" in report["streams"] and all((OUT / f"{TAG[st]}_{r}_{s}.npz").exists() for st in STREAM_CAM): | |
| log(f"[pack v2] {r}/{s}: exists — skipped"); continue | |
| uid, t = d["uid"], d["t"].astype(np.float64); N = len(t); order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1]) | |
| Z = {st: np.zeros((N, K), np.float16) for st in STREAM_CAM}; Z1b = {st: np.zeros((N, K), np.float16) for st in STREAM_CAM}; okc = {c: np.zeros(N, bool) for c in CAM_FILE} | |
| lag1 = {c: np.full(N, np.nan, np.float32) for c in CAM_FILE}; stale1 = {c: np.full(N, np.nan, np.float32) for c in CAM_FILE}; stale2 = {c: np.full(N, np.nan, np.float32) for c in CAM_FILE}; same_pick = {c: np.zeros(N, bool) for c in CAM_FILE} | |
| for a_, b_ in zip(bounds[:-1], bounds[1:]): | |
| rows_all = order[a_:b_]; u = uid[rows_all[0]] | |
| gs = {cam: store.get(u, cam) for cam in CAM_FILE} | |
| if any(g is None for g in gs.values()): continue # row_ok stays False (as v1: both cameras required) | |
| for c0 in range(0, len(rows_all), 8192): # bounded memory: <= 8192 anchors x 4224 per stream | |
| rows = rows_all[c0:c0 + 8192]; P = {} | |
| for st, cam in STREAM_CAM.items(): | |
| g = gs[cam] | |
| P[st], ok, last2 = pool_rows(g["tt"], g["F"], t[rows]); okc[cam][rows] = ok | |
| hi1 = np.searchsorted(g["ts"], t[rows] + 1e-6, side="right") - 1; v1ok = hi1 >= 0; h1 = np.clip(hi1, 0, None) | |
| lag1[cam][rows[v1ok]] = (g["ts"][h1] - g["tt"][h1])[v1ok]; stale1[cam][rows[v1ok]] = (t[rows] - g["tt"][h1])[v1ok] | |
| stale2[cam][rows[ok]] = (t[rows] - g["tt"][np.clip(last2, 0, None)])[ok]; same_pick[cam][rows] = (hi1 == last2) & ok & v1ok | |
| both = okc["front"][rows] & okc["cabin"][rows] | |
| for st in P: | |
| x = P[st][both].astype(np.float32); Z[st][rows[both]] = ((x - pca2[st][0]) @ pca2[st][1]).astype(np.float16) | |
| Z1b[st][rows[both]] = ((x - pca1[st]["mu"].astype(np.float32)) @ pca1[st]["W"].astype(np.float32)).astype(np.float16) | |
| del P, x | |
| row_ok = okc["front"] & okc["cabin"] | |
| for st in STREAM_CAM: | |
| for c in ("front", "cabin"): pass | |
| pth = pca2[st][3] | |
| np.savez(OUT / f"{TAG[st]}_{r}_{s}.npz", Z=Z[st], row_ok=row_ok, pca_file=np.array(pth.name), pca_sha256=np.array(sha256_file(pth)), source=np.array(f"s5 {STREAM_CAM[st]} library re-indexed to frame/20 (container fps table {OUT / 'container_fps.json'})"), | |
| targets=np.array(str(S12R / f"targets_{r}_{s}.npz")), n=np.array(N), stream=np.array(st), roi=np.array("fullframe_causal"), axis=np.array("frame/20 (v2)"), version=np.array(2)) | |
| # comparison with v1 pack | |
| st_rep = dict(N=int(N), row_ok_v2=int(row_ok.sum()), row_ok_v1=None, identical_to_v1_with_v1_basis={}, lag_v1={}, lag_v2={}, stale_v1={}, stale_v2={}, routes={}) | |
| for st in STREAM_CAM: | |
| z1 = np.load(PACK1 / f"{TAG[st]}_{r}_{s}.npz"); ok1 = z1["row_ok"].astype(bool); st_rep["row_ok_v1"] = int(ok1.sum()); both = ok1 & row_ok | |
| eq = np.all(z1["Z"][both] == Z1b[st][both], axis=1); st_rep["identical_to_v1_with_v1_basis"][st] = dict(n_both=int(both.sum()), frac_identical=float(eq.mean()) if both.any() else None) | |
| cam = STREAM_CAM[st]; l1 = np.abs(lag1[cam][row_ok]); l1 = l1[np.isfinite(l1)]; s1 = stale1[cam][row_ok]; s1 = s1[np.isfinite(s1)]; s2 = stale2[cam][row_ok]; s2 = s2[np.isfinite(s2)] | |
| st_rep["lag_v1"][cam] = dict(n=int(l1.size), le_0p25=float((l1 <= 0.25).mean()), ge_1s=float((l1 >= 1).mean()), ge_3s=float((l1 >= 3).mean()), max=float(l1.max()) if l1.size else None) | |
| st_rep["lag_v2"][cam] = dict(n=int(row_ok.sum()), le_0p25=1.0, ge_1s=0.0, ge_3s=0.0, max=0.0, note="label == true frame/20 time by construction") | |
| st_rep["stale_v1"][cam] = dict(future_content_frac=float((s1 < -1e-6).mean()), ge_2p5s=float((s1 >= 2.5).mean()), max=float(s1.max()), min=float(s1.min())) | |
| st_rep["stale_v2"][cam] = dict(future_content_frac=float((s2 < -1e-6).mean()), max=float(s2.max()), min=float(s2.min())) | |
| u_ok = uid[row_ok]; drift = np.abs(lag1[cam][row_ok]) >= 0.25 | |
| st_rep["routes"][cam] = dict(routes=int(len(np.unique(u_ok))), routes_changed=int(len(np.unique(u_ok[np.isfinite(lag1[cam][row_ok]) & drift]))), same_clip_pick_frac=float(same_pick[cam][row_ok].mean())) | |
| report["streams"][f"{r}_{s}"] = st_rep | |
| log(f"[pack v2] {r}/{s}: row_ok v2 {row_ok.sum()} v1 {st_rep['row_ok_v1']}; identical(v1 basis) Vf {st_rep['identical_to_v1_with_v1_basis']['Vf']['frac_identical']:.4f} Vc {st_rep['identical_to_v1_with_v1_basis']['Vc']['frac_identical']:.4f}; " | |
| f"lag_v1 front |lag|>=1s {st_rep['lag_v1']['front']['ge_1s']:.4f} >=3s {st_rep['lag_v1']['front']['ge_3s']:.4f} max {st_rep['lag_v1']['front']['max']:.2f}; cabin >=1s {st_rep['lag_v1']['cabin']['ge_1s']:.4f}; routes changed front {st_rep['routes']['front']['routes_changed']}/{st_rep['routes']['front']['routes']}") | |
| (OUT / "alignment.json").write_text(json.dumps(report, indent=1)) | |
| # ---------------------------------------------------------------- future F v2 | |
| if a.phase in ("future", "all"): | |
| rng = np.random.default_rng(a.seed); files = sorted(S5_DIR["front"].glob("*.npz")); files = [f for f in files if f.stem in tr_uids]; rng.shuffle(files); vecs = [] | |
| for f in files: | |
| z = np.load(f); F = z["features"]; vecs.append(F[rng.choice(len(F), min(400, len(F)), replace=False)].astype(np.float32)) | |
| if sum(len(v) for v in vecs) >= 100_000: break | |
| X = np.concatenate(vecs)[:100_000]; mu = X.mean(0); Xc = X - mu; sub = Xc[rng.choice(len(X), 50_000, replace=False)] | |
| _, sv, Vt = np.linalg.svd(sub, full_matrices=False); W = Vt[:K].T.astype(np.float32); evr = (sv[:K] ** 2 / (sv ** 2).sum()).astype(np.float32) | |
| pf = OUT / "future_pca_v2.npz" | |
| if not pf.exists(): np.savez(pf, mu=mu.astype(np.float32), W=W, evr=evr, n_fit=np.array(len(X)), n_svd=np.array(50_000), seed=np.array(a.seed), train_only=np.array(True), n_files=np.array(len(files)), axis=np.array("frame/20 (v2)")) | |
| report["future"]["pca"] = dict(n_fit=int(len(X)), n_train_routes_sampled=len(files), evr256=float(evr.sum()), evr_first8=[float(v) for v in evr[:8]], train_only=True, v1_fit="all routes (s12_wm/materialize_oracle_future.fit_future_pca)") | |
| log(f"[future pca v2] train-only files {len(files)} n={len(X)} evr(256 of 1408)={float(evr.sum()):.4f}") | |
| for (r, s), d in tg.items(): | |
| if f"{r}_{s}" in report["future"] and (OUT / f"future_{r}_{s}.npz").exists(): log(f"[future v2] {r}/{s}: exists — skipped"); continue | |
| uid, t = d["uid"], d["t"].astype(np.float64); N = len(t); F2 = np.zeros((N, 5, K), np.float16); ok2 = np.zeros((N, 5), bool); past1 = np.zeros((N, 5), bool); ok1c = np.zeros((N, 5), bool); same = np.zeros((N, 5), bool) | |
| order = np.argsort(uid, kind="stable"); bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1]) | |
| for a_, b_ in zip(bounds[:-1], bounds[1:]): | |
| rows = order[a_:b_]; u = uid[rows[0]]; g = store.get(u, "front") | |
| if g is None: continue | |
| proj = None | |
| for j, tau in enumerate(TAUS): | |
| tk = t[rows] + tau | |
| jj = np.searchsorted(g["tt"], tk + 1e-6) - 1; okk = (jj >= 0) & (np.abs(g["tt"][np.clip(jj, 0, None)] - tk) <= 0.6); ok2[rows, j] = okk | |
| j1 = np.searchsorted(g["ts"], tk + 1e-6) - 1; ok1 = (j1 >= 0) & (np.abs(g["ts"][np.clip(j1, 0, None)] - tk) <= 0.6); ok1c[rows, j] = ok1 | |
| past1[rows, j] = ok1 & (g["tt"][np.clip(j1, 0, None)] <= t[rows] + 1e-6) # v1 target content END <= anchor time | |
| same[rows, j] = ok1 & okk & (j1 == jj) | |
| if okk.any(): | |
| if proj is None: proj = ((g["F"].astype(np.float32) - mu) @ W).astype(np.float16) | |
| F2[rows[okk], j] = proj[jj[okk]] | |
| np.savez_compressed(OUT / f"future_{r}_{s}.npz", F=F2, ok=ok2, pca_file=np.array(pf.name), pca_sha256=np.array(sha256_file(pf)), targets=np.array(str(S12R / f"targets_{r}_{s}.npz")), axis=np.array("frame/20 (v2)"), version=np.array(2)) | |
| z1 = np.load(S12R / f"future_{r}_{s}.npz"); o1 = z1["ok"].astype(bool) | |
| report["future"][f"{r}_{s}"] = dict(N=int(N), ok_v2_frac=float(ok2.mean()), ok_v1_frac=float(o1.mean()), ok_v1_recomputed_frac=float(ok1c.mean()), ok_mask_equal_v1=float((ok2 == o1).mean()), | |
| v1_target_is_past_frac_by_k={f"+{int(k)}s": float(past1[o1[:, j], j].mean()) if o1[:, j].any() else None for j, k in enumerate(TAUS)}, | |
| v1_target_is_past_frac_all=float(past1[o1].mean()), same_clip_pick_frac=float(same[o1 & ok2].mean()) if (o1 & ok2).any() else None) | |
| log(f"[future v2] {r}/{s}: ok v2 {ok2.mean():.4f} v1 {o1.mean():.4f}; v1 F content in the PAST: {report['future'][f'{r}_{s}']['v1_target_is_past_frac_all']:.4f} (by k {report['future'][f'{r}_{s}']['v1_target_is_past_frac_by_k']}); same pick {report['future'][f'{r}_{s}']['same_clip_pick_frac']}") | |
| (OUT / "alignment.json").write_text(json.dumps(report, indent=1)) | |
| # ---------------------------------------------------------------- empirical (audit-style) check on the probe rows with v2 labels | |
| if a.phase in ("verify", "all"): | |
| aud = json.load(open(PROBES / "audit_latest.json")) if (PROBES / "audit_latest.json").exists() else {} | |
| for r in REGIMES: | |
| for s in SPLITS: | |
| pr = PROBES / f"rows_{r}_{s}.npz"; pf_ = PROBES / f"feats_{r}_{s}.npz" | |
| if not (pr.exists() and pf_.exists()): continue | |
| rows = np.load(pr)["rows"].astype(np.int64); fe = np.load(pf_); raw = fe["raw1408"]; bt = fe["bank_t16"][:, -1].astype(np.float64) | |
| d = tg[(r, s)]; uid = d["uid"][rows]; have = np.isfinite(raw[:, 0].astype(np.float32)) & np.isfinite(bt) | |
| lag = np.full(len(rows), np.nan, np.float32); best = np.full(len(rows), np.nan, np.float32) | |
| for u in np.unique(uid[have]): | |
| g = store.get(u, "front") | |
| if g is None: continue | |
| Fn = g["F"].astype(np.float32); Fn /= np.linalg.norm(Fn, axis=1, keepdims=True) + 1e-6 | |
| for i in np.flatnonzero(have & (uid == u)): | |
| m = (g["tt"] >= bt[i] - 20) & (g["tt"] <= bt[i] + 20) | |
| if m.sum() < 3: continue | |
| q = raw[i].astype(np.float32); q /= np.linalg.norm(q) + 1e-6; c = Fn[m] @ q; j = int(np.argmax(c)); best[i] = c[j] | |
| if c[j] >= 0.98: lag[i] = g["tt"][m][j] - bt[i] | |
| l = lag[np.isfinite(lag)]; al = np.abs(l) | |
| rep = dict(n_rows=int(len(rows)), n_with_bank=int(have.sum()), n_lag_estimated=int(l.size), n_unresolved=int((have & ~np.isfinite(lag)).sum()), | |
| v2=dict(le_0p25=float((al <= 0.25).mean()), ge_0p5=float((al >= 0.5).mean()), ge_1s=float((al >= 1).mean()), ge_3s=float((al >= 3).mean()), max=float(al.max()) if al.size else None, p95=float(np.percentile(al, 95)) if al.size else None, neg_frac=float((l < -1e-6).mean()) if l.size else None), | |
| v1_audit=({k: aud["A_s5_alignment"][f"{r}/{s}"].get(k) for k in ("frac_abs_lag_ge_1s", "frac_abs_lag_ge_3s", "lag_abs_max", "lag_abs_p95", "n_lag_estimated", "routes_with_drift", "routes")} if aud.get("A_s5_alignment", {}).get(f"{r}/{s}") else None)) | |
| report["empirical"][f"{r}_{s}"] = rep; log(f"[empirical v2] {r}/{s}: |lag|<=0.25 {rep['v2']['le_0p25']:.4f} >=1s {rep['v2']['ge_1s']:.4f} >=3s {rep['v2']['ge_3s']:.4f} max {rep['v2']['max']} (n {l.size}); v1 audit >=1s {rep['v1_audit'] and rep['v1_audit']['frac_abs_lag_ge_1s']}") | |
| (OUT / "alignment.json").write_text(json.dumps(report, indent=1)) | |
| # ---------------------------------------------------------------- sequence-level alignment check (robust to static scenes) | |
| if a.phase in ("seqcheck", "all"): | |
| # For every route and 300 s window: lag δ (±20 s, 0.25 s steps) maximising the mean cosine between the frame/20 frontfeat | |
| # clip sequence (s12_wm/cache/frontfeat/<uid>.npz: t = clip END on the 0.5 s grid, feats[N,1408]) and the s5 front sequence | |
| # labelled with v1 (container-fps ts) or v2 (t_true) times. Unlike the per-anchor argmax, a window-level peak is not | |
| # confused by neighbouring near-identical clips. "Resolvable" = peak − median cosine >= 0.01. | |
| FF = EXPLORE / "s12_wm" / "cache" / "frontfeat"; lags = np.arange(-20.0, 20.0 + 1e-9, 0.25); WIN = 300.0 | |
| rows_w = []; anchors_by_uid = {} | |
| for (r, s_), d in tg.items(): | |
| u_, c_ = np.unique(d["uid"], return_counts=True) | |
| for uu, cc in zip(u_, c_): anchors_by_uid[uu] = anchors_by_uid.get(uu, 0) + int(cc) | |
| for u in uids: | |
| pf_ = FF / f"{u}.npz"; g = store.get(u, "front") | |
| if not pf_.exists() or g is None: continue | |
| z = np.load(pf_); tf = z["t"].astype(np.float64); Xf = z["feats"].astype(np.float32); Xf /= np.linalg.norm(Xf, axis=1, keepdims=True) + 1e-6 | |
| Xs = g["F"].astype(np.float32); Xs /= np.linalg.norm(Xs, axis=1, keepdims=True) + 1e-6 | |
| for lab, axis in (("v1", g["ts"]), ("v2", g["tt"])): | |
| for w0 in np.arange(tf[0], tf[-1], WIN): | |
| sel = (tf >= w0) & (tf < w0 + WIN) | |
| if sel.sum() < 40: continue | |
| tw = tf[sel]; Xw = Xf[sel]; score = np.full(len(lags), np.nan) | |
| for li, dl in enumerate(lags): | |
| j = np.searchsorted(axis, tw + dl); j = np.clip(j, 1, len(axis) - 1); j = np.where(np.abs(axis[j - 1] - (tw + dl)) < np.abs(axis[j] - (tw + dl)), j - 1, j) | |
| okm = np.abs(axis[j] - (tw + dl)) <= 0.25 + 1e-6 | |
| if okm.sum() < 20: continue | |
| score[li] = float(np.mean(np.sum(Xw[okm] * Xs[j[okm]], axis=1))) | |
| if np.isfinite(score).sum() < 10: continue | |
| li = int(np.nanargmax(score)); peak = score[li]; med = float(np.nanmedian(score)) | |
| far = np.abs(lags - lags[li]) >= 0.5; sharp = float(peak - np.nanmax(score[far])) if np.isfinite(score[far]).any() else float("nan") # peak over the best lag >= 0.5 s away | |
| rows_w.append(dict(uid=u, label=lab, w0=float(w0), n_slots=int(sel.sum()), lag=float(lags[li]), peak=float(peak), margin=float(peak - med), sharp=sharp, fps20=bool(tab[u]["front"]["fps"] == 20.0), n_anchors=anchors_by_uid.get(u, 0))) | |
| import csv | |
| with open(OUT / "seqcheck_windows.csv", "w", newline="") as f: | |
| w = csv.DictWriter(f, fieldnames=list(rows_w[0].keys())); w.writeheader(); w.writerows(rows_w) | |
| SHARP = 0.01 | |
| def summ(lab, resolvable_only): | |
| R_ = [x for x in rows_w if x["label"] == lab and ((np.isfinite(x["sharp"]) and x["sharp"] >= SHARP) or not resolvable_only)] | |
| if not R_: return None | |
| al = np.array([abs(x["lag"]) for x in R_]); wts = np.array([x["n_anchors"] for x in R_], float) / np.array([sum(1 for y in R_ if y["uid"] == x["uid"]) for x in R_]) | |
| return dict(n_windows=len(R_), n_routes=len({x["uid"] for x in R_}), le_0p25=float((al <= 0.25).mean()), le_0p5=float((al <= 0.5).mean()), ge_1s=float((al >= 1).mean()), ge_3s=float((al >= 3).mean()), max=float(al.max()), | |
| le_0p25_anchor_weighted=float(np.sum(wts * (al <= 0.25)) / wts.sum()), routes_with_any_window_ge_1s=int(len({x["uid"] for x, a_ in zip(R_, al) if a_ >= 1}))) | |
| # per-route consistent offsets under v2 (median lag over sharp windows) -> flagged for review (origin mismatch between the video s5 read and today's file) | |
| offs = {} | |
| for u in {x["uid"] for x in rows_w}: | |
| ws = [x for x in rows_w if x["uid"] == u and x["label"] == "v2" and np.isfinite(x["sharp"]) and x["sharp"] >= SHARP] | |
| if len(ws) >= 2: | |
| m = float(np.median([x["lag"] for x in ws])) | |
| if abs(m) > 0.25: offs[u] = dict(median_lag_s=m, n_sharp_windows=len(ws), n_windows_off=int(sum(abs(x["lag"]) > 0.25 for x in ws)), fps=tab[u]["front"]["fps"], n_anchors=anchors_by_uid.get(u, 0)) | |
| (OUT / "route_offsets_v2.json").write_text(json.dumps(dict(definition="routes whose median v2 lag over sharp windows exceeds 0.25 s: a route-constant origin offset not explained by container fps (v1 shows it at t~0 too); NOT corrected in pack_v2, flagged for review", routes=offs), indent=1)) | |
| report["seqcheck"] = dict(definition="per route x 300 s window: lag (±20 s, 0.25 s grid) maximising mean cosine between frame/20 frontfeat clips and s5 clips labelled v1 (container fps) / v2 (t_true); resolvable = sharp peak: peak − best score at |δ−δ*| >= 0.5 s is >= 0.01 (flat curves = static scenes are unresolvable)", | |
| route_offsets_flagged=offs, n_unresolvable_windows_v2=int(sum(1 for x in rows_w if x["label"] == "v2" and not (np.isfinite(x["sharp"]) and x["sharp"] >= SHARP))), | |
| v1_all=summ("v1", False), v2_all=summ("v2", False), v1_resolvable=summ("v1", True), v2_resolvable=summ("v2", True), | |
| v2_fps20_routes=(lambda R_: dict(n_windows=len(R_), le_0p25=float(np.mean([abs(x["lag"]) <= 0.25 for x in R_]))) if R_ else None)([x for x in rows_w if x["label"] == "v2" and x["fps20"] and np.isfinite(x["sharp"]) and x["sharp"] >= SHARP]), | |
| v2_fpsne20_routes=(lambda R_: dict(n_windows=len(R_), le_0p25=float(np.mean([abs(x["lag"]) <= 0.25 for x in R_]))) if R_ else None)([x for x in rows_w if x["label"] == "v2" and not x["fps20"] and np.isfinite(x["sharp"]) and x["sharp"] >= SHARP]), | |
| v2_resolvable_excluding_flagged_routes=(lambda R_: dict(n_windows=len(R_), le_0p25=float(np.mean([abs(x["lag"]) <= 0.25 for x in R_])), le_0p25_anchor_weighted=float(np.sum([x["n_anchors"] / max(1, sum(1 for y in R_ if y["uid"] == x["uid"])) * (abs(x["lag"]) <= 0.25) for x in R_]) / max(1e-9, np.sum([x["n_anchors"] / max(1, sum(1 for y in R_ if y["uid"] == x["uid"])) for x in R_])))) if R_ else None)([x for x in rows_w if x["label"] == "v2" and np.isfinite(x["sharp"]) and x["sharp"] >= SHARP and x["uid"] not in offs])) | |
| log(f"[seqcheck] v1 resolvable {report['seqcheck']['v1_resolvable']} v2 resolvable {report['seqcheck']['v2_resolvable']}") | |
| (OUT / "alignment.json").write_text(json.dumps(report, indent=1)) | |
| # ---------------------------------------------------------------- table | |
| R = json.load(open(OUT / "alignment.json")); L = ["# pack_v2 alignment — s5 pooled streams / future targets re-indexed to frame/20", "", f"Generated {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())} by cwm3/data/build_pack_v2.py; every number from results/pack_v2/alignment.json.", "", | |
| "## Container fps table", "", f"`{json.dumps(R['fps'])}`", "", "## Analytic lag of the clip each anchor uses (front camera → Vf, cabin → Vc); lag = s5 label − true frame/20 time", "", | |
| "| regime/split | rows ok v2 / v1 | cam | v1 \\|lag\\|≤0.25 | v1 ≥1 s | v1 ≥3 s | v1 max (s) | v1 future-content | v2 \\|lag\\|≤0.25 | v2 ≥1 s | v2 max | routes changed | identical to v1 (v1 basis) |", "|---|---|---|---|---|---|---|---|---|---|---|---|---|"] | |
| for k, v in R.get("streams", {}).items(): | |
| if k.startswith("pca_"): continue | |
| for st, cam in STREAM_CAM.items(): | |
| a1, a2, s1, ro, idn = v["lag_v1"][cam], v["lag_v2"][cam], v["stale_v1"][cam], v["routes"][cam], v["identical_to_v1_with_v1_basis"][st] | |
| L.append(f"| {k} | {v['row_ok_v2']:,} / {v['row_ok_v1']:,} | {cam} ({st}) | {a1['le_0p25']:.4f} | {a1['ge_1s']:.4f} | {a1['ge_3s']:.4f} | {a1['max']:.2f} | {s1['future_content_frac']:.4f} | {a2['le_0p25']:.4f} | {a2['ge_1s']:.4f} | {a2['max']:.2f} | {ro['routes_changed']} / {ro['routes']} | {idn['frac_identical']:.4f} (n={idn['n_both']:,}) |") | |
| L += ["", "## PCA re-fit (manual train, 200k rows)", ""] + [f"- {k}: v2 evr(256) {v['evr256_v2']:.4f} vs v1 {v['evr256_v1']:.4f}, n_fit {v['n_fit']:,}" for k, v in R.get("streams", {}).items() if k.startswith("pca_")] | |
| if R.get("future"): | |
| L += ["", "## Future targets F(t+k)", "", f"- future PCA v2: {json.dumps(R['future'].get('pca'))}", "", "| regime/split | ok v2 | ok v1 | ok masks equal | v1 target content ≤ t (past) | by k | same clip pick |", "|---|---|---|---|---|---|---|"] | |
| for k, v in R["future"].items(): | |
| if k == "pca": continue | |
| L.append(f"| {k} | {v['ok_v2_frac']:.4f} | {v['ok_v1_frac']:.4f} | {v['ok_mask_equal_v1']:.4f} | {v['v1_target_is_past_frac_all']:.4f} | {json.dumps({kk: round(vv, 4) if vv is not None else None for kk, vv in v['v1_target_is_past_frac_by_k'].items()})} | {v['same_clip_pick_frac']} |") | |
| if R.get("empirical"): | |
| L += ["", "## Empirical (audit-style) check on the probe rows: best-match lag of v2-labelled s5 clips vs the frame/20 frontfeat bank", "", "| regime/split | n lag est. | v2 \\|lag\\|≤0.25 | v2 ≥1 s | v2 ≥3 s | v2 max | v1 audit ≥1 s | v1 audit ≥3 s | v1 audit max |", "|---|---|---|---|---|---|---|---|---|"] | |
| for k, v in R["empirical"].items(): | |
| a1 = v.get("v1_audit") or {} | |
| L.append(f"| {k} | {v['n_lag_estimated']:,} | {v['v2']['le_0p25']:.4f} | {v['v2']['ge_1s']:.4f} | {v['v2']['ge_3s']:.4f} | {v['v2']['max']} | {a1.get('frac_abs_lag_ge_1s')} | {a1.get('frac_abs_lag_ge_3s')} | {a1.get('lag_abs_max')} |") | |
| L += ["", "Per-anchor argmax matching is confounded on fps ≠ 20 routes (v2 true times fall between the 0.5 s bank slots → neighbouring near-identical clips tie; residuals sit at ±0.5 s symmetrically); on fps == 20 routes it gives 1.0000. See the sequence-level check below for the acceptance."] | |
| if R.get("seqcheck"): | |
| S_ = R["seqcheck"]; L += ["", "## Sequence-level alignment (route × 300 s window cross-correlation of frontfeat frame/20 clips vs s5 clips)", "", f"{S_['definition']}", "", | |
| "| labels | windows (routes) | \\|δ\\|≤0.25 | \\|δ\\|≤0.5 | ≥1 s | ≥3 s | max (s) | anchor-weighted \\|δ\\|≤0.25 | routes with a window ≥1 s |", "|---|---|---|---|---|---|---|---|---|"] | |
| for k in ("v1_all", "v1_resolvable", "v2_all", "v2_resolvable"): | |
| v = S_[k] | |
| if v: L.append(f"| {k} | {v['n_windows']:,} ({v['n_routes']}) | {v['le_0p25']:.4f} | {v['le_0p5']:.4f} | {v['ge_1s']:.4f} | {v['ge_3s']:.4f} | {v['max']:.2f} | {v['le_0p25_anchor_weighted']:.4f} | {v['routes_with_any_window_ge_1s']} |") | |
| L += ["", f"- v2 resolvable windows on fps == 20 routes: {S_['v2_fps20_routes']}; on fps ≠ 20 routes: {S_['v2_fpsne20_routes']}; unresolvable (flat) v2 windows: {S_.get('n_unresolvable_windows_v2')}", | |
| f"- routes flagged with a route-constant offset not explained by fps (results/pack_v2/route_offsets_v2.json, NOT corrected): {json.dumps({k[:50]: round(v['median_lag_s'], 2) for k, v in S_.get('route_offsets_flagged', {}).items()})}", | |
| f"- v2 resolvable windows excluding the flagged routes: {S_.get('v2_resolvable_excluding_flagged_routes')}"] | |
| acc = S_["v2_resolvable"]["le_0p25_anchor_weighted"] if S_.get("v2_resolvable") else float("nan") | |
| L += ["", f"**Acceptance (|lag| ≤ 0.25 s for ≥ 99.5 % of resolvable anchors, anchor-weighted over resolvable windows, v2 labels): {acc:.4f} → {'PASS' if acc >= 0.995 else 'FAIL'}** (analytic per-anchor lag under v2 labels is 0 by construction for 100 % of rows; see the first table)"] | |
| (TABLES / "pack_v2_alignment.md").write_text("\n".join(L) + "\n") | |
| write_run_json(OUT, "build_pack_v2", [Path(__file__)], [S12R / f"targets_{r}_{s}.npz" for r in REGIMES for s in SPLITS] + [LEAVES], a.seed, t_start, extra=dict(summary=dict(fps=R["fps"], pca={k: v for k, v in R.get("streams", {}).items() if k.startswith("pca_")}, future_pca=R.get("future", {}).get("pca"))), input_listing=False) | |
| log("PACK_V2_DONE") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 38.2 kB
- Xet hash:
- e487b3a71eb99da4ba8da4c083fb817b7ef7ee9dd570f0dfb839843840cd215c
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.