Add direct polygon label QA

scripts/render_operator_polygon_label_qa.py

@@ -0,0 +1,102 @@
+"""Render governed source polygons directly on their raster before YOLO box export."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Render direct polygon/raster alignment evidence from an operator samples manifest."
+    )
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--output-dir", type=Path, required=True)
+    parser.add_argument("--line-width", type=int, default=3)
+    parser.add_argument("--force", action="store_true")
+    return parser.parse_args()
+
+
+def geometry_rings(geometry: dict[str, Any]):
+    if geometry.get("type") == "Polygon":
+        yield from geometry.get("coordinates") or []
+    elif geometry.get("type") == "MultiPolygon":
+        for polygon in geometry.get("coordinates") or []:
+            yield from polygon
+
+
+def render(manifest_path: Path, output_dir: Path, line_width: int, force: bool) -> dict[str, Any]:
+    try:
+        import rasterio
+        from PIL import Image, ImageDraw
+        from pyproj import Transformer
+    except ImportError as exc:  # pragma: no cover - runtime dependency guard
+        raise SystemExit("Polygon QA rendering requires rasterio, Pillow and pyproj") from exc
+
+    if output_dir.exists() and any(output_dir.iterdir()) and not force:
+        raise SystemExit(f"Output directory is not empty: {output_dir}; use --force")
+    output_dir.mkdir(parents=True, exist_ok=True)
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig"))
+    evidence = []
+    for sample in manifest.get("samples") or []:
+        slug = str(sample["sample_slug"])
+        raster_path = Path(sample["raster_path"])
+        reference_path = Path(sample["reference_path"])
+        reference = json.loads(reference_path.read_text(encoding="utf-8-sig"))
+        feature_count = 0
+        with rasterio.open(raster_path) as raster:
+            if raster.crs is None:
+                raise SystemExit(f"Raster CRS is required for {slug}")
+            bands = raster.read([1, 2, 3]).transpose(1, 2, 0)
+            image = Image.fromarray(bands)
+            draw = ImageDraw.Draw(image)
+            transformer = Transformer.from_crs("EPSG:4326", raster.crs, always_xy=True)
+            for feature in reference.get("features") or []:
+                drew_feature = False
+                for ring in geometry_rings(feature.get("geometry") or {}):
+                    pixels = []
+                    for coordinate in ring:
+                        if len(coordinate) < 2:
+                            continue
+                        x, y = transformer.transform(float(coordinate[0]), float(coordinate[1]))
+                        row, column = raster.index(x, y)
+                        pixels.append((column, row))
+                    if len(pixels) >= 3:
+                        draw.line(pixels + [pixels[0]], fill=(255, 220, 0), width=line_width)
+                        drew_feature = True
+                feature_count += int(drew_feature)
+            output_path = output_dir / f"{slug}-polygon-overlay.png"
+            image.save(output_path)
+        evidence.append(
+            {
+                "sample_slug": slug,
+                "raster_path": str(raster_path),
+                "reference_path": str(reference_path),
+                "rendered_feature_count": feature_count,
+                "output_path": str(output_path),
+            }
+        )
+    summary = {
+        "schema_version": 1,
+        "status": "ok",
+        "manifest_path": str(manifest_path),
+        "label_semantics": "direct_governed_polygon_overlay_before_box_export",
+        "samples": evidence,
+    }
+    (output_dir / "operator_polygon_label_qa_summary.json").write_text(
+        json.dumps(summary, indent=2), encoding="utf-8"
+    )
+    return summary
+
+
+def main() -> int:
+    args = parse_args()
+    summary = render(args.manifest, args.output_dir, args.line_width, args.force)
+    print(json.dumps({"status": summary["status"], "sample_count": len(summary["samples"])}, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())