"""Prepare explicit real operator samples for GeoIntel detection QA. This script downloads small orthophoto and GRB building reference pairs from Digitaal Vlaanderen for documented Kempen AOIs. It is an operator/runtime helper, not an application provider integration: no GeoIntel API route calls it and no production data is fetched silently by the app. """ from __future__ import annotations import argparse import json import os import sys from dataclasses import dataclass from pathlib import Path from typing import Any WMS_URL = "https://geo.api.vlaanderen.be/omwrgbmrvl/wms" GRB_GBG_URL = "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/GBG/items" DEFAULT_OUTPUT_DIR = Path("/app/storage/operator-data") requests: Any = None rasterio: Any = None Transformer: Any = None MemoryFile: Any = None from_bounds: Any = None @dataclass(frozen=True) class OperatorSample: slug: str display_name: str center_lon: float center_lat: float half_size_m: float = 250.0 width: int = 512 height: int = 512 SAMPLES: dict[str, OperatorSample] = { "geel": OperatorSample( slug="geel", display_name="Geel center", center_lon=4.991, center_lat=51.162, ), "mol": OperatorSample( slug="mol", display_name="Mol center", center_lon=5.1167, center_lat=51.1919, ), "turnhout": OperatorSample( slug="turnhout", display_name="Turnhout center", center_lon=4.9488, center_lat=51.3225, ), } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Prepare real Digitaal Vlaanderen orthophoto/GRB building samples for GeoIntel operator QA.", ) parser.add_argument( "--output-dir", type=Path, default=Path(os.environ.get("OPERATOR_DATA_DIR", DEFAULT_OUTPUT_DIR)), help="Directory for generated GeoTIFF, GeoJSON and manifest files.", ) parser.add_argument( "--samples", default=",".join(SAMPLES), help="Comma/space separated sample slugs to prepare. Defaults to all documented samples.", ) parser.add_argument( "--force", action="store_true", help="Refetch and overwrite sample files. By default existing raster/reference pairs are reused.", ) parser.add_argument( "--manifest-name", default="operator_samples_manifest.json", help="Manifest filename written inside output-dir.", ) return parser.parse_args() def ensure_gis_dependencies() -> None: global MemoryFile, Transformer, from_bounds, rasterio, requests try: import requests as requests_module import rasterio as rasterio_module from pyproj import Transformer as transformer_class from rasterio.io import MemoryFile as memory_file_class from rasterio.transform import from_bounds as from_bounds_function except Exception as exc: # pragma: no cover - exercised only in runtime envs. raise SystemExit( "prepare_operator_real_data_samples.py requires requests, rasterio and pyproj. " "Run it inside the GeoIntel all-in-one container or an equivalent GIS Python environment." ) from exc requests = requests_module rasterio = rasterio_module Transformer = transformer_class MemoryFile = memory_file_class from_bounds = from_bounds_function def selected_samples(raw: str) -> list[OperatorSample]: slugs = [value.strip().lower() for value in raw.replace(",", " ").split() if value.strip()] if not slugs: raise SystemExit("--samples must include at least one sample slug") unknown = [slug for slug in slugs if slug not in SAMPLES] if unknown: raise SystemExit(f"Unknown sample slug(s): {', '.join(unknown)}. Known: {', '.join(SAMPLES)}") return [SAMPLES[slug] for slug in slugs] def sample_bounds(sample: OperatorSample) -> tuple[tuple[float, float, float, float], list[float]]: lambert = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True) wgs84 = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True) center_x, center_y = lambert.transform(sample.center_lon, sample.center_lat) minx = center_x - sample.half_size_m miny = center_y - sample.half_size_m maxx = center_x + sample.half_size_m maxy = center_y + sample.half_size_m corners = [ wgs84.transform(x, y) for x, y in ((minx, miny), (minx, maxy), (maxx, miny), (maxx, maxy)) ] lon_values = [point[0] for point in corners] lat_values = [point[1] for point in corners] return (minx, miny, maxx, maxy), [ min(lon_values), min(lat_values), max(lon_values), max(lat_values), ] def prepared_url(url: str, params: dict[str, str]) -> str: return requests.Request("GET", url, params=params).prepare().url def raster_summary(path: Path) -> dict[str, Any]: with rasterio.open(path) as ds: return { "path": str(path), "crs": str(ds.crs), "bounds": list(ds.bounds), "width": ds.width, "height": ds.height, "count": ds.count, "dtypes": list(ds.dtypes), } def geojson_feature_count(path: Path) -> int: payload = json.loads(path.read_text(encoding="utf-8-sig")) return len(payload.get("features") or []) def fetch_orthophoto(sample: OperatorSample, ortho_path: Path, lambert_bbox: tuple[float, float, float, float]) -> str: minx, miny, maxx, maxy = lambert_bbox wms_params = { "SERVICE": "WMS", "VERSION": "1.3.0", "REQUEST": "GetMap", "LAYERS": "Ortho", "STYLES": "", "FORMAT": "image/tiff", "CRS": "EPSG:31370", "BBOX": f"{minx},{miny},{maxx},{maxy}", "WIDTH": str(sample.width), "HEIGHT": str(sample.height), } response = requests.get(WMS_URL, params=wms_params, timeout=120) response.raise_for_status() content_type = response.headers.get("content-type", "") if "image" not in content_type.lower() and "tiff" not in content_type.lower(): raise SystemExit(f"Orthophoto WMS did not return an image for {sample.slug}: {content_type}") with MemoryFile(response.content) as memfile: with memfile.open() as src: image = src.read() profile = src.profile.copy() profile.update( driver="GTiff", width=src.width, height=src.height, count=src.count, dtype=src.dtypes[0], crs="EPSG:31370", transform=from_bounds(minx, miny, maxx, maxy, src.width, src.height), compress="deflate", tiled=False, ) with rasterio.open(ortho_path, "w", **profile) as dst: dst.write(image) dst.update_tags( source="Digitaal Vlaanderen OMWRGBMRVL WMS Ortho layer", source_url=prepared_url(WMS_URL, wms_params), attribution="Bron: Orthofotomozaiek Vlaanderen, Digitaal Vlaanderen", aoi=f"{sample.display_name} sample AOI for GeoIntel operator validation", ) return prepared_url(WMS_URL, wms_params) def fetch_reference(sample: OperatorSample, reference_path: Path, geo_bbox: list[float]) -> tuple[str, int]: ogc_params = { "f": "application/geo+json", "limit": "1000", "bbox": ",".join(f"{value:.8f}" for value in geo_bbox), } response = requests.get(GRB_GBG_URL, params=ogc_params, timeout=120) response.raise_for_status() reference = response.json() features = reference.get("features") or [] if not features: raise SystemExit(f"GRB GBG returned no building features for {sample.slug} bbox {geo_bbox}") reference["name"] = f"GRB GBG buildings - {sample.display_name} sample AOI" reference["source"] = "Digitaal Vlaanderen GRB OGC API Features collection GBG" reference["source_url"] = prepared_url(GRB_GBG_URL, ogc_params) reference["attribution"] = "Bron: Grootschalig Referentie Bestand Vlaanderen, Digitaal Vlaanderen" reference["bbox"] = geo_bbox reference["sample_slug"] = sample.slug for feature in features: props = feature.setdefault("properties", {}) props.setdefault("source_name", "grb") props.setdefault("reference_layer_name", "buildings") props.setdefault("sample_slug", sample.slug) reference_path.write_text(json.dumps(reference, ensure_ascii=False), encoding="utf-8") return prepared_url(GRB_GBG_URL, ogc_params), len(features) def prepare_sample(sample: OperatorSample, output_dir: Path, force: bool) -> dict[str, Any]: ortho_path = output_dir / f"{sample.slug}_orthophoto_wms_512.tif" reference_path = output_dir / f"{sample.slug}_grb_gbg_buildings.geojson" lambert_bbox, geo_bbox = sample_bounds(sample) skip_existing = ortho_path.exists() and reference_path.exists() and not force source_urls: dict[str, str | None] = {"orthophoto": None, "reference": None} if not skip_existing: source_urls["orthophoto"] = fetch_orthophoto(sample, ortho_path, lambert_bbox) source_urls["reference"], reference_feature_count = fetch_reference(sample, reference_path, geo_bbox) else: reference_feature_count = geojson_feature_count(reference_path) return { "sample_slug": sample.slug, "display_name": sample.display_name, "center_lon": sample.center_lon, "center_lat": sample.center_lat, "half_size_m": sample.half_size_m, "raster_path": str(ortho_path), "reference_path": str(reference_path), "reference_feature_count": reference_feature_count, "raster": raster_summary(ortho_path), "wgs84_bbox": geo_bbox, "epsg31370_bbox": list(lambert_bbox), "skip_existing": skip_existing, "source_urls": source_urls, "attribution": { "orthophoto": "Bron: Orthofotomozaiek Vlaanderen, Digitaal Vlaanderen", "reference": "Bron: Grootschalig Referentie Bestand Vlaanderen, Digitaal Vlaanderen", }, } def write_readme(output_dir: Path, samples: list[dict[str, Any]]) -> None: lines = [ "# GeoIntel operator real-data samples", "", "Generated for runtime validation, not committed to the GeoIntel repository.", "", "Sources:", "- Orthophoto rasters: Digitaal Vlaanderen OMWRGBMRVL WMS `Ortho` layer.", "- Reference buildings: Digitaal Vlaanderen GRB OGC API Features `GBG` collection.", "- Attribution: Bron: Orthofotomozaiek Vlaanderen / Grootschalig Referentie Bestand Vlaanderen, Digitaal Vlaanderen.", "", "Samples:", ] for sample in samples: lines.append( f"- `{sample['sample_slug']}`: `{Path(sample['raster_path']).name}` and " f"`{Path(sample['reference_path']).name}`, " f"{sample['reference_feature_count']} reference features." ) lines.append("") lines.append("Purpose: configured-YOLO detection + persisted QA/QC validation with operator-provided files.") (output_dir / "README.md").write_text("\n".join(lines) + "\n", encoding="utf-8") def main() -> int: args = parse_args() ensure_gis_dependencies() output_dir: Path = args.output_dir output_dir.mkdir(parents=True, exist_ok=True) samples = [prepare_sample(sample, output_dir, force=args.force) for sample in selected_samples(args.samples)] write_readme(output_dir, samples) manifest = { "schema_version": 1, "description": "GeoIntel operator real-data samples for configured-YOLO QA validation.", "output_dir": str(output_dir), "samples": samples, } manifest_path = output_dir / args.manifest_name manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8") print(json.dumps({"status": "ok", "manifest_path": str(manifest_path), "samples": samples}, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": sys.exit(main())