exoroute-cross-venue-order-.../exoroute/energi_bridge.py

44 lines
1.9 KiB
Python

from __future__ import annotations
from typing import Dict, Any, Optional
from .core import LocalProblem, SharedVariables, PlanDelta, GraphOfContractsRegistry
class EnergiBridge:
"""Minimal bridge translating ExoRoute primitives to a canonical IR (EnergiBridge-style).
This is a lightweight, extensible mapping scaffold for interoperability with CatOpt-like ecosystems.
"""
@staticmethod
def to_canonical(local: LocalProblem, shared: SharedVariables, delta: Optional[PlanDelta] = None, registry: Optional[GraphOfContractsRegistry] = None) -> Dict[str, Any]:
can = {
"Objects": {
"LocalProblem": {
"id": local.id,
"domain": local.domain,
"assets": local.assets,
"objective": local.objective,
"constraints": local.constraints,
"solver_hint": local.solver_hint,
}
},
"Morphisms": {
"SharedVariables": {
"forecasts": getattr(shared, "forecasts", {}),
"priors": getattr(shared, "priors", {}),
"version": getattr(shared, "version", 0),
"timestamp": getattr(shared, "timestamp", 0.0),
},
"DualVariables": {},
},
"PlanDelta": delta.delta if delta else {},
}
# Optional registry embedding for interoperability and provenance
if registry is not None:
can["GraphOfContractsRegistry"] = {
entry_id: {
"adapter_id": entry.adapter_id,
"supported_domains": entry.supported_domains,
"contract_version": entry.contract_version,
}
for entry_id, entry in registry.entries.items()
}
return can