Buckets:
| """tests/test_warning_fs_block.py — the F2′ future-state block of the learned warning read-out (train/warning_features_ext.fs_block; 2026-09-17, owner decision (6)): | |
| names / order agree with names_for and extra_block, values are probabilities / seconds, the remaining-off-dwell feature is NaN exactly on on-task rows, invalid pack rows are | |
| NaN with fs_valid 0, monotone constraints are +1 on the off-task quantities and 0 on fs_leave3_* / fs_valid, and the column tag is `dwfcfs`. Reads the real FS_dwell_s0 pack | |
| (skipped when absent). No openpilot DMS field is read. | |
| python -m pytest -q tests/test_warning_fs_block.py | |
| """ | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(CWM3 / "train")) | |
| import warning_features_ext as WFX # noqa: E402 | |
| FS = CWM3 / "results" / "runs" / "FS_dwell_s0" | |
| def test_names_tag_and_monotone(): | |
| assert len(WFX.FS_NAMES) == 22 and WFX.names_for(("fs",)) == WFX.FS_NAMES | |
| assert WFX.tag_for(("dwell", "face", "fs")) == "dwfcfs" and WFX.tag_for(("fs",)) == "fs" | |
| assert WFX.names_for(("dwell", "face", "fs"))[-22:] == WFX.FS_NAMES | |
| mono = {n: WFX.monotone(n) for n in WFX.FS_NAMES} | |
| assert all(mono[n] == 1 for n in WFX.FS_NAMES if n.startswith(("fs_off_", "fs_EDoff_", "fs_remE_"))) | |
| assert all(mono[n] == 0 for n in WFX.FS_NAMES if n.startswith("fs_leave3_") or n == "fs_valid") | |
| def test_fs_block_values_on_real_pack(): | |
| WFX.set_fs_run(FS) | |
| rows = np.arange(0, 4000) | |
| X, names = WFX.fs_block("manual", "val", rows) | |
| assert names == WFX.FS_NAMES and X.shape == (len(rows), 22) and X.dtype == np.float32 | |
| with np.load(FS / "fs_manual_val.npz", allow_pickle=True) as z: | |
| valid = z["valid"][rows].astype(bool) | |
| z0 = z["z0"][rows].astype(np.float32) | |
| z0v = z["z0valid"][rows].astype(bool) | |
| off = z["off_value"].astype(np.float32) | |
| pio = z["pi_off"][rows].astype(np.float32) | |
| fv = X[:, names.index("fs_valid")] | |
| assert np.array_equal(fv > 0.5, valid) | |
| if (~valid).any(): | |
| assert np.isnan(X[~valid][:, :-1]).all() | |
| v = valid | |
| for j, n in enumerate(names[:-1]): | |
| col = X[v, j] | |
| fin = np.isfinite(col) | |
| if n.startswith(("fs_off_", "fs_leave3_")): | |
| assert fin.all() and col.min() >= 0.0 and col.max() <= 1.0, n | |
| elif n.startswith("fs_EDoff_"): | |
| assert fin.all() and col.min() >= 0.0 and col.max() <= 5.0 + 1e-3, n | |
| elif n.startswith("fs_remE_"): | |
| s = WFX.FS_STATES.index(n.split("fs_remE_")[1]) | |
| is_off = z0v[v, s] & (np.abs(z0[v, s] - off[s]) < 0.5) | |
| assert np.array_equal(fin, is_off), n # defined exactly on off-task rows | |
| assert col[fin].min() >= 0.2 - 1e-3 and col[fin].max() <= 5.2 + 1e-3 | |
| # the +3 s off occupancy column equals pi_off[:, 14, s] | |
| for s, st in enumerate(WFX.FS_STATES): | |
| j = names.index(f"fs_off_{st}_3s") | |
| assert np.allclose(X[v, j], np.clip(pio[v, 14, s], 0, 1), atol=1e-6) | |
| def test_extra_block_order_matches_names_for(): | |
| WFX.set_fs_run(FS) | |
| sys.path.insert(0, str(CWM3 / "data")) | |
| import cwm3_dataset as ds | |
| ds.set_kin_source("causal_routes", "v2") | |
| rows = np.arange(100, 400) | |
| X, names = WFX.extra_block("manual", "train", rows, ("dwell", "face", "fs")) | |
| assert names == WFX.names_for(("dwell", "face", "fs")) and X.shape == (len(rows), len(names)) | |
Xet Storage Details
- Size:
- 3.64 kB
- Xet hash:
- 6f744bbadd601107ce70d62c0e47a2d1ed60edab4b0696ec3e3d9850d4a1c54b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.