133 lines
5.3 KiB
Python
133 lines
5.3 KiB
Python
"""
|
|
EnergiBridge Skeleton: canonical bridge between GridVerse primitives and a
|
|
vendor-agnostic intermediate representation (IR) inspired by CatOpt-like
|
|
intermediates.
|
|
|
|
This module provides a small, production-friendly helper set that a future
|
|
implementation can plug into the GraphContractRegistry and Adapter Marketplace
|
|
to enable cross-domain interoperability without rewriting solvers.
|
|
"""
|
|
|
|
from typing import Any, Dict, Tuple, Optional
|
|
|
|
|
|
class EnergiBridge:
|
|
"""EnergiBridge: bridge between GridVerse primitives and a canonical IR.
|
|
|
|
This is a pragmatic, production-friendly surface that downstream components
|
|
can rely on when wiring adapters. It focuses on stability and small surface
|
|
surface area, enabling predictable interoperability without pulling in a
|
|
large external IR library.
|
|
"""
|
|
|
|
@staticmethod
|
|
def to_ir(contract_type: str, payload: Dict[str, Any], metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Convert a GridVerse contract into a canonical IR payload.
|
|
|
|
The IR is deliberately lightweight but includes a stable schema
|
|
(ir_type, ir_payload) plus optional metadata. If no metadata is
|
|
supplied, a sensible default is applied.
|
|
|
|
Args:
|
|
contract_type: The name of the contract (e.g., "LocalProblem").
|
|
payload: The contract payload data.
|
|
metadata: Optional metadata dictionary (e.g., source, version).
|
|
|
|
Returns:
|
|
A dict representing the IR contract, suitable for transport or
|
|
storage in the registry.
|
|
"""
|
|
if metadata is None:
|
|
metadata = {
|
|
"source": "gridverse",
|
|
"version": "0.2",
|
|
"schema": "EnergiBridge.v0.2",
|
|
}
|
|
# Attach a small, deterministic, environment-agnostic header to aid auditing
|
|
if "timestamp" not in metadata:
|
|
try:
|
|
import time
|
|
metadata["timestamp"] = int(time.time())
|
|
except Exception:
|
|
metadata["timestamp"] = 0
|
|
return {
|
|
"ir_type": contract_type,
|
|
"ir_payload": payload,
|
|
"metadata": metadata,
|
|
}
|
|
|
|
@staticmethod
|
|
def from_ir(ir: Dict[str, Any]) -> Tuple[str, Dict[str, Any]]:
|
|
"""Reverse translation from IR to a GridVerse contract tuple."""
|
|
contract_type = ir.get("ir_type")
|
|
if contract_type is None:
|
|
contract_type = "UnknownContract"
|
|
payload = ir.get("ir_payload", {})
|
|
return contract_type, payload
|
|
|
|
@staticmethod
|
|
def map_gridverse_to_catopt(gridverse_contract: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Tiny mapper from GridVerse contract to a canonical IR representation.
|
|
|
|
This demonstrates how a contract could be re-wired into an IR-compatible
|
|
payload that adapters can consume. It keeps a thin, stable surface and
|
|
defers heavy lifting to downstream components.
|
|
"""
|
|
ct = gridverse_contract.get("type") or gridverse_contract.get("contract_type")
|
|
if ct is None:
|
|
ct = "UnknownContract"
|
|
else:
|
|
ct = str(ct)
|
|
payload = gridverse_contract.get("payload") or gridverse_contract
|
|
return EnergiBridge.to_ir(ct or "UnknownContract", payload)
|
|
|
|
@staticmethod
|
|
def create_metadata_for_audit(contract_type: str) -> Dict[str, Any]:
|
|
"""Helper to produce a minimal audit metadata block for a contract."""
|
|
import time
|
|
return {
|
|
"source": "gridverse",
|
|
"version": "0.2",
|
|
"contract_type": contract_type,
|
|
"timestamp": int(time.time()),
|
|
}
|
|
|
|
|
|
def bootstrap_contracts(registry) -> None:
|
|
"""Lightweight bootstrap helper: seed a registry with canonical contracts.
|
|
|
|
This is intended for Phase 0 MVP wiring. The function is tolerant of environments
|
|
where the registry may not yet implement the full interface; it will silently
|
|
no-op in that case to avoid breaking tests or runtime in minimal setups.
|
|
"""
|
|
try:
|
|
# Prefer meta-aware registration if available
|
|
if hasattr(registry, "register_contract_with_meta"):
|
|
registry.register_contract_with_meta(
|
|
"LocalProblem",
|
|
"0.2",
|
|
{"id": "lp-bootstrap", "name": "Bootstrap LocalProblem"},
|
|
{"source": "EnergiBridge bootstrap"},
|
|
)
|
|
registry.register_contract_with_meta(
|
|
"SharedVariables",
|
|
"0.2",
|
|
{"signals": {}},
|
|
{"source": "EnergiBridge bootstrap"},
|
|
)
|
|
registry.register_contract_with_meta(
|
|
"PlanDelta",
|
|
"0.2",
|
|
{"delta_id": "pd-bootstrap", "changes": {}},
|
|
{"source": "EnergiBridge bootstrap"},
|
|
)
|
|
else:
|
|
# Fallback for registries without meta support
|
|
if hasattr(registry, "register_contract"):
|
|
registry.register_contract("LocalProblem", "0.2", {"id": "lp-bootstrap", "name": "Bootstrap LocalProblem"})
|
|
registry.register_contract("SharedVariables", "0.2", {"signals": {}})
|
|
registry.register_contract("PlanDelta", "0.2", {"delta_id": "pd-bootstrap", "changes": {}})
|
|
except Exception:
|
|
# Never crash MVP startup on bootstrap issues
|
|
return
|