openenergysphere-federated-.../src/energysphere/energi_bridge.py

56 lines
2.0 KiB
Python

"""EnergiBridge: minimal canonical mapping for cross-domain interoperability.
This is a small MVP shim that translates Energysphere primitives into a
vendor-agnostic IR representation (very light-weight). The goal is to provide a
stable, testable layer that can be expanded to full CatOpt-like mappings while
preserving offline-first semantics and governance hooks in later iterations.
"""
from __future__ import annotations
from typing import Any, Dict
from .models import LocalProblem
class EnergiBridgeMapper:
"""Lightweight translator between Energysphere primitives and a canonical IR.
Only a tiny subset is implemented to keep MVP surface small:
- LocalProblem becomes a canonical Object with a few core fields.
- SharedVariables/PlanDelta could be extended similarly in future iterations.
"""
def __init__(self, namespace: str = "default") -> None:
self.namespace = namespace
def map_local_problem(self, lp: LocalProblem) -> Dict[str, Any]:
"""Map a LocalProblem to a canonical Object representation.
The structure is intentionally simple but stable for downstream adapters
to consume. It can be extended with versioning, provenance, and crypto
tags in future iterations.
"""
return {
"type": "Object",
"namespace": self.namespace,
"site_id": lp.site_id,
"objective": lp.objective,
"parameters": lp.parameters,
"version": 1,
}
def map_object_to_local_problem(self, obj: Dict[str, Any]) -> LocalProblem:
"""Inverse mapping from canonical Object to LocalProblem when possible."""
if obj.get("type") != "Object":
raise ValueError("Unsupported canonical type: expected Object")
# Basic reconstruction; in MVP we ignore extra fields safely.
return LocalProblem(
site_id=obj.get("site_id", ""),
objective=obj.get("objective", ""),
parameters=obj.get("parameters", {}),
)
__all__ = ["EnergiBridgeMapper"]