Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /eval /robustness_perturb.py

AsianPlayer

16 days ago

download

raw

10.8 kB

"""cwm3/eval/robustness_perturb.py — test-time INPUT perturbations of a frozen joint run (roadmap B "robustness" table; gaze/robustness agent 2026-09-12), no retraining.
The frozen run dir (e.g. results/runs/I1_all_minus_road2drv_s0_final) is READ-ONLY.  A new dir results/runs/<run>_pert_<name>/ receives copies of
config.json / BEST / TRAIN_DONE / base_ckpts.json / stats.npz / history.jsonl / params.json and a symlink to the BEST checkpoint, so that
eval/paired_compare.layout_guard accepts its layouts (same checkpoint, same cfg, same base checkpoints).  Prediction re-uses train/train_cwm3_joint.py's own
`load_raw` / `predict_split` / `write_pred_layout_joint.write_layout` (imported, unchanged); the RAW test arrays are perturbed in numpy BEFORE `predict_split`
(the TARGETS and the row validity `kin_row_ok` are never touched -> identical rows for the paired comparison; the steer-phase / pedal anchor states come from
the data layer's clean X).  perturbation.json records the spec, the numpy seed and the number of affected entries.
Perturbations (--pert):
  kin_drop20 / kin_drop50   driver kinematics dropout: 20 % / 50 % of the 40 Kh history steps of every row -> NaN (= valid bit 0, standardised value 0; a step
                            dropped at t makes K_0 the train mean, as in the trainer's own convention).  Seed: --pert-seed (default 0) x regime.
  can_noise                 CAN sensor noise, i.i.d. per step: steeringAngleDeg + N(0, 2 deg), vEgo + N(0, 0.5 m/s) (clipped at 0); gas / brake bits unchanged.
  road_off                  road stream removed: vf256 = 0 and vf_ok = 0 on every row (E's side vector gated off; the token bank and the P head inputs stay).
  kh_trunc_old2s            history truncation: the OLDEST 2 s of Kh (t-3.9..t-2.0, steps 0..19) masked -> the driver trunk sees the last 2 s only.
  kh_mask_recent2s          the MOST RECENT 2 s of Kh (t-1.9..t, steps 20..39) masked -> stale driver state (the literal reading of "last 2 s masked").
  python eval/robustness_perturb.py --run I1_all_minus_road2drv_s0_final --pert kin_drop20 [--regimes manual,engaged] [--pert-seed 0]
Then: eval/paired_compare_joint.py --a <pert dir> --b <run dir> --kin-targets causal_routes --kin-version v2 (identical rows) and eval/pressure_eval.py --a --b (P cells).
GPU inference only (~1-2 min per regime); no GPU lock (--no-lock semantics: inline), gated by the caller.  No openpilot DMS field is read.
"""
from __future__ import annotations

import os

os.environ.setdefault("OMP_NUM_THREADS", "4")
os.environ.setdefault("MKL_NUM_THREADS", "4")
import argparse                                             # noqa: E402
import json                                                 # noqa: E402
import shutil                                               # noqa: E402
import sys                                                  # noqa: E402
import time                                                 # noqa: E402
from pathlib import Path                                    # noqa: E402
from types import SimpleNamespace                           # noqa: E402

import numpy as np                                          # noqa: E402

CWM3 = Path(__file__).resolve().parents[1]
EXPLORE = CWM3.parent
for p in (CWM3, CWM3 / "data", CWM3 / "common", CWM3 / "train", CWM3 / "eval", EXPLORE / "s12_wm"):
    sys.path.insert(0, str(p))
import cwm3_dataset as ds                                   # noqa: E402
import train_cwm3 as T                                      # noqa: E402  (pred_ok)
import train_cwm3_joint as TJ                               # noqa: E402  (imported, unchanged: load_raw / base_info_numpy / load_bank / predict_split)
import write_pred_layout_joint as wplj                      # noqa: E402
from runmeta import write_run_json                          # noqa: E402

