Add leak-free regional YOLO dataset builder

scripts/build_regional_yolo_dataset.py

@@ -1,15 +1,19 @@
 #!/usr/bin/env python3
-"""Create a checksummed regional YOLO view without copying protected data."""
+"""Build a leak-free regional YOLO view over an immutable tiled corpus."""
 
 from __future__ import annotations
 
 import argparse
 import hashlib
 import json
+from collections import Counter
 from pathlib import Path
 from typing import Any
 
 
+PROTECTED_SPLITS = {"calibration", "test", "background-test"}
+
+
 def sha256(path: Path) -> str:
     digest = hashlib.sha256()
     with path.open("rb") as stream:
@@ -18,21 +22,80 @@ def sha256(path: Path) -> str:
     return digest.hexdigest()
 
 
-def select_paths(summary: dict[str, Any], manifest: dict[str, Any], region: str) -> tuple[list[str], list[str]]:
+def write_lines(path: Path, values: list[str]) -> None:
+    path.write_text("".join(f"{value}\n" for value in values), encoding="utf-8")
+
+
+def build(
+    *,
+    summary: dict[str, Any],
+    manifest: dict[str, Any],
+    region: str,
+    priority_contexts: set[str],
+    priority_repeat: int,
+    negative_repeat: int,
+) -> tuple[list[str], list[str], dict[str, Any]]:
     samples = {item["sample_slug"]: item for item in manifest["samples"]}
+    selected: list[tuple[dict[str, Any], dict[str, Any]]] = []
+    protected: list[str] = []
+    unknown: list[str] = []
+    for tile in summary["tiles"]:
+        if not tile.get("kept", True):
+            continue
+        sample = samples.get(tile["sample_slug"])
+        if sample is None:
+            unknown.append(tile["sample_slug"])
+            continue
+        split = str(tile.get("split") or sample.get("split") or "")
+        if split in PROTECTED_SPLITS:
+            protected.append(tile["sample_slug"])
+            continue
+        if sample.get("region") == region and split in {"train", "val"}:
+            selected.append((tile, sample))
+    if unknown:
+        raise ValueError(f"summary references unknown samples: {sorted(set(unknown))}")
+    if protected:
+        raise ValueError(f"summary contains protected tiles: {sorted(set(protected))}")
+
     train: list[str] = []
     val: list[str] = []
-    for tile in summary["tiles"]:
+    sample_counts: Counter[str] = Counter()
-        sample = samples[tile["sample_slug"]]
+    context_counts: Counter[str] = Counter()
-        if sample["region"] != region or not tile.get("kept", True):
+    negative_count = 0
+    for tile, sample in selected:
+        image_path = str(tile["image_path"])
+        split = str(tile.get("split") or sample.get("split"))
+        if split == "val":
+            val.append(image_path)
             continue
-        if sample["split"] == "train" and tile["split"] == "train":
+        context = str(sample.get("context") or "unknown")
-            train.append(tile["image_path"])
+        is_negative = bool(tile.get("is_negative"))
-        elif sample["split"] == "val" and tile["split"] == "val":
+        repeat = negative_repeat if is_negative else priority_repeat if context in priority_contexts else 1
-            val.append(tile["image_path"])
+        train.extend([image_path] * repeat)
+        sample_counts[sample["sample_slug"]] += repeat
+        context_counts[context] += repeat
+        negative_count += repeat if is_negative else 0
     if not train or not val:
-        raise ValueError(f"Region {region!r} must contain train and validation images")
+        raise ValueError(f"regional dataset requires non-empty train and val lists: {region}")
-    return sorted(train), sorted(val)
+    if set(train) & set(val):
+        raise ValueError("regional train/validation image leakage")
+    evidence = {
+        "schema_version": 1,
+        "status": "ok",
+        "region": region,
+        "priority_contexts": sorted(priority_contexts),
+        "priority_repeat": priority_repeat,
+        "negative_repeat": negative_repeat,
+        "train_entry_count": len(train),
+        "train_unique_image_count": len(set(train)),
+        "validation_image_count": len(val),
+        "negative_train_entry_count": negative_count,
+        "sample_entry_counts": dict(sorted(sample_counts.items())),
+        "context_entry_counts": dict(sorted(context_counts.items())),
+        "protected_samples_in_training": [],
+        "train_validation_overlap": [],
+    }
+    return train, val, evidence
 
 
 def main() -> int:
@@ -41,34 +104,44 @@ def main() -> int:
     parser.add_argument("--corpus-manifest", type=Path, required=True)
     parser.add_argument("--region", required=True)
     parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--priority-context", action="append", default=[])
+    parser.add_argument("--priority-repeat", type=int, default=2)
+    parser.add_argument("--negative-repeat", type=int, default=2)
     args = parser.parse_args()
