Buckets:
| """cwm3/data/build_holdout_targets.py — the s12 anchor grid + CAN context for the BATON-holdout leaves (WP5), built with the FROZEN | |
| code paths of the benchmark and a `known` split override that pins every holdout dongle to 'holdout'. | |
| The benchmark targets (s12_wm/results/targets_{regime}_{split}.npz) come from two frozen scripts: | |
| 1. manual anchors — explore/e23_onset_baseline/build_labels.process_route (s3_can_baseline/build_labels_s3.py runs it with cfg23 paths | |
| redirected): 50 Hz siglib grid, anchors every 0.2 s from L=4 s after the start to H=3 s before the end, kept when [t-4, t+3] is | |
| entirely manual (canonical engaged mask, 1 s debounce, min-episode cleanup) and vEgo(t) > 0.5 m/s; t = round(grid[a] - grid[0], 2). | |
| 2. s12_wm/build_targets_s12.main — for those rows: Y/M via traj_at on the 10 Hz canspec_v2 grid, M &= ~engaged over the horizon, rows | |
| with M.any() kept, X = context_at, slice_tag = slice_tags; engaged anchors: [t-4, t) fully engaged and pedal-free, vEgo > 0.5, on the | |
| even 10 Hz steps, Y/M via traj_at (no censoring), split = split_of(dongle, known). | |
| This script imports those functions (siglib / cfg23 / build_targets_s12) and applies them to the holdout leaves in the same order; the | |
| only deliberate difference is `known = {dongle: 'holdout'}` for the 60 dongles (pre-registration 2026-09-16T06:46:35Z). It never | |
| touches the six benchmark files. Outputs: s12_wm/results/targets_{manual,engaged}_holdout.npz (same keys / dtypes) + run json. | |
| --verify-benchmark N rebuilds N benchmark routes (test split, local CSV) with exactly this code and asserts uid / t / Y / M / X / | |
| slice_tag equality against the frozen files (bit-for-bit reproduction check; writes results/holdout/targets_verify.json, nothing else). | |
| """ | |
| import argparse, json, sys, time | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1]; EXPLORE = ROOT.parent | |
| E23 = Path("/home/USER/Drive/explore/e23_onset_baseline") | |
| for p in (ROOT / "common", EXPLORE / "s12_wm", E23, E23.parent): sys.path.insert(0, str(p)) | |
| from runmeta import write_run_json # noqa: E402 | |
| import cfg23, siglib # noqa: E402 | |
| import canspec_v2 as cs # noqa: E402 | |
| import build_targets_s12 as bt # noqa: E402 | |
| LROOT = Path("/home/USER/BATON"); S12R = EXPLORE / "s12_wm" / "results"; OUT = ROOT / "results" / "holdout" | |
| IDX = bt.IDX | |
| def leaf_of_uid(u): p = u.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:]) | |
| def manual_anchor_times(leaf): | |
| """build_labels.process_route, anchor part only (base_ok rows): -> t float64[n] (relative to the 50 Hz grid start), or None.""" | |
| sig = siglib.load_route(str(leaf)) | |
| if sig is None: return None | |
| fs, grid = sig["fs"], sig["grid"]; n = len(grid) | |
| manual = ~siglib.engaged_mask(sig, canonical=True) | |
| iL, iH = int(round(cfg23.L_HIST_S * fs)), int(round(cfg23.H_FUT_S * fs)); step = int(round(cfg23.ANCHOR_STRIDE_S * fs)) | |
| a_idx = np.arange(iL, n - iH, step) | |
| if len(a_idx) == 0: return np.zeros(0) | |
| cs_ = np.concatenate([[0], np.cumsum(manual.astype(np.int64))]) | |
| manual_ok = (cs_[a_idx + iH + 1] - cs_[a_idx - iL]) == (a_idx + iH + 1 - (a_idx - iL)) | |
| veg_ok = sig["vEgo"][a_idx] > cfg23.VEGO_MIN_MPS | |
| keep = manual_ok & veg_ok | |
| return np.round(grid[a_idx][keep] - grid[0], 2) | |
| def route_targets(uid, dongle, leaf, known): | |
| """-> {('manual'|'engaged', split): part dict} for one route, exactly as build_targets_s12.main assembles them.""" | |
| out = {} | |
| ta = manual_anchor_times(leaf) | |
| sig = cs.route_signals(str(leaf)) | |
| if sig is None: return out | |
| t10 = sig["t"] | |
| if ta is not None and len(ta): | |
| ta = ta.astype(np.float64) | |
| Y, M = bt.traj_at(sig, ta) | |
| k = np.round((ta - t10[0]) * 10).astype(np.int64); offs = (np.arange(1, bt.H_STEPS + 1) * 2)[None, :] | |
| ki = np.clip(k[:, None] + offs, 0, len(t10) - 1); M &= ~sig["engaged"][ki]; keep = M.any(1) | |
| if keep.any(): | |
| out[("manual", bt.split_of(dongle, known))] = dict(uid=uid, dongle=dongle, t=ta[keep], Y=Y[keep], M=M[keep], X=bt.context_at(sig, ta[keep]), tag=bt.slice_tags(Y[keep], M[keep])) | |
| eng, spd = sig["engaged"], sig["Y"][:, IDX["vEgo"]]; gas = sig["Y"][:, IDX["gasPressed"]] > 0.5; brk = sig["Y"][:, IDX["brakePressed"]] > 0.5 | |
| n = len(t10); Lw = int(bt.L_S * 10); ii = np.arange(Lw, n) | |
| csum = np.cumsum(np.r_[0, eng]); eng_hist = (csum[ii] - csum[ii - Lw]) == Lw | |
| psum = np.cumsum(np.r_[0, gas | brk]); ped_hist = (psum[ii] - psum[ii - Lw]) == 0 | |
| valid = eng_hist & ped_hist & (spd[ii] > 0.5) & eng[ii]; valid &= (ii % 2 == 0) | |
| te = t10[ii[valid]] | |
| if len(te): | |
| Ye, Me = bt.traj_at(sig, te); out[("engaged", bt.split_of(dongle, known))] = dict(uid=uid, dongle=dongle, t=te, Y=Ye, M=Me, X=bt.context_at(sig, te), tag=bt.slice_tags(Ye, Me)) | |
| return out | |
| def assemble(parts): | |
| uid = np.concatenate([[p["uid"]] * len(p["t"]) for p in parts]); don = np.concatenate([[p["dongle"]] * len(p["t"]) for p in parts]) | |
| return dict(uid=uid, dongle=don, t=np.concatenate([p["t"] for p in parts]), Y=np.concatenate([p["Y"] for p in parts]), M=np.concatenate([p["M"] for p in parts]), | |
| X=np.concatenate([p["X"] for p in parts]), slice_tag=np.concatenate([p["tag"] for p in parts])) | |
| def verify_benchmark(n_routes): | |
| """rebuild n benchmark test routes and compare with the frozen targets row by row (per route, both regimes).""" | |
| t0 = time.time(); z = {r: np.load(S12R / f"targets_{r}_test.npz", allow_pickle=True) for r in ("manual", "engaged")} | |
| uids = sorted(set(np.unique(z["manual"]["uid"]).tolist()) | set(np.unique(z["engaged"]["uid"]).tolist())) | |
| uids = [u for u in uids if (LROOT / leaf_of_uid(u) / "vehicle_dynamics.csv").exists()][:n_routes] | |
| rep = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), n_routes=len(uids), routes={}); n_ok = 0 | |
| for u in uids: | |
| dongle = u.split("__")[1]; parts = route_targets(u, dongle, LROOT / leaf_of_uid(u), {dongle: "test"}); rr = {} | |
| for reg in ("manual", "engaged"): | |
| d = z[reg]; rows = np.flatnonzero(d["uid"] == u); p = parts.get((reg, "test")) | |
| if p is None and len(rows) == 0: rr[reg] = dict(n=0, equal=True); continue | |
| if p is None or len(rows) == 0: rr[reg] = dict(n_frozen=int(len(rows)), n_rebuilt=(0 if p is None else int(len(p["t"]))), equal=False); continue | |
| o = np.argsort(d["t"][rows], kind="stable"); rows = rows[o]; o2 = np.argsort(p["t"], kind="stable") | |
| eq = dict(n_frozen=int(len(rows)), n_rebuilt=int(len(p["t"])), t=bool(len(rows) == len(p["t"]) and np.array_equal(d["t"][rows], p["t"][o2]))) | |
| if eq["t"]: | |
| eq.update(Y=bool(np.array_equal(d["Y"][rows], p["Y"][o2])), M=bool(np.array_equal(d["M"][rows], p["M"][o2])), X=bool(np.array_equal(d["X"][rows], p["X"][o2])), slice_tag=bool(np.array_equal(d["slice_tag"][rows], p["tag"][o2]))) | |
| eq["max_abs_dY"] = float(np.abs(d["Y"][rows] - p["Y"][o2]).max()) if len(rows) else 0.0; eq["max_abs_dX"] = float(np.abs(d["X"][rows] - p["X"][o2]).max()) if len(rows) else 0.0 | |
| eq["equal"] = bool(eq["t"] and eq.get("Y") and eq.get("M") and eq.get("X") and eq.get("slice_tag")); rr[reg] = eq | |
| rep["routes"][u] = rr; ok = all(v["equal"] for v in rr.values()); n_ok += ok | |
| print(f"{'OK ' if ok else 'DIFF'} {u[:70]} manual n={rr['manual'].get('n_frozen', rr['manual'].get('n'))} engaged n={rr['engaged'].get('n_frozen', rr['engaged'].get('n'))}", flush=True) | |
| rep["n_routes_identical"] = n_ok; rep["all_identical"] = bool(n_ok == len(uids)); rep["wall_s"] = round(time.time() - t0, 1) | |
| OUT.mkdir(parents=True, exist_ok=True); (OUT / "targets_verify.json").write_text(json.dumps(rep, indent=1)) | |
| print(f"VERIFY {n_ok}/{len(uids)} routes bit-identical (uid, t, Y, M, X, slice_tag) -> {OUT / 'targets_verify.json'}", flush=True) | |
| return rep["all_identical"] | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--draw", default=str(OUT / "draw_v1.json")); ap.add_argument("--verify-benchmark", type=int, default=0); ap.add_argument("--split", default="holdout") | |
| ap.add_argument("--force", action="store_true"); a = ap.parse_args(); t0 = time.time() | |
| if a.verify_benchmark: sys.exit(0 if verify_benchmark(a.verify_benchmark) else 5) | |
| assert a.split not in ("train", "val", "test"), "never write into the benchmark splits" | |
| draw = json.load(open(a.draw)); leaves = draw["leaves"]; known = {c["dongle"]: a.split for c in leaves} | |
| outs = {r: S12R / f"targets_{r}_{a.split}.npz" for r in ("manual", "engaged")} | |
| if any(p.exists() for p in outs.values()) and not a.force: raise SystemExit(f"{list(outs.values())} exist — refuse to overwrite (use --force only before the decisions row)") | |
| parts = {("manual", a.split): [], ("engaged", a.split): []}; per_route = {}; missing = [] | |
| for c in leaves: | |
| leaf = LROOT / c["leaf"] | |
| if not (leaf / "vehicle_dynamics.csv").exists(): missing.append(c["leaf"]); continue | |
| rt = route_targets(c["uid"], c["dongle"], leaf, known); rec = {} | |
| for (reg, sp), p in rt.items(): | |
| assert sp == a.split, (reg, sp); parts[(reg, sp)].append(p); rec[reg] = int(len(p["t"])) | |
| per_route[c["uid"]] = rec; print(f"{c['uid'][:70]} manual={rec.get('manual', 0)} engaged={rec.get('engaged', 0)}", flush=True) | |
| summ = dict(split=a.split, n_leaves=len(leaves), n_missing_csv=len(missing), missing_csv=missing, known_override=f"{len(known)} dongles -> '{a.split}'", per_route=per_route) | |
| for (reg, sp), ps in parts.items(): | |
| if not ps: summ[f"{reg}_{sp}"] = dict(n_rows=0, routes=0); continue | |
| d = assemble(ps); np.savez_compressed(outs[reg], uid=d["uid"], dongle=d["dongle"], t=d["t"], Y=d["Y"], M=d["M"], X=d["X"], slice_tag=d["slice_tag"]) | |
| summ[f"{reg}_{sp}"] = dict(n_rows=int(len(d["t"])), routes=len(ps), devices=int(len(np.unique(d["dongle"]))), hours=round(len(d["t"]) * 0.2 / 3600, 3), tags={int(v): int((d["slice_tag"] == v).sum()) for v in range(4)}, file=str(outs[reg])) | |
| print(f"{reg}/{sp}: n={len(d['t'])} routes={len(ps)} -> {outs[reg]}", flush=True) | |
| OUT.mkdir(parents=True, exist_ok=True); (OUT / f"targets_build_{a.split}.json").write_text(json.dumps(summ, indent=1)) | |
| write_run_json(OUT, f"build_holdout_targets_{a.split}", [Path(__file__), Path(bt.__file__), Path(cs.__file__), Path(siglib.__file__), Path(cfg23.__file__)], [Path(a.draw)], None, t0, extra=dict(summary={k: v for k, v in summ.items() if k != "per_route"}), input_listing=False) | |
| print("HOLDOUT_TARGETS_DONE", json.dumps({k: v for k, v in summ.items() if k != "per_route"}), flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 10.7 kB
- Xet hash:
- 28cb49fb47fbf7f24df83b8511bd6813a2540a061817dca884a49784e87b13e6
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.