Buckets:
| """cwm3/tests/test_no_dms_inputs.py — no cwm3 model / train / eval / data / common code reads an openpilot DMS field or | |
| grid10 columns 0-11, and none of those packages imports weak/ (the only place DMS fields may be read). | |
| (a) static tokens : the forbidden DMS tokens never appear in cwm3/{models,train,eval,data,common}/*.py (this test excepted) | |
| (b) grid10 column rule : a file in those packages may mention `grid10` only if every grid10 column name it contains as a string | |
| literal has index >= 12 in the AST-recovered grid_s14 layout (i.e. is a canspec CAN column or an R-block column) and it | |
| never indexes G with a literal column < 12. (The raw token ban of the first version broke on docstrings of the causal-pose | |
| scripts that read the whitelisted vEgo column; the column rule is the actual invariant.) | |
| (c) no weak/ import : models/ train/ eval/ data/ common/ never import from weak/ (regex on import statements and sys.path edits) | |
| (d) single DMS reader : across the whole cwm3 tree (tests/docs excepted) the DMS tokens appear only in weak/dms_teacher_targets.py | |
| and weak/dms_teacher_pretrain.py (the declared readers) | |
| (e) the data layer's compact contract carries no forbidden key and no DMS file path | |
| Run: cd cwm3 && python -m pytest tests/test_no_dms_inputs.py -q | |
| """ | |
| from __future__ import annotations | |
| import ast | |
| import re | |
| import sys | |
| from pathlib import Path | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| sys.path.insert(0, str(CWM3 / "data")) | |
| FORBIDDEN = ["driver_state", "faceProb", "isDistracted", "face_yaw", "face_pitch", "face_roll", "blinkProb", "dmonitoring", | |
| "driverState", "awarenessStatus", "distractedPct", "dstate", "leftEyeProb", "rightEyeProb"] | |
| DIRS = ("models", "train", "eval", "data", "common") | |
| DMS_READERS = {CWM3 / "weak" / "dms_teacher_targets.py", CWM3 / "weak" / "dms_teacher_pretrain.py", | |
| CWM3 / "weak" / "face_vs_dms_diag.py"} # 2026-09-12: DIAGNOSTIC correlation of our landmark channels with the derived DMS channels (train/val anchors; never a feature) | |
| N_FORBIDDEN_COLS = 12 | |
| def _py_files(dirs=DIRS): | |
| out = [] | |
| for d in dirs: | |
| out += sorted((CWM3 / d).glob("*.py")) | |
| return out | |
| def _grid_cols(): | |
| sys.path.insert(0, str(EXPLORE / "s3_can_baseline")) | |
| import canspec | |
| tree = ast.parse((EXPLORE / "s14_corr" / "grid_s14.py").read_text()) | |
| lists = {} | |
| for node in tree.body: | |
| if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name) and node.targets[0].id in ("DCOLS", "PCOLS", "RCOLS"): | |
| lists[node.targets[0].id] = [ast.literal_eval(e) for e in node.value.elts] | |
| return lists["DCOLS"] + lists["PCOLS"] + list(canspec.FEATURES) + lists["RCOLS"] | |
| def test_static_no_dms_tokens(): | |
| pat = re.compile("|".join(map(re.escape, FORBIDDEN))) | |
| hits = [] | |
| for p in _py_files(): | |
| for i, line in enumerate(p.read_text().splitlines(), 1): | |
| if pat.search(line) and "FORBIDDEN" not in line and "never read" not in line.lower(): | |
| hits.append(f"{p.relative_to(CWM3)}:{i}: {line.strip()[:100]}") | |
| assert not hits, "forbidden DMS tokens in cwm3 code:\n" + "\n".join(hits) | |
| def _grid10_reader(path): | |
| """A file READS grid10 if it names the directory as a path component (quoted 'grid10' or /grid10/), outside guard lists.""" | |
| pat = re.compile(r"[\"']grid10[\"']|/grid10/|cache/grid10") | |
| return any(pat.search(l) and "FORBIDDEN" not in l for l in path.read_text().splitlines()) | |
| def _string_literals_outside_guards(path): | |
| lines = path.read_text().splitlines() | |
| out = [] | |
| for node in ast.walk(ast.parse(path.read_text())): | |
| if isinstance(node, ast.Constant) and isinstance(node.value, str) and "FORBIDDEN" not in lines[node.lineno - 1]: | |
| out.append(node.value) | |
| return out | |
| def test_grid10_readers_touch_only_can_and_r_columns(): | |
| cols = _grid_cols() | |
| forbidden_names = set(cols[:N_FORBIDDEN_COLS]) | |
| assert forbidden_names >= {"face_yaw", "faceProb", "isDistracted", "hands_on_wheel"} | |
| g_index = re.compile(r"\bG\s*\[\s*[^\]]*?,\s*(\d+)\s*\]") | |
| bad, readers = [], [] | |
| for p in _py_files(): | |
| if not _grid10_reader(p): | |
| continue | |
| readers.append(p.name) | |
| for c in sorted(set(_string_literals_outside_guards(p)) & set(cols)): | |
| if c in forbidden_names: | |
| bad.append(f"{p.relative_to(CWM3)}: names grid10 column {c!r} (index {cols.index(c)} < {N_FORBIDDEN_COLS})") | |
| for m in g_index.finditer(p.read_text()): | |
| if int(m.group(1)) < N_FORBIDDEN_COLS: | |
| bad.append(f"{p.relative_to(CWM3)}: literal G[:, {m.group(1)}] indexes the DMS/pose block") | |
| assert "build_pressure_targets.py" in readers, readers # the rule must actually see the known readers | |
| assert not bad, "\n".join(bad) | |
| def test_no_weak_imports_in_model_train_eval_data(): | |
| pat = re.compile(r"^\s*(from\s+weak[\s.]|import\s+weak\b|from\s+\.?weak\b)|sys\.path\.(insert|append)\([^)]*[\"'/]weak[\"'/)]|[\"']weak[\"']\s*\)?\s*/|/\s*[\"']weak[\"']") | |
| hits = [] | |
| for p in _py_files(): | |
| for i, line in enumerate(p.read_text().splitlines(), 1): | |
| if pat.search(line): | |
| hits.append(f"{p.relative_to(CWM3)}:{i}: {line.strip()[:100]}") | |
| assert not hits, "models/train/eval/data/common must never import weak/:\n" + "\n".join(hits) | |
| def test_single_dms_reader_in_tree(): | |
| pat = re.compile("|".join(map(re.escape, FORBIDDEN))) | |
| hits = [] | |
| for p in CWM3.rglob("*.py"): | |
| rel = p.relative_to(CWM3).parts | |
| if rel[0] in ("tests", "docs", "aide", "results", "release", "demo") or "__pycache__" in rel or p.resolve() in {q.resolve() for q in DMS_READERS}: | |
| continue | |
| for i, line in enumerate(p.read_text().splitlines(), 1): | |
| if pat.search(line) and "FORBIDDEN" not in line and "never read" not in line.lower(): | |
| hits.append(f"{'/'.join(rel)}:{i}: {line.strip()[:100]}") | |
| assert not hits, "DMS tokens outside the declared readers:\n" + "\n".join(hits) | |
| def test_dms_reader_declares_itself(): | |
| p = CWM3 / "weak" / "dms_teacher_targets.py" | |
| if not p.exists(): | |
| return | |
| head = "\n".join(p.read_text().splitlines()[:40]) | |
| assert "ONLY" in head and "DMS" in head, "the DMS reader must state in its header that it is the only DMS reader" | |
| def test_data_layer_files_are_dms_free(): | |
| import cwm3_dataset as ds | |
| for regime in ds.REGIMES: | |
| for split in ds.SPLITS: | |
| for k, p in ds.files(regime, split).items(): | |
| s = str(p) | |
| assert "driver_state" not in s and "grid10" not in s and "dms" not in s.lower(), (k, s) | |
| for k in ds.FORBIDDEN_KEYS: | |
| assert k not in ds.KIN_CHANNELS | |
Xet Storage Details
- Size:
- 6.79 kB
- Xet hash:
- fbe85bd812fb4412f78e61a358a4c3e9b4047317a042590f5bafcc826144e27b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.