Add deterministic portfolio spec composition

scripts/combine_belgium_building_portfolio_specs.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Combine completed Belgian portfolio specs with fail-closed provenance."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def combine_specs(paths: list[Path]) -> dict[str, Any]:
+    if not paths:
+        raise ValueError("At least one portfolio spec is required")
+    samples: list[dict[str, Any]] = []
+    seen: set[str] = set()
+    side_m: float | None = None
+    resolution_m: float | None = None
+    sources = []
+    for path in paths:
+        payload = json.loads(path.read_text(encoding="utf-8-sig"))
+        if payload.get("status") != "complete":
+            raise ValueError(f"Portfolio spec is not complete: {path}")
+        current_side = float(payload["side_m"])
+        current_resolution = float(payload["resolution_m"])
+        if side_m is not None and current_side != side_m:
+            raise ValueError(f"Portfolio side_m differs: {path}")
+        if resolution_m is not None and current_resolution != resolution_m:
+            raise ValueError(f"Portfolio resolution_m differs: {path}")
+        side_m, resolution_m = current_side, current_resolution
+        for sample in payload.get("samples") or []:
+            slug = str(sample.get("sample_slug") or sample.get("slug") or "")
+            if not slug:
+                raise ValueError(f"Portfolio sample has no slug: {path}")
+            if slug in seen:
+                raise ValueError(f"Duplicate portfolio sample slug: {slug}")
+            seen.add(slug)
+            normalized = dict(sample)
+            normalized["sample_slug"] = slug
+            normalized.pop("slug", None)
+            samples.append(normalized)
+        sources.append({"path": str(path), "sha256": sha256(path), "sample_count": len(payload.get("samples") or [])})
+    return {
+        "schema_version": 2,
+        "status": "complete",
+        "side_m": side_m,
+        "resolution_m": resolution_m,
+        "sample_count": len(samples),
+        "source_specs": sources,
+        "samples": samples,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--spec", action="append", type=Path, required=True)
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    payload = combine_specs(args.spec)
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(json.dumps({"status": "ok", "sample_count": payload["sample_count"], "output": str(args.output)}))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())