83 lines
3.4 KiB
Python
83 lines
3.4 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", {}),
|
|
)
|
|
|
|
# --- Additional helpers for richer IR coverage ---
|
|
def map_shared_variable(self, sv: Any) -> Dict[str, Any]: # be resilient to different shapes
|
|
"""Map a SharedVariable (or similar) to a canonical IR representation."""
|
|
return {
|
|
"type": "SharedVariable",
|
|
"name": getattr(sv, "name", ""),
|
|
"value": getattr(sv, "value", None),
|
|
"version": getattr(sv, "version", 1),
|
|
"privacy_bound": getattr(sv, "privacy_bound", None),
|
|
"source_contract_id": getattr(sv, "source_contract_id", None),
|
|
}
|
|
|
|
def map_plan_delta(self, delta: Any) -> Dict[str, Any]: # be resilient to different shapes
|
|
"""Map a PlanDelta to a canonical IR representation."""
|
|
return {
|
|
"type": "PlanDelta",
|
|
"delta_id": getattr(delta, "delta_id", ""),
|
|
"contract_id": getattr(delta, "contract_id", None),
|
|
"changes": getattr(delta, "changes", {}),
|
|
"timestamp": getattr(delta, "timestamp", 0.0),
|
|
# Prefer an explicit author field if present; otherwise fall back to delta_id
|
|
"author": getattr(delta, "author", getattr(delta, "delta_id", "")),
|
|
"provenance": getattr(delta, "provenance", {}),
|
|
"privacy_budget": getattr(delta, "privacy_budget", None),
|
|
"signature": getattr(delta, "signature", None),
|
|
}
|
|
|
|
|
|
__all__ = ["EnergiBridgeMapper"]
|