Buckets:
| """cwm3/data/holdout_pipeline.py — run the FROZEN cwm3 data builders on the BATON-holdout split (WP5) without touching any benchmark artefact. | |
| Every step imports the frozen builder module unchanged and only (a) restricts its route enumeration to the 60 holdout uids | |
| (results/holdout/holdout_uids.txt), (b) points its OUTPUT directory into results/holdout/<same dir name>/ where the builder would | |
| otherwise overwrite a shared file (route_refs.npz, front_tok.npz, alignment.json, container_fps.json, build logs), and (c) selects the | |
| split 'holdout' (CWM3_S12_SPLITS=holdout for the three builders whose uid enumeration was hard-coded to train/val/test — | |
| build_kin_causal / build_face_causal / build_can8, additive env override with the benchmark default). Anchor targets live in | |
| s12_wm/results/targets_{regime}_holdout.npz (data/build_holdout_targets.py, pre-registration wording) and the 10 Hz grid in | |
| s14_corr/cache/grid10/<uid>.npz (new uid files: the consumers read that path). s5 / frontfeat features are written into their libraries | |
| (new uid files). Nothing existing is modified or deleted. | |
| python data/holdout_pipeline.py grid10 # CPU s14_corr/grid_s14.build for the 60 routes (needs the CSVs only) | |
| python data/holdout_pipeline.py can8 # CPU results/holdout/pack/xpre_{r}_holdout.npz | |
| python data/holdout_pipeline.py pose [--max-minutes M] [--decode-workers W] # GPU results/holdout/pose_causal_routes/<uid>.npz (exit 3 = routes remaining) | |
| python data/holdout_pipeline.py kin # CPU results/holdout/targets_kin_causal_routes_v2/ | |
| python data/holdout_pipeline.py face # CPU results/holdout/targets_face_causal_routes_v2/ (frozen TRAIN thresholds) | |
| python data/holdout_pipeline.py pressure|response|behaviour|demand # CPU results/holdout/targets_*/ | |
| python data/holdout_pipeline.py s5 # GPU s5_features/vjepa2g384/{front,cabin}/<uid>.npz for the 60 routes | |
| python data/holdout_pipeline.py frontfeat # GPU s12_wm/cache/frontfeat/<uid>.npz (anchor-window clips, front cam) | |
| python data/holdout_pipeline.py bank # CPU results/holdout/banks/{front_tok.npz, anchor_index_{r}_holdout.npy} | |
| python data/holdout_pipeline.py status # what exists | |
| (pack v2 is data/build_pack_v2_holdout.py; prediction is eval/predict_holdout.py — run by the coordinator, once.) | |
| """ | |
| import argparse, json, os, sys, time | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1]; EXPLORE = CWM3.parent | |
| H = CWM3 / "results" / "holdout"; S12R = EXPLORE / "s12_wm" / "results"; LROOT = Path("/home/USER/BATON") | |
| LEDGER = CWM3 / "results" / "baton_local_fetch_ledger.jsonl" | |
| SPLIT = "holdout" | |
| sys.path.insert(0, str(CWM3 / "common")); sys.path.insert(0, str(CWM3 / "data")) | |
| def holdout_uids(): return [l.strip() for l in open(H / "holdout_uids.txt") if l.strip()] | |
| def leaf_of(uid): p = uid.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:]) | |
| def ledger_verified(): | |
| """{(leaf, file): bytes} of verified ledger rows.""" | |
| out = {} | |
| for l in open(LEDGER): | |
| if l.strip(): | |
| r = json.loads(l) | |
| if r.get("verified"): out[(r["leaf"], r["file"])] = r["bytes"] | |
| return out | |
| def videos_complete(uids, files=("dcamera.mp4",)): | |
| led = ledger_verified(); missing = [] | |
| for u in uids: | |
| leaf = leaf_of(u) | |
| for f in files: | |
| p = LROOT / leaf / f | |
| if (leaf, f) not in led or not p.exists() or p.stat().st_size != led[(leaf, f)]: missing.append(f"{leaf}/{f}") | |
| return missing | |
| def set_argv(*args): sys.argv = [sys.argv[0]] + [str(a) for a in args] | |
| # ------------------------------------------------------------------------------------------------ steps | |
| def step_grid10(a): | |
| sys.path.insert(0, str(EXPLORE / "s14_corr")); import grid_s14 as g # noqa | |
| U = set(holdout_uids()); routes = [r for r in g.discover_routes() if r["uid"] in U] | |
| st = {}; t0 = time.time() | |
| for r in routes: | |
| try: s = g.build(r) | |
| except Exception as e: s = f"err:{str(e)[:60]}" | |
| st[s] = st.get(s, 0) + 1; print(f"{r['uid'][:70]} {s}", flush=True) | |
| print(f"GRID10_HOLDOUT_DONE routes {len(routes)}/{len(U)} {st} {time.time()-t0:.0f}s (out {g.OUT})", flush=True) | |
| (H / "grid10_status.json").write_text(json.dumps(dict(n_uids=len(U), n_discovered=len(routes), status=st, out=str(g.OUT), utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())), indent=1)) | |
| def step_can8(a): | |
| os.environ["CWM3_S12_SPLITS"] = SPLIT; import build_can8 as m # noqa | |
| m.PACK = H / "pack"; set_argv("--workers", a.workers); m.main() | |
| def step_pose(a): | |
| import extract_pose_causal_routes as m # noqa | |
| uids = holdout_uids(); miss = videos_complete(uids) | |
| if miss and not a.ignore_ledger: print(f"POSE_WAIT videos not ledger-complete: {len(miss)} missing, e.g. {miss[:3]}", flush=True); sys.exit(7) | |
| U = set(uids) | |
| def anchors_by_uid_holdout(): | |
| A = {} | |
| for reg in ("manual", "engaged"): | |
| z = np.load(S12R / f"targets_{reg}_{SPLIT}.npz", allow_pickle=True) | |
| for u, t in zip(z["uid"], z["t"]): | |
| if u in U: A.setdefault(u, []).append(float(t)) | |
| return {u: np.sort(np.array(v)) for u, v in A.items()} | |
| def resolve_video_local(leaf, leaves, stats): | |
| p = LROOT / leaf / "dcamera.mp4"; led = ledger_verified() | |
| assert p.exists() and (leaf, "dcamera.mp4") in led and p.stat().st_size == led[(leaf, "dcamera.mp4")], f"{leaf}: dcamera.mp4 not ledger-verified locally" | |
| return str(p), "local_ledger_verified" | |
| m.anchors_by_uid = anchors_by_uid_holdout; m.resolve_video = resolve_video_local | |
| out = H / "pose_causal_routes"; out.mkdir(parents=True, exist_ok=True) | |
| set_argv("--uids", H / "holdout_uids.txt", "--out-dir", out, "--population-prior", "auto", "--max-minutes", a.max_minutes, "--decode-workers", a.decode_workers, "--vram-check-gb", a.vram_check_gb) | |
| m.main() | |
| def step_kin(a): | |
| os.environ["CWM3_S12_SPLITS"] = SPLIT; import build_kin_causal as m # noqa | |
| set_argv("--source", "routes", "--pose-dir", H / "pose_causal_routes", "--out-dir", H / "targets_kin_causal_routes_v2", "--splits", SPLIT); m.main_routes() | |
| def step_face(a): | |
| os.environ["CWM3_S12_SPLITS"] = SPLIT; import build_face_causal as m # noqa | |
| frozen = CWM3 / "results" / "targets_face_causal_routes_v2" / "facehist_causal_manual_train.npz" | |
| set_argv("--source", "routes", "--pose-dir", H / "pose_causal_routes", "--out-dir", H / "targets_face_causal_routes_v2", "--splits", SPLIT, "--thresholds-from", frozen); m.main() | |
| def step_pressure(a): | |
| import build_pressure_targets as m # noqa | |
| m.OUT = H / "targets_pressure"; set_argv("--splits", SPLIT); m.main() | |
| def step_response(a): | |
| import build_response_targets as m # noqa | |
| m.OUT = H / "targets_response"; set_argv("--splits", SPLIT); m.main() | |
| def step_behaviour(a): | |
| import build_behaviour_targets as m # noqa | |
| m.OUT = H / "targets_behaviour"; m.KIN_ROUTES = H / "targets_kin_causal_routes_v2"; set_argv("--splits", SPLIT, "--use-routes"); m.main() | |
| def step_demand(a): | |
| sys.path.insert(0, str(EXPLORE / "cwm2" / "data")); sys.path.insert(0, str(EXPLORE / "cwm2")); import build_demand_targets as m # noqa | |
| m.OUT = H / "targets_demand"; set_argv("--splits", SPLIT); m.main() | |
| def step_s5(a): | |
| sys.path.insert(0, str(EXPLORE / "s5_features")); import s5_extract_v4 as m # noqa | |
| U = set(holdout_uids()); miss = videos_complete(U, ("dcamera.mp4", "qcamera.mp4")) | |
| if miss and not a.ignore_ledger: print(f"S5_WAIT videos not ledger-complete: {len(miss)} missing", flush=True); sys.exit(7) | |
| orig = m.discover_routes; m.discover_routes = lambda: [r for r in orig() if r["uid"] in U] | |
| set_argv("--max-routes", a.max_routes); m.main() | |
| def step_frontfeat(a): | |
| sys.path.insert(0, str(EXPLORE / "s12_wm")); import extract_crop as m # noqa | |
| U = set(holdout_uids()); miss = videos_complete(U, ("qcamera.mp4",)) | |
| if miss and not a.ignore_ledger: print(f"FRONTFEAT_WAIT videos not ledger-complete: {len(miss)} missing", flush=True); sys.exit(7) | |
| def route_intervals_holdout(): | |
| iv = {} | |
| for reg in ("manual", "engaged"): | |
| d = np.load(S12R / f"targets_{reg}_{SPLIT}.npz", allow_pickle=True); uids, ts = d["uid"], d["t"] | |
| order = np.argsort(uids, kind="stable"); bounds = np.flatnonzero(np.r_[1, uids[order][1:] != uids[order][:-1], 1]) | |
| for x, y in zip(bounds[:-1], bounds[1:]): | |
| u = uids[order[x]] | |
| if u in U: iv.setdefault(u, []).append(np.sort(ts[order[x:y]])) | |
| merged = {} | |
| for u, parts in iv.items(): | |
| ts = np.sort(np.concatenate(parts)); cuts = np.where(np.diff(ts) > m.PAD_S)[0]; st = np.concatenate([[0], cuts + 1]); en = np.concatenate([cuts, [len(ts) - 1]]) | |
| merged[u] = [(max(0.0, ts[i] - m.PAD_S), ts[j]) for i, j in zip(st, en)] | |
| return merged | |
| orig = m.discover_routes; m.discover_routes = lambda: [r for r in orig() if r["uid"] in U]; m.route_intervals = route_intervals_holdout | |
| set_argv("--cam", "front", "--max-routes", a.max_routes); m.main() | |
| def step_bank(a): | |
| import token_bank as m # noqa | |
| U = holdout_uids(); links = H / "frontfeat_links"; links.mkdir(parents=True, exist_ok=True); n = 0 | |
| for u in U: | |
| src = EXPLORE / "s12_wm" / "cache" / "frontfeat" / f"{u}.npz"; dst = links / f"{u}.npz" | |
| if src.exists(): | |
| n += 1 | |
| if not dst.exists(): dst.symlink_to(src) | |
| print(f"frontfeat present for {n}/{len(U)} holdout routes -> {links}", flush=True) | |
| m.BANK_DIR = H / "banks"; m.FD = links; m.BANK_DIR.mkdir(parents=True, exist_ok=True); logp = m.BANK_DIR / "token_bank.log" | |
| def log(msg): | |
| print(msg, flush=True) | |
| with open(logp, "a") as f: f.write(msg + "\n") | |
| t0 = time.time(); m.build(a.workers, log) | |
| with np.load(m.BANK_DIR / "front_tok.npz", allow_pickle=True) as z: | |
| bank_t, names, off = z["t"], [str(u) for u in z["uid_index"]], z["route_off"] | |
| route_of = {u: i for i, u in enumerate(names)} | |
| for regime in ("manual", "engaged"): | |
| out = m.BANK_DIR / f"anchor_index_{regime}_{SPLIT}.npy" | |
| if out.exists(): log(f"[index] {out.name} exists — skipped"); continue | |
| with np.load(S12R / f"targets_{regime}_{SPLIT}.npz", allow_pickle=True) as z: uid, t = z["uid"], z["t"].astype(np.float64) | |
| idx = m.anchor_index_for(bank_t, route_of, off, uid, t); np.save(out, idx); ok = idx >= 0 | |
| rec = {"file": out.name, "N": int(len(t)), "rows_all16": float(ok.all(1).mean()), "rows_last_slot": float(ok[:, -1].mean()), "rows_none": float((~ok.any(1)).mean()), "causality_rule": m.CAUSALITY_RULE, "created_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())} | |
| with open(m.BANK_DIR / "anchor_index_meta.jsonl", "a") as f: f.write(json.dumps(rec) + "\n") | |
| log(f"[index] {out.name}: N={len(t):,} all16={rec['rows_all16']:.4f} last={rec['rows_last_slot']:.4f} none={rec['rows_none']:.4f}") | |
| from runmeta import write_run_json | |
| write_run_json(m.BANK_DIR, "token_bank_holdout", [Path(__file__), Path(m.__file__)], [S12R / f"targets_{r}_{SPLIT}.npz" for r in ("manual", "engaged")], 0, t0, extra=dict(n_routes_with_frontfeat=n, pca_dir=str(m.PCA_DIR)), input_listing=False) | |
| log("TOKEN_BANK_HOLDOUT_DONE") | |
| def step_status(a): | |
| U = holdout_uids(); led = videos_complete(U, ("dcamera.mp4", "qcamera.mp4")) | |
| st = dict(uids=len(U), videos_missing_or_unverified=len(led), targets={r: (S12R / f"targets_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, | |
| grid10=sum((EXPLORE / "s14_corr" / "cache" / "grid10" / f"{u}.npz").exists() for u in U), xpre={r: (H / "pack" / f"xpre_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, | |
| pose=sum((H / "pose_causal_routes" / f"{u}.npz").exists() for u in U), kin={r: (H / "targets_kin_causal_routes_v2" / f"kin_causal_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, | |
| face={r: (H / "targets_face_causal_routes_v2" / f"facehist_causal_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, | |
| pressure={r: (H / "targets_pressure" / f"pressure_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, response={r: (H / "targets_response" / f"response_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, | |
| behaviour={r: (H / "targets_behaviour" / f"behaviour_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, demand={r: (H / "targets_demand" / f"demand_{r}_{SPLIT}.npz").exists() for r in ("manual", "engaged")}, | |
| s5_front=sum((EXPLORE / "s5_features" / "vjepa2g384" / "front" / f"{u}.npz").exists() for u in U), s5_cabin=sum((EXPLORE / "s5_features" / "vjepa2g384" / "cabin" / f"{u}.npz").exists() for u in U), | |
| frontfeat=sum((EXPLORE / "s12_wm" / "cache" / "frontfeat" / f"{u}.npz").exists() for u in U), bank=(H / "banks" / "front_tok.npz").exists(), | |
| pack_v2={k: (H / "pack_v2" / f"{k}_manual_{SPLIT}.npz").exists() for k in ("vf256", "vc256", "future")}) | |
| print(json.dumps(st, indent=1)) | |
| STEPS = dict(grid10=step_grid10, can8=step_can8, pose=step_pose, kin=step_kin, face=step_face, pressure=step_pressure, response=step_response, behaviour=step_behaviour, demand=step_demand, s5=step_s5, frontfeat=step_frontfeat, bank=step_bank, status=step_status) | |
| if __name__ == "__main__": | |
| ap = argparse.ArgumentParser(); ap.add_argument("step", choices=sorted(STEPS)); ap.add_argument("--workers", type=int, default=4); ap.add_argument("--max-minutes", type=float, default=10.0) | |
| ap.add_argument("--decode-workers", type=int, default=4); ap.add_argument("--vram-check-gb", type=float, default=4.0); ap.add_argument("--max-routes", type=int, default=0); ap.add_argument("--ignore-ledger", action="store_true") | |
| a = ap.parse_args(); STEPS[a.step](a) | |
Xet Storage Details
- Size:
- 14 kB
- Xet hash:
- a61de3c820ee8f76e13f1b4cd7591dfe287cf52d6760191a8de0506f07f8cc62
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.