Buckets:

AsianPlayer
/

TriDrive

Files
xet
AsianPlayer/TriDrive / code /baselines_dms /run_full_extraction.py
AsianPlayer
16 days ago
download
raw
71.9 kB
#!/usr/bin/env python3
"""cwm3/baselines_dms/run_full_extraction.py — ONE shared native decode per route feeding every extractor of a model set (full-exposure DMS-baseline extraction, 2026-09-14).
Why      layer 1 trains the U-base on ALL TRAIN rows and layer 3 needs alerts/h over the whole test exposure, so every baseline needs per-frame features on the entire
         s12 5 Hz grid of every route (local cabin video: 5.84 M frames @ 10 Hz over the merged [t-8.2, t+0.2] spans of 1,066 routes).  Six independent extractors
         (each with its own decode; Phase 1 measured 22.6-31.2 ms per native frame per worker) would cost ~300 h; here a route is decoded ONCE per model set and
         every decoded frame is handed to every selected model.
Sets     --models cpu = ov_gaze, ov_action, mp_face (openvino / mediapipe, 4 threads each, inferred one after the other in the main process)
         --models gpu = openface3, l2cs_gaze, videomae2 (cuda:0; with --l2cs-detector shared RetinaFace runs ONCE per frame: openface3.detect_faces -> the same box
         goes to openface3.infer(dets=...) and l2cs_gaze.infer_with_boxes, L2CS's own face cut / resize / gaze transform unchanged)
Frames   per (tier, route): the union over the route's anchors of [t - 8.2, t + 0.2] (video_io.spans_from_anchors, merged — the s12 rows are 0.2 s apart but the packs
         have gaps, so a route has several spans) on the 10 Hz frame/20 grid.  Window models add a causal lead-in so that every 5 Hz window inside the spans is
         complete: videomae2 3.0 s (videomae2.lead_in_spans, as in Phase 1), ov_action 1.5 s (15 frames at 10 Hz; Phase 1 had none -> its first 7 windows per span were
         invalid by construction).  The decode covers the union; each model consumes only the frames inside its own spans (per-frame models never see the lead-in).
Decode   video_io.FrameStream / decode_block (index-exact, native resolution, `workers` spawn processes with 2 ffmpeg threads, blocks of `block` frames, box in native
         px).  A model with a decode_wh (ov_action 960x600; videomae2 aspect-preserving 600 px) gets cv2.resize(native, INTER_AREA) of the SAME frame = exactly what
         decode_block did for it in Phase 1, and the driver box scaled exactly as FrameStream.box_for_frame; the crop is video_io.driver_crop (+10 %) in every case.
         Batches are flushed at every contiguous-span end (Phase 1 ran run_route once per span for openface3 / ov_action), so batch compositions match Phase 1.
Caches   results/baselines_dms/cache/<b>/<uid>.npz — cache_schema.RouteCache with the Phase-1 channels / meta keys plus shared_decode, model_set, face_detector,
         lead_in_s, tier, requested_spans; window caches results/baselines_dms/cache/<b>/windows/<uid>.npz (ov_action: OvActionExtractor.decode_windows over the per-frame
         embeddings, 16 frames at 10 Hz ending at every 5 Hz step; videomae2: WindowAssembler, 16 frames at 5 Hz).  summarise.write_summary reads them unchanged
         (cache_dir = cs.cache_dir(b), window_dir = cs.cache_dir(b) / "windows").
Resume   a route whose caches already cover the requested spans is skipped; a partial cache (time budget / kill / --checkpoint-frames) is written as <uid>.part.npz
         (+ windows/<uid>.part.npz; spans truncated at the last decoded frame) and EXTENDED on the next run — only the uncovered spans are decoded
         (cache_schema.spans_subtract; videomae2 re-decodes the 3.0 s lead-in before each resumed span so its clips are complete) and merged (RouteCache.merged_with).
         The complete file is written and the .part removed only when the cache covers the request, so summarise.py never sees a partial cache under <uid>.npz.
Queue    results/baselines_dms/full/route_queue.csv (written before the first route; --build-queue rewrites it): tier A = the anchor windows of the routes carrying the
         979 TRAIN label anchors (data.labels.train_label_table), the clean gold anchors (warning_gold.build_gold, clean_test + clean_val) and the newtest manifest
         clips (small -> layer 2 first); tier B = val exposure; tier C = test exposure; tier D = train exposure (tier-A routes reappear there and are EXTENDED).
         results/baselines_dms/full/progress.json holds one record per set (routes / frames done per tier, rolling fps, ETA per tier), rewritten after every route.
Exit     0 = every queued route complete for the selected models; 3 = --max-minutes reached with work remaining (relaunch); 4 = a route failed (the others are done);
         5 = resources not available at start (GPU gate / RAM).  Guards before every route: MemAvailable >= 28 GB and 1-min load < 21 (waits, never kills anything).
Sharding (2026-09-15) two processes of the SAME model set may walk the queue concurrently: (i) every cache write is atomic (cache_schema.savez_atomic: tmp + os.replace),
         (ii) a route is CLAIMED before it is processed (<cache>/<first model>/<uid>.claim, O_EXCL, json pid / utc / tag; removed in finally; a claim whose pid is dead is
         taken over) and a route claimed by a live other pid is deferred (status claimed_by_other), (iii) with --skip-fresh-part-min M a route whose .part.npz was written
         by another tag within the last M minutes is deferred too (status in_progress_elsewhere: a process that predates the claim files is working on it),
         (iv) --reverse walks the filtered queue from the end so that two shards meet in the middle.  Deferred routes are re-polled (every 120 s) after the queue until they
         exist or the budget ends; they count as remaining (exit 3) when still held.  Without the new flags and with a single process nothing changes.
  python baselines_dms/run_full_extraction.py --build-queue
  python baselines_dms/run_full_extraction.py --models cpu --workers 4 --max-minutes 600 [--tiers A,B,C,D] [--uids u1,u2] [--out-root DIR] [--limit-frames N]
  python baselines_dms/run_full_extraction.py --models gpu --l2cs-detector shared|own --workers 4 --max-minutes 600
  python baselines_dms/run_full_extraction.py --models cpu --workers 3 --reverse --tiers C,D --tag cpu_shard2 --max-load1 19 --skip-fresh-part-min 20   # second shard
  python baselines_dms/run_full_extraction.py --verify-phase1 --models cpu|gpu --uids u1,u2,u3   # manifest windows -> full/verify_phase1/, per-model max |diff| vs phase1 caches
  python baselines_dms/run_full_extraction.py --l2cs-detector-check                              # 45 Phase-1 routes: shared RetinaFace box vs the package detector (rho / median |d|)
No openpilot field is read; no network; nothing here trains or tunes anything.
"""
from __future__ import annotations

import argparse
import contextlib
import csv
import fcntl
import importlib
import json
import os
import sys
import time
from pathlib import Path

import numpy as np

os.environ.setdefault("OMP_NUM_THREADS", "4")
os.environ.setdefault("MKL_NUM_THREADS", "4")
PKG = Path(__file__).resolve().parent
CWM3 = PKG.parent
EXT = PKG / "extractors"
for _p in (str(CWM3), str(PKG), str(EXT)):
    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 base as B                                            # noqa: E402  (extractors/base.py: fail_codes_for, exit codes)

FULL_DIR = cs.RESULTS_ROOT / "full"
QUEUE_CSV = FULL_DIR / "route_queue.csv"
PROGRESS_JSON = FULL_DIR / "progress.json"
PHASE1 = cs.RESULTS_ROOT / "phase1"
MANIFEST = PHASE1 / "manifest_v1.csv"
NEWTEST_MANIFEST = CWM3 / "results" / "newtest" / "manifest_v1.csv"
SETS = {"cpu": ("ov_gaze", "ov_action", "mp_face"), "gpu": ("openface3", "l2cs_gaze", "videomae2")}
ALL_MODELS = SETS["cpu"] + SETS["gpu"]
WINDOW_MODELS = ("ov_action", "videomae2")
LEAD_IN_S = {"ov_action": 1.5, "videomae2": 3.0}
TIER_PACKS = {"B": ("manual_val", "engaged_val"), "C": ("manual_test", "engaged_test"), "D": ("manual_train", "engaged_train")}
TIERS = ("A", "B", "C", "D")
EXIT_DONE, EXIT_BUDGET, EXIT_FAIL, EXIT_RESOURCES = B.EXIT_DONE, B.EXIT_BUDGET, B.EXIT_FAIL, 5
NEED_RAM_GB, MAX_LOAD1, VRAM_HEADROOM_GB = 28.0, 21.0, 4.0
DEFER_POLL_S = 120.0                                                # re-poll interval for routes deferred to another process
NEED_VRAM_GB = {"openface3": 2.0, "l2cs_gaze": 3.0, "videomae2": 2.5}
PHASE1_WINDOW_DIR = {"ov_action": "cache_win", "videomae2": "windows"}
FACE_DETECTOR = {"openface3": "openface3 RetinaFace mnet0.25 (Alignment_RetinaFace.pth sha256 2979b33f...; top-scoring anchor, score >= 0.5)",
                 "l2cs_own": "own: face_detection package RetinaFace mnet0.25 (mobilenet0.25_Final.pth, byte-identical weights; NMS, largest box, score >= 0.5)",
                 "l2cs_shared": "shared: openface3.detect_faces box (top-scoring anchor, score >= 0.5) -> L2CS face_crop (int bounds, clip 0, BGR->RGB, 224x224)",
                 "ov_gaze": "omz face-detection-adas-0001 (fd_mode reshape, conf >= 0.5)", "mp_face": "mediapipe BlazeFace short-range (Face Landmarker task)",
                 "ov_action": "none (driver crop)", "videomae2": "none (driver crop)"}