+    if args.priority_repeat < 1 or args.negative_repeat < 1:
+        parser.error("repeat counts must be positive")
     summary = json.loads(args.summary.read_text(encoding="utf-8"))
     manifest = json.loads(args.corpus_manifest.read_text(encoding="utf-8"))
-    train, val = select_paths(summary, manifest, args.region)
+    train, val, evidence = build(
-    args.output_dir.mkdir(parents=True, exist_ok=True)
+        summary=summary,
-    train_list = args.output_dir / "train.txt"
+        manifest=manifest,
-    val_list = args.output_dir / "val.txt"
+        region=args.region,
-    train_list.write_text("\n".join(train) + "\n", encoding="utf-8")
+        priority_contexts=set(args.priority_context),
-    val_list.write_text("\n".join(val) + "\n", encoding="utf-8")
+        priority_repeat=args.priority_repeat,
-    dataset_yaml = args.output_dir / "dataset.yaml"
+        negative_repeat=args.negative_repeat,
-    dataset_yaml.write_text(
+    )
-        f"path: {args.output_dir}\ntrain: {train_list}\nval: {val_list}\nnames:\n  0: building\n",
+    args.output_dir.mkdir(parents=True, exist_ok=False)
+    train_path = args.output_dir / "train.txt"
+    val_path = args.output_dir / "val.txt"
+    write_lines(train_path, train)
+    write_lines(val_path, val)
+    yaml_path = args.output_dir / "dataset.yaml"
+    yaml_path.write_text(
+        f"path: /\ntrain: {train_path}\nval: {val_path}\nnames:\n  0: building\n",
         encoding="utf-8",
     )
-    evidence = {
+    evidence.update({
-        "schema_version": 1,
+        "source_summary": str(args.summary),
-        "status": "ok",
+        "source_summary_sha256": sha256(args.summary),
-        "region": args.region,
-        "train_image_count": len(train),
-        "validation_image_count": len(val),
-        "summary": str(args.summary),
-        "summary_sha256": sha256(args.summary),
         "corpus_manifest": str(args.corpus_manifest),
         "corpus_manifest_sha256": sha256(args.corpus_manifest),
-        "dataset_yaml": str(dataset_yaml),
+        "train_sha256": sha256(train_path),
-        "protected_splits_in_training": [],
+        "validation_sha256": sha256(val_path),
-    }
+        "dataset_yaml": str(yaml_path),
-    (args.output_dir / "regional-dataset.json").write_text(json.dumps(evidence, indent=2), encoding="utf-8")
+    })
+    (args.output_dir / "regional-dataset-evidence.json").write_text(
+        json.dumps(evidence, indent=2), encoding="utf-8"
+    )
     print(json.dumps(evidence, indent=2))
     return 0
 

tests/test_build_regional_yolo_dataset.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import pytest
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "build_regional_yolo_dataset.py"
+SPEC = importlib.util.spec_from_file_location("build_regional_yolo_dataset", SCRIPT)
+assert SPEC and SPEC.loader
+module = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(module)
+
+
+def test_regional_dataset_balances_contexts_and_keeps_validation_closed() -> None:
+    manifest = {"samples": [
+        {"sample_slug": "f-train", "region": "flanders", "split": "train", "context": "ribbon"},
+        {"sample_slug": "f-val", "region": "flanders", "split": "val", "context": "mixed"},
+        {"sample_slug": "w-train", "region": "wallonia", "split": "train", "context": "ribbon"},
+    ]}
+    summary = {"tiles": [
+        {"sample_slug": "f-train", "split": "train", "image_path": "/f-pos.png", "kept": True},
+        {"sample_slug": "f-train", "split": "train", "image_path": "/f-neg.png", "is_negative": True},
+        {"sample_slug": "f-val", "split": "val", "image_path": "/f-val.png"},
+        {"sample_slug": "w-train", "split": "train", "image_path": "/w.png"},
+    ]}
+    train, val, evidence = module.build(
+        summary=summary, manifest=manifest, region="flanders",
+        priority_contexts={"ribbon"}, priority_repeat=3, negative_repeat=2,
+    )
+    assert train == ["/f-pos.png"] * 3 + ["/f-neg.png"] * 2
+    assert val == ["/f-val.png"]
+    assert evidence["negative_train_entry_count"] == 2
+    assert evidence["protected_samples_in_training"] == []
+
+
+def test_regional_dataset_rejects_protected_tiles() -> None:
+    manifest = {"samples": [
+        {"sample_slug": "f-cal", "region": "flanders", "split": "calibration", "context": "ribbon"},
+    ]}
+    summary = {"tiles": [{"sample_slug": "f-cal", "split": "calibration", "image_path": "/cal.png"}]}
+    with pytest.raises(ValueError, match="protected"):
+        module.build(
+            summary=summary, manifest=manifest, region="flanders",
+            priority_contexts=set(), priority_repeat=1, negative_repeat=1,
+        )