62 lines
2.6 KiB
Python
62 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Tuple
|
|
|
|
from .core_contracts import LocalProblem, SharedVariables, PlanDelta, ConstraintSet, DeviceInfo
|
|
|
|
|
|
class GraphContractRegistry:
|
|
"""Lightweight in-memory registry for contracts and adapters.
|
|
|
|
This is a minimal scaffold to support MVP development of a graph-contract
|
|
oriented interoperability bridge. It stores basic contract schemas and
|
|
adapter interface descriptors, plus a simple conformance test harness.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._contracts: Dict[str, Dict[str, Any]] = {}
|
|
self._adapters: Dict[str, Dict[str, Any]] = {}
|
|
|
|
# Contracts
|
|
def register_contract(self, name: str, version: str, schema: Dict[str, Any]) -> None:
|
|
self._contracts.setdefault(name, {})[version] = schema
|
|
|
|
def get_contract(self, name: str, version: str) -> Dict[str, Any]:
|
|
return self._contracts.get(name, {}).get(version, {})
|
|
|
|
# Adapters
|
|
def register_adapter(self, name: str, version: str, iface: Dict[str, Any]) -> None:
|
|
self._adapters.setdefault(name, {})[version] = iface
|
|
|
|
def get_adapter(self, name: str, version: str) -> Dict[str, Any]:
|
|
return self._adapters.get(name, {}).get(version, {})
|
|
|
|
# Conformance harness (stub)
|
|
def conformance_test(self, adapter_iface: Dict[str, Any], contract_schema: Dict[str, Any]) -> bool:
|
|
# Minimal check: ensure required keys exist in both sides
|
|
required_adapter_keys = {"name", "version", "interface"}
|
|
required_contract_keys = {"name", "version", "schema"}
|
|
has_adapter = required_adapter_keys.issubset(set(adapter_iface.keys()))
|
|
has_contract = required_contract_keys.issubset(set(contract_schema.keys()))
|
|
return bool(has_adapter and has_contract)
|
|
|
|
# --- Convenience helpers for introspection and tooling ---
|
|
def list_contracts(self) -> Dict[str, list]:
|
|
"""Return a map of contract names to available versions.
|
|
|
|
Example: { "LocalProblem": ["0.1", "0.2"], ... }
|
|
"""
|
|
return {name: list(versions.keys()) for name, versions in self._contracts.items()}
|
|
|
|
def list_adapters(self) -> Dict[str, list]:
|
|
"""Return a map of adapter names to available versions.
|
|
|
|
Example: { "DERController": ["0.1"], ... }
|
|
"""
|
|
return {name: list(versions.keys()) for name, versions in self._adapters.items()}
|
|
|
|
# Backwards-compatibility alias for older code/tests
|
|
# Some clients import ContractRegistry from gridverse.registry.
|
|
# Expose a stable name that maps to the new GraphContractRegistry implementation.
|
|
ContractRegistry = GraphContractRegistry
|