"""Build an immutable geographic validation scope bound to exact model bytes.""" from __future__ import annotations import argparse from hashlib import sha256 import json from pathlib import Path from typing import Any from shapely.geometry import mapping, shape from shapely.ops import unary_union SCHEMA_VERSION = "geointel.model-validation-scope/v1" def file_sha256(path: Path) -> str: digest = sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def read_scope_geometry(path: Path): payload = json.loads(path.read_text(encoding="utf-8")) payload_type = payload.get("type") if isinstance(payload, dict) else None if payload_type == "FeatureCollection": geometries = [shape(feature.get("geometry")) for feature in payload.get("features", [])] geometry = unary_union([item for item in geometries if not item.is_empty]) elif payload_type == "Feature": geometry = shape(payload.get("geometry")) else: geometry = shape(payload) if geometry.is_empty or not geometry.is_valid or geometry.geom_type not in {"Polygon", "MultiPolygon"}: raise ValueError("Scope input must resolve to one non-empty valid Polygon or MultiPolygon") return geometry def build_manifest(args: argparse.Namespace) -> dict[str, Any]: model_path = args.model.expanduser().resolve() scope_path = args.scope_geojson.expanduser().resolve() if not model_path.is_file(): raise FileNotFoundError(f"Model file not found: {model_path}") if not scope_path.is_file(): raise FileNotFoundError(f"Scope GeoJSON not found: {scope_path}") return { "schema_version": SCHEMA_VERSION, "model_id": args.model_id, "model_sha256": file_sha256(model_path), "scope_key": args.scope_key, "crs": "EPSG:4326", "geometry": mapping(read_scope_geometry(scope_path)), "source": { "scope_geojson_path": str(scope_path), "scope_geojson_sha256": file_sha256(scope_path), "authority": args.authority, "snapshot_date": args.snapshot_date, }, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--model", type=Path, required=True) parser.add_argument("--model-id", required=True) parser.add_argument("--scope-geojson", type=Path, required=True) parser.add_argument("--scope-key", required=True) parser.add_argument("--authority", required=True) parser.add_argument("--snapshot-date", required=True) parser.add_argument("--output", type=Path, required=True) return parser.parse_args() def main() -> int: args = parse_args() output = args.output.expanduser().resolve() if output.exists(): raise FileExistsError(f"Refusing to overwrite immutable scope manifest: {output}") output.parent.mkdir(parents=True, exist_ok=True) encoded = (json.dumps(build_manifest(args), ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8") output.write_bytes(encoded) print(json.dumps({"manifest_path": str(output), "manifest_sha256": sha256(encoded).hexdigest()}, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())