105 lines
2.8 KiB
Python
105 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Any, Optional, List
|
|
|
|
# Reuse core CatOpt primitives for minimal compatibility in MVP
|
|
from .catopt_bridge import LocalProblem, SharedVariable, DualVariable, PlanDelta
|
|
|
|
|
|
@dataclass
|
|
class LocalProblemEP:
|
|
problem_id: str
|
|
assets: List[str]
|
|
objective: str
|
|
constraints: List[str]
|
|
data_contracts: Dict[str, Any]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"problem_id": self.problem_id,
|
|
"assets": self.assets,
|
|
"objective": self.objective,
|
|
"constraints": self.constraints,
|
|
"data_contracts": self.data_contracts,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class EnergiSignal:
|
|
channel: str
|
|
value: Any
|
|
version: int = 0
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"channel": self.channel, "value": self.value, "version": self.version}
|
|
|
|
|
|
@dataclass
|
|
class SharedVariableEP:
|
|
channel: str
|
|
version: int
|
|
payload: Dict[str, Any]
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"channel": self.channel, "version": self.version, "payload": self.payload}
|
|
|
|
@dataclass
|
|
class DualVariableEP:
|
|
channel: str
|
|
version: int
|
|
payload: Dict[str, Any]
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"channel": self.channel, "version": self.version, "payload": self.payload}
|
|
|
|
@dataclass
|
|
class PlanDeltaEP:
|
|
delta_id: str
|
|
changes: Dict[str, Any]
|
|
timestamp: float
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {"delta_id": self.delta_id, "changes": self.changes, "timestamp": self.timestamp}
|
|
|
|
|
|
class EnergiBridge:
|
|
"""Minimal EnergiBridge canonical bridge implementation.
|
|
|
|
Maps CosmosMesh primitives into a vendor-agnostic Intermediary Representation
|
|
suitable for cross-domain adapters.
|
|
"""
|
|
|
|
def to_energi(
|
|
self,
|
|
lp: LocalProblemEP,
|
|
shared: List[SharedVariableEP],
|
|
duals: List[DualVariableEP],
|
|
deltas: Optional[List[PlanDeltaEP]] = None,
|
|
) -> Dict[str, Any]:
|
|
data = {
|
|
"Version": "0.1",
|
|
"Objects": {
|
|
"LocalProblem": lp.to_dict(),
|
|
},
|
|
"Morphisms": [
|
|
{"Morphisms": {"SharedVariable": sv.to_dict()}} for sv in shared
|
|
] + [
|
|
{"Morphisms": {"DualVariable": dv.to_dict()}} for dv in duals
|
|
],
|
|
}
|
|
if deltas:
|
|
data["Objects"]["PlanDeltas"] = [p.to_dict() for p in deltas]
|
|
return data
|
|
|
|
# Backwards-compatible API expected by existing tests
|
|
def to_ir(
|
|
self,
|
|
lp: LocalProblemEP,
|
|
shared: List[SharedVariableEP],
|
|
duals: List[DualVariableEP],
|
|
deltas: Optional[List[PlanDeltaEP]] = None,
|
|
) -> Dict[str, Any]:
|
|
return self.to_energi(lp, shared, duals, deltas)
|
|
|
|
|
|
|
|
__all__ = ["EnergiBridge", "EnergiLocalProblem", "EnergiSignal"]
|