scorevision: push artifact

training/poll_manako.py

@@ -0,0 +1,132 @@
+"""
+Continuous poller for in-domain frames from the manako/scoredata API for Detect-crime.
+
+The console.scorevision.io endpoint surfaces ONE `latestAnnotatedChallenge` per element.
+As the runner produces new challenges this single record rotates. By polling every couple
+of minutes we accumulate distinct in-domain frames + the king's predictions for each.
+
+Use this as a long-running collector alongside `build_dataset.py`. Run it for a few hours
+or days, then re-run `build_dataset.py --extra-dir <out>/images` (or `--manako`) to fold
+the new frames into the YOLO dataset.
+
+Usage:
+  python poll_manako.py --out ./manako_pool --interval 120 --hours 8
+  python poll_manako.py --out ./manako_pool --interval 120 --forever  # until killed
+
+Output structure (each unique frame = one image + one preds JSON):
+  manako_pool/
+    images/manako_ch{cid}_f{fid}_{sha16}.png
+    preds/ manako_ch{cid}_f{fid}_{sha16}.json
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import sys
+import time
+from pathlib import Path
+from urllib.parse import quote
+from urllib.request import Request, urlopen
+
+API = "https://console.scorevision.io/api/v2"
+ELEMENT_ID = "manak0/Detect-crime"
+
+
+def http_get(url: str, timeout: int = 30, retries: int = 3) -> bytes:
+    last = None
+    for i in range(retries):
+        try:
+            req = Request(url, headers={"User-Agent": "poll_manako/1.0"})
+            with urlopen(req, timeout=timeout) as r:
+                return r.read()
+        except Exception as e:
+            last = e
+            time.sleep(1.0 * (i + 1))
+    raise last  # type: ignore
+
+
+def sha16(b: bytes) -> str:
+    return hashlib.sha256(b).hexdigest()[:16]
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--out", required=True, help="output dir for accumulated frames")
+    ap.add_argument("--interval", type=int, default=120, help="seconds between polls")
+    ap.add_argument("--hours", type=float, default=4.0, help="duration to run (ignored with --forever)")
+    ap.add_argument("--forever", action="store_true", help="run until killed")
+    args = ap.parse_args()
+
+    out = Path(args.out).resolve()
+    img_dir = out / "images"
+    pred_dir = out / "preds"
+    img_dir.mkdir(parents=True, exist_ok=True)
+    pred_dir.mkdir(parents=True, exist_ok=True)
+
+    deadline = time.time() + args.hours * 3600 if not args.forever else float("inf")
+    seen: set[str] = {p.stem.rsplit("_", 1)[-1] for p in img_dir.glob("manako_*.png")}
+    n_polls = 0
+    n_new = 0
+    print(f"[poll] starting; existing frames in pool: {len(seen)}", file=sys.stderr)
+
+    while time.time() < deadline:
+        n_polls += 1
+        try:
+            raw = http_get(f"{API}/elements/{quote(ELEMENT_ID, safe='')}?lookback_days=7")
+        except Exception as e:
+            print(f"[poll {n_polls}] API error: {e}", file=sys.stderr)
+            time.sleep(args.interval)
+            continue
+
+        elt = json.loads(raw)
+        lac = elt.get("latestAnnotatedChallenge")
+        if not lac:
+            print(f"[poll {n_polls}] no latestAnnotatedChallenge", file=sys.stderr)
+            time.sleep(args.interval)
+            continue
+
+        challenge_id = lac.get("challenge_id")
+        frames = (lac.get("response") or {}).get("frames") or []
+        preds = ((lac.get("response") or {}).get("predictions") or {}).get("frames") or []
+        preds_by_id = {p.get("frame_id"): p for p in preds}
+
+        added = 0
+        for f in frames:
+            url = f.get("url")
+            fid = f.get("frame_id")
+            if not url:
+                continue
+            try:
+                data = http_get(url)
+            except Exception as e:
+                print(f"[poll] image fetch failed {url}: {e}", file=sys.stderr)
+                continue
+            digest = sha16(data)
+            if digest in seen:
+                continue
+            seen.add(digest)
+            (img_dir / f"manako_ch{challenge_id}_f{fid}_{digest}.png").write_bytes(data)
+            preds_obj = preds_by_id.get(fid)
+            if preds_obj is not None:
+                (pred_dir / f"manako_ch{challenge_id}_f{fid}_{digest}.json").write_text(
+                    json.dumps(preds_obj), encoding="utf-8"
+                )
+            added += 1
+            n_new += 1
+
+        print(f"[poll {n_polls}] cid={challenge_id} frames={len(frames)} new={added} "
+              f"total_unique={len(seen)} (cumulative new this run: {n_new})", file=sys.stderr)
+
+        if time.time() >= deadline:
+            break
+        time.sleep(args.interval)
+
+    print(f"[poll] done. polls={n_polls}, new frames added this run: {n_new}, "
+          f"pool size: {len(seen)}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())