Buckets:
| """cwm3/tests/test_targets_causal.py — causality and contract tests for the P / R / B target builders. | |
| (1) No future leak: on a synthetic route, perturbing every 10 Hz sample AFTER the anchor changes only the future targets | |
| (P, R) — never the anchor state (P0, M0, S0, R0, Bits0) — with the route thresholds held fixed (the badas route quantile | |
| is a label definition that reads the whole route and is therefore passed in explicitly here). | |
| (2) Steer-rate lag: R at the anchor ignores steer_rate samples k-1 and k (the grid10 steer rate peeks 0.2 s ahead) and | |
| reads sample k-2. | |
| (3) Perturbing samples BEFORE the anchor may change P0/R0 (the tests have teeth), and the future targets do change under (1). | |
| (4) Contract against cwm2 targets_demand (val, when built): identical row order (uid, t), shapes, Resp chain bit-identical, | |
| and the three shared families ttc / ttc_drop / badas identical to D. | |
| (5) Behaviour file layout / source codes / causal-only copy consistency. | |
| Run: cd cwm3 && python -m pytest tests/test_targets_causal.py -q | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(CWM3 / "data")) | |
| import build_pressure_targets as bp # noqa: E402 | |
| import build_response_targets as br # noqa: E402 | |
| DEM = CWM3.parent / "cwm2" / "results" / "targets_demand" | |
| CASES = [(r, s) for r in ("manual", "engaged") for s in ("val", "train", "test")] | |
| def synthetic_route(T=600, seed=0): | |
| """A plausible 10 Hz route in grid10 layout: moving, lead appearing/vanishing, lane lines, badas, brake/steer bursts.""" | |
| rng = np.random.default_rng(seed) | |
| G = np.full((T, len(bp.COLS)), np.nan, np.float32) | |
| C = bp.CI | |
| t = np.arange(T) / 10.0 | |
| v = 15 + 5 * np.sin(t / 7) + rng.normal(0, 0.2, T) | |
| G[:, C["vEgo"]] = v | |
| G[:, C["aEgo"]] = np.gradient(v, 0.1) + rng.normal(0, 0.3, T) | |
| G[:, C["aEgo"]][100:110] = -3.0 # hard braking burst | |
| G[:, C["aEgo"]][300:306] = -1.2 # mild braking burst | |
| ang = np.cumsum(rng.normal(0, 0.3, T)) # gentle steering: |steer_rate| ~ 3 deg/s | |
| G[:, C["steeringAngleDeg"]] = ang | |
| sr = np.gradient(ang, 0.1) | |
| sr[200:205] = 140.0 # hard steer burst | |
| sr[400:404] = 45.0 # mild steer burst | |
| G[:, C["steer_rate"]] = sr | |
| G[:, C["brakePressed"]] = 0.0 | |
| G[:, C["brakePressed"]][100:112] = 1.0 | |
| lead = np.zeros(T, bool) | |
| lead[50:250] = True | |
| lead[350:520] = True | |
| dRel = np.where(lead, 40 - 0.05 * np.arange(T), 200.0) | |
| dRel[380:400] -= 20 # target switch (jump) | |
| vRel = np.where(lead, -2.0 + rng.normal(0, 0.1, T), 0.0) | |
| vLead = np.where(lead, v + vRel, 0.0) | |
| vLead[450:500] -= np.linspace(0, 6, 50) # lead braking | |
| G[:, C["lead_valid"]] = lead | |
| G[:, C["lead_dRel"]] = np.clip(dRel, 0, 200) | |
| G[:, C["lead_vRel"]] = vRel | |
| G[:, C["lead_vLead"]] = vLead | |
| ttc = dRel / np.maximum(-vRel, 0.1) | |
| ttc = np.clip(ttc, 0, 30) | |
| ttc[~lead | (vRel >= 0)] = np.nan | |
| G[:, C["ttc"]] = ttc | |
| lane_valid = np.ones(T, bool) | |
| lane_valid[520:560] = False | |
| lw = 3.4 + rng.normal(0, 0.05, T) | |
| lo = 0.6 * np.sin(t / 3) + rng.normal(0, 0.05, T) | |
| G[:, C["lane_valid"]] = lane_valid | |
| G[:, C["lane_width"]] = np.where(lane_valid, lw, 0.0) | |
| G[:, C["lane_center_offset"]] = np.where(lane_valid, lo, 0.0) | |
| G[:, C["lane_off"]] = np.where(lane_valid, lo, np.nan) | |
| G[:, C["badas_risk"]] = np.clip(0.5 + 0.4 * np.sin(t / 5) + rng.normal(0, 0.05, T), 0, 1) | |
| return G | |
| THR_FIXED = dict(badas=0.85) | |
| def _anchor_outputs(G, k): | |
| rd = bp.route_pressure(G, THR_FIXED) | |
| rd["t0"] = 0.0 | |
| gp = bp.gather_route(rd, np.array([k])) | |
| rr = br.route_response(G) | |
| rr["t0"] = 0.0 | |
| gr = br.gather_response(rr, np.array([k])) | |
| return gp, gr | |
| def test_no_future_leak_into_anchor_state(): | |
| G = synthetic_route() | |
| rng = np.random.default_rng(1) | |
| anchors = [60, 120, 210, 260, 385, 460] | |
| any_future_changed = False | |
| for k in anchors: | |
| gp0, gr0 = _anchor_outputs(G, k) | |
| G2 = G.copy() | |
| fut = slice(k + 1, None) # every sample strictly after the anchor | |
| G2[fut] = G2[fut] + rng.normal(0, 1.0, G2[fut].shape).astype(np.float32) * 3.0 | |
| G2[fut, bp.CI["lead_valid"]] = rng.integers(0, 2, G2[fut].shape[0]) | |
| G2[fut, bp.CI["lane_valid"]] = rng.integers(0, 2, G2[fut].shape[0]) | |
| G2[fut, bp.CI["brakePressed"]] = rng.integers(0, 2, G2[fut].shape[0]) | |
| gp1, gr1 = _anchor_outputs(G2, k) | |
| for key in ("P0", "P0valid", "S0", "M0", "M0valid", "Moving0"): | |
| assert np.array_equal(gp0[key], gp1[key]), (k, key) | |
| for key in ("R0", "R0valid", "Bits0", "Moving0"): | |
| assert np.array_equal(gr0[key], gr1[key]), (k, key) | |
| any_future_changed |= (not np.array_equal(gp0["P"], gp1["P"])) or (not np.array_equal(gr0["R"], gr1["R"])) | |
| assert any_future_changed, "perturbing the future must change some future target (test has no teeth otherwise)" | |
| def test_past_perturbation_can_change_anchor_state(): | |
| G = synthetic_route() | |
| k = 385 # lead present, approaching -> P0[ttc] True | |
| gp0, gr0 = _anchor_outputs(G, k) | |
| assert gp0["P0"][0, 0], "premise: ttc pressure active at the anchor" | |
| G2 = G.copy() | |
| G2[k - 5:k + 1, bp.CI["ttc"]] = np.nan # samples <= k: the anchor's own present/past | |
| G2[k - 5:k + 1, bp.CI["lead_valid"]] = 0.0 | |
| G2[k - 5:k + 1, bp.CI["lead_dRel"]] = 200.0 | |
| G2[k - 1, bp.CI["aEgo"]] = -5.0 # hard decel one sample before the anchor | |
| gp1, gr1 = _anchor_outputs(G2, k) | |
| assert not gp1["P0"][0, 0] and not np.array_equal(gp0["M0"], gp1["M0"]), "anchor pressure state / margins must follow samples <= k" | |
| assert gr1["R0"][0] == 2 and gr1["Bits0"][0] & br.BIT["aego_hard"], "aEgo at sample k-1 belongs to the anchor step" | |
| def test_steer_rate_lag_two_samples(): | |
| G = synthetic_route() | |
| k = next(k for k in range(130, 190) if _anchor_outputs(G, k)[1]["R0"][0] == 0 and _anchor_outputs(G, k)[1]["R0valid"][0]) | |
| _, gr0 = _anchor_outputs(G, k) # a quiet anchor: R0 == 0 | |
| for j in (k, k - 1): # samples k and k-1 are the 0.2 s look-ahead of the central gradient | |
| G2 = G.copy() | |
| G2[j, bp.CI["steer_rate"]] = 500.0 | |
| _, gr = _anchor_outputs(G2, k) | |
| assert gr["R0"][0] == 0, f"steer_rate sample {j} leaked into R0 at anchor {k}" | |
| G2 = G.copy() | |
| G2[k - 2, bp.CI["steer_rate"]] = 500.0 # sample k-2 -> lagged sr_c[k] -> read by the anchor step | |
| _, gr = _anchor_outputs(G2, k) | |
| assert gr["R0"][0] == 2 and (gr["Bits0"][0] & br.BIT["sr_hard"]), "sr_c[k] = steer_rate[k-2] must be read" | |
| G2 = G.copy() | |
| G2[k - 3, bp.CI["steer_rate"]] = 500.0 # k-3 -> sr_c[k-1] -> the step's first sample: also read (2-sample step) | |
| _, gr = _anchor_outputs(G2, k) | |
| assert gr["R0"][0] == 2 | |
| G2 = G.copy() | |
| G2[k - 4, bp.CI["steer_rate"]] = 500.0 # k-4 -> sr_c[k-2] -> previous step: not read | |
| _, gr = _anchor_outputs(G2, k) | |
| assert gr["R0"][0] == 0 | |
| def test_future_steps_read_only_their_own_past(): | |
| """P/R at future step j (grid index k+2j) must not depend on samples after k+2j.""" | |
| G = synthetic_route() | |
| k, j = 60, 7 | |
| gp0, gr0 = _anchor_outputs(G, k) | |
| G2 = G.copy() | |
| G2[k + 2 * j + 1:] += 5.0 | |
| G2[k + 2 * j + 1:, bp.CI["lead_valid"]] = 0.0 | |
| gp1, gr1 = _anchor_outputs(G2, k) | |
| assert np.array_equal(gp0["P"][0, :j + 1], gp1["P"][0, :j + 1]) | |
| assert np.array_equal(gp0["Pvalid"][0, :j + 1], gp1["Pvalid"][0, :j + 1]) | |
| assert np.array_equal(gr0["R"][0, :j + 1], gr1["R"][0, :j + 1]) | |
| def test_grid_columns_are_outside_dms_and_pose_blocks(): | |
| for c in bp.READ_COLS + br.READ_COLS: | |
| assert bp.CI[c] >= bp.N_FORBIDDEN | |
| assert bp.N_FORBIDDEN == 12 | |
| assert bp.COLS[:7] == ["face_yaw", "face_pitch", "face_roll", "blink", "eye", "isDistracted", "faceProb"] # the never-read block | |
| # ----------------------------------------------------------------------------- contract against the built files | |
| def test_pressure_response_contract(reg, sp): | |
| pp = CWM3 / "results" / "targets_pressure" / f"pressure_{reg}_{sp}.npz" | |
| pr = CWM3 / "results" / "targets_response" / f"response_{reg}_{sp}.npz" | |
| pd_ = DEM / f"demand_{reg}_{sp}.npz" | |
| if not (pp.exists() and pr.exists()): | |
| pytest.skip("targets not built yet") | |
| P, R = np.load(pp, allow_pickle=True), np.load(pr, allow_pickle=True) | |
| N = len(P["t"]) | |
| assert P["P"].shape == (N, 25, 6) and P["Pvalid"].shape == (N, 25, 6) and P["P0"].shape == (N, 6) and P["M0"].shape == (N, 10) | |
| assert P["P_first"].shape == (N, 6) and P["P_first"].min() >= -2 and P["P_first"].max() <= 24 | |
| assert list(P["families"]) == ["ttc", "ttc_drop", "cut_in", "lead_braking", "lane_excursion", "badas"] | |
| assert R["R"].shape == (N, 25) and R["R"].dtype == np.uint8 and R["R"].max() <= 2 and R["Rvalid"].shape == (N, 25) | |
| assert np.array_equal(R["Resp"], (R["R"] >= 1) & R["Rvalid"]) | |
| assert np.array_equal(P["uid"], R["uid"]) and np.array_equal(P["t"], R["t"]) | |
| assert not (P["P"] & ~P["Pvalid"]).any(), "fires only where valid" | |
| if pd_.exists(): | |
| D = np.load(pd_, allow_pickle=True) | |
| assert np.array_equal(D["uid"], P["uid"]) and np.array_equal(D["t"], P["t"]), "row order differs from targets_demand" | |
| assert np.array_equal(D["Resp"] & D["RespValid"], R["Resp"]) and np.array_equal(D["RespValid"], R["Rvalid"]) | |
| assert np.array_equal(D["Resp0"] & D["Resp0valid"], (R["R0"] >= 1) & R["R0valid"]) and np.array_equal(D["Resp0valid"], R["R0valid"]) | |
| fd = list(map(str, D["families"])) | |
| fp = list(map(str, P["families"])) | |
| for fam in ("ttc", "ttc_drop", "badas"): | |
| a, b = fd.index(fam), fp.index(fam) | |
| assert np.array_equal(D["D"][:, :, a], P["P"][:, :, b]) and np.array_equal(D["Dvalid"][:, :, a], P["Pvalid"][:, :, b]), fam | |
| assert np.array_equal(D["D0"][:, a], P["P0"][:, b]) and np.array_equal(D["D0valid"][:, a], P["P0valid"][:, b]), fam | |
| assert np.array_equal(D["Moving"], P["Moving"]) and np.array_equal(D["InRange"], P["InRange"]) | |
| def test_behaviour_contract(reg, sp): | |
| pb = CWM3 / "results" / "targets_behaviour" / f"behaviour_{reg}_{sp}.npz" | |
| if not pb.exists(): | |
| pytest.skip("behaviour targets not built yet") | |
| B = np.load(pb, allow_pickle=True) | |
| s12 = np.load(CWM3.parent / "s12_wm" / "results" / f"targets_{reg}_{sp}.npz", allow_pickle=True) | |
| N = len(s12["t"]) | |
| assert np.array_equal(B["uid"], s12["uid"]) and np.allclose(B["t"], s12["t"]) | |
| assert B["B"].shape == (N, 25, 3) and B["Bvalid"].shape == (N, 25, 3) and B["B_source"].shape == (N,) and B["B_first"].shape == (N, 3) | |
| assert not (B["B"].astype(bool) & ~B["Bvalid"]).any() | |
| causal = np.isin(B["B_source"], [1, 3]) | |
| assert np.array_equal(B["Bc_only"][causal], B["B"][causal]) and not B["Bc_valid"][~causal].any() | |
| assert not B["Bvalid"][B["B_source"] == 0].any() | |
| kc = CWM3 / "results" / "targets_kin_causal" / f"kin_causal_{reg}_{sp}.npz" | |
| if kc.exists(): | |
| k = np.load(kc, allow_pickle=True) | |
| rows = B["B_source"] == 1 | |
| assert np.array_equal(B["B"][rows], k["E_c"][rows]) and np.array_equal(B["Bvalid"][rows], k["EM_c"][rows]) | |
| assert rows.sum() <= k["row_ok"].sum() | |
Xet Storage Details
- Size:
- 11.8 kB
- Xet hash:
- 58bb45855b96f3fdb69afdcb09dcb5c08fa71b06ebaccd781f1068129a19f720
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.