103 lines
4.1 KiB
Python
103 lines
4.1 KiB
Python
"""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())
|