Reject blank positive imagery in training loop

scripts/run_belgium_building_training_loop.py

@@ -0,0 +1,210 @@
+#!/usr/bin/env python3
+"""Run checkpointed CUDA train/evaluate iterations until gates pass or a batch yields."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import shutil
+import subprocess
+import sys
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def write_json(path: Path, value: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(value, indent=2), encoding="utf-8")
+    temporary.replace(path)
+
+
+def training_command(
+    yolo: str,
+    *,
+    model: Path,
+    data: Path,
+    project: Path,
+    name: str,
+    epochs: int,
+    seed: int,
+    batch: int,
+    workers: int,
+) -> list[str]:
+    return [
+        yolo,
+        "train",
+        f"model={model}",
+        f"data={data}",
+        f"epochs={epochs}",
+        "imgsz=640",
+        f"batch={batch}",
+        "device=0",
+        f"workers={workers}",
+        "patience=35",
+        "cache=disk",
+        "close_mosaic=20",
+        f"seed={seed}",
+        "deterministic=True",
+        f"project={project}",
+        f"name={name}",
+        "exist_ok=True",
+    ]
+
+
+def run(command: list[str], log_path: Path | None = None, *, allowed: set[int] = {0}) -> int:
+    if log_path:
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        with log_path.open("a", encoding="utf-8") as log:
+            completed = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, check=False)
+    else:
+        completed = subprocess.run(command, check=False)
+    if completed.returncode not in allowed:
+        raise RuntimeError(f"Command failed ({completed.returncode}): {' '.join(command)}")
+    return completed.returncode
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--initial-model", type=Path, required=True)
+    parser.add_argument("--train-yaml", type=Path, required=True)
+    parser.add_argument("--calibration-summary", type=Path, required=True)
+    parser.add_argument("--test-summary", type=Path, required=True)
+    parser.add_argument("--background-summary", type=Path, required=True)
+    parser.add_argument("--corpus-manifest", type=Path, required=True)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--iterations", type=int, default=1)
+    parser.add_argument("--epochs", type=int, default=160)
+    parser.add_argument("--batch", type=int, default=2)
+    parser.add_argument("--workers", type=int, default=4)
+    parser.add_argument("--seed", type=int, default=20260731)
+    parser.add_argument("--yolo", default="yolo")
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+    if args.iterations < 1:
+        raise SystemExit("--iterations must be positive")
+
+    state_path = args.output_dir / "training-loop-state.json"
+    state: dict[str, Any] = {
+        "schema_version": 1,
+        "status": "running",
+        "started_at": datetime.now(UTC).isoformat(),
+        "initial_model": str(args.initial_model),
+        "train_yaml": str(args.train_yaml),
+        "corpus_manifest": str(args.corpus_manifest),
+        "iterations": [],
+    }
+    if state_path.is_file():
+        state = json.loads(state_path.read_text(encoding="utf-8"))
+        state["status"] = "running"
+    model = Path(state.get("next_model") or args.initial_model)
+    first_index = len(state["iterations"]) + 1
+    scripts_dir = Path(__file__).resolve().parent
+
+    for offset in range(args.iterations):
+        index = first_index + offset
+        name = f"iteration-{index:03d}"
+        iteration_dir = args.output_dir / name
+        train_run = args.output_dir / "runs" / name
+        command = training_command(
+            args.yolo,
+            model=model,
+            data=args.train_yaml,
+            project=args.output_dir / "runs",
+            name=name,
+            epochs=args.epochs,
+            seed=args.seed + index,
+            batch=args.batch,
+            workers=args.workers,
+        )
+        if args.dry_run:
+            print(json.dumps({"training_command": command}, indent=2))
+            return 0
+        run(command, iteration_dir / "training.log")
+        best = train_run / "weights" / "best.pt"
+        if not best.is_file():
+            raise RuntimeError(f"Training produced no best checkpoint: {best}")
+        candidate = iteration_dir / "candidate.pt"
+        shutil.copy2(best, candidate)
+
+        reports: dict[str, Path] = {}
+        for role, summary in (
+            ("calibration", args.calibration_summary),
+            ("test", args.test_summary),
+            ("background", args.background_summary),
+        ):
+            report = iteration_dir / f"{role}.json"
+            reports[role] = report
+            run(
+                [
+                    sys.executable,
+                    str(scripts_dir / "evaluate_belgium_building_candidate.py"),
+                    "--model",
+                    str(candidate),
+                    "--summary",
+                    str(summary),
+                    "--corpus-manifest",
+                    str(args.corpus_manifest),
+                    "--output",
+                    str(report),
+                    "--device",
+                    "cuda:0",
+                ],
+                iteration_dir / f"{role}.log",
+            )
+        assessment = iteration_dir / "assessment.json"
+        run(
+            [
+                sys.executable,
+                str(scripts_dir / "assess_belgium_building_training_iteration.py"),
+                "--calibration",
+                str(reports["calibration"]),
+                "--test",
+                str(reports["test"]),
+                "--background",
+                str(reports["background"]),
+                "--output",
+                str(assessment),
+            ],
+            iteration_dir / "assessment.log",
+            allowed={0, 2},
+        )
+        decision = json.loads(assessment.read_text(encoding="utf-8"))
+        record = {
+            "iteration": index,
+            "candidate": str(candidate),
+            "candidate_sha256": sha256(candidate),
+            "assessment": str(assessment),
+            "status": decision["status"],
+            "failures": decision["failures"],
+        }
+        state["iterations"].append(record)
+        state["next_model"] = str(candidate)
+        if decision["status"] == "training_complete":
+            state["status"] = "training_complete"
+            state["completed_at"] = datetime.now(UTC).isoformat()
+            write_json(state_path, state)
+            print(json.dumps(state, indent=2))
+            return 0
+        model = candidate
+        write_json(state_path, state)
+
+    state["status"] = "continue_training_loop"
+    state["yielded_at"] = datetime.now(UTC).isoformat()
+    write_json(state_path, state)
+    print(json.dumps(state, indent=2))
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())