76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
"""EnergiBridge: Canonical interoperability bridge for GridResilience primitives.
|
|
|
|
This module provides a minimal, production-friendly starting point for
|
|
interconnecting GridResilience primitives with a vendor-agnostic intermediate
|
|
representation (IR). It is intentionally lightweight and focused on stability so
|
|
other components can build on top of it without pulling heavy dependencies.
|
|
|
|
- Objects -> LocalProblems
|
|
- Morphisms -> SharedSignals
|
|
- PlanDelta -> Incremental actions
|
|
- DualVariables, AuditLog, PrivacyBudget, RegistryEntry
|
|
|
|
The bridge stores a lightweight registry of adapters with per-message metadata
|
|
to support replay protection and auditing in a real deployment.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict
|
|
|
|
|
|
class EnergiBridge:
|
|
"""Canonical bridge for GridResilience primitives.
|
|
|
|
This is a minimal skeleton designed for extension. It maps high-level inputs
|
|
to a simple, serializable IR and maintains a per-adapter registry.
|
|
"""
|
|
|
|
def __init__(self, registry: Dict[str, Any] | None = None) -> None:
|
|
# Registry maps adapter_id -> RegistryEntry-like dict
|
|
self._registry: Dict[str, Dict[str, Any]] = registry or {}
|
|
|
|
# --- Local problem mapping (Objects) ---
|
|
def map_local_problem(self, local_problem: Any) -> Dict[str, Any]:
|
|
"""Direct mapping (Objects) for compatibility."""
|
|
return {"type": "Objects", "local_problem": local_problem}
|
|
|
|
def _map_local_problem(self, local_problem: Any) -> Dict[str, Any]:
|
|
return {"type": "Objects", "local_problem": local_problem}
|
|
|
|
# --- Shared signals mapping (Morphisms) ---
|
|
def map_shared_signals(self, shared_signals: Any) -> Dict[str, Any]:
|
|
"""Map SharedSignals to the IR morphism form."""
|
|
return self._map_shared_signals(shared_signals)
|
|
|
|
def _map_shared_signals(self, shared_signals: Any) -> Dict[str, Any]:
|
|
return {"type": "Morphisms", "shared_signals": shared_signals}
|
|
|
|
# --- Plan delta mapping ---
|
|
def map_plan_delta(self, plan_delta: Any) -> Dict[str, Any]:
|
|
"""Map a PlanDelta to the incremental action form."""
|
|
return self._map_plan_delta(plan_delta)
|
|
|
|
def _map_plan_delta(self, plan_delta: Any) -> Dict[str, Any]:
|
|
return {"type": "PlanDelta", "delta": plan_delta}
|
|
|
|
# --- Adapter registry management (GoC-like) ---
|
|
def register_adapter(self, adapter_id: str, contract_version: str, data_contract: Any) -> Dict[str, Any]:
|
|
"""Register an adapter with its contract metadata.
|
|
Returns the stored registry entry for confirmation.
|
|
"""
|
|
entry = {
|
|
"adapter_id": adapter_id,
|
|
"contract_version": contract_version,
|
|
"data_contract": data_contract,
|
|
}
|
|
self._registry[adapter_id] = entry
|
|
return entry
|
|
|
|
def get_registry(self) -> Dict[str, Dict[str, Any]]:
|
|
"""Return the current adapter registry."""
|
|
return self._registry
|
|
|
|
|
|
__all__ = ["EnergiBridge"]
|