Buckets:
| """cwm3/data/build_response_targets.py — graded VEHICLE RESPONSE R[N,25] ∈ {0 none, 1 mild, 2 hard} on the s12 anchor grid. | |
| The binary response chain of cwm2 (targets_demand Resp / RespValid / Resp0 / Resp0valid, build_demand_targets.py | |
| 2026-09-09) is reproduced BIT-FOR-BIT here as `Resp = R >= 1` (asserted against the cwm2 file when present), so every | |
| old evaluator keeps working; R adds the grade and first-passage indices. | |
| 10 Hz rules (CAN whitelist columns only; grid10 C block = canspec.can_features output): | |
| sr_c[i] = steer_rate[i-2] steer rate LAGGED 2 samples: grid10 steer_rate is a central gradient of a 0.15 s | |
| box-smoothed angle and peeks 0.2 s ahead — with the lag, sample i reads angle samples <= i | |
| resp_c = brakePressed > 0.5 | aEgo < -0.8 m/s^2 | |sr_c| > 30 deg/s (= cwm2 Resp rule, analyze_safety.py:72-77) | |
| hard_c = aEgo < -2.5 m/s^2 | |sr_c| > 100 deg/s (100 deg/s = the D steer_rate floor; | |
| -2.5 m/s^2 ~ 0.25 g, between the D decel | |
| floor -2.0 and mine_anomalies STRICT -3.0) | |
| R10 = 2 where hard_c, 1 where resp_c, else 0 | |
| R[k] = max(R10[j-1], R10[j]) over the two 10 Hz samples of the 0.2 s step ending at grid index j (cwm2's resp2 rule) | |
| Rvalid = vEgo, aEgo, sr_c finite at both samples (= cwm2 RespValid) | |
| R0 = R at the anchor (samples k-1, k) (= cwm2 Resp0 grade) | |
| Bits[N,25] u8 keep the components: 1 brakePressed, 2 aEgo<-0.8, 4 |sr_c|>30, 8 aEgo<-2.5, 16 |sr_c|>100 (max over the | |
| two samples), so any other grading can be re-derived without rereading grid10. | |
| First passage: R_first_any = first step with R >= 1 given R0 == 0 (-1 none, -2 R0 invalid or already responding); | |
| R_first_hard = first step with R == 2 given R0 < 2. Also `gas_on` bits? — no: gas is a drive action, not a response; | |
| gas chains stay in the s12 Y / episodes pedal chains. | |
| Outputs results/targets_response/response_{regime}_{split}.npz (row-aligned with s12; uid, t asserted): | |
| R[N,25] u8, Rvalid[N,25] bool, R0[N] u8, R0valid[N] bool, Bits[N,25] u8, Bits0[N] u8, Resp[N,25] bool (= R>=1), | |
| R_first_any[N] int16, R_first_hard[N] int16, InRange[N,25], Moving0[N], uid, t, dongle, grades | |
| Never reads grid10 columns 0–11 (DMS + pose blocks): READ_COLS asserted >= 12 through build_pressure_targets. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import sys | |
| import time | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "8") | |
| import numpy as np # noqa: E402 | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "data")) | |
| sys.path.insert(0, str(ROOT / "common")) | |
| import build_pressure_targets as bp # noqa: E402 | |
| from runmeta import write_run_json # noqa: E402 | |
| OUT = ROOT / "results" / "targets_response" | |
| CWM2_DEM = ROOT.parent / "cwm2" / "results" / "targets_demand" | |
| READ_COLS = ["vEgo", "aEgo", "steer_rate", "brakePressed"] | |
| bp.allow_columns(READ_COLS) | |
| GRADES = ["none", "mild", "hard"] | |
| STEER_LAG = 2 | |
| RESP = dict(aEgo_lt=-0.8, steer_rate_abs_gt=30.0, brakePressed_gt=0.5) # cwm2 (analyze_safety.py:72-77) | |
| HARD = dict(aEgo_lt=-2.5, steer_rate_abs_gt=100.0) | |
| BIT = dict(brake=1, aego_mild=2, sr_mild=4, aego_hard=8, sr_hard=16) | |
| FS, H_STEPS = bp.FS, bp.H_STEPS | |
| def route_response(G, hard=None): | |
| """Label-side response arrays for one route on its 10 Hz grid (all CAUSAL at their sample).""" | |
| hard = dict(HARD, **(hard or {})) | |
| T = G.shape[0] | |
| v = bp.col(G, "vEgo") | |
| a = bp.col(G, "aEgo") | |
| sr_raw = bp.col(G, "steer_rate") | |
| sr_c = np.r_[np.full(STEER_LAG, np.nan), sr_raw[:-STEER_LAG]] | |
| brake = bp.col(G, "brakePressed") | |
| vfin = np.isfinite(v) | |
| mv = vfin & (v > bp.MOVING_MS) | |
| b_brake = np.nan_to_num(brake) > RESP["brakePressed_gt"] | |
| b_am = np.nan_to_num(a) < RESP["aEgo_lt"] | |
| b_sm = np.abs(np.nan_to_num(sr_c)) > RESP["steer_rate_abs_gt"] | |
| b_ah = np.nan_to_num(a) < hard["aEgo_lt"] | |
| b_sh = np.abs(np.nan_to_num(sr_c)) > hard["steer_rate_abs_gt"] | |
| bits10 = (b_brake * BIT["brake"] + b_am * BIT["aego_mild"] + b_sm * BIT["sr_mild"] + b_ah * BIT["aego_hard"] + b_sh * BIT["sr_hard"]).astype(np.uint8) | |
| resp_c = b_brake | b_am | b_sm | |
| hard_c = b_ah | b_sh | |
| r10 = np.where(hard_c, 2, np.where(resp_c, 1, 0)).astype(np.uint8) | |
| R = np.maximum(r10, np.r_[0, r10[:-1]]) # max over the step's two samples (cwm2 resp2 = OR) | |
| bits = bits10 | np.r_[0, bits10[:-1]] | |
| rv1 = vfin & np.isfinite(a) & np.isfinite(sr_c) | |
| rvalid = rv1 & np.r_[False, rv1[:-1]] | |
| return dict(T=T, R=R, Rvalid=rvalid, bits=bits, mv=mv, hard=hard) | |
| def gather_response(rd, k): | |
| T = rd["T"] | |
| kic, inr = bp.gather_steps(None, k, T) | |
| k0ok = (k >= 0) & (k < T) | |
| k0c = np.clip(k, 0, T - 1) | |
| R = np.where(inr, rd["R"][kic], 0).astype(np.uint8) | |
| Rv = rd["Rvalid"][kic] & inr | |
| R0 = np.where(k0ok, rd["R"][k0c], 0).astype(np.uint8) | |
| R0v = rd["Rvalid"][k0c] & k0ok | |
| any_ = bp.first_onset(R0 >= 1, R0v, R >= 1, Rv) | |
| hard = bp.first_onset(R0 >= 2, R0v, R >= 2, Rv) | |
| return dict(R=R, Rvalid=Rv, R0=R0, R0valid=R0v, Bits=np.where(inr, rd["bits"][kic], 0).astype(np.uint8), | |
| Bits0=np.where(k0ok, rd["bits"][k0c], 0).astype(np.uint8), Resp=(R >= 1) & Rv, | |
| R_first_any=any_, R_first_hard=hard, InRange=inr, Moving0=rd["mv"][k0c] & k0ok) | |
| def check_against_cwm2(regime, split, d): | |
| """Bit-for-bit reproduction of the cwm2 binary response chain (when the cwm2 file exists).""" | |
| p = CWM2_DEM / f"demand_{regime}_{split}.npz" | |
| if not p.exists(): | |
| return None | |
| z = np.load(p, allow_pickle=True) | |
| assert np.array_equal(z["uid"], d["uid"]) and np.array_equal(z["t"], d["t"]) | |
| ok = dict(RespValid=bool(np.array_equal(z["RespValid"], d["Rvalid"])), | |
| Resp=bool(np.array_equal(z["Resp"] & z["RespValid"], (d["R"] >= 1) & d["Rvalid"])), | |
| Resp0valid=bool(np.array_equal(z["Resp0valid"], d["R0valid"])), | |
| Resp0=bool(np.array_equal(z["Resp0"] & z["Resp0valid"], (d["R0"] >= 1) & d["R0valid"]))) | |
| return ok | |
| def split_summary(d): | |
| Rv, R = d["Rvalid"], d["R"] | |
| tot = int(Rv.sum()) | |
| sel_any = (d["R_first_any"] != -2) & Rv[:, :15].all(1) | |
| sel_hard = (d["R_first_hard"] != -2) & Rv[:, :15].all(1) | |
| return dict(valid_frac_steps=float(Rv.mean()), grade_none=float((R[Rv] == 0).mean()), grade_mild=float((R[Rv] == 1).mean()), | |
| grade_hard=float((R[Rv] == 2).mean()), prev_R0_any=float((d["R0"][d["R0valid"]] >= 1).mean()), | |
| prev_R0_hard=float((d["R0"][d["R0valid"]] == 2).mean()), | |
| n_inactive_anchors_3s_any=int(sel_any.sum()), | |
| onset_anchors_3s_any=int(((d["R_first_any"][sel_any] >= 0) & (d["R_first_any"][sel_any] < 15)).sum()), | |
| n_inactive_anchors_3s_hard=int(sel_hard.sum()), | |
| onset_anchors_3s_hard=int(((d["R_first_hard"][sel_hard] >= 0) & (d["R_first_hard"][sel_hard] < 15)).sum()), | |
| bit_frac_brake=float(((d["Bits"][Rv] & BIT["brake"]) > 0).mean()), bit_frac_aego_mild=float(((d["Bits"][Rv] & BIT["aego_mild"]) > 0).mean()), | |
| bit_frac_sr_mild=float(((d["Bits"][Rv] & BIT["sr_mild"]) > 0).mean()), bit_frac_aego_hard=float(((d["Bits"][Rv] & BIT["aego_hard"]) > 0).mean()), | |
| bit_frac_sr_hard=float(((d["Bits"][Rv] & BIT["sr_hard"]) > 0).mean()), n_valid_steps=tot) | |
| def sensitivity(regime, split): | |
| rows = [] | |
| for key, vals in (("aEgo_lt", [-2.0, -2.5, -3.0]), ("steer_rate_abs_gt", [80.0, 100.0, 150.0])): | |
| for val in vals: | |
| cache, thr_rows = {}, [] | |
| d, _ = bp.build_split(regime, split, cache, thr_rows, route_fn=lambda G, key=key, val=val: route_response(G, {key: val}), | |
| gather_fn=gather_response) | |
| s = split_summary(d) | |
| rows.append(dict(regime=regime, split=split, param=key, value=val, grade_hard=s["grade_hard"], prev_R0_hard=s["prev_R0_hard"], | |
| onset_anchors_3s_hard=s["onset_anchors_3s_hard"], n_inactive_anchors_3s_hard=s["n_inactive_anchors_3s_hard"])) | |
| print(f"[sens] hard {key}={val}: grade_hard={s['grade_hard']:.5f} onsets_3s={s['onset_anchors_3s_hard']}/{s['n_inactive_anchors_3s_hard']}", flush=True) | |
| return rows | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--splits", default="train,val,test") | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--sensitivity", action="store_true") | |
| a = ap.parse_args() | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| t_start = time.time() | |
| if a.sensitivity: | |
| import csv | |
| rows = sensitivity("manual", "val") + sensitivity("engaged", "val") | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| p = OUT / f"sensitivity_val_{stamp}.csv" | |
| with open(p, "w", newline="") as f: | |
| w = csv.DictWriter(f, fieldnames=list(rows[0])) | |
| w.writeheader() | |
| w.writerows(rows) | |
| write_run_json(OUT, "response_sensitivity", [Path(__file__), Path(bp.__file__)], [bp.S12 / "targets_manual_val.npz", bp.S12 / "targets_engaged_val.npz"], | |
| 0, t_start, extra=dict(csv=str(p), hard_default=HARD, dms_fields_used=[])) | |
| print("wrote", p) | |
| return | |
| cache, thr_rows, summary, files, checks = {}, [], {}, [], {} | |
| for regime in a.regimes.split(","): | |
| for split in a.splits.split(","): | |
| d, wa = bp.build_split(regime, split, cache, thr_rows, route_fn=route_response, gather_fn=gather_response) | |
| src = np.load(bp.S12 / f"targets_{regime}_{split}.npz", allow_pickle=True) | |
| assert np.array_equal(src["uid"], d["uid"]) and np.array_equal(src["t"], d["t"]), "row alignment" | |
| chk = check_against_cwm2(regime, split, d) | |
| checks[f"{regime}_{split}"] = chk | |
| assert chk is None or all(chk.values()), f"cwm2 Resp chain not reproduced: {chk}" | |
| d.update(grades=np.array(GRADES)) | |
| p = OUT / f"response_{regime}_{split}.npz" | |
| bp.write_npz(p, d, a.force) | |
| files.append(str(p)) | |
| s = split_summary(d) | |
| summary[f"{regime}_{split}"] = dict(N=int(len(d["t"])), align=wa, cwm2_resp_identical=chk, **s) | |
| print(f"[{regime}/{split}] N={len(d['t'])} align={wa:.1e} cwm2_identical={chk} grades none/mild/hard = " | |
| f"{s['grade_none']:.4f}/{s['grade_mild']:.4f}/{s['grade_hard']:.4f} R0 any/hard {s['prev_R0_any']:.4f}/{s['prev_R0_hard']:.4f} " | |
| f"onsets3s any {s['onset_anchors_3s_any']}/{s['n_inactive_anchors_3s_any']} hard {s['onset_anchors_3s_hard']}/{s['n_inactive_anchors_3s_hard']}", flush=True) | |
| del d | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| (OUT / f"build_stats_{stamp}.json").write_text(json.dumps(summary, indent=1)) | |
| (OUT / "build_stats_latest.json").write_text(json.dumps(summary, indent=1)) | |
| inputs = [bp.S12 / f"targets_{r}_{s}.npz" for r in a.regimes.split(",") for s in a.splits.split(",")] + [bp.GRID_SRC] | |
| inputs += sorted(bp.GRID / f"{u}.npz" for u in cache) | |
| write_run_json(OUT, "build_response_targets", [Path(__file__), Path(bp.__file__)], inputs, 0, t_start, | |
| extra=dict(files_written=files, grades=GRADES, resp_rule=RESP, hard_rule=HARD, steer_lag_samples=STEER_LAG, bits=BIT, | |
| grid_columns_read=[(c, bp.CI[c]) for c in READ_COLS], grid_columns_never_read=[(bp.COLS[i], i) for i in range(bp.N_FORBIDDEN)], | |
| cwm2_resp_identical=checks, n_routes=len(cache), dms_fields_used=[])) | |
| print("RESPONSE_DONE", f"{time.time() - t_start:.0f}s", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 12 kB
- Xet hash:
- ec52259afe2557004970c0cdbb5f922d860f916d6e322741c2d8090ba55fe52f
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.