Buckets:
| #!/usr/bin/env python3 | |
| """cwm3/baselines_dms/tier_status.py — is extraction tier X complete for the given model sets? (layer-2 launcher gate, 2026-09-14) | |
| Completion of tier T for a set S (cpu = ov_gaze, ov_action, mp_face; gpu = openface3, l2cs_gaze, videomae2; models taken from the set's progress record) holds | |
| when ANY of these is true — they are OR-ed because each alone has a hole: | |
| (a) progress.json[S].tiers[T].routes_done == routes_total (every queued route of the tier finished in the current chunk; fails / no-pose skips break it) | |
| (b) progress.json[S].note in {ALL_DONE, FAILURES} (the chunk walked the whole queue) | |
| (c) sum of routes_done over the tiers AFTER T > 0 (run_full_extraction walks the queue strictly in tier order A -> B -> C -> D and re-visits | |
| finished routes as `exists` at every chunk start, so progress in a later tier means every | |
| row of T was visited — done, failed or skipped — in this chunk) | |
| (d) cache-based: every route_queue.csv row of tier T with video_local & pose_cache has a COMPLETE cache <uid>.npz (no <uid>.part.npz) for every model of the set, | |
| plus windows/<uid>.npz for the window models (ov_action, videomae2) — independent of progress.json (which resets at every chunk start) | |
| Routes without a pose cache are never extracted (every frame would be fail 1) and are excluded from (d). Exit 0 = complete for ALL requested sets; 1 otherwise. | |
| Nothing here reads a video, a model or any openpilot field. | |
| python baselines_dms/tier_status.py --tier A --sets cpu,gpu [--json] [--verbose] | |
| python baselines_dms/tier_status.py --tiers A,B,C --sets cpu,cpu_shard2,gpu --cache-only # layer-3 gate (2026-09-15, additive) | |
| Layer-3 gate (2026-09-15, additive): `--tiers T1,T2,..` checks several tiers at once (exit 0 only when every tier is complete for every set) and `--cache-only` | |
| counts a (set, tier) as complete ONLY by criterion (d). Reason: the second CPU process (`cpu_shard2`, run_full_extraction --reverse --tiers C,D) walks the queue | |
| BACKWARDS, so its criterion (c) ("progress in a later tier") is true for tier C as soon as it has touched tier D, and (a) / (b) are per progress key while the two | |
| CPU keys (`cpu`, `cpu_shard2`) share one cache directory per model. (d) is progress-key independent (cache files + windows/ for ov_action / videomae2), so with | |
| `--sets cpu,cpu_shard2,gpu --cache-only` the six models' caches for every route of the requested tiers decide, whichever process wrote them. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import csv | |
| import json | |
| import sys | |
| from pathlib import Path | |
| PKG = Path(__file__).resolve().parent | |
| CWM3 = PKG.parent | |
| FULL = CWM3 / "results" / "baselines_dms" / "full" | |
| CACHE = CWM3 / "results" / "baselines_dms" / "cache" | |
| TIERS = ("A", "B", "C", "D") | |
| DEFAULT_MODELS = {"cpu": ["ov_gaze", "ov_action", "mp_face"], "gpu": ["openface3", "l2cs_gaze", "videomae2"]} | |
| WINDOW_MODELS = ("ov_action", "videomae2") | |
| def load_progress(path=FULL / "progress.json"): | |
| try: | |
| return json.loads(Path(path).read_text()) | |
| except Exception as e: # noqa: BLE001 (a half-written file under the writer's lock: treat as unknown) | |
| return {"_error": f"{type(e).__name__}: {e}"} | |
| def queue_rows(tier, path=FULL / "route_queue.csv"): | |
| with open(path, newline="") as fh: | |
| return [r for r in csv.DictReader(fh) if r["tier"] == tier] | |
| def cache_complete(uid, model): | |
| d = CACHE / model | |
| if not (d / f"{uid}.npz").exists() or (d / f"{uid}.part.npz").exists(): | |
| return False | |
| if model in WINDOW_MODELS and (not (d / "windows" / f"{uid}.npz").exists() or (d / "windows" / f"{uid}.part.npz").exists()): | |
| return False | |
| return True | |
| def set_status(set_name, tier, prog, rows, cache_only=False): | |
| rec = prog.get(set_name) or {} | |
| models = rec.get("models") or DEFAULT_MODELS.get(set_name, []) | |
| tiers = rec.get("tiers") or {} | |
| tv = tiers.get(tier) or {} | |
| later = TIERS[TIERS.index(tier) + 1:] | |
| a = bool(tv) and tv.get("routes_total", -1) > 0 and tv.get("routes_done", 0) >= tv.get("routes_total", -1) | |
| b = rec.get("note") in ("ALL_DONE", "FAILURES") | |
| c = sum(int((tiers.get(t) or {}).get("routes_done", 0)) for t in later) > 0 | |
| eligible = [r["uid"] for r in rows if int(r["video_local"]) and int(r["pose_cache"])] | |
| missing = {m: [u for u in eligible if not cache_complete(u, m)] for m in models} | |
| d = bool(models) and all(len(v) == 0 for v in missing.values()) | |
| return {"set": set_name, "tier": tier, "models": models, "record_present": bool(rec), "note": rec.get("note"), "current_route": rec.get("current_route"), | |
| "chunk_started_utc": rec.get("chunk_started_utc"), "tier_record": tv, "criteria": {"a_all_routes_done": a, "b_chunk_walked_queue": b, "c_later_tier_progress": c, "d_caches_complete": d}, | |
| "eligible_routes": len(eligible), "routes_no_pose": len(rows) - len(eligible), "missing_caches": {m: len(v) for m, v in missing.items()}, | |
| "missing_examples": {m: v[:3] for m, v in missing.items() if v}, "cache_only": bool(cache_only), "complete": bool(d) if cache_only else bool(a or b or c or d)} | |
| def status(tier, sets, cache_only=False): | |
| prog = load_progress() | |
| rows = queue_rows(tier) | |
| out = {"tier": tier, "queue_rows": len(rows), "progress_updated_utc": prog.get("updated_utc"), "progress_error": prog.get("_error"), "cache_only": bool(cache_only), | |
| "sets": {s: set_status(s, tier, prog, rows, cache_only) for s in sets}} | |
| out["complete"] = all(v["complete"] for v in out["sets"].values()) | |
| return out | |
| def status_multi(tiers, sets, cache_only=False): | |
| """several tiers at once (layer-3 gate): complete only when every requested tier is complete for every requested set.""" | |
| out = {"tiers": {t: status(t, sets, cache_only) for t in tiers}, "cache_only": bool(cache_only)} | |
| out["complete"] = all(v["complete"] for v in out["tiers"].values()) | |
| return out | |
| def main(argv=None): | |
| ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) | |
| ap.add_argument("--tier", default="A", choices=TIERS) | |
| ap.add_argument("--tiers", default="", help="comma list of tiers checked together (overrides --tier); exit 0 only when ALL are complete for all sets") | |
| ap.add_argument("--sets", default="cpu,gpu") | |
| ap.add_argument("--cache-only", action="store_true", help="a (set, tier) is complete only by criterion (d) (cache files); progress-key independent") | |
| ap.add_argument("--json", action="store_true") | |
| ap.add_argument("--verbose", action="store_true") | |
| a = ap.parse_args(argv) | |
| sets = [s.strip() for s in a.sets.split(",") if s.strip()] | |
| tiers = [t.strip() for t in a.tiers.split(",") if t.strip()] or [a.tier] | |
| for t in tiers: | |
| assert t in TIERS, t | |
| multi = status_multi(tiers, sets, a.cache_only) | |
| if a.json or a.verbose: | |
| print(json.dumps(multi if len(tiers) > 1 else multi["tiers"][tiers[0]], indent=1, default=str)) | |
| else: | |
| for t in tiers: | |
| st = multi["tiers"][t] | |
| for s, v in st["sets"].items(): | |
| tv = v["tier_record"] | |
| print(f"tier {t} {s}: complete={v['complete']}{' (cache-only)' if a.cache_only else ''} criteria={v['criteria']} routes_done {tv.get('routes_done')}/{tv.get('routes_total')} note={v['note']} missing_caches={v['missing_caches']}") | |
| print(f"tier {t} complete for all sets: {st['complete']}") | |
| if len(tiers) > 1: | |
| print(f"tiers {','.join(tiers)} complete for all sets: {multi['complete']}") | |
| return 0 if multi["complete"] else 1 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 7.84 kB
- Xet hash:
- 59087b1a41d1d193ca548d6d5c9babcea84204d9905b740cd3cd4eb2be386163
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.