PERTS = ("kin_drop20", "kin_drop50", "can_noise", "road_off", "kh_trunc_old2s", "kh_mask_recent2s")
COPY = ("config.json", "BEST", "TRAIN_DONE", "base_ckpts.json", "stats.npz", "history.jsonl", "params.json", "label_rows.npz")
STEER_SD_DEG, VEGO_SD_MPS = 2.0, 0.5


def perturb(raw, name, rng):
    """in place on the raw dict of ONE (regime, split); -> dict of counts for perturbation.json."""
    N = len(raw["t"])
    rec = {"n_rows": int(N)}
    if name.startswith("kin_drop"):
        p = int(name[len("kin_drop"):]) / 100.0
        Kh = raw["Kh"]                                       # f16 [N,40,11], NaN = invalid
        m = rng.random(Kh.shape[:2]) < p
        was_valid = np.isfinite(Kh[:, :, 0].astype(np.float32))
        Kh[m] = np.nan
        rec.update(p=p, steps_dropped=int(m.sum()), steps_total=int(m.size), valid_steps_before=int(was_valid.sum()), valid_steps_after=int(np.isfinite(Kh[:, :, 0].astype(np.float32)).sum()),
                   last_step_dropped_frac=float(m[:, -1].mean()))
    elif name == "can_noise":
        X = raw["X"]                                         # f32 [N,40,5]: steer, gasPressed, brakePressed, vEgo, aEgo
        X[:, :, 0] += rng.normal(0, STEER_SD_DEG, X.shape[:2]).astype(np.float32)
        v = X[:, :, 3] + rng.normal(0, VEGO_SD_MPS, X.shape[:2]).astype(np.float32)
        X[:, :, 3] = np.maximum(v, 0.0)
        rec.update(steer_sd_deg=STEER_SD_DEG, vego_sd_mps=VEGO_SD_MPS, vego_clipped_at_zero=int((v < 0).sum()), note="i.i.d. per step; phase0 / gas0 / brake0 anchor states from the clean X (data layer)")
    elif name == "road_off":
        ok_before = int(raw["vf256_ok"].sum())
        raw["vf256"][:] = 0
        raw["vf256_ok"][:] = False
        rec.update(vf_ok_before=ok_before, vf_ok_after=0)
    elif name == "kh_trunc_old2s":
        Kh = raw["Kh"]
        was = int(np.isfinite(Kh[:, :20, 0].astype(np.float32)).sum())
        Kh[:, :20, :] = np.nan
        rec.update(steps_masked="0..19 (t-3.9..t-2.0)", valid_steps_removed=was)
    elif name == "kh_mask_recent2s":
        Kh = raw["Kh"]
        was = int(np.isfinite(Kh[:, 20:, 0].astype(np.float32)).sum())
        Kh[:, 20:, :] = np.nan
        rec.update(steps_masked="20..39 (t-1.9..t)", valid_steps_removed=was, note="K_0 = last history step becomes the train mean (valid bit 0) on every row")
    else:
        raise ValueError(name)
    return rec


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--run", required=True, help="frozen joint run dir name under results/runs (read-only)")
    ap.add_argument("--pert", required=True, choices=PERTS)
    ap.add_argument("--regimes", default="manual,engaged")
    ap.add_argument("--split", default="test")
    ap.add_argument("--pert-seed", type=int, default=0)
    ap.add_argument("--cpu", action="store_true")
    a = ap.parse_args()
    t0 = time.time()
    src = CWM3 / "results" / "runs" / a.run
    assert (src / "PRED_DONE").exists() and (src / "TRAIN_DONE").exists(), f"{src}: not a finished run"
    out = CWM3 / "results" / "runs" / f"{a.run}_pert_{a.pert}"
    out.mkdir(parents=True, exist_ok=True)
    for f in COPY:
        if (src / f).exists() and not (out / f).exists():
            shutil.copy2(src / f, out / f)
    best = (src / "BEST").read_text().strip()
    if best != "base" and not (out / best).exists():
        os.symlink(src / best, out / best)                  # same checkpoint; never copied (the frozen file is the single source)
    cfgj = json.loads((out / "config.json").read_text())
    cfg, cargs = cfgj["cfg"], cfgj["args"]
    seed = int(cargs["seed"])
    ds.set_pack_version(cfg.get("pack_version", "v1"))
    ds.set_kin_source(cfg.get("kin_source", "cwm2"), cfg.get("kin_version", "v1"))
    args = SimpleNamespace(out=str(out), seed=seed, cpu=a.cpu, no_amp=False, minutes=600.0, synthetic=0, max_anchors=0, no_label_rows=True, no_lock=True)
    TJ.log(f"=== robustness_perturb {a.run} pert={a.pert} seed={seed} regimes={a.regimes} out={out} ===", args.out)
    info = TJ.base_info_numpy(cfg, seed)
    rec_bc = json.loads((out / "base_ckpts.json").read_text())["base_ckpts"]
    assert info["base_ckpts"] == rec_bc, f"base checkpoints changed since the run: {info['base_ckpts']} vs {rec_bc}"
    st = {k: v for k, v in np.load(out / "stats.npz").items()}
    bank = TJ.load_bank(args, cfg, info)
    pj = out / "perturbation.json"
    prec = json.loads(pj.read_text()) if pj.exists() else {"run": a.run, "pert": a.pert, "pert_seed": a.pert_seed, "regimes": {}}
    for regime in a.regimes.split(","):
        pf = out / f"pred_{regime}_{a.split}.npz"
        if pf.exists() and not T.pred_ok(pf):                 # a crash mid-write leaves a truncated npz -> renamed *.corrupt by pred_ok, re-predicted
            TJ.log(f"{pf.name} was corrupt -> re-predicting", args.out)
        if pf.exists():
            TJ.log(f"{pf.name} exists — skipped", args.out)
        else:
            raw = TJ.load_raw(args, cfg, a.split, subset=False, regimes=[regime])
            rng = np.random.default_rng(10_000 * a.pert_seed + {"manual": 1, "engaged": 2}[regime])
            rec = perturb(raw, a.pert, rng)
            prec["regimes"][f"{regime}_{a.split}"] = rec
            pj.write_text(json.dumps(prec, indent=1))
            TJ.log(f"perturbed {regime}/{a.split}: {rec}", args.out)
            code = TJ.predict_split(args, cfg, raw, st, bank, info, regime, a.split)
            assert code == 0, f"predict_split exit {code}"
            del raw
        lay = out / f"pred_layout_{regime}_{a.split}.npz"
        if not lay.exists():
            code_files = sorted((CWM3 / "models").glob("*.py")) + [CWM3 / "train" / "train_cwm3_joint.py", CWM3 / "train" / "train_cwm3.py", CWM3 / "train" / "batch.py",
                                                                    CWM3 / "eval" / "write_pred_layout_joint.py", CWM3 / "data" / "cwm3_dataset.py", EXPLORE / "s12_wm/eval_s13.py"]
            wplj.write_layout(str(out), regime, a.split, {"stage": cfg["stage"], "variant": cfg["variant"], "cfg": cfg}, code_files, log=lambda m: TJ.log(m, args.out))
    (out / "PRED_DONE").write_text(json.dumps({"splits": [a.split], "regimes": a.regimes.split(","), "perturbation": a.pert, "source_run": a.run}))
    write_run_json(str(out), f"robustness_perturb_{a.pert}", [Path(__file__), CWM3 / "train" / "train_cwm3_joint.py", CWM3 / "eval" / "write_pred_layout_joint.py"],
                   [src / best] if best != "base" else [], seed=seed, t_start=t0, input_listing=False,
                   extra={"source_run": a.run, "perturbation": a.pert, "pert_seed": a.pert_seed, "records": prec["regimes"], "base_ckpts": info["base_ckpts"], "dms_fields_used": [],
                          "note": "test-time input perturbation of the frozen run; targets and kin_row_ok untouched; layouts pair against the unperturbed run on identical rows"})
    TJ.log(f"ROBUSTNESS_PERTURB_DONE {a.run} {a.pert} ({time.time() - t0:.0f}s)", args.out)


