Buckets:
| """cwm3/eval/pc_joint_deferred.py — eval/paired_compare_joint.py with a DEFERRED append to the shared results/paired_compare.csv (2026-09-17, ECAN Table 3 lane). | |
| Why: eval/pressure_eval.py appends to the shared CSV without the flock of eval/pc_joint_locked.py, so while a sequential P-cell batch is running no other | |
| writer may touch the shared file (an 86-row append spans several write() calls; an unlocked 6-row append landing in between would split a row). This | |
| wrapper lets joint pairings run meanwhile: the comparison runs unchanged (same arguments as paired_compare_joint.py) but its rows stay in a PRIVATE file | |
| results/tmp_pc/deferred/<tag>_<regime>.csv; `--merge` later appends every deferred file to results/paired_compare.csv under the exclusive flock on | |
| results/paired_compare.csv.lock (exactly pc_joint_locked's append) and deletes them. Nothing about the statistics changes. CPU only; no openpilot DMS field. | |
| python eval/pc_joint_deferred.py --regime manual --a results/runs/A --b results/runs/B --tag T --kin-targets causal_routes --kin-version v2 | |
| python eval/pc_joint_deferred.py --merge | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import fcntl | |
| import os | |
| import sys | |
| import time | |
| from pathlib import Path | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(CWM3 / "eval")) | |
| import paired_compare_joint as PCJ # noqa: E402 | |
| SHARED = CWM3 / "results" / "paired_compare.csv" | |
| LOCK = CWM3 / "results" / "paired_compare.csv.lock" | |
| DEFER = CWM3 / "results" / "tmp_pc" / "deferred" | |
| def merge(): | |
| files = sorted(DEFER.glob("*.csv")) if DEFER.exists() else [] | |
| if not files: | |
| print("nothing deferred") | |
| return | |
| rows = [] | |
| for f in files: | |
| with open(f, newline="") as fh: | |
| rows += list(csv.DictReader(fh)) | |
| t0 = time.time() | |
| with open(LOCK, "a") as lk: | |
| fcntl.flock(lk, fcntl.LOCK_EX) | |
| new = not SHARED.exists() or SHARED.stat().st_size == 0 | |
| with open(SHARED, "a", newline="") as f: | |
| w = csv.DictWriter(f, fieldnames=PCJ.COLS) | |
| if new: | |
| w.writeheader() | |
| for r in rows: | |
| w.writerow(r) | |
| f.flush() | |
| os.fsync(f.fileno()) | |
| fcntl.flock(lk, fcntl.LOCK_UN) | |
| for f in files: | |
| f.unlink() | |
| print(f"locked-append {len(rows)} deferred rows from {len(files)} files to {SHARED} ({time.time() - t0:.1f}s)") | |
| def main(): | |
| args = sys.argv[1:] | |
| if args == ["--merge"]: | |
| merge() | |
| return | |
| tag = args[args.index("--tag") + 1] if "--tag" in args else "untagged" | |
| reg = args[args.index("--regime") + 1] if "--regime" in args else "x" | |
| DEFER.mkdir(parents=True, exist_ok=True) | |
| private = DEFER / f"{tag}_{reg}.csv" | |
| if private.exists(): | |
| private.unlink() | |
| PCJ.OUT = private # the module-level path paired_compare_joint.main() appends to | |
| sys.argv = [str(CWM3 / "eval" / "paired_compare_joint.py")] + args | |
| PCJ.main() | |
| if not private.exists(): | |
| print(f"no rows written for {tag} {reg}") | |
| return | |
| with open(private, newline="") as f: | |
| n = sum(1 for _ in csv.DictReader(f)) | |
| print(f"deferred {n} rows ({tag} {reg}) in {private}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 3.31 kB
- Xet hash:
- e1844f59ac0dbf7db8c0ef7ad704e7d1dc95c9fbeaa6eb4aa6f855a5e4fe9835
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.