scorevision: push artifact

training/milestones_watch.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""Poll Ultralytics results.csv and append one summary line per N-epoch milestone."""
+from __future__ import annotations
+
+import argparse
+import csv
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def fmt_row(row: dict) -> str:
+    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    ep = int(float(row["epoch"]))
+    return (
+        f"{ts} | epoch={ep} | "
+        f"mAP50={float(row['metrics/mAP50(B)']):.4f} mAP={float(row['metrics/mAP50-95(B)']):.4f} | "
+        f"P={float(row['metrics/precision(B)']):.4f} R={float(row['metrics/recall(B)']):.4f} | "
+        f"train_box={float(row['train/box_loss']):.4f} train_cls={float(row['train/cls_loss']):.4f}"
+    )
+
+
+def recover_done(out_path: Path) -> set[int]:
+    done: set[int] = set()
+    if not out_path.exists():
+        return set()
+    for line in out_path.read_text().splitlines():
+        if "| epoch=" not in line:
+            continue
+        try:
+            part = line.split("| epoch=")[1].split()[0]
+            done.add(int(part))
+        except (IndexError, ValueError):
+            continue
+    return done
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--csv", required=True, help="path to results.csv (created by Ultralytics)")
+    ap.add_argument("--every", type=int, default=10)
+    ap.add_argument("--max-epoch", type=int, default=200)
+    ap.add_argument("--out", default=None, help="append logs here (default: <run-dir>/milestones_every_N.log)")
+    ap.add_argument("--poll", type=float, default=15.0, help="seconds between reads")
+    ap.add_argument("--idle-exit-sec", type=float, default=180.0,
+                    help="exit if results.csv is untouched this long and milestones caught up")
+    args = ap.parse_args()
+
+    csv_path = Path(args.csv)
+    run_dir = csv_path.parent
+    out_path = Path(args.out) if args.out else run_dir / f"milestones_every_{args.every}.log"
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    if not out_path.exists():
+        ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+        out_path.write_text(
+            f"{ts} | milestones_watch | status=started next_report=epoch_{args.every} csv={csv_path}\n"
+        )
+        print(f"[milestones_watch] created {out_path} (metric lines appear every {args.every} epochs)", flush=True)
+    done = recover_done(out_path)
+
+    while True:
+        if not csv_path.exists():
+            time.sleep(args.poll)
+            continue
+        try:
+            with csv_path.open(newline="") as f:
+                rows = list(csv.DictReader(f))
+        except Exception:
+            time.sleep(args.poll)
+            continue
+        if not rows:
+            time.sleep(args.poll)
+            continue
+
+        last_ep = int(float(rows[-1]["epoch"]))
+        for m in range(args.every, min(last_ep, args.max_epoch) + 1, args.every):
+            if m in done:
+                continue
+            match = [r for r in rows if int(float(r["epoch"])) == m]
+            if not match:
+                continue
+            line = fmt_row(match[0])
+            out_path.parent.mkdir(parents=True, exist_ok=True)
+            with out_path.open("a") as o:
+                o.write(line + "\n")
+            print(line, flush=True)
+            done.add(m)
+
+        mtime = csv_path.stat().st_mtime
+        idle = time.time() - mtime
+        max_reportable = (last_ep // args.every) * args.every
+        need = list(range(args.every, max_reportable + 1, args.every))
+        caught_up = all(m in done for m in need)
+        if idle >= args.idle_exit_sec and caught_up and last_ep >= 1:
+            print(f"milestones_watch: idle {idle:.0f}s and milestones caught up → exit", flush=True)
+            break
+
+        time.sleep(args.poll)
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())