Buckets:
| """cwm3/data/build_can8.py — the extra 4 s of causal CAN history needed for the 8 s vehicle-branch variant. | |
| Fact (verified 2026-09-10 on targets_manual_val: dv/aEgo ~ 0.1 s per step, X[i+1,-3] == X[i,-1] for 0.2 s-apart anchors): | |
| s12 X[N,40,5] is t-3.9..t at 10 Hz (build_targets_s12.context_at), NOT 8 s @ 5 Hz. This script rebuilds the 10 Hz | |
| signal grid of every route with the SAME whitelist entry point the targets were built from (s12_wm/canspec_v2.route_signals: | |
| vehicle_dynamics.csv -> [steeringAngleDeg, gasPressed, brakePressed, vEgo, aEgo] interpolated to the frame-anchored 10 Hz axis), | |
| ASSERTS that the stored X is reproduced (|diff| <= 1e-3 on every anchor of the route; otherwise the route is marked invalid | |
| and logged), and writes the preceding 40 steps: | |
| results/pack/xpre_{regime}_{split}.npz {Xpre f32[N,40,5] = t-7.9..t-4.0 (edge-padded like context_at), ok bool[N]} | |
| ok = route reproduced AND k-79 >= 0 (a full 8 s of real history exists). 8 s @ 5 Hz for the TCN = concat(Xpre, X)[:, 1::2]. | |
| Routes whose CSV is no longer local (BATON was partially cleaned 2026-09-10) are fetched from the backup host into | |
| results/can_csv_cache/<uid>/vehicle_dynamics.csv when --fetch-remote is given (subprocess ssh via the cwm2 ControlMaster; | |
| never touches the remote), otherwise their rows get ok = False. | |
| CAN whitelist: only vehicle_dynamics.csv columns read by canspec_v2 (no other CSV is opened here). No DMS field. | |
| python build_can8.py [--fetch-remote] [--workers 8] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import subprocess | |
| import sys | |
| import time | |
| from multiprocessing import Pool | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| import os | |
| S12_SPLITS = tuple(os.environ.get("CWM3_S12_SPLITS", "train,val,test").split(",")) # 2026-09-16 WP5: split enumeration override (default = the benchmark; the holdout lane sets CWM3_S12_SPLITS=holdout) | |
| PACK = CWM3 / "results" / "pack" | |
| CACHE = CWM3 / "results" / "can_csv_cache" | |
| sys.path.insert(0, str(EXPLORE / "s12_wm")) | |
| sys.path.insert(0, "/home/USER/Drive/explore") | |
| sys.path.insert(0, str(CWM3 / "common")) | |
| import canspec_v2 as cs # noqa: E402 (the s12 whitelist reader; same as build_targets_s12) | |
| from common.routes import discover_routes # noqa: E402 | |
| from runmeta import write_run_json # noqa: E402 | |
| REMOTE, RROOT = "USER@REMOTE_HOST", "/data/backup/BATON" | |
| CTL = "/tmp/baton_bak_ctl" | |
| SSH = ["ssh", "-o", f"ControlPath={CTL}", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"] | |
| HIST, PRE = 40, 40 | |
| ATOL = 1e-3 | |
| def ensure_master(): | |
| r = subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True, text=True) | |
| if r.returncode != 0: | |
| subprocess.run(["bash", str(EXPLORE / "cwm2" / "private" / "ssh_master.sh")], check=False, capture_output=True) | |
| time.sleep(2) | |
| r = subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True, text=True) | |
| return r.returncode == 0 | |
| def fetch_remote(uid, log): | |
| """scp <remote leaf>/vehicle_dynamics.csv -> results/can_csv_cache/<uid>/ (read-only on the remote).""" | |
| parts = uid.split("__") | |
| if len(parts) != 4: | |
| return None | |
| dst = CACHE / uid | |
| if (dst / "vehicle_dynamics.csv").exists(): | |
| return dst | |
| dst.mkdir(parents=True, exist_ok=True) | |
| rpath = f"{RROOT}/{parts[0]}/{parts[1]}/{parts[2]}/{parts[3]}/vehicle_dynamics.csv" | |
| r = subprocess.run(["scp", "-o", f"ControlPath={CTL}", "-o", "BatchMode=yes", f"{REMOTE}:{rpath!s}", str(dst / "vehicle_dynamics.csv")], | |
| capture_output=True, text=True, timeout=600) | |
| if r.returncode != 0: | |
| log(f" fetch failed {uid[:60]}: {r.stderr.strip()[:120]}") | |
| return None | |
| return dst | |
| def _route_job(args): | |
| uid, leaf, t_anchor, X_stored = args | |
| sig = cs.route_signals(leaf) | |
| n = len(t_anchor) | |
| out = np.zeros((n, PRE, 5), np.float32) | |
| ok = np.zeros(n, bool) | |
| if sig is None: | |
| return uid, out, ok, "no_signal", 0.0 | |
| t10, Y10 = sig["t"], sig["Y"] | |
| k = np.round((t_anchor - t10[0]) * 10).astype(np.int64) | |
| ki = np.clip(k[:, None] + np.arange(-39, 1)[None, :], 0, len(t10) - 1) | |
| Xre = Y10[ki].astype(np.float32) | |
| err = float(np.max(np.abs(Xre - X_stored))) | |
| if err > ATOL: | |
| return uid, out, ok, "mismatch", err | |
| kp = np.clip(k[:, None] + np.arange(-79, -39)[None, :], 0, len(t10) - 1) | |
| out[:] = Y10[kp].astype(np.float32) | |
| ok[:] = (k - 79) >= 0 | |
| return uid, out, ok, "ok", err | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--fetch-remote", action="store_true") | |
| ap.add_argument("--workers", type=int, default=8) | |
| a = ap.parse_args() | |
| t_start = time.time() | |
| PACK.mkdir(parents=True, exist_ok=True) | |
| logp = PACK / "build_can8.log" | |
| def log(m): | |
| line = f"[{time.strftime('%H:%M:%S')}] {m}" | |
| print(line, flush=True) | |
| with open(logp, "a") as f: | |
| f.write(line + "\n") | |
| leaf_of = {r["uid"]: r["path"] for r in discover_routes() if Path(r["path"], "vehicle_dynamics.csv").exists()} | |
| for p in CACHE.glob("*/vehicle_dynamics.csv"): | |
| leaf_of.setdefault(p.parent.name, str(p.parent)) | |
| master = ensure_master() if a.fetch_remote else False | |
| log(f"local routes with vehicle_dynamics.csv: {len(leaf_of)}; remote master: {master}") | |
| summary = {} | |
| for regime in ("manual", "engaged"): | |
| for split in S12_SPLITS: | |
| out = PACK / f"xpre_{regime}_{split}.npz" | |
| if out.exists(): | |
| log(f"{out.name} exists — skipped") | |
| continue | |
| with np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| uid, t, X = z["uid"], z["t"].astype(np.float64), z["X"].astype(np.float32) | |
| N = len(t) | |
| Xpre = np.zeros((N, PRE, 5), np.float32) | |
| ok = np.zeros(N, bool) | |
| order = np.argsort(uid, kind="stable") | |
| bounds = np.flatnonzero(np.r_[1, uid[order][1:] != uid[order][:-1], 1]) | |
| jobs, rows_of, missing = [], {}, [] | |
| for s, e in zip(bounds[:-1], bounds[1:]): | |
| u = uid[order[s]] | |
| rows = order[s:e] | |
| leaf = leaf_of.get(u) | |
| if leaf is None and master: | |
| d = fetch_remote(u, log) | |
| if d is not None: | |
| leaf = str(d) | |
| leaf_of[u] = leaf | |
| if leaf is None: | |
| missing.append(u) | |
| continue | |
| rows_of[u] = rows | |
| jobs.append((u, leaf, t[rows], X[rows])) | |
| status = {"ok": 0, "mismatch": 0, "no_signal": 0, "missing_csv": len(missing)} | |
| worst = 0.0 | |
| with Pool(a.workers) as pool: | |
| for u, xp, okr, st, err in pool.imap_unordered(_route_job, jobs, chunksize=1): | |
| rows = rows_of[u] | |
| Xpre[rows] = xp | |
| ok[rows] = okr | |
| status[st] += 1 | |
| worst = max(worst, err if st == "ok" else 0.0) | |
| if st != "ok": | |
| log(f" {regime}/{split} route {u[:70]}: {st} (max|dX|={err:.4g})") | |
| np.savez_compressed(out, Xpre=Xpre, ok=ok, n=np.array(N), atol=np.array(ATOL), | |
| rule=np.array("Xpre[k] = Y10[clip(k_anchor-79+k)] for k=0..39 (t-7.9..t-4.0 @10 Hz, canspec_v2.route_signals grid); " | |
| "ok = stored X reproduced on the route (|diff|<=1e-3) and k_anchor-79>=0"), | |
| missing_routes=np.array(missing)) | |
| rec = {"file": out.name, "N": N, "rows_ok": float(ok.mean()), "routes": status, "max_reproduction_err": worst, | |
| "missing_routes": missing} | |
| summary[f"{regime}_{split}"] = rec | |
| log(f"{out.name}: N={N:,} ok={ok.mean():.4f} routes={status} max|X-Xre|={worst:.2e} ({time.time() - t_start:.0f}s)") | |
| write_run_json(PACK, "build_can8", [Path(__file__), EXPLORE / "s12_wm" / "canspec_v2.py", EXPLORE / "s12_wm" / "build_targets_s12.py"], | |
| [S12R / f"targets_{r}_{s}.npz" for r in ("manual", "engaged") for s in S12_SPLITS], | |
| seed=0, t_start=t_start, input_listing=False, | |
| extra={"summary": summary, "fetch_remote": a.fetch_remote, "remote_master": master, "dms_fields_used": [], | |
| "can_columns_read": "vehicle_dynamics.csv via canspec_v2._read_dyn whitelist only"}) | |
| log("DONE") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 8.68 kB
- Xet hash:
- 9fcfd8cbbc943869e577cbcd52beab396421c67db70e8a3f26d38fc8e7aee2d0
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.