diff --git a/src/energiamesh/catopt_bridge.py b/src/energiamesh/catopt_bridge.py new file mode 100644 index 0000000..3f4f8a3 --- /dev/null +++ b/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"] diff --git a/test.sh b/test.sh old mode 100644 new mode 100755 diff --git a/tests/test_catopt_bridge.py b/tests/test_catopt_bridge.py new file mode 100644 index 0000000..925768d --- /dev/null +++ b/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)