catopt-graph-graph-calculus.../core/energi_bridge.py

67 lines
2.9 KiB
Python

from __future__ import annotations
from typing import Any, Dict, Optional
# Minimal EnergiBridge: bridge between domain LocalProblem representations and
# a canonical CatOpt-Graph IR. This is intentionally small and MVP-friendly to
# accelerate adapters onboarding and interoperability testing.
class EnergiBridge:
"""Static helpers to map between LocalProblem domain objects and canonical IR."""
@staticmethod
def to_canonical(local_problem: Any) -> Dict[str, Any]:
"""Convert a domain LocalProblem object into a canonical LocalProblem IR.
The function is defensive: it accepts either a dict-like object or a
simple object with attributes. It returns a dict compatible with the
tests in this repository and the existing bridge layer.
"""
# If it's already a dict-like object
if isinstance(local_problem, dict):
lp_id = local_problem.get("id") or local_problem.get("asset_id")
domain = local_problem.get("domain", "unknown")
objective = local_problem.get("objective")
variables = local_problem.get("variables", {})
payload = local_problem.get("payload", {})
else:
lp_id = getattr(local_problem, "asset_id", None) or getattr(local_problem, "id", None)
domain = getattr(local_problem, "domain", "unknown")
objective = getattr(local_problem, "objective", None)
# Prefer explicit variables attribute if present, otherwise payload
variables = getattr(local_problem, "variables", None)
if variables is None:
variables = getattr(local_problem, "payload", {})
payload = getattr(local_problem, "payload", {})
return {
"LocalProblem": {
"id": lp_id,
"domain": domain,
"objective": objective,
"variables": variables,
}
}
@staticmethod
def from_canonical(canonical: Dict[str, Any]) -> Any:
"""Convert a canonical IR LocalProblem back into a domain LocalProblem object.
If the canonical representation is not compatible, returns a simple
dictionary fallback.
"""
lp = canonical.get("LocalProblem") if canonical else None
if isinstance(lp, dict):
asset_id = lp.get("id") or lp.get("object_id") or lp.get("asset_id")
payload = lp.get("variables") if "variables" in lp else lp.get("payload", {})
# Import LocalProblem from core.contracts lazily to avoid circular imports
try:
from .contracts import LocalProblem as DomainLocalProblem # type: ignore
return DomainLocalProblem(asset_id=asset_id, payload=payload or {})
except Exception:
return {"asset_id": asset_id, "payload": payload or {}}
# Fallback: return the input as-is
return canonical