Initial public release

scripts/derive_yolo_nonoverlap_evaluation_view.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+"""Derive a checksum-bound, training-disabled non-overlapping YOLO val view."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import shutil
+from collections import defaultdict
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from PIL import Image
+
+try:
+    from scripts.audit_yolo_cross_tile_repetition import tile_offsets
+except ModuleNotFoundError:  # Standalone operator-tool copy beside the auditor.
+    from audit_yolo_cross_tile_repetition import tile_offsets
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def select_nonoverlap_tiles(
+    tiles: list[dict[str, Any]], tile_size: int
+) -> list[dict[str, Any]]:
+    selected: list[dict[str, Any]] = []
+    for tile in tiles:
+        if not isinstance(tile, dict) or not tile.get("kept", True):
+            continue
+        if str(tile.get("split") or "") != "val":
+            continue
+        row_offset, column_offset = tile_offsets(str(tile.get("image_path") or ""))
+        if row_offset % tile_size == 0 and column_offset % tile_size == 0:
+            selected.append(tile)
+    return sorted(
+        selected,
+        key=lambda tile: (
+            str(tile.get("sample_slug") or ""),
+            int(tile.get("tile_index") or 0),
+        ),
+    )
+
+
+def aggregate_hash(paths: list[Path], root: Path) -> str:
+    digest = hashlib.sha256()
+    for path in sorted(paths):
+        digest.update(path.relative_to(root).as_posix().encode("utf-8"))
+        digest.update(b"\0")
+        digest.update(sha256_file(path).encode("ascii"))
+        digest.update(b"\n")
+    return digest.hexdigest()
+
+
+def image_has_low_visual_variance(image_path: Path, blank_range_threshold: int) -> bool:
+    image = Image.open(image_path).convert("L")
+    minimum, maximum = image.getextrema()
+    return maximum - minimum <= blank_range_threshold
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--source-dir", required=True, type=Path)
+    parser.add_argument("--output-dir", required=True, type=Path)
+    parser.add_argument(
+        "--exclude-low-variance",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+    )
+    parser.add_argument("--blank-range-threshold", type=int, default=3)
+    args = parser.parse_args()
+
+    source_dir = args.source_dir.expanduser().resolve(strict=True)
+    output_dir = args.output_dir.expanduser().resolve(strict=False)
+    if output_dir.exists():
+        parser.error(f"output directory already exists: {output_dir}")
+    source_summary_path = source_dir / "yolo_tile_dataset_summary.json"
+    source_summary = json.loads(source_summary_path.read_text(encoding="utf-8"))
+    tiles = source_summary.get("tiles")
+    tile_size = source_summary.get("tile_size")
+    if not isinstance(tiles, list):
+        raise ValueError("source summary must contain a tiles list")
+    if not isinstance(tile_size, int) or tile_size <= 0:
+        raise ValueError("source summary must contain a positive integer tile_size")
+    selected_candidates = select_nonoverlap_tiles(tiles, tile_size)
+    if not selected_candidates:
+        raise ValueError("no non-overlapping validation tiles were selected")
+
+    selected: list[dict[str, Any]] = []
+    excluded_tiles: list[dict[str, Any]] = []
+    for tile in selected_candidates:
+        source_image = Path(str(tile.get("image_path") or "")).resolve(strict=True)
+        if args.exclude_low_variance and image_has_low_visual_variance(
+            source_image, args.blank_range_threshold
+        ):
+            excluded_tiles.append(
+                {
+                    "sample_slug": str(tile.get("sample_slug") or "unknown"),
+                    "tile_index": int(tile.get("tile_index") or 0),
+                    "source_image_path": str(source_image),
+                    "reason": "low_visual_variance_no_data",
+                }
+            )
+            continue
+        selected.append(tile)
+    if not selected:
+        raise ValueError("every non-overlapping validation tile was excluded")
+
+    offsets_by_sample: dict[str, set[tuple[int, int]]] = defaultdict(set)
+    source_val_counts: dict[str, int] = defaultdict(int)
+    for tile in tiles:
+        if (
+            isinstance(tile, dict)
+            and tile.get("kept", True)
+            and tile.get("split") == "val"
+        ):
+            source_val_counts[str(tile.get("sample_slug") or "unknown")] += 1
+    for tile in selected:
+        offsets_by_sample[str(tile.get("sample_slug") or "unknown")].add(
+            tile_offsets(str(tile.get("image_path") or ""))
+        )
+    accounted_samples = set(offsets_by_sample) | {
+        tile["sample_slug"] for tile in excluded_tiles
+    }
+    if accounted_samples != set(source_val_counts):
+        raise ValueError("non-overlap selection has an unaccounted validation sample")
+
+    output_dir.mkdir(parents=True)
+    (output_dir / "images" / "train").mkdir(parents=True)
+    (output_dir / "labels" / "train").mkdir(parents=True)
+    output_tiles: list[dict[str, Any]] = []
+    output_files: list[Path] = []
+    for tile in selected:
+        source_image = Path(str(tile.get("image_path") or "")).resolve(strict=True)
+        source_label = Path(str(tile.get("label_path") or "")).resolve(strict=True)
+        target_image = output_dir / "images" / "val" / source_image.name
+        target_label = output_dir / "labels" / "val" / source_label.name
+        target_image.parent.mkdir(parents=True, exist_ok=True)
+        target_label.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(source_image, target_image)
+        shutil.copy2(source_label, target_label)
+        output_files.extend((target_image, target_label))
+        derived_tile = dict(tile)
+        derived_tile.update(
+            {
+                "image_path": str(target_image),
+                "label_path": str(target_label),
+                "derived_from_image_path": str(source_image),
+                "derived_from_label_path": str(source_label),
+            }
+        )
+        output_tiles.append(derived_tile)
+
+    dataset_yaml = output_dir / "dataset.yaml"
+    dataset_yaml.write_text(
+        "\n".join(
+            (
+                f"path: {output_dir}",
+                "train: images/train",
+                "val: images/val",
+                "names:",
+                "  0: building",
+                "",
+            )
+        ),
+        encoding="utf-8",
+    )
+    no_training = {
+        "schema_version": 1,
+        "training_prohibited": True,
+        "reason": "Immutable non-overlapping evaluation view; train directories are intentionally empty.",
+        "source_summary_sha256": sha256_file(source_summary_path),
+    }
+    no_training_path = output_dir / "NO_TRAINING.json"
+    no_training_path.write_text(
+        json.dumps(no_training, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    summary = {
+        "schema_version": 1,
+        "status": "ready_evaluation_only",
+        "claim_boundary": (
+            "Non-protected, non-overlapping tile-level validation view; no "
+            "training, protected-test, object-level independence or promotion claim."
+        ),
+        "dataset_yaml": str(dataset_yaml),
+        "output_dir": str(output_dir),
+        "source_dataset_dir": str(source_dir),
+        "source_summary_path": str(source_summary_path),
+        "source_summary_sha256": sha256_file(source_summary_path),
+        "class_names": source_summary.get("class_names", ["building"]),
+        "tile_size": tile_size,
+        "stride": tile_size,
+        "source_stride": source_summary.get("stride"),
+        "training_prohibited": True,
+        "training_marker": str(no_training_path),
+        "tile_count": len(output_tiles),
+        "val_tile_count": len(output_tiles),
+        "train_tile_count": 0,
+        "positive_tile_count": sum(
+            int(tile.get("label_count") or 0) > 0 for tile in output_tiles
+        ),
+        "negative_tile_count": sum(
+            int(tile.get("label_count") or 0) == 0 for tile in output_tiles
+        ),
+        "label_count": sum(int(tile.get("label_count") or 0) for tile in output_tiles),
+        "selected_offsets_by_sample": {
+            sample: [list(offset) for offset in sorted(offsets)]
+            for sample, offsets in sorted(offsets_by_sample.items())
+        },
+        "source_val_tile_counts_by_sample": dict(sorted(source_val_counts.items())),
+        "excluded_tile_count": len(excluded_tiles),
+        "excluded_tiles": excluded_tiles,
+        "excluded_sample_slugs": sorted(
+            {tile["sample_slug"] for tile in excluded_tiles}
+        ),
+        "exclude_low_variance": args.exclude_low_variance,
+        "blank_range_threshold": args.blank_range_threshold,
+        "tiles": output_tiles,
+    }
+    summary_path = output_dir / "yolo_tile_dataset_summary.json"
+    summary_path.write_text(
+        json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    manifest = {
+        "schema_version": 1,
+        "generated_at": datetime.now(UTC).isoformat(),
+        "status": "complete_evaluation_only",
+        "source_summary_path": str(source_summary_path),
+        "source_summary_sha256": sha256_file(source_summary_path),
+        "output_summary_path": str(summary_path),
+        "output_summary_sha256": sha256_file(summary_path),
+        "dataset_yaml_sha256": sha256_file(dataset_yaml),
+        "evaluation_file_set_aggregate_sha256": aggregate_hash(
+            output_files, output_dir
+        ),
+        "training_marker_sha256": sha256_file(no_training_path),
+        "training_prohibited": True,
+        "selected_tile_count": len(output_tiles),
+        "selected_sample_count": len(offsets_by_sample),
+        "excluded_tile_count": len(excluded_tiles),
+        "excluded_tiles": excluded_tiles,
+    }
+    manifest_path = output_dir / "evaluation_view_manifest.json"
+    manifest_path.write_text(
+        json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    print(json.dumps(manifest, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())