if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 10.8 kB
Xet hash:: 25a8292f16eac4033328568c81a9ec562d2227b1880e5a5a37fa3bf39bd17d80

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""cwm3/eval/robustness_perturb.py — test-time INPUT perturbations of a frozen joint run (roadmap B "robustness" table; gaze/robustness agent 2026-09-12), no retraining.

	The frozen run dir (e.g. results/runs/I1_all_minus_road2drv_s0_final) is READ-ONLY. A new dir results/runs/<run>_pert_<name>/ receives copies of
	config.json / BEST / TRAIN_DONE / base_ckpts.json / stats.npz / history.jsonl / params.json and a symlink to the BEST checkpoint, so that
	eval/paired_compare.layout_guard accepts its layouts (same checkpoint, same cfg, same base checkpoints). Prediction re-uses train/train_cwm3_joint.py's own
	`load_raw` / `predict_split` / `write_pred_layout_joint.write_layout` (imported, unchanged); the RAW test arrays are perturbed in numpy BEFORE `predict_split`
	(the TARGETS and the row validity `kin_row_ok` are never touched -> identical rows for the paired comparison; the steer-phase / pedal anchor states come from
	the data layer's clean X). perturbation.json records the spec, the numpy seed and the number of affected entries.
	Perturbations (--pert):
	kin_drop20 / kin_drop50 driver kinematics dropout: 20 % / 50 % of the 40 Kh history steps of every row -> NaN (= valid bit 0, standardised value 0; a step
	dropped at t makes K_0 the train mean, as in the trainer's own convention). Seed: --pert-seed (default 0) x regime.
	can_noise CAN sensor noise, i.i.d. per step: steeringAngleDeg + N(0, 2 deg), vEgo + N(0, 0.5 m/s) (clipped at 0); gas / brake bits unchanged.
	road_off road stream removed: vf256 = 0 and vf_ok = 0 on every row (E's side vector gated off; the token bank and the P head inputs stay).
	kh_trunc_old2s history truncation: the OLDEST 2 s of Kh (t-3.9..t-2.0, steps 0..19) masked -> the driver trunk sees the last 2 s only.
	kh_mask_recent2s the MOST RECENT 2 s of Kh (t-1.9..t, steps 20..39) masked -> stale driver state (the literal reading of "last 2 s masked").
	python eval/robustness_perturb.py --run I1_all_minus_road2drv_s0_final --pert kin_drop20 [--regimes manual,engaged] [--pert-seed 0]
	Then: eval/paired_compare_joint.py --a <pert dir> --b <run dir> --kin-targets causal_routes --kin-version v2 (identical rows) and eval/pressure_eval.py --a --b (P cells).
	GPU inference only (~1-2 min per regime); no GPU lock (--no-lock semantics: inline), gated by the caller. No openpilot DMS field is read.
	"""
	from __future__ import annotations

	import os

	os.environ.setdefault("OMP_NUM_THREADS", "4")
	os.environ.setdefault("MKL_NUM_THREADS", "4")
	import argparse # noqa: E402
	import json # noqa: E402
	import shutil # noqa: E402
	import sys # noqa: E402
	import time # noqa: E402
	from pathlib import Path # noqa: E402
	from types import SimpleNamespace # noqa: E402

	import numpy as np # noqa: E402

	CWM3 = Path(__file__).resolve().parents[1]
	EXPLORE = CWM3.parent
	for p in (CWM3, CWM3 / "data", CWM3 / "common", CWM3 / "train", CWM3 / "eval", EXPLORE / "s12_wm"):
	sys.path.insert(0, str(p))
	import cwm3_dataset as ds # noqa: E402
	import train_cwm3 as T # noqa: E402 (pred_ok)
	import train_cwm3_joint as TJ # noqa: E402 (imported, unchanged: load_raw / base_info_numpy / load_bank / predict_split)
	import write_pred_layout_joint as wplj # noqa: E402
	from runmeta import write_run_json # noqa: E402

	PERTS = ("kin_drop20", "kin_drop50", "can_noise", "road_off", "kh_trunc_old2s", "kh_mask_recent2s")
	COPY = ("config.json", "BEST", "TRAIN_DONE", "base_ckpts.json", "stats.npz", "history.jsonl", "params.json", "label_rows.npz")
	STEER_SD_DEG, VEGO_SD_MPS = 2.0, 0.5


	def perturb(raw, name, rng):
	"""in place on the raw dict of ONE (regime, split); -> dict of counts for perturbation.json."""
	N = len(raw["t"])
	rec = {"n_rows": int(N)}
	if name.startswith("kin_drop"):
	p = int(name[len("kin_drop"):]) / 100.0
	Kh = raw["Kh"] # f16 [N,40,11], NaN = invalid
	m = rng.random(Kh.shape[:2]) < p
	was_valid = np.isfinite(Kh[:, :, 0].astype(np.float32))
	Kh[m] = np.nan
	rec.update(p=p, steps_dropped=int(m.sum()), steps_total=int(m.size), valid_steps_before=int(was_valid.sum()), valid_steps_after=int(np.isfinite(Kh[:, :, 0].astype(np.float32)).sum()),
	last_step_dropped_frac=float(m[:, -1].mean()))
	elif name == "can_noise":
	X = raw["X"] # f32 [N,40,5]: steer, gasPressed, brakePressed, vEgo, aEgo
	X[:, :, 0] += rng.normal(0, STEER_SD_DEG, X.shape[:2]).astype(np.float32)
	v = X[:, :, 3] + rng.normal(0, VEGO_SD_MPS, X.shape[:2]).astype(np.float32)
	X[:, :, 3] = np.maximum(v, 0.0)
	rec.update(steer_sd_deg=STEER_SD_DEG, vego_sd_mps=VEGO_SD_MPS, vego_clipped_at_zero=int((v < 0).sum()), note="i.i.d. per step; phase0 / gas0 / brake0 anchor states from the clean X (data layer)")
	elif name == "road_off":
	ok_before = int(raw["vf256_ok"].sum())
	raw["vf256"][:] = 0
	raw["vf256_ok"][:] = False
	rec.update(vf_ok_before=ok_before, vf_ok_after=0)
	elif name == "kh_trunc_old2s":
	Kh = raw["Kh"]
	was = int(np.isfinite(Kh[:, :20, 0].astype(np.float32)).sum())
	Kh[:, :20, :] = np.nan
	rec.update(steps_masked="0..19 (t-3.9..t-2.0)", valid_steps_removed=was)
	elif name == "kh_mask_recent2s":
	Kh = raw["Kh"]
	was = int(np.isfinite(Kh[:, 20:, 0].astype(np.float32)).sum())
	Kh[:, 20:, :] = np.nan
	rec.update(steps_masked="20..39 (t-1.9..t)", valid_steps_removed=was, note="K_0 = last history step becomes the train mean (valid bit 0) on every row")
	else:
	raise ValueError(name)
	return rec


	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("--run", required=True, help="frozen joint run dir name under results/runs (read-only)")
	ap.add_argument("--pert", required=True, choices=PERTS)
	ap.add_argument("--regimes", default="manual,engaged")
	ap.add_argument("--split", default="test")
	ap.add_argument("--pert-seed", type=int, default=0)
	ap.add_argument("--cpu", action="store_true")
	a = ap.parse_args()
	t0 = time.time()
	src = CWM3 / "results" / "runs" / a.run
	assert (src / "PRED_DONE").exists() and (src / "TRAIN_DONE").exists(), f"{src}: not a finished run"
	out = CWM3 / "results" / "runs" / f"{a.run}_pert_{a.pert}"
	out.mkdir(parents=True, exist_ok=True)
	for f in COPY:
	if (src / f).exists() and not (out / f).exists():
	shutil.copy2(src / f, out / f)
	best = (src / "BEST").read_text().strip()
	if best != "base" and not (out / best).exists():
	os.symlink(src / best, out / best) # same checkpoint; never copied (the frozen file is the single source)
	cfgj = json.loads((out / "config.json").read_text())
	cfg, cargs = cfgj["cfg"], cfgj["args"]
	seed = int(cargs["seed"])
	ds.set_pack_version(cfg.get("pack_version", "v1"))
	ds.set_kin_source(cfg.get("kin_source", "cwm2"), cfg.get("kin_version", "v1"))
	args = SimpleNamespace(out=str(out), seed=seed, cpu=a.cpu, no_amp=False, minutes=600.0, synthetic=0, max_anchors=0, no_label_rows=True, no_lock=True)
	TJ.log(f"=== robustness_perturb {a.run} pert={a.pert} seed={seed} regimes={a.regimes} out={out} ===", args.out)
	info = TJ.base_info_numpy(cfg, seed)
	rec_bc = json.loads((out / "base_ckpts.json").read_text())["base_ckpts"]
	assert info["base_ckpts"] == rec_bc, f"base checkpoints changed since the run: {info['base_ckpts']} vs {rec_bc}"
	st = {k: v for k, v in np.load(out / "stats.npz").items()}
	bank = TJ.load_bank(args, cfg, info)
	pj = out / "perturbation.json"
	prec = json.loads(pj.read_text()) if pj.exists() else {"run": a.run, "pert": a.pert, "pert_seed": a.pert_seed, "regimes": {}}
	for regime in a.regimes.split(","):
	pf = out / f"pred_{regime}_{a.split}.npz"
	if pf.exists() and not T.pred_ok(pf): # a crash mid-write leaves a truncated npz -> renamed *.corrupt by pred_ok, re-predicted
	TJ.log(f"{pf.name} was corrupt -> re-predicting", args.out)
	if pf.exists():
	TJ.log(f"{pf.name} exists — skipped", args.out)
	else:
	raw = TJ.load_raw(args, cfg, a.split, subset=False, regimes=[regime])
	rng = np.random.default_rng(10_000 * a.pert_seed + {"manual": 1, "engaged": 2}[regime])
	rec = perturb(raw, a.pert, rng)
	prec["regimes"][f"{regime}_{a.split}"] = rec
	pj.write_text(json.dumps(prec, indent=1))
	TJ.log(f"perturbed {regime}/{a.split}: {rec}", args.out)
	code = TJ.predict_split(args, cfg, raw, st, bank, info, regime, a.split)
	assert code == 0, f"predict_split exit {code}"
	del raw
	lay = out / f"pred_layout_{regime}_{a.split}.npz"
	if not lay.exists():
	code_files = sorted((CWM3 / "models").glob("*.py")) + [CWM3 / "train" / "train_cwm3_joint.py", CWM3 / "train" / "train_cwm3.py", CWM3 / "train" / "batch.py",
	CWM3 / "eval" / "write_pred_layout_joint.py", CWM3 / "data" / "cwm3_dataset.py", EXPLORE / "s12_wm/eval_s13.py"]
	wplj.write_layout(str(out), regime, a.split, {"stage": cfg["stage"], "variant": cfg["variant"], "cfg": cfg}, code_files, log=lambda m: TJ.log(m, args.out))
	(out / "PRED_DONE").write_text(json.dumps({"splits": [a.split], "regimes": a.regimes.split(","), "perturbation": a.pert, "source_run": a.run}))
	write_run_json(str(out), f"robustness_perturb_{a.pert}", [Path(__file__), CWM3 / "train" / "train_cwm3_joint.py", CWM3 / "eval" / "write_pred_layout_joint.py"],
	[src / best] if best != "base" else [], seed=seed, t_start=t0, input_listing=False,
	extra={"source_run": a.run, "perturbation": a.pert, "pert_seed": a.pert_seed, "records": prec["regimes"], "base_ckpts": info["base_ckpts"], "dms_fields_used": [],
	"note": "test-time input perturbation of the frozen run; targets and kin_row_ok untouched; layouts pair against the unperturbed run on identical rows"})
	TJ.log(f"ROBUSTNESS_PERTURB_DONE {a.run} {a.pert} ({time.time() - t0:.0f}s)", args.out)


	if __name__ == "__main__":
	main()