Buckets:
| """cwm3/tests/test_road_dynamics_targets.py — data/build_road_dynamics_targets.py contract: strict future, exact grid alignment, side-stream composition. | |
| (1) Strict future: perturbing every 10 Hz sample AFTER the anchor changes only Yf / Yw / Yvalid — never H / Hvalid / M0 / ecan (the route thresholds are | |
| held fixed; the badas route quantile is a label definition and is not an input here anyway). | |
| (2) Perturbing samples BEFORE the anchor changes H / ecan (the test has teeth) and never Yf. | |
| (3) Exact alignment: Yf[:, j] equals the grid sample k + 2(j+1); H[:, -1] equals the anchor sample; Hvalid / Yvalid follow the validity channels. | |
| (4) ecan feature list has 40 names, the M0 block equals build_pressure_targets' M0, the summaries respect the sentinels. | |
| (5) vfcan256 / can256 composition on a synthetic pack (Z / row_ok contract, 216 PCA + 40 standardised features, row_ok rule). | |
| (6) When the real packs exist (val): Yf at k=1..25 equals the grid sample, and E equals the pressure pack's ttc / lane_excursion / lead_braking families. | |
| Run: cd cwm3 && python -m pytest tests/test_road_dynamics_targets.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")) | |
| sys.path.insert(0, str(CWM3 / "tests")) | |
| import build_pressure_targets as bp # noqa: E402 | |
| import build_road_dynamics_targets as rdyn # noqa: E402 | |
| from test_targets_causal import synthetic_route # noqa: E402 | |
| def _route(G): | |
| return rdyn.route_dynamics(G) | |
| def test_strict_future_and_alignment(): | |
| G = synthetic_route(T=800, seed=1) | |
| k = np.array([100, 160, 230, 400, 460, 520]) | |
| rd = _route(G) | |
| base = rdyn.gather_dynamics(rd, k) | |
| # (3) alignment | |
| for j in range(rdyn.H_STEPS): | |
| for c, name in enumerate(rdyn.TARGET_CHANNELS): | |
| assert np.allclose(base["Yf"][:, j, c], np.nan_to_num(G[k + 2 * (j + 1), bp.CI[name]]), atol=1e-4), (j, name) | |
| for c, name in enumerate(rdyn.HIST_CHANNELS): | |
| assert np.allclose(base["H"][:, -1, c].astype(np.float32), np.nan_to_num(G[k, bp.CI[name]]), rtol=1e-2, atol=1e-2), name | |
| assert np.allclose(base["H"][:, 0, c].astype(np.float32), np.nan_to_num(G[k - 78, bp.CI[name]]), rtol=1e-2, atol=1e-2), name | |
| assert base["Hvalid"].all() and base["ecan_ok"].all() | |
| lead_k = G[k[:, None] + 2 * np.arange(1, 26)[None, :], bp.CI["lead_valid"]] > 0.5 | |
| assert np.array_equal(base["Yvalid"][:, :, 0], lead_k) and np.array_equal(base["Yvalid"][:, :, 2], lead_k) | |
| # (1) strict future: perturb everything after the LAST anchor's... no — after EACH anchor separately | |
| rng = np.random.default_rng(0) | |
| for i, ki in enumerate(k): | |
| G2 = G.copy() | |
| for name in rdyn.GRID_READ: | |
| G2[ki + 1:, bp.CI[name]] = rng.normal(0, 5, G2.shape[0] - ki - 1) + 5 | |
| G2[ki + 1:, bp.CI["lead_valid"]] = rng.random(G2.shape[0] - ki - 1) > 0.5 | |
| G2[ki + 1:, bp.CI["lane_valid"]] = rng.random(G2.shape[0] - ki - 1) > 0.5 | |
| G2[ki + 1:, bp.CI["badas_risk"]] = rng.random(G2.shape[0] - ki - 1) | |
| rd2 = rdyn.route_dynamics(G2) | |
| rd2["thr"] = rd["thr"] | |
| g2 = rdyn.gather_dynamics(rd2, k[i:i + 1]) | |
| for key in ("H", "Hvalid", "M0", "M0valid", "ecan", "ecan_ok"): | |
| assert np.array_equal(np.nan_to_num(g2[key].astype(np.float64)), np.nan_to_num(base[key][i:i + 1].astype(np.float64))), (key, ki) | |
| assert not np.allclose(g2["Yf"], base["Yf"][i:i + 1]), ki # the future targets DID change | |
| # (2) perturbing BEFORE the anchor changes the history / ecan, never Yf | |
| ki = k[3] | |
| G3 = G.copy() | |
| G3[ki - 60:ki - 10, bp.CI["lead_dRel"]] += 30.0 | |
| G3[ki - 60:ki - 10, bp.CI["lead_valid"]] = 1.0 | |
| rd3 = rdyn.route_dynamics(G3) | |
| g3 = rdyn.gather_dynamics(rd3, k[3:4]) | |
| assert not np.array_equal(g3["H"], base["H"][3:4]) and not np.array_equal(g3["ecan"], base["ecan"][3:4]) | |
| assert np.array_equal(g3["Yf"], base["Yf"][3:4]) and np.array_equal(g3["Yvalid"], base["Yvalid"][3:4]) | |
| def test_ecan_block_and_sentinels(): | |
| assert len(rdyn.ECAN_NAMES) == rdyn.N_ECAN == 40 and rdyn.ECAN_NAMES[:10] == [f"m0_{n}" for n in bp.MARGIN_NAMES] | |
| G = synthetic_route(T=800, seed=2) | |
| k = np.array([120, 300, 600]) | |
| rd = rdyn.route_dynamics(G) | |
| g = rdyn.gather_dynamics(rd, k) | |
| pm = bp.gather_route(bp.route_pressure(G), k) | |
| assert np.allclose(g["M0"], pm["M0"], atol=1e-4) and np.array_equal(g["M0valid"], pm["M0valid"]) | |
| assert np.allclose(g["ecan"][:, :10], g["M0"]) | |
| # a route with NO lead at all: lead summaries sit on the sentinels, lead fraction 0, ttc_min 30, headway_min 20 | |
| G0 = G.copy() | |
| G0[:, bp.CI["lead_valid"]] = 0.0 | |
| G0[:, bp.CI["lead_dRel"]] = bp.LEAD_CAP if hasattr(bp, "LEAD_CAP") else 200.0 | |
| G0[:, bp.CI["lead_vRel"]] = 0.0 | |
| G0[:, bp.CI["lead_vLead"]] = 0.0 | |
| g0 = rdyn.gather_dynamics(rdyn.route_dynamics(G0), k) | |
| E = {n: i for i, n in enumerate(rdyn.ECAN_NAMES)} | |
| assert np.allclose(g0["ecan"][:, E["h8_dRel_mean"]], 200.0) and np.allclose(g0["ecan"][:, E["h8_dRel_slope"]], 0.0) | |
| assert np.allclose(g0["ecan"][:, E["h8_lead_frac"]], 0.0) and np.allclose(g0["ecan"][:, E["h8_ttc_min"]], 30.0) and np.allclose(g0["ecan"][:, E["h8_headway_min"]], 20.0) | |
| assert np.allclose(g0["ecan"][:, E["h8_frac_approach"]], 0.0) and np.allclose(g0["ecan"][:, E["h8_n_lead_switch"]], 0.0) | |
| # anchors before the 8 s window fills: partial validity, features finite | |
| g1 = rdyn.gather_dynamics(rd, np.array([5, 30])) | |
| assert np.isfinite(g1["ecan"]).all() and g1["Hvalid"][0].sum() == 3 and g1["Hvalid"][1].sum() == 16 | |
| def test_masked_stats_slope(): | |
| dt = np.arange(-9, 1) / 10.0 | |
| x = (2.0 * dt + 1.0)[None, :].repeat(2, 0) | |
| w = np.ones_like(x) | |
| w[1, :7] = 0 | |
| mean, slope, mn, mx = rdyn._masked_stats(x, w, dt, -1, -1, -1) | |
| assert np.allclose(slope, 2.0) and np.allclose(mn[0], x[0].min()) and np.allclose(mx[1], x[1, 7:].max()) and np.allclose(mean[1], x[1, 7:].mean()) | |
| mean, slope, mn, mx = rdyn._masked_stats(x, np.zeros_like(x), dt, -1, -2, -3) | |
| assert np.allclose(mean, -1) and np.allclose(slope, 0) and np.allclose(mn, -2) and np.allclose(mx, -3) | |
| def test_side_stream_composition(tmp_path, monkeypatch): | |
| N = 50 | |
| rng = np.random.default_rng(0) | |
| out = tmp_path / "rd" | |
| pack = tmp_path / "pack_v2" | |
| out.mkdir(), pack.mkdir() | |
| monkeypatch.setattr(rdyn, "OUT", out) | |
| monkeypatch.setattr(rdyn, "PACK_V2", pack) | |
| for r in ("manual",): | |
| for s in ("train", "val"): | |
| e = rng.normal(3, 2, (N, 40)).astype(np.float32) | |
| ok = rng.random(N) > 0.1 | |
| e[~ok] = 0 | |
| np.savez(out / f"road_dyn_{r}_{s}.npz", ecan=e, ecan_ok=ok, uid=np.array([f"u{i}" for i in range(N)]), t=np.arange(N) * 0.2) | |
| Z = rng.normal(0, 3, (N, 256)).astype(np.float16) | |
| rok = rng.random(N) > 0.2 | |
| np.savez(pack / f"vf256_{r}_{s}.npz", Z=Z, row_ok=rok, n=np.int64(N)) | |
| files, st = rdyn.standardise_and_write_side(["manual"], ["train", "val"], force=False) | |
| assert len(files) == 4 and len(st["mu"]) == 40 | |
| with np.load(out / "road_dyn_manual_val.npz") as z, np.load(pack / "vf256_manual_val.npz") as v, np.load(out / "vfcan256_manual_val.npz") as a, np.load(out / "can256_manual_val.npz") as b: | |
| e, ok, Z, rok = z["ecan"], z["ecan_ok"], v["Z"].astype(np.float32), v["row_ok"] | |
| Za, Zb = a["Z"].astype(np.float32), b["Z"].astype(np.float32) | |
| assert Za.shape == Zb.shape == (N, 256) and int(a["n"]) == N and str(a["stream"]) == "vfcan256" and str(b["stream"]) == "can256" | |
| assert np.array_equal(a["row_ok"], rok | ok) and np.array_equal(b["row_ok"], ok) | |
| assert np.allclose(Za[:, :216], Z[:, :216] * rok[:, None], atol=2e-2) and np.all(Zb[:, :216] == 0) | |
| mu, sd = np.array(st["mu"], np.float32), np.array(st["sd"], np.float32) | |
| exp = np.clip((e - mu) / sd, -8, 8) * ok[:, None] | |
| assert np.allclose(Za[:, 216:], exp, atol=2e-2) and np.allclose(Zb[:, 216:], exp, atol=2e-2) | |
| with pytest.raises(FileExistsError): | |
| rdyn.standardise_and_write_side(["manual"], ["val"], force=False) | |
| def test_real_pack_alignment(regime): | |
| p = rdyn.OUT / f"road_dyn_{regime}_val.npz" | |
| if not p.exists(): | |
| pytest.skip("road_dyn pack missing") | |
| with np.load(p, allow_pickle=True) as z, np.load(rdyn.PRESS / f"pressure_{regime}_val.npz", allow_pickle=True) as pz: | |
| assert np.array_equal(z["uid"], pz["uid"]) and np.allclose(z["t"], pz["t"]) | |
| assert np.array_equal(z["E"], pz["P"][:, :, rdyn.EVENT_COLS]) and np.array_equal(z["E_first"], pz["P_first"][:, rdyn.EVENT_COLS]) | |
| assert np.allclose(z["M0"][z["M0valid"] & pz["M0valid"]], pz["M0"][z["M0valid"] & pz["M0valid"]], atol=1e-3) | |
| assert [str(c) for c in z["ecan_names"]] == rdyn.ECAN_NAMES and z["ecan"].shape[1] == 40 and z["H"].shape[1:] == (40, 8) and z["Yf"].shape[1:] == (25, 4) | |
| uid, t = z["uid"], z["t"] | |
| rows = np.random.default_rng(0).choice(len(t), 40, replace=False) | |
| for i in rows: | |
| G, tg = bp.load_grid(uid[i]) | |
| k = int(round((t[i] - tg[0]) * 10)) | |
| for j in (0, 4, 14, 24): | |
| if 0 <= k + 2 * (j + 1) < len(tg): | |
| for c, name in enumerate(rdyn.TARGET_CHANNELS): | |
| assert np.isclose(z["Yf"][i, j, c], np.nan_to_num(G[k + 2 * (j + 1), bp.CI[name]]), atol=1e-3), (uid[i], j, name) | |
| assert np.isclose(float(z["H"][i, -1, 0]), float(np.nan_to_num(G[k, bp.CI["lead_dRel"]])), rtol=1e-2, atol=0.2) | |
| def test_holdout_override_is_additive(tmp_path): | |
| """2026-09-17 WP5: `--holdout` path override — defaults unchanged, override touches only OUT / PRESS / PACK_V2, the frozen-stats loader refuses a foreign | |
| file, and standardise_and_write_side(frozen_stats=...) writes NO ecan_stats.npz and standardises with the given statistics.""" | |
| assert rdyn.OUT == CWM3 / "results" / "targets_road_dyn" and rdyn.PRESS == CWM3 / "results" / "targets_pressure" and rdyn.PACK_V2 == CWM3 / "results" / "pack_v2" | |
| import inspect | |
| assert inspect.signature(rdyn.standardise_and_write_side).parameters["frozen_stats"].default is None | |
| saved = (rdyn.OUT, rdyn.PRESS, rdyn.PACK_V2) | |
| try: | |
| p = rdyn.configure_holdout(tmp_path) | |
| assert p["OUT"] == rdyn.OUT == tmp_path / "targets_road_dyn" and rdyn.PRESS == tmp_path / "targets_pressure" and rdyn.PACK_V2 == tmp_path / "pack_v2" | |
| assert bp.S12 == CWM3.parent / "s12_wm" / "results" and bp.GRID == CWM3.parent / "s14_corr" / "cache" / "grid10" # untouched | |
| finally: | |
| rdyn.OUT, rdyn.PRESS, rdyn.PACK_V2 = saved | |
| assert rdyn.OUT == CWM3 / "results" / "targets_road_dyn" | |
| # frozen statistics: pinned sha256 + run-json equality; a re-saved copy (different bytes) is refused | |
| if rdyn.FROZEN_ECAN_STATS.exists(): | |
| mu, sd, n = rdyn.load_frozen_ecan_stats() | |
| assert mu.shape == sd.shape == (40,) and n == 1457322 and np.all(sd > 0) | |
| bad = tmp_path / "ecan_stats_copy.npz" | |
| np.savez(bad, mu=mu.astype(np.float32), sd=sd.astype(np.float32), n=np.int64(n), names=np.array(rdyn.ECAN_NAMES)) | |
| with pytest.raises(AssertionError): | |
| rdyn.load_frozen_ecan_stats(bad) | |
| # frozen path on a synthetic pack: no ecan_stats.npz written, the given mu / sd used, meta records the frozen source | |
| N = 40 | |
| rng = np.random.default_rng(3) | |
| out, pack = tmp_path / "rd", tmp_path / "pack_v2" | |
| out.mkdir(), pack.mkdir() | |
| rdyn.OUT, rdyn.PACK_V2 = out, pack | |
| try: | |
| e = rng.normal(3, 2, (N, 40)).astype(np.float32) | |
| ok = rng.random(N) > 0.1 | |
| e[~ok] = 0 | |
| np.savez(out / "road_dyn_manual_holdout.npz", ecan=e, ecan_ok=ok, uid=np.array([f"u{i}" for i in range(N)]), t=np.arange(N) * 0.2) | |
| Z, rok = rng.normal(0, 3, (N, 256)).astype(np.float16), rng.random(N) > 0.2 | |
| np.savez(pack / "vf256_manual_holdout.npz", Z=Z, row_ok=rok, n=np.int64(N)) | |
| mu0, sd0 = rng.normal(0, 1, 40), rng.uniform(0.5, 2, 40) | |
| files, st = rdyn.standardise_and_write_side(["manual"], ["holdout"], force=False, frozen_stats=(mu0, sd0, 7)) | |
| assert len(files) == 2 and not (out / "ecan_stats.npz").exists() and st["n"] == 7 | |
| with np.load(out / "vfcan256_manual_holdout.npz") as a: | |
| exp = np.clip((e - mu0.astype(np.float32)) / sd0.astype(np.float32), -8, 8) * ok[:, None] | |
| assert np.allclose(a["Z"].astype(np.float32)[:, 216:], exp, atol=2e-2) and np.array_equal(a["row_ok"], rok | ok) | |
| assert bool(a["frozen_stats"]) and str(a["ecan_stats_sha256"]) == rdyn.FROZEN_ECAN_STATS_SHA256 | |
| finally: | |
| rdyn.OUT, rdyn.PRESS, rdyn.PACK_V2 = saved | |
Xet Storage Details
- Size:
- 12.8 kB
- Xet hash:
- 331598b67e7a0df10939f20884cd909f9851ce884b954dbbc3b431570047ab6c
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.