Buckets:
| """cwm3/tests/test_full_extraction.py — contracts of the shared-decode full-extraction runner (baselines_dms/run_full_extraction.py, 2026-09-14). | |
| (a) planning : plan_job gives the Phase-1 spans per model (per-frame = the anchor spans; ov_action + 1.5 s lead-in; videomae2 lead_in_spans), subtracts what an | |
| existing (partial) cache covers, and re-decodes the 3.0 s clip lead-in before a resumed videomae2 span; a complete pair of caches plans nothing. | |
| (b) geometry : scaled_box equals video_io.FrameStream.box_for_frame (the box a model with a decode_wh saw in Phase 1), and the resized view is the same cv2 call. | |
| (c) windows : merge_window_caches unions two window caches (later rows win); videomae2.WindowAssembler reproduces the run_route_windows clip rule on a dummy model | |
| (16 frames at 5 Hz ending at every k5, fail 5 before the decoded start, fail 1 when a frame has no crop). | |
| (d) queue : queue_summary / load_queue round trip on a synthetic queue; every tier present; frame counts add up. | |
| (e) hygiene : the runner / launchers contain no forbidden DMS token and the runner is never imported by models/ train/ eval/ data/ common/. | |
| (f) atomic : RouteCache.save / WindowCache.save write tmp + os.replace: a failed write leaves the old file intact and no *.tmp.npz behind (2026-09-15). | |
| (g) claims : <uid>.claim held by a live other pid -> deferred / dead pid -> taken over / only our own claim is released; a fresh .part.npz by another tag defers | |
| a route (--skip-fresh-part-min), our own or an old one does not; run_queue with two 'processes' (stubbed process_route). | |
| (h) sharding : select_rows(--reverse) flips the filtered queue; the argv of the running launchers plans exactly the old selection with the old gates. | |
| No video, no model weights, no GPU: the tests stub what they need. | |
| Run: cd cwm3 && python -m pytest -q tests/test_full_extraction.py | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import json | |
| import os | |
| import re | |
| import sys | |
| import tempfile | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pytest | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| PKG = CWM3 / "baselines_dms" | |
| for p in (str(CWM3), str(PKG), str(PKG / "extractors"), str(CWM3 / "tests")): | |
| if p not in sys.path: | |
| sys.path.insert(0, p) | |
| import cache_schema as cs # noqa: E402 | |
| import video_io as vio # noqa: E402 | |
| import run_full_extraction as RF # noqa: E402 | |
| # ----------------------------------------------------------------------------- (a) planning | |
| def test_plan_per_frame_model_fresh_and_resume(): | |
| req = np.array([[100.0, 120.0], [300.0, 310.0]]) | |
| p = RF.plan_job("ov_gaze", req, None, None, 100000) | |
| assert np.allclose(p["rc_req"], req) and np.allclose(p["todo_dec"], req) and p["wc_req"] is None and p["windows_complete"] | |
| p2 = RF.plan_job("mp_face", req, np.array([[100.0, 110.0]]), None, 100000) | |
| assert np.allclose(p2["todo_dec"], [[110.0, 120.0], [300.0, 310.0]], atol=1e-4) | |
| p3 = RF.plan_job("ov_gaze", req, req, None, 100000) | |
| assert len(p3["todo_dec"]) == 0 | |
| def test_plan_ov_action_lead_in_and_windows(): | |
| req = np.array([[100.0, 120.0]]) | |
| p = RF.plan_job("ov_action", req, None, None, 100000) | |
| assert np.allclose(p["rc_req"], [[98.5, 120.0]]) and np.allclose(p["wc_req"], req) and not p["windows_complete"] | |
| p2 = RF.plan_job("ov_action", req, np.array([[98.5, 120.0]]), np.array([[100.0, 120.0]]), 100000) | |
| assert len(p2["todo_dec"]) == 0 and p2["windows_complete"] | |
| p3 = RF.plan_job("ov_action", np.array([[0.5, 10.0]]), None, None, 100000) | |
| assert p3["rc_req"][0, 0] == 0.0, "lead-in clipped at the video start" | |
| def test_plan_videomae2_lead_in_resume_and_complete(): | |
| import videomae2 as VM | |
| req = np.array([[100.0, 120.0], [300.0, 310.0]]) | |
| p = RF.plan_job("videomae2", req, None, None, 100000, VM) | |
| assert np.allclose(p["rc_req"], [[97.0, 120.0], [297.0, 310.0]]) and p["k5"][0] == 500 and p["k5"][-1] == 1550 and len(p["k5"]) == 152 | |
| # resumed: per-frame rows cover [97, 110], windows up to 110 -> the next clips need the 3 s before 110 again | |
| p2 = RF.plan_job("videomae2", req, np.array([[97.0, 110.0]]), np.array([[100.0, 110.0]]), 100000, VM) | |
| assert p2["k5"][0] == 550 and len(p2["k5"]) == 102 and np.allclose(p2["todo_dec"], [[107.0, 120.0], [297.0, 310.0]]) and not p2["windows_complete"] | |
| p3 = RF.plan_job("videomae2", req, np.array([[97.0, 120.0], [297.0, 310.0]]), req, 100000, VM) | |
| assert len(p3["todo_dec"]) == 0 and p3["windows_complete"] | |
| # ----------------------------------------------------------------------------- (b) geometry | |
| def test_scaled_box_matches_framestream_and_view_resize_is_the_same_call(): | |
| class FS: | |
| pass | |
| fs = FS() | |
| fs.decode_wh, fs.W, fs.H, fs.box_coords = (960, 600), 1928, 1208, "frame" | |
| rng = np.random.default_rng(0) | |
| for _ in range(50): | |
| b = (rng.uniform([0, 0, 900, 700], [900, 700, 1928, 1208])).astype(np.float32) | |
| fs.box = b[None] | |
| assert np.array_equal(vio.FrameStream.box_for_frame(fs, 0), RF.scaled_box(b, 1928, 1208, 960, 600)) | |
| nan = np.full(4, np.nan, np.float32) | |
| assert np.isnan(RF.scaled_box(nan, 1928, 1208, 960, 600)).all() | |
| cv2 = pytest.importorskip("cv2") | |
| fr = rng.integers(0, 255, (1208, 1928, 3), dtype=np.uint8) | |
| a = cv2.resize(fr, (960, 600), interpolation=cv2.INTER_AREA) # what decode_block does in the worker for a decode_wh model | |
| b2 = cv2.resize(fr, (960, 600), interpolation=cv2.INTER_AREA) # what the runner does in the main process on the native frame | |
| assert np.array_equal(a, b2) | |
| # ----------------------------------------------------------------------------- (c) windows | |
| def test_merge_window_caches_later_rows_win(): | |
| m = cs.make_meta("x", [], decode_wh=(1, 1), native_wh=(1, 1), code_sha="0" * 64) | |
| e1 = np.ones((3, 2), np.float16) | |
| e1[2] = np.nan | |
| w1 = cs.WindowCache("u", np.array([1.0, 1.2, 1.4], np.float32), e1, np.array([1, 1, 0], bool), 16, np.array([[1.0, 1.4]], np.float32), m, {}) | |
| w2 = cs.WindowCache("u", np.array([1.4, 1.6], np.float32), np.full((2, 2), 2, np.float16), np.array([1, 1], bool), 16, np.array([[1.4, 1.6]], np.float32), m, {}) | |
| wm = RF.merge_window_caches(w1, w2).validate() | |
| assert np.allclose(wm.win_t_end, [1.0, 1.2, 1.4, 1.6]) and wm.emb.astype(float)[:, 0].tolist() == [1.0, 1.0, 2.0, 2.0] and wm.win_valid.all() and wm.covers([[1.0, 1.6]]) | |
| def test_window_assembler_clip_rule(): | |
| import videomae2 as VM | |
| class Dummy: | |
| D, batch, device = 4, 3, "cpu" | |
| def infer_windows(self, clips): # clips u8 [B, 16, 224, 224, 3] -> mean gray of each of the 4 quarter-clips | |
| B_ = clips.shape[0] | |
| g = clips.astype(np.float32).mean((2, 3, 4)) # [B, 16] | |
| return np.stack([g[:, i * 4:(i + 1) * 4].mean(1) for i in range(4)], 1) | |
| k5 = np.arange(20, 26) # windows ending at t = 4.0 .. 5.0 s | |
| asm = VM.WindowAssembler(Dummy(), k5) | |
| for k in range(10, 51): # 10 Hz steps 1.0 .. 5.0 s; the first window needs steps 10, 12, .., 40 -> complete; step 12 has no crop | |
| crop = None if k == 12 else np.full((40, 30, 3), k, np.uint8) | |
| asm.feed(k, True, crop) | |
| emb, wv, wf, infer_ms, pre_ms = asm.finish() | |
| assert emb.shape == (6, 4) and wf.tolist() == [1, 1, 0, 0, 0, 0], wf.tolist() # the windows ending at k5 = 20 (steps 10..40) and 21 (12..42) contain step 12 (no crop) -> fail 1 | |
| assert wv.tolist() == [False, False, True, True, True, True] and np.isnan(emb[:2]).all() and np.isfinite(emb[2:]).all() | |
| assert emb[2, 0] < emb[2, 3], "the oldest quarter of the clip must be darker (smaller step number) than the newest" | |
| asm2 = VM.WindowAssembler(Dummy(), np.array([15])) # window ending at 3.0 s fed from 2.0 s only -> 5 (not decoded) | |
| for k in range(20, 31): | |
| asm2.feed(k, True, np.zeros((8, 8, 3), np.uint8)) | |
| assert asm2.finish()[2].tolist() == [5] | |
| # ----------------------------------------------------------------------------- (d) queue | |
| def test_queue_roundtrip_and_summary(): | |
| rows = [] | |
| for i, tier in enumerate(["A", "A", "B", "C", "D"]): | |
| sp = np.array([[10.0 * i, 10.0 * i + 8.4]]) | |
| rows.append({"order": i, "tier": tier, "uid": f"CAR__d{i}__seg__route_0-{i}", "sources": "test", "n_anchor_rows": 1, "n_spans": 1, "span_s": 8.4, "n_frames10": 85, | |
| "video_local": int(i != 1), "pose_cache": 1, "n_video_frames": 1000, "spans": json.dumps(sp.tolist())}) | |
| with tempfile.TemporaryDirectory() as d: | |
| p = Path(d) / "q.csv" | |
| with open(p, "w", newline="") as f: | |
| w = csv.DictWriter(f, fieldnames=list(rows[0])) | |
| w.writeheader() | |
| w.writerows(rows) | |
| back = RF.load_queue(p) | |
| assert len(back) == 5 and back[0]["spans_arr"].shape == (1, 2) and back[3]["n_frames10"] == 85 and isinstance(back[0]["video_local"], int) | |
| s = RF.queue_summary(back) | |
| assert set(s["tiers"]) == set(RF.TIERS) and s["tiers"]["A"]["routes"] == 2 and s["tiers"]["A"]["routes_local"] == 1 and s["frames10_local_total"] == 4 * 85 | |
| def test_set_names_and_model_parsing(): | |
| assert RF.set_name_of(["ov_gaze", "mp_face"]) == "cpu" and RF.set_name_of(list(RF.SETS["gpu"])) == "gpu" and RF.set_name_of(["ov_gaze", "openface3"]) == "custom" | |
| assert RF.parse_models("cpu") == list(RF.SETS["cpu"]) and RF.parse_models("l2cs_gaze,videomae2") == ["l2cs_gaze", "videomae2"] | |
| with pytest.raises(AssertionError): | |
| RF.parse_models("nope") | |
| # ----------------------------------------------------------------------------- (e) hygiene | |
| def test_runner_and_launchers_have_no_forbidden_tokens_and_core_never_imports_the_runner(): | |
| import test_no_dms_inputs as TN | |
| pat = re.compile("|".join(map(re.escape, TN.FORBIDDEN))) | |
| files = [PKG / "run_full_extraction.py", Path(__file__)] + sorted((CWM3 / "results" / "baselines_dms" / "logs").glob("full_*.sh")) | |
| hits = [f"{p.name}:{i}" for p in files if p.exists() for i, line in enumerate(p.read_text().splitlines(), 1) if pat.search(line) and "FORBIDDEN" not in line] | |
| assert not hits, hits | |
| imp = re.compile(r"run_full_extraction") | |
| bad = [str(p) for d in ("models", "train", "eval", "data", "common") for p in (CWM3 / d).glob("*.py") if imp.search(p.read_text())] | |
| assert not bad, bad | |
| # ----------------------------------------------------------------------------- (f) atomic cache writes (2026-09-15 sharding) | |
| def _tmp_leftovers(d): | |
| return sorted(p.name for p in Path(d).glob("*.tmp.npz")) + sorted(p.name for p in (Path(d) / "windows").glob("*.tmp.npz") if (Path(d) / "windows").exists()) | |
| def test_atomic_save_leaves_no_partial_file_on_simulated_failure(monkeypatch): | |
| rc = cs.synthetic_route_cache(t1=20.0, seed=1) | |
| with tempfile.TemporaryDirectory() as d: | |
| p = Path(d) / "u.npz" | |
| assert rc.save(p) == p and p.exists() and _tmp_leftovers(d) == [] | |
| before = p.read_bytes() | |
| real = cs.np.savez_compressed | |
| def broken(path, **arrays): # writes a truncated file, then dies half-way | |
| Path(path).write_bytes(b"PK\x03\x04 truncated") | |
| raise RuntimeError("disk full (simulated)") | |
| monkeypatch.setattr(cs.np, "savez_compressed", broken) | |
| rc2 = cs.synthetic_route_cache(t1=30.0, seed=2) | |
| with pytest.raises(RuntimeError): | |
| rc2.save(p) | |
| assert p.read_bytes() == before, "the existing complete cache must be untouched by a failed write" | |
| assert _tmp_leftovers(d) == [], "no <uid>.<pid>.tmp.npz may survive a failed write" | |
| assert cs.RouteCache.load(p).n == rc.n | |
| with pytest.raises(RuntimeError): # a first-time write that fails leaves nothing behind at all | |
| rc2.save(Path(d) / "fresh.npz") | |
| assert not (Path(d) / "fresh.npz").exists() and _tmp_leftovers(d) == [] | |
| monkeypatch.setattr(cs.np, "savez_compressed", real) | |
| # window cache: same contract; np.savez_compressed's ".npz appended when missing" rule is preserved | |
| m = cs.make_meta("x", [], decode_wh=(1, 1), native_wh=(1, 1), code_sha="0" * 64) | |
| wc = cs.WindowCache("u", np.array([1.0, 1.2], np.float32), np.ones((2, 2), np.float16), np.ones(2, bool), 16, np.array([[1.0, 1.2]], np.float32), m, {}) | |
| pw = wc.save(Path(d) / "windows" / "u") | |
| assert pw == Path(d) / "windows" / "u.npz" and cs.WindowCache.load(pw).m == 2 and _tmp_leftovers(d) == [] | |
| monkeypatch.setattr(cs.np, "savez_compressed", broken) | |
| with pytest.raises(RuntimeError): | |
| wc.save(pw) | |
| assert cs.WindowCache.load(pw).m == 2 and _tmp_leftovers(d) == [] | |
| # ----------------------------------------------------------------------------- (g) claims / fresh .part deferral | |
| def _dead_pid(): | |
| import subprocess | |
| pr = subprocess.Popen([sys.executable, "-c", "pass"]) | |
| pr.wait() # reaped -> that pid is dead | |
| return pr.pid | |
| def test_claim_alive_pid_skips_dead_pid_is_taken_over_and_release_only_own(): | |
| with tempfile.TemporaryDirectory() as d: | |
| cp = Path(d) / "u.claim" | |
| alive = os.getppid() | |
| cp.write_text(json.dumps({"pid": alive, "utc": "x", "tag": "cpu"})) | |
| got, owner = RF.try_claim(cp, "cpu_shard2") | |
| assert got is False and owner["pid"] == alive and owner["tag"] == "cpu" and json.loads(cp.read_text())["pid"] == alive | |
| assert RF.release_claim(cp) is False and cp.exists(), "a claim held by another live pid is never removed" | |
| dead = _dead_pid() | |
| assert not RF.pid_alive(dead) and RF.pid_alive(os.getpid()) | |
| cp.write_text(json.dumps({"pid": dead, "utc": "x", "tag": "cpu"})) | |
| got, owner = RF.try_claim(cp, "cpu_shard2") | |
| me = json.loads(cp.read_text()) | |
| assert got is True and owner is None and me["pid"] == os.getpid() and me["tag"] == "cpu_shard2" and "utc" in me | |
| assert RF.try_claim(cp, "cpu_shard2")[0] is True, "our own (re-entrant) claim is re-taken" | |
| assert RF.release_claim(cp) is True and not cp.exists() | |
| cp.write_text("not json") | |
| assert RF.try_claim(cp, "t")[0] is True, "an unreadable claim is stale" | |
| RF.release_claim(cp) | |
| def test_fresh_part_by_another_tag_is_deferred_own_or_old_is_not(): | |
| from types import SimpleNamespace | |
| with tempfile.TemporaryDirectory() as d: | |
| slots = [SimpleNamespace(name="ov_gaze", cache_dir=Path(d) / "ov_gaze", win_dir=Path(d) / "ov_gaze" / "windows", is_window=False), | |
| SimpleNamespace(name="ov_action", cache_dir=Path(d) / "ov_action", win_dir=Path(d) / "ov_action" / "windows", is_window=True)] | |
| uid = "CAR__d0__seg__route_0-0" | |
| assert RF.fresh_part_elsewhere(slots, uid, "cpu_shard2", 20) is None, "no .part -> nothing" | |
| rc = cs.synthetic_route_cache(uid=uid, t1=10.0, meta={"baseline": "ov_gaze", "writer_tag": "cpu"}) | |
| p = rc.save(slots[0].cache_dir / f"{uid}.part.npz") | |
| info = RF.fresh_part_elsewhere(slots, uid, "cpu_shard2", 20) | |
| assert info is not None and info["writer_tag"] == "cpu" and info["age_min"] < 1 and info["path"] == str(p) | |
| assert RF.fresh_part_elsewhere(slots, uid, "cpu", 20) is None, "our own tag (previous chunk) -> resume, not defer" | |
| assert RF.fresh_part_elsewhere(slots, uid, "cpu_shard2", 0) is None, "0 = off (default path)" | |
| old = time.time() - 31 * 60 | |
| os.utime(p, (old, old)) | |
| assert RF.fresh_part_elsewhere(slots, uid, "cpu_shard2", 20) is None, "a 31 min old .part is stale -> take it" | |
| assert RF.fresh_part_elsewhere(slots, uid, "cpu_shard2", 60) is not None | |
| os.unlink(p) | |
| rc0 = cs.synthetic_route_cache(uid=uid, t1=10.0, meta={"baseline": "ov_gaze"}) # a writer that predates the claim files (no writer_tag) | |
| rc0.save(slots[0].cache_dir / f"{uid}.part.npz") | |
| assert RF.fresh_part_elsewhere(slots, uid, "cpu_shard2", 20)["writer_tag"] is None | |
| os.unlink(slots[0].cache_dir / f"{uid}.part.npz") | |
| m = cs.make_meta("ov_action", [], decode_wh=(1, 1), native_wh=(1, 1), code_sha="0" * 64, writer_tag="gpu") | |
| cs.WindowCache(uid, np.array([1.0], np.float32), np.ones((1, 2), np.float16), np.ones(1, bool), 16, np.array([[1.0, 1.0]], np.float32), m, {}).save(slots[1].win_dir / f"{uid}.part.npz") | |
| assert RF.fresh_part_elsewhere(slots, uid, "cpu_shard2", 20)["writer_tag"] == "gpu", "window .part files count too" | |
| def _synthetic_rows(): | |
| rows = [] | |
| for i, tier in enumerate(["A", "A", "B", "C", "C", "D", "D", "D"]): | |
| sp = np.array([[10.0 * i, 10.0 * i + 8.4]]) | |
| rows.append({"order": i, "tier": tier, "uid": f"CAR__d{i}__seg__route_0-{i}", "sources": "test", "n_anchor_rows": 1, "n_spans": 1, "span_s": 8.4, "n_frames10": 85 + i, | |
| "video_local": int(i != 1 and i != 6), "pose_cache": 1, "n_video_frames": 1000, "spans": json.dumps(sp.tolist()), "spans_arr": sp}) | |
| return rows | |
| def test_run_queue_defers_live_claim_and_takes_over_dead_claim(monkeypatch): | |
| """two 'processes' on one queue: a route claimed by a live other pid is deferred (not processed, not credited, exit 3 after the straggler poll), a dead claim is taken | |
| over; our claims are removed after every route; the default path (no other process) processes everything and exits 0.""" | |
| import argparse | |
| from types import SimpleNamespace | |
| rows = RF.select_rows(_synthetic_rows(), "C,D") | |
| assert [r["uid"][6] for r in rows] == ["3", "4", "5", "7"] | |
| calls = [] | |
| def fake_process_route(slots, uid, req_spans, tier, workers, block, deadline, model_set, **kw): | |
| calls.append((uid, kw.get("tag"))) | |
| assert RF.claim_path(slots, uid).exists() and json.loads(RF.claim_path(slots, uid).read_text())["pid"] == os.getpid(), "the claim is held while the route runs" | |
| return RF.EXIT_DONE, {"uid": uid, "tier": tier, "status": "done", "n_frames_decoded": 85, "wall_s": 1.0, "fps": 85.0, "models": {}} | |
| monkeypatch.setattr(RF, "process_route", fake_process_route) | |
| monkeypatch.setattr(RF, "wait_for_resources", lambda deadline, log=print, **kw: True) | |
| monkeypatch.setattr(RF, "DEFER_POLL_S", 0.05) | |
| recs = [] | |
| monkeypatch.setattr(RF, "update_progress", lambda name, rec, path=None: recs.append((name, rec))) | |
| with tempfile.TemporaryDirectory() as d: | |
| monkeypatch.setattr(RF, "FULL_DIR", Path(d) / "full") | |
| slots = [SimpleNamespace(name=n, cache_dir=Path(d) / "cache" / n, win_dir=Path(d) / "cache" / n / "windows", is_window=(n == "ov_action")) for n in ("ov_gaze", "ov_action", "mp_face")] | |
| args = argparse.Namespace(tag="cpu_shard2", workers=1, block=32, l2cs_detector="own", force=False, limit_frames=0, checkpoint_frames=0, max_load1=19.0, skip_fresh_part_min=20.0) | |
| held = RF.claim_path(slots, rows[1]["uid"]) | |
| held.parent.mkdir(parents=True) | |
| held.write_text(json.dumps({"pid": os.getppid(), "utc": "x", "tag": "cpu"})) | |
| code = RF.run_queue(slots, rows, args, time.time() + 1.0, lambda m: None) | |
| assert code == RF.EXIT_BUDGET, "a route still held elsewhere counts as remaining -> exit 3 (the launcher relaunches later)" | |
| assert [c[0][6] for c in calls] == ["3", "5", "7"] and all(c[1] == "cpu_shard2" for c in calls) | |
| assert json.loads(held.read_text())["pid"] == os.getppid(), "the other process's claim is untouched" | |
| assert sorted(p.name for p in (Path(d) / "cache" / "ov_gaze").glob("*.claim")) == [held.name], "our own claims are released" | |
| summ = json.loads(next((Path(d) / "full" / "logs").glob("chunk_cpu_shard2_*.json")).read_text()) | |
| assert summ["n_deferred_elsewhere"] == 1 and summ["deferred_elsewhere"][0]["status"] == "claimed_by_other" and summ["routes_remaining"] == 1 | |
| assert recs[-1][0] == "cpu_shard2" and recs[-1][1]["routes_done"] == 3 and recs[-1][1]["tiers"]["C"]["routes_done"] == 1 and recs[-1][1]["tiers"]["D"]["routes_done"] == 2 | |
| assert any("waiting for 1 route" in r["note"] for _, r in recs), "the straggler poll ran" | |
| # dead owner -> taken over; and the default path (tag cpu, no other process, fresh-part rule off) does everything -> exit 0 | |
| held.write_text(json.dumps({"pid": _dead_pid(), "utc": "x", "tag": "cpu"})) | |
| calls.clear() | |
| args2 = argparse.Namespace(tag="", workers=1, block=32, l2cs_detector="own", force=False, limit_frames=0, checkpoint_frames=0, max_load1=RF.MAX_LOAD1, skip_fresh_part_min=0.0) | |
| code = RF.run_queue(slots, rows, args2, time.time() + 5.0, lambda m: None) | |
| assert code == RF.EXIT_DONE and [c[0][6] for c in calls] == ["3", "4", "5", "7"] and all(c[1] == "cpu" for c in calls) | |
| assert not list((Path(d) / "cache" / "ov_gaze").glob("*.claim")) | |
| # fresh .part written under another tag -> deferred (the process that predates the claims will finish it); after 20 min it is ours | |
| rc = cs.synthetic_route_cache(uid=rows[2]["uid"], t1=10.0, meta={"baseline": "ov_gaze"}) | |
| pp = rc.save(slots[0].cache_dir / f"{rows[2]['uid']}.part.npz") | |
| calls.clear() | |
| code = RF.run_queue(slots, rows, args, time.time() + 0.5, lambda m: None) | |
| assert code == RF.EXIT_BUDGET and [c[0][6] for c in calls] == ["3", "4", "7"] | |
| old = time.time() - 25 * 60 | |
| os.utime(pp, (old, old)) | |
| calls.clear() | |
| assert RF.run_queue(slots, rows, args, time.time() + 5.0, lambda m: None) == RF.EXIT_DONE and [c[0][6] for c in calls] == ["3", "4", "5", "7"] | |
| # ----------------------------------------------------------------------------- (h) --reverse and the unchanged default argv path | |
| def _old_selection(rows, tiers_csv="A,B,C,D", uids=None, limit_routes=0): | |
| """the pre-2026-09-15 inline selection of main() (verbatim logic).""" | |
| tiers = [t.strip() for t in tiers_csv.split(",") if t.strip()] | |
| rows = [r for r in rows if r["tier"] in tiers and r["video_local"]] | |
| if uids: | |
| keep = set(uids) | |
| rows = [r for r in rows if r["uid"] in keep] | |
| if limit_routes: | |
| rows = rows[:limit_routes] | |
| return rows | |
| def test_select_rows_reverse_and_default_equal_old_selection(): | |
| rows = _synthetic_rows() | |
| ids = lambda rr: [r["uid"] for r in rr] # noqa: E731 | |
| assert ids(RF.select_rows(rows)) == ids(_old_selection(rows)) == [r["uid"] for r in rows if r["video_local"]] | |
| assert ids(RF.select_rows(rows, reverse=True)) == ids(_old_selection(rows))[::-1] | |
| assert ids(RF.select_rows(rows, "C,D", reverse=True)) == ids(_old_selection(rows, "C,D"))[::-1] and ids(RF.select_rows(rows, "C,D", reverse=True))[0] == rows[7]["uid"] | |
| assert ids(RF.select_rows(rows, "C,D", reverse=True, limit_routes=1)) == [rows[7]["uid"]], "--limit-routes applies after --reverse (= the last N of the queue)" | |
| u = [rows[7]["uid"], rows[3]["uid"]] | |
| assert ids(RF.select_rows(rows, "C,D", u)) == [rows[3]["uid"], rows[7]["uid"]] and ids(RF.select_rows(rows, "C,D", u, reverse=True)) == [rows[7]["uid"], rows[3]["uid"]] | |
| assert ids(RF.select_rows(rows, "A,B,C,D", u, False, 1)) == ids(_old_selection(rows, "A,B,C,D", u, 1)) | |
| def test_default_argv_of_the_running_launchers_is_unchanged(): | |
| """the argv of full_cpu.sh / full_gpu.sh (re-executed at every chunk boundary) must plan the same queue and the same gates as before the sharding flags existed.""" | |
| rows = _synthetic_rows() | |
| for argv in (["--models", "ov_gaze,ov_action,mp_face", "--workers", "4", "--block", "32", "--max-minutes", "600", "--tag", "cpu"], | |
| ["--models", "openface3,l2cs_gaze,videomae2", "--workers", "4", "--block", "32", "--max-minutes", "600", "--tag", "gpu", "--device", "cuda:0", "--l2cs-detector", "shared"]): | |
| a = RF.build_parser().parse_args(argv) | |
| assert a.reverse is False and a.skip_fresh_part_min == 0.0 and a.max_load1 == RF.MAX_LOAD1 == 21.0 | |
| assert a.tiers == "A,B,C,D" and a.uids == "" and a.limit_routes == 0 and a.checkpoint_frames == 6000 and a.force is False and a.max_minutes == 600.0 | |
| uids = [u.strip() for u in a.uids.split(",") if u.strip()] or None | |
| planned = RF.select_rows(rows, a.tiers, uids, a.reverse, a.limit_routes) | |
| assert [r["uid"] for r in planned] == [r["uid"] for r in _old_selection(rows, a.tiers, uids, a.limit_routes)] | |
| b = RF.build_parser().parse_args(["--models", "cpu", "--workers", "3", "--reverse", "--tiers", "C,D", "--tag", "cpu_shard2", "--max-load1", "19", "--skip-fresh-part-min", "20"]) | |
| assert b.reverse and b.max_load1 == 19.0 and b.skip_fresh_part_min == 20.0 and RF.parse_models(b.models) == list(RF.SETS["cpu"]) | |
Xet Storage Details
- Size:
- 24.9 kB
- Xet hash:
- 8d7a3e72c4b69e1c2d0d68922245026fd42c3a48d2306804a4b113009be0ac9a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.