diff --git a/scripts/combine_belgium_building_portfolio_specs.py b/scripts/combine_belgium_building_portfolio_specs.py new file mode 100644 index 00000000..34603f8f --- /dev/null +++ b/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()) diff --git a/tests/test_combine_belgium_building_portfolio_specs.py b/tests/test_combine_belgium_building_portfolio_specs.py new file mode 100644 index 00000000..f4a28c9a --- /dev/null +++ b/tests/test_combine_belgium_building_portfolio_specs.py @@ -0,0 +1,38 @@ +import json +from pathlib import Path + +import pytest + +from scripts.combine_belgium_building_portfolio_specs import combine_specs + + +def write_spec(path: Path, slug: str, *, resolution: float = 0.25, status: str = "complete") -> Path: + path.write_text(json.dumps({ + "status": status, + "side_m": 256.0, + "resolution_m": resolution, + "samples": [{"sample_slug": slug}], + }), encoding="utf-8") + return path + + +def test_combine_specs_records_provenance(tmp_path: Path) -> None: + payload = combine_specs([write_spec(tmp_path / "one.json", "one"), write_spec(tmp_path / "two.json", "two")]) + assert payload["sample_count"] == 2 + assert [sample["sample_slug"] for sample in payload["samples"]] == ["one", "two"] + assert all(len(source["sha256"]) == 64 for source in payload["source_specs"]) + + +def test_combine_specs_rejects_duplicate_slugs(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="Duplicate"): + combine_specs([write_spec(tmp_path / "one.json", "same"), write_spec(tmp_path / "two.json", "same")]) + + +def test_combine_specs_rejects_mismatched_resolution(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="resolution"): + combine_specs([write_spec(tmp_path / "one.json", "one"), write_spec(tmp_path / "two.json", "two", resolution=0.5)]) + + +def test_combine_specs_rejects_incomplete_source(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="not complete"): + combine_specs([write_spec(tmp_path / "one.json", "one", status="in_progress")])