103 lines
3.2 KiB
Python
103 lines
3.2 KiB
Python
"""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,
|
|
}
|