# ----------------------------------------------------------------------------- resources
def mem_available_gb():
    for line in open("/proc/meminfo"):
        if line.startswith("MemAvailable"):
            return int(line.split()[1]) / 2**20
    return float("nan")


def load1():
    return float(open("/proc/loadavg").read().split()[0])


def gpu_free_gb():
    import subprocess
    try:
        s = subprocess.run(["nvidia-smi", "--query-gpu=memory.free", "--format=csv,noheader,nounits"], capture_output=True, text=True, timeout=20).stdout
        return float(s.strip().splitlines()[0]) / 1024.0
    except Exception:
        return float("nan")


def wait_for_resources(deadline, log=print, need_ram=NEED_RAM_GB, max_load=MAX_LOAD1, poll_s=60):
    """block (never kill anything) until MemAvailable >= need_ram GB and the 1-min load < max_load, or the deadline passes; -> True when the resources are there."""
    waited = False
    while time.time() < deadline:
        ram, ld = mem_available_gb(), load1()
        if ram >= need_ram and ld < max_load:
            return True
        if not waited:
            log(f"waiting for resources: MemAvailable {ram:.1f} GB (need {need_ram}), load1 {ld:.1f} (max {max_load})")
        waited = True
        time.sleep(poll_s)
    return False


# ----------------------------------------------------------------------------- queue
def tier_a_anchors(log=print):
    """-> ({uid: sorted f64 anchor times}, {uid: [sources]}, counts) over the 979 TRAIN label anchors, the clean gold anchors and the newtest manifest clips."""
    import pandas as pd
    vio._shim_labels()
    from data.labels import train_label_table
    import warning_gold as WG
    A, src = {}, {}

    def add(uids, ts, tag):
        for u, t in zip(uids, ts):
            A.setdefault(str(u), []).append(float(t))
            src.setdefault(str(u), set()).add(tag)
    T = train_label_table(("consensus",))
    add(T.uid, T.t_anchor, "train979")
    EP = WG.episode_table()
    lab = EP[EP.label != "unsure"].reset_index(drop=True)
    packs = {(r, s): WG.load_pack(r, s) for r in ("manual", "engaged") for s in ("test", "val")}
    Gs = []
    for seed in (0, 1, 2):
        legacy = WG.merge_legacy(seed)
        if legacy is not None:
            Gs.append(WG.build_gold(lab, packs, legacy, "legacy_first"))
    Gs.append(WG.build_gold(lab, packs, None, "s12"))
    GA = pd.concat(Gs, ignore_index=True)
    GA = GA[GA.clean_test | GA.clean_val]
    E = GA.drop_duplicates(["uid", "t_anchor"])
    add(E.uid, E.t_anchor, "gold")
    M = pd.read_csv(NEWTEST_MANIFEST)
    add(M.uid, M.t_anchor, "newtest")
    counts = {"train979_rows": int(len(T)), "gold_rows_clean_test": int(GA[GA.clean_test].drop_duplicates("idx").shape[0]), "gold_rows_clean_val": int(GA[GA.clean_val].drop_duplicates("idx").shape[0]),
              "gold_uid_t_unique": int(len(E)), "newtest_rows": int(len(M)), "newtest_uids": int(M.uid.nunique())}
    log(f"tier A anchors: {json.dumps(counts)}")
    return {u: np.unique(np.asarray(v, np.float64)) for u, v in A.items()}, {u: sorted(s) for u, s in src.items()}, counts


def route_row(tier, uid, ts, sources):
    has = vio.has_video(uid)
    n_video = 0
    if has:
        try:
            n_video = int(vio.video_info(vio.video_path(uid))[2])
        except Exception:
            has = False
    sp = vio.spans_from_anchors(ts, vio.HIST_S, vio.POST_S)
    k = vio.frames_in_spans(sp, cs.HZ, n_video if has else None)[0]
    return {"tier": tier, "uid": uid, "sources": "+".join(sources), "n_anchor_rows": int(len(ts)), "n_spans": int(len(sp)), "span_s": round(float(np.sum(sp[:, 1] - sp[:, 0])), 1),
            "n_frames10": int(len(k)), "video_local": int(has), "pose_cache": int(vio.pose_path(uid) is not None), "n_video_frames": n_video,
            "spans": json.dumps(np.round(sp.astype(np.float64), 3).tolist())}


def build_queue(path=QUEUE_CSV, log=print):
    """write the ordered (tier, route) queue; tier A sorted by frame count (smallest first), tiers B-D by uid.  -> list of rows."""
    t0 = time.time()
    A, src, counts = tier_a_anchors(log)
    rows = [route_row("A", u, A[u], src[u]) for u in sorted(A)]
    rows.sort(key=lambda r: (r["n_frames10"], r["uid"]))
    for tier, packs in TIER_PACKS.items():
        AB = vio.anchors_by_uid(packs)
        per = {p: set(vio.anchors_by_uid((p,))) for p in packs}
        rows += [route_row(tier, u, AB[u], [p for p in packs if u in per[p]]) for u in sorted(AB)]
    for i, r in enumerate(rows):
        r["order"] = i
    path.parent.mkdir(parents=True, exist_ok=True)
    cols = ["order", "tier", "uid", "sources", "n_anchor_rows", "n_spans", "span_s", "n_frames10", "video_local", "pose_cache", "n_video_frames", "spans"]
    with open(path, "w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=cols)
        w.writeheader()
        for r in rows:
            w.writerow({c: r[c] for c in cols})
    summ = queue_summary(rows)
    summ.update(anchor_counts=counts, written_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), wall_s=round(time.time() - t0, 1), path=str(path))
    (path.with_suffix(".json")).write_text(json.dumps(summ, indent=1))
    log(f"queue: {json.dumps(summ['tiers'])}")
    return rows


def queue_summary(rows):
    out = {"tiers": {}}
    for tier in TIERS:
        rr = [r for r in rows if r["tier"] == tier]
        loc = [r for r in rr if int(r["video_local"])]
        out["tiers"][tier] = {"routes": len(rr), "routes_local": len(loc), "frames10_local": int(sum(int(r["n_frames10"]) for r in loc)),
                              "frames10_all": int(sum(int(r["n_frames10"]) for r in rr)), "anchor_rows": int(sum(int(r["n_anchor_rows"]) for r in rr)),
                              "hours_local": round(sum(int(r["n_frames10"]) for r in loc) / 36000.0, 2)}
    out["routes_local_unique"] = len({r["uid"] for r in rows if int(r["video_local"])})
    out["frames10_local_total"] = int(sum(v["frames10_local"] for v in out["tiers"].values()))
    return out


def load_queue(path=QUEUE_CSV):
    with open(path) as f:
        rows = list(csv.DictReader(f))
    for r in rows:
        for k in ("order", "n_anchor_rows", "n_spans", "n_frames10", "video_local", "pose_cache", "n_video_frames"):
            r[k] = int(r[k])
        r["span_s"] = float(r["span_s"])
        r["spans_arr"] = np.asarray(json.loads(r["spans"]), np.float64).reshape(-1, 2)
    return rows


