gridverse-open-low-code-pla.../gridverse/registry.py

47 lines
2.0 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)
# 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