build(agent): molt-y#23e5c8 iteration

src/energiamesh/catopt_bridge.py

@@ -0,0 +1,74 @@
+"""Minimal CatOpt bridge for EnergiaMesh primitives.
+
+This module provides a lightweight, language-agnostic translation layer
+that maps EnergiaMesh core primitives (LocalProblem, SharedVariables,
+PlanDelta, DualVariables, AuditLog) into a canonical CatOpt-like
+representation. The goal is to bootstrap interoperability while keeping
+the implementation tiny and well-scoped for the MVP.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict
+
+from energiamesh.core import LocalProblem, SharedVariables, PlanDelta, DualVariables, AuditLog
+
+
+@dataclass
+class CatOptBridge:
+    """Tiny bridge translating EnergiaMesh primitives to CatOpt-like dicts."""
+
+    version: str = "0.1"
+
+    def to_catopt(self, obj: Any) -> Dict[str, Any]:
+        """Translate a supported EnergiaMesh object to a CatOpt-like dict.
+
+        This is intentionally minimal and focuses on the structural
+        information needed for cross-domain interoperability in the MVP.
+        """
+        if isinstance(obj, LocalProblem):
+            return {
+                "type": "LocalProblem",
+                "version": self.version,
+                "site_id": obj.site_id,
+                "objective": obj.objective,
+                "variables": obj.variables,
+                "constraints": obj.constraints,
+                "status": obj.status,
+            }
+        if isinstance(obj, SharedVariables):
+            return {
+                "type": "SharedVariables",
+                "version": self.version,
+                "signals": obj.signals,
+                "version_tag": obj.version,
+                "timestamp": obj.timestamp,
+            }
+        if isinstance(obj, PlanDelta):
+            return {
+                "type": "PlanDelta",
+                "version": self.version,
+                "delta_id": obj.delta_id,
+                "updates": obj.updates,
+                "metadata": obj.metadata,
+                "timestamp": obj.timestamp,
+            }
+        if isinstance(obj, DualVariables):
+            return {
+                "type": "DualVariables",
+                "version": self.version,
+                "multipliers": obj.multipliers,
+                "primal": obj.primal,
+                "timestamp": obj.timestamp,
+            }
+        if isinstance(obj, AuditLog):
+            return {
+                "type": "AuditLog",
+                "version": self.version,
+                "entries": obj.entries,
+            }
+        raise TypeError(f"Unsupported object type for CatOptBridge: {type(obj)!r}")
+
+
+__all__ = ["CatOptBridge"]

tests/test_catopt_bridge.py

@@ -0,0 +1,36 @@
+from energiamesh.catopt_bridge import CatOptBridge
+from energiamesh.core import LocalProblem, SharedVariables, PlanDelta, DualVariables, AuditLog
+
+
+def test_catopt_bridge_local_problem():
+    lp = LocalProblem(site_id="SiteA", objective="minimize_cost")
+    bridge = CatOptBridge()
+    catopt = bridge.to_catopt(lp)
+    assert catopt["type"] == "LocalProblem"
+    assert catopt["site_id"] == "SiteA"
+
+
+def test_catopt_bridge_shared_variables():
+    sv = SharedVariables()
+    sv.update("forecast", {"temp": 20})
+    bridge = CatOptBridge()
+    catopt = bridge.to_catopt(sv)
+    assert catopt["type"] == "SharedVariables"
+    assert "forecast" in catopt["signals"]
+
+
+def test_catopt_bridge_plan_delta_and_dual_variables():
+    pd = PlanDelta(delta_id="d1", updates={"x": 1})
+    dv = DualVariables(multipliers={"p": 0.5}, primal={"y": 2})
+    bridge = CatOptBridge()
+    assert bridge.to_catopt(pd)["type"] == "PlanDelta"
+    assert bridge.to_catopt(dv)["type"] == "DualVariables"
+
+
+def test_catopt_bridge_audit_log():
+    al = AuditLog()
+    al.add_entry({"event": "start"})
+    bridge = CatOptBridge()
+    catopt = bridge.to_catopt(al)
+    assert catopt["type"] == "AuditLog"
+    assert isinstance(catopt["entries"], list)