# ----------------------------------------------------------------------------- progress
def update_progress(set_name, record, path=PROGRESS_JSON):
    """read-modify-write of the per-set progress record under a file lock (the cpu and gpu runners share the file)."""
    path.parent.mkdir(parents=True, exist_ok=True)
    lock = path.with_suffix(".lock")
    with open(lock, "w") as lf:
        fcntl.flock(lf, fcntl.LOCK_EX)
        try:
            P = json.loads(path.read_text()) if path.exists() else {}
        except Exception:
            P = {}
        P[set_name] = record
        P["updated_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        tmp = path.with_suffix(".tmp")
        tmp.write_text(json.dumps(P, indent=1, default=str))
        os.replace(tmp, path)
        fcntl.flock(lf, fcntl.LOCK_UN)


# ----------------------------------------------------------------------------- model slots
def scaled_box(b, W, H, dw, dh):
    """the driver box in the coordinates of a frame resized from (W, H) to (dw, dh) — the same expression as video_io.FrameStream.box_for_frame (box_coords 'frame')."""
    if np.isfinite(b).all():
        sx, sy = dw / W, dh / H
        return np.array([b[0] * sx, b[1] * sy, b[2] * sx, b[3] * sy], np.float32)
    return b.copy()


class Slot:
    """one model of the set: the Phase-1 extractor (setup unchanged), its cache directories, decode size and detector convention."""

    def __init__(self, name, device="cuda:0", out_root=cs.RESULTS_ROOT, l2cs_detector="own"):
        assert name in ALL_MODELS, name
        self.name, self.device = name, device
        self.mod = importlib.import_module(name)
        self.cache_dir = cs.cache_dir(name, out_root)
        self.win_dir = self.cache_dir / "windows"
        self.is_window = name in WINDOW_MODELS
        self.lead_in = LEAD_IN_S.get(name, 0.0)
        self.l2cs_detector = l2cs_detector if name == "l2cs_gaze" else None
        self.ex = None

    def setup(self, log=print):
        t0 = time.time()
        m, n = self.mod, self.name
        if n == "ov_gaze":
            self.ex = m.make_extractor("reshape")
        elif n == "mp_face":
            self.ex = m.MPFaceExtractor().setup()
        elif n == "ov_action":
            self.ex = m.OvActionExtractor(n_threads=4).setup(None)
        elif n == "openface3":
            self.ex = m.OpenFace3Extractor(self.device).setup()
        elif n == "l2cs_gaze":
            self.ex = m.L2CSGazeExtractor().setup(device=self.device)
        elif n == "videomae2":
            self.ex = m.VideoMAE2Extractor().setup(device=self.device)
        self.batch = int(self.ex.batch)
        log(f"[{n}] setup {time.time() - t0:.1f}s; channels {self.ex.C}; batch {self.batch}; decode_wh {self.ex.decode_wh}; detector {self.face_detector()}")
        return self

    @property
    def channels(self):
        return list(self.ex.channels)

    def decode_wh_for(self, native_wh):
        if self.name == "videomae2":
            return tuple(self.mod.decode_wh_for(native_wh))
        return None if self.ex.decode_wh is None else tuple(int(v) for v in self.ex.decode_wh)

    def face_detector(self):
        if self.name == "l2cs_gaze":
            return FACE_DETECTOR["l2cs_shared" if self.l2cs_detector == "shared" else "l2cs_own"]
        return FACE_DETECTOR[self.name]

    def code_sha(self):
        return cs.code_sha256(list(self.ex.code_files()) + [Path(__file__)])


def plan_job(name, req_spans, have_rc_spans, have_wc_spans, n_video, mod=None):
    """pure planning of one model on one route -> dict(rc_req, wc_req, todo_rc, todo_dec, k5, windows_complete).  req_spans = the route's anchor spans; have_* = spans of the
    existing (possibly partial) caches or None.  Window models: rc_req adds the causal lead-in; todo_dec adds the lead-in before every window span still to compute
    (videomae2: the clip needs the 15 earlier 5 Hz frames, so a resumed span re-decodes 3.0 s)."""
    req = cs.merge_spans(req_spans)
    if name == "videomae2":
        mod = mod or importlib.import_module("videomae2")
        rc_req = cs.merge_spans(mod.lead_in_spans(req, n_video, True)) if len(req) else req
        wc_req = req
    elif name == "ov_action":
        rc_req = cs.merge_spans([[max(0.0, a - LEAD_IN_S["ov_action"]), b] for a, b in req.astype(np.float64)]) if len(req) else req
        wc_req = req
    else:
        rc_req, wc_req = req, None
    todo_rc = rc_req if have_rc_spans is None else cs.spans_subtract(rc_req, have_rc_spans)
    todo_dec, k5, windows_complete = todo_rc, None, True
    if name == "videomae2":
        wc_todo = wc_req if have_wc_spans is None else cs.spans_subtract(wc_req, have_wc_spans)
        k5, _ = mod.window_grid(wc_todo, n_video)
        windows_complete = len(k5) == 0
        if len(k5):
            todo_dec = cs.merge_spans(np.concatenate([todo_rc.reshape(-1, 2), np.asarray(mod.lead_in_spans(wc_todo, n_video, True), np.float32).reshape(-1, 2)]))
    elif name == "ov_action":
        windows_complete = have_wc_spans is not None and cs.spans_cover(have_wc_spans, wc_req)
    return {"rc_req": rc_req, "wc_req": wc_req, "todo_rc": todo_rc, "todo_dec": todo_dec, "k5": k5, "windows_complete": windows_complete}


def merge_window_caches(old: cs.WindowCache, new: cs.WindowCache) -> cs.WindowCache:
    """union of two window caches of one route (rows of `new` win on a duplicate win_t_end); spans merged; meta / runtime of `new`."""
    k = np.round(np.concatenate([old.win_t_end, new.win_t_end]).astype(np.float64) * cs.HZ).astype(np.int64)
    order = np.argsort(k, kind="stable")
    ks = k[order]
    keep_last = np.r_[ks[1:] != ks[:-1], True]
    sel = order[keep_last]

    def cat(a, b):
        return np.concatenate([a, b])[sel]
    return cs.WindowCache(new.uid, cat(old.win_t_end, new.win_t_end), cat(old.emb, new.emb), cat(old.win_valid, new.win_valid), int(new.win_len),
                          cs.merge_spans(np.concatenate([old.spans, new.spans])), dict(new.meta), dict(new.runtime))


def _load_cache(p_full, p_part, cls, force, log):
    """the complete cache if present (unless force), else the partial one; unreadable -> None (rebuilt)."""
    if force:
        return None
    for p in (p_full, p_part):
        if p is not None and p.exists():
            try:
                return cls.load(p)
            except Exception as e:
                log(f"{p.name}: cannot read ({repr(e)[:80]}); recomputing")
                return None
    return None


class SlotJob:
    """per-route state of one model: the plan, the existing caches, the frame mask over the shared decode, the pending batch and the feature rows."""

    def __init__(self, slot: Slot, uid, req_spans, n_video, native_wh, tier, force=False, log=print):
        self.slot, self.uid, self.tier = slot, uid, tier
        self.decode_wh = slot.decode_wh_for(native_wh)
        self.native_wh = tuple(int(v) for v in native_wh)
        self.p_rc, self.p_rc_part = slot.cache_dir / f"{uid}.npz", slot.cache_dir / f"{uid}.part.npz"
        self.p_wc = slot.win_dir / f"{uid}.npz" if slot.is_window else None
        self.p_wc_part = slot.win_dir / f"{uid}.part.npz" if slot.is_window else None
        self.have = _load_cache(self.p_rc, self.p_rc_part, cs.RouteCache, force, log)
        if self.have is not None and self.have.channels != slot.channels:
            log(f"[{slot.name}] {uid[:40]}: existing cache channels differ; recomputing")
            self.have = None
        self.have_wc = _load_cache(self.p_wc, self.p_wc_part, cs.WindowCache, force, log) if slot.is_window else None
        if self.have_wc is not None and (self.have is None or self.have_wc.D != (slot.ex.D if slot.name == "videomae2" else slot.mod.N_CLASSES)):
            self.have_wc = None
        plan = plan_job(slot.name, req_spans, None if self.have is None else self.have.spans, None if self.have_wc is None else self.have_wc.spans, n_video, slot.mod)
        self.rc_req, self.wc_req, self.todo_dec, self.k5 = plan["rc_req"], plan["wc_req"], plan["todo_dec"], plan["k5"]
        self.windows_complete = plan["windows_complete"]
        self.needs_decode = len(self.todo_dec) > 0
        self.needs_work = self.needs_decode or (slot.is_window and not self.windows_complete)
        self.mask = None
        self.asm = None
        self.pend_idx, self.pend_crops, self.infer_ms = [], [], []
        self.batch = slot.batch

    def bind(self, fs):
        N = len(fs.frame_idx)
        kj = vio.frames_in_spans(self.todo_dec, cs.HZ, fs.n_frames)[0] if self.needs_decode else np.zeros(0, np.int64)
        self.mask = np.isin(fs.k, kj)
        self.feat = np.full((N, self.slot.ex.C), np.nan, np.float32)
        self.had_frame = np.zeros(N, bool)
        if self.slot.name == "videomae2" and self.k5 is not None and len(self.k5):
            self.asm = self.slot.mod.WindowAssembler(self.slot.ex, self.k5)

    def feed(self, i, k, fr, box_native, get_view):
        """one decoded frame -> crop for this model (only inside its own spans); -> True when the pending batch is full."""
        if not self.mask[i]:
            return False
        crop = None
        if fr is not None:
            self.had_frame[i] = True
            view, box_v = get_view(self.decode_wh)
            crop = vio.driver_crop(view, box_v)
        if crop is not None:
            if self.slot.name == "videomae2":                       # per-frame channels = crop geometry (cheap, no model)
                self.feat[i] = self.slot.ex.infer([crop])[0]
            else:
                self.pend_idx.append(i)
                self.pend_crops.append(crop)
        if self.asm is not None:
            self.asm.feed(k, fr is not None, crop)
        return len(self.pend_idx) >= self.batch

    def store(self, rows):
        assert rows.shape == (len(self.pend_idx), self.slot.ex.C), (rows.shape, len(self.pend_idx), self.slot.ex.C)
        self.feat[np.asarray(self.pend_idx)] = rows
        self.pend_idx.clear()
        self.pend_crops.clear()


def flush_jobs(jobs, coupled, force_all=False):
    """run the pending batches: `coupled` = (openface3 job, l2cs job) sharing one RetinaFace pass (both pending lists are identical by construction)."""
    of3, l2 = coupled if coupled else (None, None)
    for j in jobs:
        if not j.pend_idx or (not force_all and len(j.pend_idx) < j.batch):
            continue
        if j is l2 and of3 is not None:
            continue                                                # handled together with openface3 below
        t0 = time.time()
        if j is of3 and l2 is not None:
            assert j.pend_idx == l2.pend_idx, "shared detector: openface3 / l2cs pending frames differ"
            dets = j.slot.ex.detect_faces(j.pend_crops)
            rows = j.slot.ex.infer(j.pend_crops, dets=dets)
            n = len(j.pend_idx)
            j.store(np.asarray(rows, np.float32).reshape(n, -1))
            t1 = time.time()
            rows2 = l2.slot.ex.infer_with_boxes(l2.pend_crops, dets)
            l2.store(np.asarray(rows2, np.float32).reshape(n, -1))
            j.infer_ms.append(1000.0 * (t1 - t0) / n)
            l2.infer_ms.append(1000.0 * (time.time() - t1) / n)
            continue
        n = len(j.pend_idx)
        rows = np.asarray(j.slot.ex.infer(j.pend_crops), np.float32).reshape(n, -1)
        j.store(rows)
        j.infer_ms.append(1000.0 * (time.time() - t0) / n)


def finalize_job(job: SlotJob, fs, n_done, partial, model_set, wall_s, log=print, tag=""):
    """rows so far -> RouteCache (+ WindowCache) merged with the existing cache; written as <uid>.npz when it covers the request, else <uid>.part.npz.  -> info dict.
    meta.writer_pid / writer_tag record the writing process (the fresh-.part rule of run_queue reads writer_tag)."""
    slot, ex = job.slot, job.slot.ex
    info = {"model": slot.name}
    if job.needs_decode:
        rows_idx = np.flatnonzero(job.mask[:n_done])
        feat = job.feat[rows_idx].copy()
        had_frame = job.had_frame[rows_idx]
        had_box = np.isfinite(fs.box[rows_idx]).all(1)
        fail = B.fail_codes_for(feat, had_frame, had_box, True)
        valid = fail == 0
        feat[~valid] = np.nan
        if partial:
            t_last = float(fs.t[n_done - 1]) if n_done else float(job.todo_dec[0, 0]) - 1.0
            done_spans = np.array([[a, min(b, t_last)] for a, b in job.todo_dec.astype(np.float64) if a <= t_last], np.float32).reshape(-1, 2)
        else:
            done_spans = job.todo_dec
        extra = dict(pose_source=(fs.boxes or {}).get("source", "none"), needs_crop=True, crop_expand=vio.MARGIN, hz=cs.HZ, partial=bool(partial), video_path=str(fs.vpath),
                     n_video_frames=int(fs.n_frames), registry_entry=getattr(ex, "registry_entry", None), shared_decode=True, runner="baselines_dms/run_full_extraction.py",
                     model_set=list(model_set), face_detector=slot.face_detector(), lead_in_s=float(slot.lead_in), tier=job.tier, requested_spans=np.round(job.rc_req.astype(np.float64), 3).tolist(),
                     writer_pid=os.getpid(), writer_tag=str(tag or ""), **fs.decode_meta())
        if slot.name == "videomae2":
            m = slot.mod
            extra.update(decode_h=m.DECODE_H, letterbox=f"long side -> {m.IMG}, centred, pad RGB {m.PAD_RGB}", colour="BGR->RGB, /255, ImageNet mean {} std {}".format(m.MEAN_RGB, m.STD_RGB),
                         win_len=m.L, win_hz=m.WIN_HZ, win_span_s=m.WIN_SPAN_S, frame_stride_10hz=m.STEP10, emb_dim=ex.D, feature="forward_features = fc_norm(mean over tokens), pre-logit, fp16",
                         n_params=getattr(ex, "n_params", None), win_fail_codes=m.WIN_FAIL, channel_roles=ex.spec())
        meta = cs.make_meta(ex.baseline, ex.model_ids, ex.weight_sha256, ex.lib_versions, job.decode_wh or job.native_wh, job.native_wh, slot.code_sha(), "local", **extra)
        runtime = cs.make_runtime(fs.stats["decode_ms_per_frame"], job.infer_ms, len(rows_idx), wall_s=wall_s)
        new = cs.RouteCache(job.uid, fs.t[rows_idx].astype(np.float32), fs.frame_idx[rows_idx].astype(np.int64), fs.box[rows_idx].astype(np.float32), valid, fail,
                            feat.astype(np.float16), slot.channels, cs.merge_spans(done_spans), meta, runtime, moving=fs.moving[rows_idx].astype(bool))
        if job.have is not None and job.have.n:
            new = job.have.merged_with(new)
            new.meta["resumed_from_partial"] = True
        rc = new
    else:
        rc = job.have
    complete = rc is not None and rc.covers(job.rc_req)
    rc.meta["partial"] = not complete
    info.update(n_rows=int(rc.n), n_new_rows=int(len(np.flatnonzero(job.mask[:n_done]))) if job.needs_decode else 0, valid_rate=float(rc.valid.mean()) if rc.n else None,
                fail_hist={cs.FAIL_CODES[int(a)]: int(b) for a, b in zip(*np.unique(rc.fail, return_counts=True))}, complete=bool(complete))
    # ---- windows
    wc, wc_complete = None, True
    if slot.name == "ov_action":
        if complete:
            wc = ex.decode_windows(rc, job.wc_req)
            wc.meta.update(shared_decode=True, lead_in_s=float(slot.lead_in), tier=job.tier)
        else:
            wc_complete = False                                     # rebuilt from the complete per-frame cache at the end
    elif slot.name == "videomae2":
        if job.asm is not None:
            emb, wv, wf, infer_ms, pre_ms = job.asm.finish()
            m = slot.mod
            k_last = int(fs.k[n_done - 1]) if (partial and n_done) else None
            sel = np.ones(len(job.k5), bool) if k_last is None else (job.k5 * m.STEP10 <= k_last)
            t_end = (job.k5[sel] / m.WIN_HZ).astype(np.float32)
            emb_s = emb[sel].astype(np.float16)
            wc_spans = job.wc_req if (job.have_wc is None) else cs.spans_subtract(job.wc_req, job.have_wc.spans)
            if k_last is not None:
                t_l = k_last / cs.HZ
                wc_spans = np.array([[a, min(b, t_l)] for a, b in wc_spans.astype(np.float64) if a <= t_l], np.float32).reshape(-1, 2)
            rt = cs.make_runtime(fs.stats["decode_ms_per_frame"], infer_ms, len(np.flatnonzero(job.mask[:n_done])), wall_s=wall_s)
            wfail = {m.WIN_FAIL.get(int(a), str(int(a))): int(b) for a, b in zip(*np.unique(wf[sel], return_counts=True))}
            rt.update(infer_unit="window (ms per 16-frame window, batched)", n_windows=int(sel.sum()), n_windows_valid=int(wv[sel].sum()), batch=int(ex.batch),
                      preprocess_ms_p50=float(np.percentile(pre_ms, 50)) if pre_ms else float("nan"), preprocess_ms_p95=float(np.percentile(pre_ms, 95)) if pre_ms else float("nan"), win_fail_hist=wfail)
            wc = cs.WindowCache(job.uid, t_end, emb_s, wv[sel].astype(bool), m.L, cs.merge_spans(wc_spans) if len(wc_spans) else np.zeros((0, 2), np.float32), dict(rc.meta, cache_kind="windows"), rt)
            if job.have_wc is not None and job.have_wc.m:
                wc = merge_window_caches(job.have_wc, wc)
        else:
            wc = job.have_wc
        wc_complete = wc is not None and wc.covers(job.wc_req)
        if wc is not None:
            wc.meta["partial"] = not wc_complete
    # ---- write
    all_complete = complete and wc_complete
    if all_complete:
        rc.save(job.p_rc)
        with contextlib.suppress(FileNotFoundError):                # another process may have removed it first
            job.p_rc_part.unlink()
        if wc is not None:
            wc.save(job.p_wc)
            with contextlib.suppress(FileNotFoundError):
                job.p_wc_part.unlink()
    else:
        if job.needs_decode:
            rc.save(job.p_rc_part)
        if wc is not None and slot.name == "videomae2":
            wc.save(job.p_wc_part)
    info.update(windows=(None if wc is None else int(wc.m)), windows_valid=(None if wc is None else int(wc.win_valid.sum())), status=("done" if all_complete else "partial"))
    return info


def process_route(slots, uid, req_spans, tier, workers, block, deadline, model_set, l2cs_shared=False, force=False, limit_frames=0, checkpoint_frames=0, log=print, tag=""):
    """decode the route ONCE over the union of what the selected models still need and feed every frame to them.  -> (code, info)."""
    t_route = time.time()
    info = {"uid": uid, "tier": tier, "status": "", "n_frames_decoded": 0, "wall_s": 0.0, "models": {}}
    try:
        vp = vio.video_path(uid)
        W, H, n_video = vio.video_info(vp)
    except (FileNotFoundError, RuntimeError) as e:
        info.update(status="no_video", error=str(e))
        return EXIT_FAIL, info
    jobs = [SlotJob(s, uid, req_spans, n_video, (W, H), tier, force=force, log=log) for s in slots]
    work = [j for j in jobs if j.needs_work]
    if not work:
        info.update(status="exists", wall_s=round(time.time() - t_route, 2))
        return EXIT_DONE, info
    coupled = None
    if l2cs_shared:                                                 # one RetinaFace pass for both: the two jobs must see the SAME frames (union of what either still needs)
        of3 = next((j for j in jobs if j.slot.name == "openface3"), None)
        l2 = next((j for j in jobs if j.slot.name == "l2cs_gaze"), None)
        if of3 is not None and l2 is not None and l2.needs_decode:
            both = cs.merge_spans(np.concatenate([of3.todo_dec.reshape(-1, 2), l2.todo_dec.reshape(-1, 2)]))
            of3.todo_dec, l2.todo_dec = both, both
            of3.needs_decode = of3.needs_work = True
            l2.batch = of3.batch
            coupled = (of3, l2)
            work = [j for j in jobs if j.needs_work]
    dec = [j for j in jobs if j.needs_decode]
    if not dec:                                                     # only windows to rebuild from complete per-frame caches (ov_action)
        for j in work:
            if j.slot.name == "ov_action":
                info["models"][j.slot.name] = _rebuild_windows_only(j, log)
        info.update(status="done", wall_s=round(time.time() - t_route, 2))
        return EXIT_DONE, info
    todo_all = cs.merge_spans(np.concatenate([j.todo_dec.reshape(-1, 2) for j in dec]))
    try:
        fs = vio.FrameStream(uid, todo_all, hz=cs.HZ, workers=workers, decode_wh=None, block=block, box_coords="native")
    except FileNotFoundError as e:
        info.update(status="no_video", error=str(e))
        return EXIT_FAIL, info
    for j in dec:
        j.bind(fs)
    N = len(fs.frame_idx)
    if N == 0:
        for j in work:
            info["models"][j.slot.name] = finalize_job(j, fs, 0, False, model_set, time.time() - t_route, log, tag=tag)
        info.update(status="done", n_frames_decoded=0, wall_s=round(time.time() - t_route, 2))
        return EXIT_DONE, info
    import cv2
    budget_hit = False
    it = iter(fs)
    i = n_done = 0
    prev_k = None
    last_ckpt = 0
    for t_, fi_, fr_, bx_ in it:
        k = int(fs.k[i])
        if prev_k is not None and k - prev_k > 1:
            flush_jobs(dec, coupled, force_all=True)              # contiguous-span end: Phase 1 ran one run_route per span
        views = {}

        def get_view(wh):
            if wh is None or (wh[0] == W and wh[1] == H):
                return fr_, bx_
            if wh not in views:
                views[wh] = (cv2.resize(fr_, (int(wh[0]), int(wh[1])), interpolation=cv2.INTER_AREA), scaled_box(bx_, W, H, wh[0], wh[1]))
            return views[wh]
        need_flush = False
        for j in dec:
            if j.feed(i, k, fr_, bx_, get_view):
                need_flush = True
        if need_flush:
            flush_jobs(dec, coupled)
        prev_k = k
        i += 1
        n_done = i
        if i % block == 0 and i < N:
            if time.time() > deadline or (limit_frames and i >= limit_frames):
                budget_hit = True
                break
            if checkpoint_frames and i - last_ckpt >= checkpoint_frames:
                flush_jobs(dec, coupled, force_all=True)
                for j in work:
                    finalize_job(j, fs, i, True, model_set, time.time() - t_route, log, tag=tag)
                last_ckpt = i
                log(f"  checkpoint {uid[:40]} {i}/{N} frames")
    flush_jobs(dec, coupled, force_all=True)
    if budget_hit:
        it.close()
    for j in work:
        info["models"][j.slot.name] = finalize_job(j, fs, n_done, budget_hit, model_set, time.time() - t_route, log, tag=tag)
    wall = time.time() - t_route
    info.update(status=("budget" if budget_hit else "done"), n_frames_decoded=int(n_done), n_frames_requested=int(N), wall_s=round(wall, 2), fps=round(n_done / max(wall, 1e-9), 2),
                decode_worker_fps=round(fs.stats["n_grab"] / max(fs.stats["decode_worker_s"], 1e-9), 1), decode_ms_p50=float(np.median(fs.stats["decode_ms_per_frame"])) if fs.stats["decode_ms_per_frame"] else None,
                decode_meta=fs.decode_meta(), spans_decoded=np.round(todo_all.astype(np.float64), 2).tolist()[:8])
    return (EXIT_BUDGET if budget_hit else EXIT_DONE), info


def _rebuild_windows_only(job: SlotJob, log=print):
    """ov_action: complete per-frame cache but no (complete) window cache -> decode_windows without any decode."""
    assert job.slot.name == "ov_action" and job.have is not None
    wc = job.slot.ex.decode_windows(job.have, job.wc_req)
    wc.meta.update(shared_decode=True, lead_in_s=float(job.slot.lead_in), tier=job.tier)
    wc.save(job.p_wc)
    with contextlib.suppress(FileNotFoundError):
        job.p_wc_part.unlink()
    return {"model": job.slot.name, "status": "done", "windows_rebuilt": int(wc.m), "windows_valid": int(wc.win_valid.sum()), "complete": True}


# ----------------------------------------------------------------------------- claims (two shards of one model set)
def pid_alive(pid):
    try:
        os.kill(int(pid), 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except (TypeError, ValueError, OverflowError):
        return False
    return True


def claim_path(slots, uid):
    """<cache>/<first model of the set>/<uid>.claim — one claim per (model set, route); the GPU set has its own cache dirs, so it never collides."""
    return Path(slots[0].cache_dir) / f"{uid}.claim"


def read_claim(path):
    try:
        return json.loads(Path(path).read_text())
    except (FileNotFoundError, ValueError, OSError):
        return None


def try_claim(path, tag=""):
    """take <uid>.claim with O_EXCL.  -> (True, None) when we hold it; (False, owner) when a LIVE other pid holds it.  A claim whose pid is dead (or unreadable) is
    removed and re-taken once; losing that second race -> (False, owner).  Any other OSError is logged by the caller and treated as ours (claims are advisory)."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps({"pid": os.getpid(), "utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "tag": str(tag or "")})
    for attempt in range(2):
        try:
            fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
        except FileExistsError:
            owner = read_claim(path)
            if owner is not None and int(owner.get("pid", -1)) != os.getpid() and pid_alive(owner.get("pid", -1)):
                return False, owner
            if attempt == 1:
                return False, owner
            again = read_claim(path)                                # stale (pid dead / unreadable): take over, unless someone replaced it meanwhile
            if again is not None and owner is not None and again.get("pid") != owner.get("pid"):
                return False, again
            with contextlib.suppress(FileNotFoundError):
                os.unlink(path)
            continue
        with os.fdopen(fd, "w") as f:
            f.write(payload)
        return True, None
    return False, read_claim(path)


def release_claim(path):
    """remove our claim (only when the file still carries our pid)."""
    path = Path(path)
    owner = read_claim(path)
    if owner is not None and int(owner.get("pid", -1)) != os.getpid():
        return False
    with contextlib.suppress(FileNotFoundError):
        os.unlink(path)
    return True


def _npz_meta(path):
    try:
        with np.load(path, allow_pickle=False) as z:
            return json.loads(str(z["meta"])) if "meta" in z.files else {}
    except Exception:
        return {}


def fresh_part_elsewhere(slots, uid, tag, minutes, now=None):
    """-> info dict when some <uid>.part.npz of the set (frame or window cache) was modified within `minutes` and was NOT written under our tag (meta.writer_tag missing =
    a writer that predates the claim files, or another tag): that route is in progress in another process.  None otherwise (or when minutes <= 0)."""
    if not minutes or minutes <= 0:
        return None
    now = time.time() if now is None else now
    for s in slots:
        for p in (Path(s.cache_dir) / f"{uid}.part.npz",) + ((Path(s.win_dir) / f"{uid}.part.npz",) if s.is_window else ()):
            try:
                age_min = (now - p.stat().st_mtime) / 60.0
            except FileNotFoundError:
                continue
            if age_min > minutes:
                continue
            wt = _npz_meta(p).get("writer_tag")
            if wt is not None and str(wt) == str(tag or ""):
                continue                                            # our own earlier chunk
            return {"path": str(p), "age_min": round(age_min, 1), "writer_tag": wt}
    return None


# ----------------------------------------------------------------------------- the queue loop
def set_name_of(models):
    for k, v in SETS.items():
        if set(models) <= set(v):
            return k
    return "custom"


def run_queue(slots, rows, args, deadline, log):
    t0 = time.time()
    models = [s.name for s in slots]
    set_name = args.tag or set_name_of(models)
    tiers = {t: {"routes_total": 0, "frames_total": 0, "routes_done": 0, "frames_done": 0, "routes_skipped_existing": 0} for t in TIERS}
    for r in rows:
        tiers[r["tier"]]["routes_total"] += 1
        tiers[r["tier"]]["frames_total"] += r["n_frames10"]
    recent = []                                                     # (frames newly processed, wall s) of the last routes for the rolling fps
    frames_new_total, fails, budget = 0, [], False
    skipped = []                                                    # routes without a pose cache: not work, not a failure (every frame would be fail 1)
    deferred = {}                                                   # uid -> info of routes held by another process (claim / fresh .part); re-polled after the queue
    infos = []
    max_load = float(getattr(args, "max_load1", MAX_LOAD1) or MAX_LOAD1)
    fresh_min = float(getattr(args, "skip_fresh_part_min", 0) or 0)
    stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
    log_dir = FULL_DIR / "logs"
    log_dir.mkdir(parents=True, exist_ok=True)

    def progress(current=None, note=""):
        rec = {"set": set_name, "models": models, "pid": os.getpid(), "chunk_started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(t0)), "wall_s": round(time.time() - t0, 1),
               "routes_total": sum(v["routes_total"] for v in tiers.values()), "routes_done": sum(v["routes_done"] for v in tiers.values()),
               "frames_total": sum(v["frames_total"] for v in tiers.values()), "frames_done": sum(v["frames_done"] for v in tiers.values()), "frames_new_this_chunk": frames_new_total,
               "fps_rolling": None, "fps_overall_chunk": round(frames_new_total / max(time.time() - t0, 1e-9), 2), "current_route": current, "note": note,
               "routes_deferred_elsewhere": len(deferred), "tiers": {}}
        fr = sum(a for a, _ in recent[-20:])
        wl = sum(b for _, b in recent[-20:])
        fps = fr / wl if wl > 0 and fr > 0 else None
        rec["fps_rolling"] = None if fps is None else round(fps, 2)
        for t, v in tiers.items():
            rem = v["frames_total"] - v["frames_done"]
            v2 = dict(v, frames_remaining=rem, eta_h=(None if not fps else round(rem / fps / 3600.0, 2)))
            rec["tiers"][t] = v2
        rec["eta_h_all"] = None if not fps else round(sum(v["frames_total"] - v["frames_done"] for v in tiers.values()) / fps / 3600.0, 2)
        update_progress(set_name, rec)
        return rec
    def one_route(r, n_i, n_rows):
        """claim -> process -> release one queue row; -> (code, status).  Deferred routes (held elsewhere) return (None, status) and touch no counter."""
        nonlocal frames_new_total
        fresh = fresh_part_elsewhere(slots, r["uid"], set_name, fresh_min)
        if fresh is not None:
            log(f"[{r['tier']} {n_i + 1}/{n_rows}] {r['uid'][:60]} in_progress_elsewhere (.part {fresh['age_min']} min old, writer_tag {fresh['writer_tag']}) -> deferred")
            deferred[r["uid"]] = {"uid": r["uid"], "tier": r["tier"], "status": "in_progress_elsewhere", **fresh}
            return None, "in_progress_elsewhere"
        cp = claim_path(slots, r["uid"])
        try:
            got, owner = try_claim(cp, set_name)
        except OSError as e:                                        # claims are advisory: an unwritable claim dir must not stop the extraction
            log(f"claim {cp.name}: {repr(e)[:80]} -> proceeding without a claim")
            got, owner, cp = True, None, None
        if not got:
            log(f"[{r['tier']} {n_i + 1}/{n_rows}] {r['uid'][:60]} claimed_by_other (pid {owner.get('pid') if owner else '?'}, tag {owner.get('tag') if owner else '?'}) -> deferred")
            deferred[r["uid"]] = {"uid": r["uid"], "tier": r["tier"], "status": "claimed_by_other", "owner": owner}
            return None, "claimed_by_other"
        deferred.pop(r["uid"], None)
        try:
            code, info = process_route(slots, r["uid"], r["spans_arr"], r["tier"], args.workers, args.block, deadline, models, l2cs_shared=(args.l2cs_detector == "shared"),
                                       force=args.force, limit_frames=args.limit_frames, checkpoint_frames=args.checkpoint_frames, log=log, tag=set_name)
        finally:
            if cp is not None:
                release_claim(cp)
        infos.append({k: v for k, v in info.items() if k != "models"} | {"models": {m: {kk: vv for kk, vv in d.items() if kk in ("status", "n_rows", "n_new_rows", "valid_rate", "windows", "windows_valid")} for m, d in info.get("models", {}).items()}})
        st = info["status"]
        tv = tiers[r["tier"]]
        if code == EXIT_FAIL:
            fails.append({"uid": r["uid"], "status": st, "error": info.get("error")})
        elif code == EXIT_DONE:
            tv["routes_done"] += 1
            tv["frames_done"] += r["n_frames10"]
            if st == "exists":
                tv["routes_skipped_existing"] += 1
        if st in ("done", "budget"):
            n_new = int(info.get("n_frames_decoded", 0))
            frames_new_total += n_new
            recent.append((n_new, float(info.get("wall_s", 0.0))))
            if code == EXIT_BUDGET:                                 # credit the partial frames
                tv["frames_done"] += n_new
        mods = " ".join(f"{m}:{d.get('status')}/{d.get('valid_rate') if d.get('valid_rate') is None else round(d['valid_rate'], 3)}" for m, d in info.get("models", {}).items())
        log(f"[{r['tier']} {n_i + 1}/{n_rows}] {r['uid'][:60]} {st} frames {info.get('n_frames_decoded', 0)}/{r['n_frames10']} wall {info.get('wall_s')}s fps {info.get('fps')} | {mods}")
        progress(current=r["uid"])
        return code, st
    progress(note="chunk start")
    for n_i, r in enumerate(rows):
        if time.time() > deadline:
            budget = True
            break
        if not wait_for_resources(deadline, log, max_load=max_load):
            budget = True
            break
        if not r["pose_cache"]:
            log(f"[{r['tier']}] {r['uid'][:60]}: no pose cache -> skipped (every frame would be fail 1)")
            skipped.append({"uid": r["uid"], "tier": r["tier"], "status": "no_pose"})
            continue
        code, st = one_route(r, n_i, len(rows))
        if code == EXIT_BUDGET:
            budget = True
            break
    by_uid = {r["uid"]: r for r in rows}
    while deferred and not budget and time.time() < deadline:      # stragglers held by another process: re-poll until they exist / are released or the budget ends
        log(f"{len(deferred)} route(s) held elsewhere -> re-poll in {DEFER_POLL_S:.0f} s: {', '.join(u[:40] for u in list(deferred)[:5])}")
        progress(note=f"waiting for {len(deferred)} route(s) held by another process")
        time.sleep(min(DEFER_POLL_S, max(1.0, deadline - time.time())))
        if time.time() > deadline:
            break
        for u in list(deferred):
            if time.time() > deadline or not wait_for_resources(deadline, log, max_load=max_load):
                budget = True
                break
            code, st = one_route(by_uid[u], rows.index(by_uid[u]), len(rows))
            if code == EXIT_BUDGET:
                budget = True
                break
    remaining = sum(v["routes_total"] - v["routes_done"] for v in tiers.values()) - len(fails) - len(skipped)
    code = EXIT_BUDGET if (budget or remaining > 0) else (EXIT_FAIL if fails else EXIT_DONE)
    summ = {"set": set_name, "models": models, "exit_code": code, "budget_hit": budget, "routes_remaining": int(max(remaining, 0)), "n_failed": len(fails), "fails": fails[:50],
            "n_skipped_no_pose": len(skipped), "skipped_no_pose": skipped[:50], "n_deferred_elsewhere": len(deferred), "deferred_elsewhere": list(deferred.values())[:50],
            "frames_new_this_chunk": frames_new_total, "wall_s": round(time.time() - t0, 1), "tiers": tiers, "args": {k: v for k, v in vars(args).items()}, "routes": infos[-400:]}
    (log_dir / f"chunk_{set_name}_{stamp}.json").write_text(json.dumps(summ, indent=1, default=str))
    progress(note=("BUDGET_EXIT" if code == EXIT_BUDGET else ("ALL_DONE" if code == EXIT_DONE else "FAILURES")))
    log(("BUDGET_EXIT " if code == EXIT_BUDGET else ("ALL_DONE " if code == EXIT_DONE else "FAILURES ")) + json.dumps({k: summ[k] for k in ("routes_remaining", "n_failed", "frames_new_this_chunk", "wall_s")}))
    return code


# ----------------------------------------------------------------------------- Phase-1 equality check
def manifest_spans(uids):
    import pandas as pd
    M = pd.read_csv(MANIFEST)
    return {u: cs.merge_spans(M.loc[M.uid == u, ["win_start", "win_end"]].to_numpy(np.float64)) for u in uids}, M


def compare_frame_caches(a: cs.RouteCache, b: cs.RouteCache):
    """a = new, b = Phase 1 -> dict(n_common, n_valid_both, max_abs_diff (f32 of the f16 values), n_rows_differ_any (> f16 ulp), valid_agree share, only_in_new / only_in_old)."""
    ka, kb = a.frame_idx, b.frame_idx
    common, ia, ib = np.intersect1d(ka, kb, return_indices=True)
    va, vb = a.valid[ia], b.valid[ib]
    both = va & vb
    Fa, Fb = a.feat[ia].astype(np.float32), b.feat[ib].astype(np.float32)
    d = np.abs(Fa[both] - Fb[both])
    ulp = np.maximum(np.abs(Fb[both]), 1e-3) * 2.0 ** -10           # one f16 ulp of the stored value
    out = {"n_common": int(len(common)), "n_valid_both": int(both.sum()), "valid_agree_share": float((va == vb).mean()) if len(common) else None,
           "n_valid_only_new": int((va & ~vb).sum()), "n_valid_only_old": int((~va & vb).sum()), "n_only_in_new": int(len(ka) - len(common)), "n_only_in_old": int(len(kb) - len(common)),
           "max_abs_diff": float(d.max()) if d.size else 0.0, "median_abs_diff": float(np.median(d)) if d.size else 0.0,
           "rows_exact": int((d.max(1) == 0).sum()) if d.size else 0, "rows_within_1_f16_ulp": int((d <= ulp).all(1).sum()) if d.size else 0,
           "max_abs_diff_per_channel": {c: float(d[:, j].max()) for j, c in enumerate(a.channels)} if d.size and a.C <= 40 else None}
    if d.size and a.C > 40:
        out["max_abs_diff_top5_channels"] = {a.channels[j]: float(d[:, j].max()) for j in np.argsort(-d.max(0))[:5]}
    return out


def compare_window_caches(a: cs.WindowCache, b: cs.WindowCache):
    ka, kb = np.round(a.win_t_end.astype(np.float64) * cs.HZ).astype(np.int64), np.round(b.win_t_end.astype(np.float64) * cs.HZ).astype(np.int64)
    common, ia, ib = np.intersect1d(ka, kb, return_indices=True)
    va, vb = a.win_valid[ia], b.win_valid[ib]
    both = va & vb
    Ea, Eb = a.emb[ia].astype(np.float32), b.emb[ib].astype(np.float32)
    d = np.abs(Ea[both] - Eb[both])
    ulp = np.maximum(np.abs(Eb[both]), 1e-3) * 2.0 ** -10
    return {"n_common": int(len(common)), "n_valid_both": int(both.sum()), "n_valid_only_new": int((va & ~vb).sum()), "n_valid_only_old": int((~va & vb).sum()),
            "max_abs_diff": float(d.max()) if d.size else 0.0, "median_abs_diff": float(np.median(d)) if d.size else 0.0, "rows_exact": int((d.max(1) == 0).sum()) if d.size else 0,
            "rows_within_1_f16_ulp": int((d <= ulp).all(1).sum()) if d.size else 0, "emb_abs_p50": float(np.median(np.abs(Eb[both]))) if d.size else None}


def verify_phase1(models, uids, args, log=print):
    out_root = FULL_DIR / "verify_phase1"
    spans, M = manifest_spans(uids)
    det = args.l2cs_detector or "own"
    if det == "shared" and "l2cs_gaze" in models and "openface3" not in models:
        models = ["openface3"] + list(models)
    slots = [Slot(m, args.device, out_root, det).setup(log) for m in models]
    deadline = time.time() + args.max_minutes * 60
    rep = {"written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "models": list(models), "uids": list(uids), "out_root": str(out_root), "l2cs_detector": det, "routes": {}, "per_model": {}}
    for u in uids:
        code, info = process_route(slots, u, spans[u], "V", args.workers, args.block, deadline, models, l2cs_shared=(det == "shared"), force=True, log=log)
        rep["routes"][u] = {k: v for k, v in info.items() if k in ("status", "n_frames_decoded", "wall_s", "fps", "decode_worker_fps")}
        log(f"verify {u[:60]} {info['status']} frames {info.get('n_frames_decoded')} wall {info.get('wall_s')}s")
    for s in slots:
        per = {}
        for u in uids:
            p_new, p_old = s.cache_dir / f"{u}.npz", PHASE1 / s.name / "cache" / f"{u}.npz"
            if not (p_new.exists() and p_old.exists()):
                per[u] = {"error": f"missing {'new' if not p_new.exists() else 'phase1'} cache"}
                continue
            per[u] = {"frames": compare_frame_caches(cs.RouteCache.load(p_new), cs.RouteCache.load(p_old))}
            if s.is_window:
                pw_new, pw_old = s.win_dir / f"{u}.npz", PHASE1 / s.name / PHASE1_WINDOW_DIR[s.name] / f"{u}.npz"
                if pw_new.exists() and pw_old.exists():
                    per[u]["windows"] = compare_window_caches(cs.WindowCache.load(pw_new), cs.WindowCache.load(pw_old))
        fr = [v["frames"] for v in per.values() if "frames" in v]
        wn = [v["windows"] for v in per.values() if "windows" in v]
        rep["per_model"][s.name] = {"routes": per,
                                    "frames_max_abs_diff": max([f["max_abs_diff"] for f in fr] or [None]), "frames_n_valid_both": sum(f["n_valid_both"] for f in fr), "frames_rows_exact": sum(f["rows_exact"] for f in fr),
                                    "frames_rows_within_1_f16_ulp": sum(f["rows_within_1_f16_ulp"] for f in fr), "frames_valid_only_new": sum(f["n_valid_only_new"] for f in fr), "frames_valid_only_old": sum(f["n_valid_only_old"] for f in fr),
                                    "windows_max_abs_diff": (max(w["max_abs_diff"] for w in wn) if wn else None), "windows_n_valid_both": (sum(w["n_valid_both"] for w in wn) if wn else None),
                                    "windows_rows_exact": (sum(w["rows_exact"] for w in wn) if wn else None), "windows_rows_within_1_f16_ulp": (sum(w["rows_within_1_f16_ulp"] for w in wn) if wn else None),
                                    "windows_valid_only_new": (sum(w["n_valid_only_new"] for w in wn) if wn else None), "windows_valid_only_old": (sum(w["n_valid_only_old"] for w in wn) if wn else None)}
    out_root.mkdir(parents=True, exist_ok=True)
    tag = args.tag or set_name_of(models)
    (out_root / f"verify_phase1_{tag}.json").write_text(json.dumps(rep, indent=1, default=str))
    lines = [f"l2cs detector: {det}", "", "| model | rows valid in both | rows exact | rows within 1 f16 ulp | max abs diff | valid only new / only phase1 | windows valid both | windows exact / within ulp | windows max abs diff |", "|---|---|---|---|---|---|---|---|---|"]
    for m, v in rep["per_model"].items():
        lines.append(f"| {m} | {v['frames_n_valid_both']} | {v['frames_rows_exact']} | {v['frames_rows_within_1_f16_ulp']} | {v['frames_max_abs_diff']} | {v['frames_valid_only_new']} / {v['frames_valid_only_old']} | "
                     f"{v['windows_n_valid_both']} | {v['windows_rows_exact']} / {v['windows_rows_within_1_f16_ulp']} | {v['windows_max_abs_diff']} |")
    (out_root / f"verify_phase1_{tag}.md").write_text("\n".join(lines) + "\n")
    print("\n".join(lines))
    for s in slots:
        s.ex.teardown()
    return 0


# ----------------------------------------------------------------------------- l2cs shared-detector check
def l2cs_detector_check(args, log=print):
    """the 45 Phase-1 routes: openface3.detect_faces box -> l2cs infer_with_boxes vs the Phase-1 l2cs caches (own detector): Spearman rho, median |d|, box IoU."""
    from scipy.stats import spearmanr
    import pandas as pd
    M = pd.read_csv(MANIFEST)
    uids = list(dict.fromkeys(M.uid))
    spans, _ = manifest_spans(uids)
    of3 = Slot("openface3", args.device).setup(log)
    l2 = Slot("l2cs_gaze", args.device, l2cs_detector="shared").setup(log)
    rows = []                                                       # (uid, k, yaw_sh, pitch_sh, conf_sh, bw_sh, iou, own_found, yaw_own, pitch_own, conf_own)
    t_all = time.time()
    det_ms, own_ms = [], []
    for n_u, u in enumerate(uids):
        fs = vio.FrameStream(u, spans[u], hz=cs.HZ, workers=args.workers, decode_wh=None, block=args.block, box_coords="native")
        pend = []

        def flush():
            if not pend:
                return
            crops = [c for _, c in pend]
            t0 = time.time()
            dets = of3.ex.detect_faces(crops)
            t1 = time.time()
            sh = l2.ex.infer_with_boxes(crops, dets)
            own = l2.ex.detect_largest(crops)
            t2 = time.time()
            det_ms.append(1000.0 * (t1 - t0) / len(crops))
            own_ms.append(1000.0 * (t2 - t1) / len(crops))
            for (k, _), d, r, o in zip(pend, dets, sh, own):
                iou = np.nan
                if o is not None and np.isfinite(d).all():
                    a, b = d[:4], o[0]
                    ix = max(0.0, min(a[2], b[2]) - max(a[0], b[0]))
                    iy = max(0.0, min(a[3], b[3]) - max(a[1], b[1]))
                    inter = ix * iy
                    ua = (a[2] - a[0]) * (a[3] - a[1]) + (b[2] - b[0]) * (b[3] - b[1]) - inter
                    iou = inter / ua if ua > 0 else np.nan
                rows.append((u, k, float(r[1]), float(r[2]), float(r[0]), float(r[3]), float(iou), o is not None, float(d[4]) if np.isfinite(d[4]) else np.nan))
            pend.clear()
        for t_, fi_, fr_, bx_ in fs:
            k = int(round(float(t_) * cs.HZ))
            crop = vio.driver_crop(fr_, bx_) if fr_ is not None else None
            if crop is not None:
                pend.append((k, np.ascontiguousarray(crop)))
                if len(pend) >= 64:
                    flush()
        flush()
        log(f"[{n_u + 1}/{len(uids)}] {u[:60]} frames {len(fs.frame_idx)} wall {fs.stats['wall_s']:.1f}s")
    # compare with the Phase-1 caches
    R = pd.DataFrame(rows, columns=["uid", "k", "yaw_sh", "pitch_sh", "conf_sh", "bw_sh", "iou", "own_found", "score_of3"])
    old = []
    for u in uids:
        rc = cs.RouteCache.load(PHASE1 / "l2cs_gaze" / "cache" / f"{u}.npz")
        kk = np.round(rc.t.astype(np.float64) * cs.HZ).astype(np.int64)
        F = rc.feat.astype(np.float32)
        old.append(pd.DataFrame({"uid": u, "k": kk, "valid_old": rc.valid, "yaw_old": F[:, 1], "pitch_old": F[:, 2], "conf_old": F[:, 0], "bw_old": F[:, 3], "had_box": np.isfinite(rc.box).all(1)}))
    O = pd.concat(old, ignore_index=True)
    J = O.merge(R, on=["uid", "k"], how="left")
    valid_sh = np.isfinite(J.yaw_sh.to_numpy()) & np.isfinite(J.pitch_sh.to_numpy())
    valid_old = J.valid_old.to_numpy().astype(bool)
    both = valid_sh & valid_old
    def circ(a, b):
        return ((a - b + 180.0) % 360.0) - 180.0
    y_sh, y_old, p_sh, p_old = J.yaw_sh.to_numpy()[both], J.yaw_old.to_numpy()[both], J.pitch_sh.to_numpy()[both], J.pitch_old.to_numpy()[both]
    dy, dp = circ(y_sh, y_old), circ(p_sh, p_old)
    y_sh_unwrapped = y_old + dy                                     # the shared value expressed next to the Phase-1 value (undoes the +-180 wrap)
    rep = {"written_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "n_routes": len(uids), "n_frames_phase1": int(len(J)), "n_frames_with_box": int(J.had_box.sum()),
           "n_valid_phase1_own_detector": int(valid_old.sum()), "n_valid_shared_detector": int(valid_sh.sum()), "n_valid_both": int(both.sum()),
           "n_valid_only_own": int((valid_old & ~valid_sh).sum()), "n_valid_only_shared": int((valid_sh & ~valid_old).sum()),
           "spearman_rho_yaw_raw": float(spearmanr(y_sh, y_old).statistic), "spearman_rho_yaw_unwrapped": float(spearmanr(y_sh_unwrapped, y_old).statistic),
           "spearman_rho_pitch": float(spearmanr(p_sh, p_old).statistic),
           "median_abs_d_yaw_deg_circular": float(np.median(np.abs(dy))), "p90_abs_d_yaw_deg": float(np.percentile(np.abs(dy), 90)), "p99_abs_d_yaw_deg": float(np.percentile(np.abs(dy), 99)),
           "median_abs_d_pitch_deg": float(np.median(np.abs(dp))), "p90_abs_d_pitch_deg": float(np.percentile(np.abs(dp), 90)), "p99_abs_d_pitch_deg": float(np.percentile(np.abs(dp), 99)),
           "share_abs_d_yaw_le_1deg": float(np.mean(np.abs(dy) <= 1.0)), "share_abs_d_yaw_le_5deg": float(np.mean(np.abs(dy) <= 5.0)),
           "box_iou_median_both_found": float(np.nanmedian(J.iou.to_numpy()[both])), "box_iou_p10_both_found": float(np.nanpercentile(J.iou.to_numpy()[both], 10)),
           "box_w_median_abs_d_px": float(np.median(np.abs(J.bw_sh.to_numpy()[both] - J.bw_old.to_numpy()[both]))),
           "conf_median_abs_d": float(np.median(np.abs(J.conf_sh.to_numpy()[both] - J.conf_old.to_numpy()[both]))),
           "ms_per_crop_openface3_detector_batched_64": float(np.median(det_ms)) if det_ms else None, "ms_per_crop_l2cs_gaze_plus_own_detector": float(np.median(own_ms)) if own_ms else None,
           "wall_s": round(time.time() - t_all, 1), "criterion": "adopt the shared box iff Spearman rho >= 0.95 for yaw (raw) and pitch on frames valid under both detectors"}
    rep["decision"] = "shared" if (rep["spearman_rho_yaw_raw"] >= 0.95 and rep["spearman_rho_pitch"] >= 0.95) else "own"
    FULL_DIR.mkdir(parents=True, exist_ok=True)
    (FULL_DIR / "l2cs_shared_detector_check.json").write_text(json.dumps(rep, indent=1, default=str))
    (FULL_DIR / "l2cs_detector_decision.json").write_text(json.dumps({"l2cs_detector": rep["decision"], "rho_yaw_raw": rep["spearman_rho_yaw_raw"], "rho_yaw_unwrapped": rep["spearman_rho_yaw_unwrapped"],
                                                                        "rho_pitch": rep["spearman_rho_pitch"], "median_abs_d_yaw": rep["median_abs_d_yaw_deg_circular"], "median_abs_d_pitch": rep["median_abs_d_pitch_deg"],
                                                                        "n_valid_both": rep["n_valid_both"], "written_utc": rep["written_utc"]}, indent=1))
    print(json.dumps({k: v for k, v in rep.items() if not isinstance(v, (list, dict))}, indent=1))
    of3.ex.teardown()
    l2.ex.teardown()
    return 0


# ----------------------------------------------------------------------------- CLI
def parse_models(s):
    if s in SETS:
        return list(SETS[s])
    ms = [m.strip() for m in s.split(",") if m.strip()]
    bad = [m for m in ms if m not in ALL_MODELS]
    assert not bad, f"unknown models {bad}; known {ALL_MODELS}"
    return ms


def select_rows(rows, tiers="A,B,C,D", uids=None, reverse=False, limit_routes=0):
    """the (tier, route) rows this process walks, in order: tier filter + local video -> optional --uids filter (queue order kept) -> optional --reverse -> optional
    --limit-routes.  Pure; the default arguments reproduce the pre-2026-09-15 selection exactly."""
    tiers = [t.strip() for t in str(tiers).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 reverse:
        rows = rows[::-1]
    if limit_routes:
        rows = rows[:limit_routes]
    return rows


def build_parser():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--models", default="cpu", help="cpu | gpu | comma list of " + ",".join(ALL_MODELS))
    ap.add_argument("--workers", type=int, default=4, help="decode worker processes (2 ffmpeg threads each)")
    ap.add_argument("--block", type=int, default=32, help="native frames per decode block (2 x workers blocks in flight; 32 x 8 x 7 MB ~ 1.8 GB)")
    ap.add_argument("--max-minutes", type=float, default=600.0)
    ap.add_argument("--tiers", default="A,B,C,D")
    ap.add_argument("--uids", default="", help="comma list or file: restrict to these routes (queue order kept)")
    ap.add_argument("--out-root", default="", help="cache root (default results/baselines_dms -> cache/<b>/); smoke runs use another root")
    ap.add_argument("--device", default="cuda:0")
    ap.add_argument("--l2cs-detector", default="", choices=["", "own", "shared"], help="default: full/l2cs_detector_decision.json if present, else own")
    ap.add_argument("--build-queue", action="store_true", help="(re)write results/baselines_dms/full/route_queue.csv and exit")
    ap.add_argument("--verify-phase1", action="store_true", help="manifest windows of --uids -> full/verify_phase1/<b>/, compared with the Phase-1 caches")
    ap.add_argument("--l2cs-detector-check", action="store_true")
    ap.add_argument("--limit-routes", type=int, default=0)
    ap.add_argument("--limit-frames", type=int, default=0, help="stop a route after this many decoded frames (smoke: writes .part caches)")
    ap.add_argument("--checkpoint-frames", type=int, default=6000, help="write .part caches every N decoded frames inside a route (0 = only at the budget)")
    ap.add_argument("--force", action="store_true")
    ap.add_argument("--tag", default="", help="progress record name (default cpu / gpu / custom); also the claim / writer tag")
    ap.add_argument("--no-gpu-gate", action="store_true")
    ap.add_argument("--reverse", action="store_true", help="walk the filtered queue from the end (second shard of a set)")
    ap.add_argument("--max-load1", type=float, default=MAX_LOAD1, help=f"per-route gate: wait while the 1-min load is >= this (default {MAX_LOAD1})")
    ap.add_argument("--skip-fresh-part-min", type=float, default=0.0,
                    help="defer a route whose .part.npz was written by another tag within the last N minutes (in progress in a process that predates the claims); 0 = off")
    return ap


def main():
    a = build_parser().parse_args()
    t_start = time.time()

    def log(m):
        print(f"{time.strftime('%H:%M:%S')} {m}", flush=True)
    if a.build_queue:
        build_queue(log=log)
        return 0
    models = parse_models(a.models)
    if a.l2cs_detector_check:
        return l2cs_detector_check(a, log)
    import cv2
    cv2.setNumThreads(4)
    if any(m in SETS["gpu"] for m in models) and str(a.device).startswith("cuda"):
        import torch
        torch.set_num_threads(int(os.environ.get("OMP_NUM_THREADS", "4")))
        need = sum(NEED_VRAM_GB[m] for m in models if m in NEED_VRAM_GB) + VRAM_HEADROOM_GB
        free = gpu_free_gb()
        if not a.no_gpu_gate and not (free >= need):
            log(f"GPU gate: free VRAM {free:.1f} GB < need {need:.1f} GB -> exit {EXIT_RESOURCES}")
            return EXIT_RESOURCES
        if mem_available_gb() < NEED_RAM_GB:
            log(f"RAM gate: MemAvailable {mem_available_gb():.1f} GB < {NEED_RAM_GB} -> exit {EXIT_RESOURCES}")
            return EXIT_RESOURCES
    uids = None
    if a.uids:
        uids = [l.strip() for l in open(a.uids)] if Path(a.uids).exists() else [u.strip() for u in a.uids.split(",")]
        uids = [u for u in uids if u]
    if not a.l2cs_detector:
        dec = FULL_DIR / "l2cs_detector_decision.json"
        a.l2cs_detector = json.loads(dec.read_text())["l2cs_detector"] if dec.exists() else "own"
    if a.verify_phase1:
        assert uids, "--verify-phase1 needs --uids"
        a.max_minutes = max(a.max_minutes, 240.0)
        return verify_phase1(models, uids, a, log)
    out_root = Path(a.out_root) if a.out_root else cs.RESULTS_ROOT
    if not QUEUE_CSV.exists():
        build_queue(log=log)
    rows = select_rows(load_queue(), a.tiers, uids, a.reverse, a.limit_routes)
    log(f"queue: {len(rows)} (tier, route) entries, {sum(r['n_frames10'] for r in rows)} frames @ 10 Hz{' (REVERSED)' if a.reverse else ''}; models {models}; "
        f"l2cs detector {a.l2cs_detector}; out_root {out_root}; max load1 {a.max_load1}; skip fresh .part {a.skip_fresh_part_min} min")
    if a.l2cs_detector == "shared" and "l2cs_gaze" in models and "openface3" not in models:
        models = ["openface3"] + models                             # the shared box needs the openface3 detector in the process
        log("shared l2cs detector: openface3 added to the set")
    slots = [Slot(m, a.device, out_root, a.l2cs_detector).setup(log) for m in models]
    deadline = t_start + a.max_minutes * 60.0
    code = run_queue(slots, rows, a, deadline, log)
    for s in slots:
        try:
            s.ex.teardown()
        except Exception:
            pass
    return code


if __name__ == "__main__":
    sys.exit(main())
Xet Storage Details

Size:: 71.9 kB
Xet hash:: 38b5c7850db27afab861849c278572105e98a1507589471cbe1e4c6a2120cd89
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.