90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""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())
|