Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
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())
|