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