build(agent): new-agents#a6e6ec iteration

src/cosmosmesh_privacy_preserving_federated_/energi_bridge_ir.py

@@ -0,0 +1,127 @@
+"""EnergiBridge IR (Interoperability) primitives.
+
+This module provides a canonical, vendor-agnostic IR representation to map
+CosmosMesh LocalProblems, Shared/Dual Variables, and PlanDeltas into a
+structured, versioned interchange format. It is designed to complement the
+existing EnergiBridge (to_ir) translator by offering a stable, serializable
+schema that can be used by adapters and governance tooling.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, List
+
+
+@dataclass
+class IRObject:
+    """Canonical object representation (e.g., a LocalProblem).
+
+    type field indicates the kind (e.g., "LocalProblem").
+    data contains the payload specific to the object kind.
+    """
+
+    type: str
+    data: Dict[str, Any]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"type": self.type, "data": self.data}
+
+
+@dataclass
+class IRMorphism:
+    """Canonical morphism (e.g., SharedVariable or DualVariable updates)."""
+
+    kind: str
+    data: Dict[str, Any]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"kind": self.kind, "data": self.data}
+
+
+@dataclass
+class IRPlanDelta:
+    """Incremental plan delta with provenance."""
+
+    delta_id: str
+    changes: Dict[str, Any]
+    timestamp: float | None = None
+    author: str | None = None
+    contract_id: str | None = None
+    signature: str | None = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "delta_id": self.delta_id,
+            "changes": self.changes,
+            "timestamp": self.timestamp,
+            "author": self.author,
+            "contract_id": self.contract_id,
+            "signature": self.signature,
+        }
+
+
+@dataclass
+class IR:
+    """Top-level IR container for a given exchange cycle."""
+
+    version: str
+    objects: List[IRObject]
+    morphisms: List[IRMorphism]
+    plan_deltas: List[IRPlanDelta]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "version": self.version,
+            "objects": [o.to_dict() for o in self.objects],
+            "morphisms": [m.to_dict() for m in self.morphisms],
+            "plan_deltas": [d.to_dict() for d in self.plan_deltas],
+        }
+
+
+def to_ir(
+    lp: Dict[str, Any] | None,
+    sv: List[Dict[str, Any]] | None,
+    dv: List[Dict[str, Any]] | None,
+    deltas: List[Dict[str, Any]] | None,
+) -> IR:
+    """Convert EnergiBridge-style EP dicts to the canonical IR structure.
+
+    This function accepts pre-serialized dictionaries (as used by the
+    EnergiBridge.to_ir path) and normalizes them into a stable IR format.
+    """
+
+    objects: List[IRObject] = []
+    morphisms: List[IRMorphism] = []
+    plan_deltas: List[IRPlanDelta] = []
+
+    if lp:
+        objects.append(IRObject(type="LocalProblem", data=lp))
+
+    for s in (sv or []):
+        morphisms.append(IRMorphism(kind="SharedVariable", data=s))
+    for d in (dv or []):
+        morphisms.append(IRMorphism(kind="DualVariable", data=d))
+
+    for delta in (deltas or []):
+        plan_deltas.append(
+            IRPlanDelta(
+                delta_id=delta.get("delta_id"),
+                changes=delta.get("changes", {}),
+                timestamp=delta.get("timestamp"),
+                author=delta.get("author"),
+                contract_id=delta.get("contract_id"),
+                signature=delta.get("signature"),
+            )
+        )
+
+    return IR(version="0.1", objects=objects, morphisms=morphisms, plan_deltas=plan_deltas)
+
+
+__all__ = [
+    "IRObject",
+    "IRMorphism",
+    "IRPlanDelta",
+    "IR",
+    "to_ir",
+]