Create sovereign_capacity_seal.py

sovereign_capacity_seal.py

@@ -0,0 +1,194 @@
+
+# sovereign_capacity_seal.py
+from __future__ import annotations
+
+import json
+from datetime import datetime, timezone
+from typing import Any, Dict, Optional
+
+from sovereign_crypto_seal import SovereignCryptoSeal, build_manifest
+from sovereign_capacity_report import generate_capacity_report
+
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+class SovereignCapacitySeal:
+    """
+    Seals and verifies Sovereign capacity artifacts.
+
+    Primary use:
+    - capacity report
+    - capacity manifest
+    """
+
+    def __init__(self, sealer: Optional[SovereignCryptoSeal] = None) -> None:
+        self.sealer = sealer or SovereignCryptoSeal()
+
+    # ---------------------------------------------------------
+    # Capacity report sealing
+    # ---------------------------------------------------------
+    def seal_capacity_report(
+        self,
+        report: Dict[str, Any],
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        sealed = self.sealer.seal_object(
+            report,
+            object_type="capacity_report",
+            metadata=metadata or {
+                "domain": "banking",
+                "artifact_class": "capacity_report",
+            },
+        )
+        return {
+            "ok": True,
+            "sealed": True,
+            "report": report,
+            "crypto_seal": sealed["seal"],
+        }
+
+    def verify_capacity_report(
+        self,
+        report: Dict[str, Any],
+        seal: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        verified = self.sealer.verify_object(report, seal)
+        return {
+            "ok": bool(verified.get("ok")),
+            "verified": bool(verified.get("verified")),
+            "artifact_type": "capacity_report",
+            "verification": verified,
+        }
+
+    # ---------------------------------------------------------
+    # Generate + seal convenience
+    # ---------------------------------------------------------
+    def generate_and_seal_capacity_report(
+        self,
+        iterations_per_case: int = 25,
+        warmup_runs: int = 2,
+        output_dir: str = "sovereign_capacity_outputs",
+        base_name: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        generated = generate_capacity_report(
+            iterations_per_case=iterations_per_case,
+            warmup_runs=warmup_runs,
+            output_dir=output_dir,
+            base_name=base_name,
+        )
+
+        report = generated.get("report", {}) if isinstance(generated, dict) else {}
+        sealed_bundle = self.seal_capacity_report(
+            report=report,
+            metadata=metadata or {
+                "domain": "banking",
+                "artifact_class": "capacity_report",
+                "generated_at": _utc_now(),
+                "iterations_per_case": iterations_per_case,
+                "warmup_runs": warmup_runs,
+            },
+        )
+
+        return {
+            "ok": True,
+            "generated_at": _utc_now(),
+            "report": report,
+            "files": generated.get("files", {}) if isinstance(generated, dict) else {},
+            "crypto_seal": sealed_bundle["crypto_seal"],
+            "sealed": True,
+        }
+
+    # ---------------------------------------------------------
+    # Capacity manifest
+    # ---------------------------------------------------------
+    def build_capacity_manifest(
+        self,
+        *,
+        subject_name: str,
+        subject_version: str,
+        capacity_report: Dict[str, Any],
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
+        manifest_bundle = build_manifest(
+            manifest_type="sovereign_capacity_bundle",
+            subject_name=subject_name,
+            subject_version=subject_version,
+            artifacts={
+                "capacity_report": capacity_report,
+            },
+            metadata=metadata or {
+                "domain": "banking",
+                "artifact_class": "capacity_bundle",
+            },
+        )
+        return {
+            "ok": True,
+            "manifest_bundle": manifest_bundle,
+            "artifacts_included": ["capacity_report"],
+            "artifact_count": 1,
+        }
+
+
+SEALER = SovereignCapacitySeal()
+
+
+def seal_capacity_report(
+    report: Dict[str, Any],
+    metadata: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    return SEALER.seal_capacity_report(report, metadata)
+
+
+def verify_capacity_report(
+    report: Dict[str, Any],
+    seal: Dict[str, Any],
+) -> Dict[str, Any]:
+    return SEALER.verify_capacity_report(report, seal)
+
+
+def generate_and_seal_capacity_report(
+    iterations_per_case: int = 25,
+    warmup_runs: int = 2,
+    output_dir: str = "sovereign_capacity_outputs",
+    base_name: Optional[str] = None,
+    metadata: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    return SEALER.generate_and_seal_capacity_report(
+        iterations_per_case=iterations_per_case,
+        warmup_runs=warmup_runs,
+        output_dir=output_dir,
+        base_name=base_name,
+        metadata=metadata,
+    )
+
+
+def build_capacity_manifest(
+    *,
+    subject_name: str,
+    subject_version: str,
+    capacity_report: Dict[str, Any],
+    metadata: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    return SEALER.build_capacity_manifest(
+        subject_name=subject_name,
+        subject_version=subject_version,
+        capacity_report=capacity_report,
+        metadata=metadata,
+    )
+
+
+if __name__ == "__main__":
+    out = generate_and_seal_capacity_report(iterations_per_case=10, warmup_runs=1)
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+
+    manifest = build_capacity_manifest(
+        subject_name="Sovereign Bank Runtime",
+        subject_version="v1",
+        capacity_report=out.get("report", {}),
+        metadata={"pack_type": "capacity_identity_seed"},
+    )
+    print(json.dumps(manifest, indent=2, ensure_ascii=False))
+