72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
from typing import Dict, Any
|
|
from datetime import datetime
|
|
from .primitives import LocalProblem, SharedVariables, PlanDelta, DualVariables
|
|
|
|
|
|
def _map_to_ir(lp: LocalProblem, sv: SharedVariables, pd: PlanDelta, dv: DualVariables) -> Dict[str, Any]:
|
|
"""Minimal canonical IR mapping (toy implementation).
|
|
This provides a structural bridge from ExoRoute primitives to a CatOpt-like IR.
|
|
Returns a dictionary with keys expected by tests: object, morphisms, plan_delta.
|
|
"""
|
|
ts = pd.timestamp
|
|
if isinstance(ts, (int, float)):
|
|
ts_iso = datetime.fromtimestamp(ts).isoformat()
|
|
else:
|
|
ts_iso = ts.isoformat()
|
|
# Payload for the LocalProblem
|
|
payload = {
|
|
"id": lp.id,
|
|
"domain": lp.domain,
|
|
"assets": lp.assets,
|
|
"objective": lp.objective,
|
|
"constraints": lp.constraints,
|
|
}
|
|
ir = {
|
|
"object": {"type": "LocalProblem", "payload": payload},
|
|
"morphisms": {
|
|
"shared_variables": _shared_to_dict(sv),
|
|
"dual_variables": {
|
|
"lambda_latency": dv.lambda_latency,
|
|
"lambda_fees": dv.lambda_fees,
|
|
},
|
|
},
|
|
"plan_delta": {
|
|
"delta": pd.delta,
|
|
"timestamp": ts_iso,
|
|
"author": pd.author,
|
|
"contract_id": pd.contract_id,
|
|
"signature": pd.signature,
|
|
},
|
|
}
|
|
return ir
|
|
|
|
|
|
def _shared_to_dict(sv: SharedVariables) -> Dict[str, Any]:
|
|
return {
|
|
"forecasts": sv.forecasts,
|
|
"priors": sv.priors,
|
|
"version": sv.version,
|
|
}
|
|
|
|
|
|
class EnergiBridge:
|
|
"""Tiny wrapper exposing a stable API for bridging primitives to IR."""
|
|
|
|
@staticmethod
|
|
def map(lp: LocalProblem, sv: SharedVariables, pd: PlanDelta, dv: DualVariables) -> Dict[str, Any]:
|
|
return map_to_ir(lp, sv, pd, dv)
|
|
|
|
# Backwards-compat alias expected by tests
|
|
@staticmethod
|
|
def map_to_ir(lp: LocalProblem, sv: SharedVariables, pd: PlanDelta, dv: DualVariables = None) -> Dict[str, Any]:
|
|
return map_to_ir(lp, sv, pd, dv)
|
|
|
|
|
|
def map_to_ir(lp: LocalProblem, sv: SharedVariables, pd: PlanDelta, dv: DualVariables = None) -> Dict[str, Any]:
|
|
if dv is None:
|
|
dv = DualVariables()
|
|
return _map_to_ir(lp, sv, pd, dv)
|
|
|
|
|
|
__all__ = ["map_to_ir", "EnergiBridge"]
|