diff --git a/spacesafeml_certification_benchmark_and_/__init__.py b/spacesafeml_certification_benchmark_and_/__init__.py index 51ddf18..9fe149c 100644 --- a/spacesafeml_certification_benchmark_and_/__init__.py +++ b/spacesafeml_certification_benchmark_and_/__init__.py @@ -4,6 +4,7 @@ __version__ = "0.1.0" # Re-export commonly used components for convenience (optional) from .dsl import LocalCapabilities, SafetyPreConditions, SafetyPostConditions, ResourceBudgets, DataSharingPolicy +from .interoperability import contract_to_ir, contract_to_ir_json from .verification import VerificationEngine from .governance.ledger import Ledger @@ -16,4 +17,6 @@ __all__ = [ "DataSharingPolicy", "VerificationEngine", "Ledger", + "contract_to_ir", + "contract_to_ir_json", ] diff --git a/spacesafeml_certification_benchmark_and_/interoperability.py b/spacesafeml_certification_benchmark_and_/interoperability.py new file mode 100644 index 0000000..5f27ad0 --- /dev/null +++ b/spacesafeml_certification_benchmark_and_/interoperability.py @@ -0,0 +1,102 @@ +"""Canonical interoperability spine for SpaceSafeML. + +Provides a minimal EnergiBridge-like bridge that maps SB-DSL primitives +to a vendor-agnostic intermediate representation (IR). + +This module is intentionally lightweight (MVP) but designed to be extended +with richer mappings and pluggable backends as the ecosystem evolves. +""" +from __future__ import annotations + +from dataclasses import dataclass, field, asdict +from typing import List, Dict, Any + +from .dsl import SafetyContract, LocalCapabilities + + +@dataclass +class IRObject: + id: str + type: str + properties: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class IRMorphism: + signal: str + source: str + target: str + data_schema: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class IRPlanDelta: + delta: Dict[str, Any] + timestamp: str + author: str + contract_id: str + attestation_hash: str + + +@dataclass +class IRDualVariables: + resource_prices: Dict[str, float] = field(default_factory=dict) + privacy_budget: Dict[str, Any] = field(default_factory=dict) + audit_log_ref: str = "" + + +@dataclass +class IRCube: + objects: List[IRObject] = field(default_factory=list) + morphisms: List[IRMorphism] = field(default_factory=list) + plan_delta: IRPlanDelta | None = None + dual_variables: IRDualVariables | None = None + + +def contract_to_ir(contract: SafetyContract) -> IRCube: + """Convert a SafetyContract into a minimal IRCube representation. + + This does not try to be feature-complete; it provides a deterministic, + cross-vendor mapping suitable for early-stage interoperability and + auditing pipelines. + """ + objects: List[IRObject] = [] + for cap in contract.local_capabilities or []: + obj = IRObject( + id=f"cap-{cap.name}", + type="AgentCapability", + properties={"name": cap.name, "capabilities": cap.capabilities, "metadata": cap.metadata}, + ) + objects.append(obj) + + morphisms: List[IRMorphism] = [] + # Create a minimal set of telemeter signals from contract metadata + morphisms.append(IRMorphism(signal="telemetry.metrics", source="Agent", target="Monitor", data_schema={"fields": ["cpu", "memory", "energy"]})) + + # Plan delta is created as a separate container; leave default None here + plan_delta = IRPlanDelta( + delta={"contract_id": contract.contract_id, "note": "initial"}, + timestamp="0", + author="system", + contract_id=contract.contract_id, + attestation_hash="", + ) + + dual = IRDualVariables( + resource_prices={"cpu_cores": 1.0}, + privacy_budget={"remaining": 0}, + audit_log_ref="", + ) + + return IRCube(objects=objects, morphisms=morphisms, plan_delta=plan_delta, dual_variables=dual) + + +def contract_to_ir_json(contract: SafetyContract) -> Dict[str, Any]: + """Serialize the IRCube representation to JSON-friendly dict.""" + ir = contract_to_ir(contract) + return { + "objects": [asdict(o) for o in ir.objects], + "morphisms": [asdict(m) for m in ir.morphisms], + "plan_delta": asdict(ir.plan_delta) if ir.plan_delta else None, + "dual_variables": asdict(ir.dual_variables) if ir.dual_variables else None, + } diff --git a/tests/test_interoperability.py b/tests/test_interoperability.py new file mode 100644 index 0000000..28040b5 --- /dev/null +++ b/tests/test_interoperability.py @@ -0,0 +1,23 @@ +import json +from spacesafeml_certification_benchmark_and_.interoperability import contract_to_ir_json +from spacesafeml_certification_benchmark_and_.dsl import SafetyContract, SafetyPreConditions, SafetyPostConditions, ResourceBudgets, LocalCapabilities, DataSharingPolicy + + +def test_interoperability_contract_to_ir_json_struct(): + contract = SafetyContract( + contract_id="contract-interop-1", + local_capabilities=[LocalCapabilities(name="Planner", capabilities=["plan"] )], + budgets=ResourceBudgets(cpu_cores=2.0, memory_gb=4.0, energy_wh=150.0, time_seconds=3600), + data_policy=DataSharingPolicy(policy_id="policy-interop", allowed_data=["telemetry"]), + version="1.0", + ) + + ir_json = contract_to_ir_json(contract) + # Ensure canonical keys exist and are serializable to JSON + assert "objects" in ir_json + assert "morphisms" in ir_json + assert isinstance(ir_json["objects"], list) + assert isinstance(ir_json["morphisms"], list) + # PlanDelta may be None or a dict depending on mapping; we accept either + # Ensure JSON can be dumped without errors + json.dumps(ir_json)