diff --git a/README.md b/README.md index d0408c1..aa73878 100644 --- a/README.md +++ b/README.md @@ -1,19 +1,35 @@ -# GridGuard-Inspired EnergiBridge Skeleton (MVP) +# GridGuard-Inspired EnergiBridge Skeleton -This repository provides a production-ready skeleton for a canonical interop layer (EnergiBridge-like) to enable secure, verifiable telemetry and orchestration across multi-asset power facilities. +This repository implements a compact, verifiable telemetry orchestration layer for multi-asset power facilities. -- Phase 0 MVP: skeleton protocol plus two starter adapters (solar_meter and der_aggregator) over TLS (stubbed), a lightweight ADMM-lite local solver, deterministic delta-sync with replay capability, and a toy mesh-energy-balance objective. -- Phase 1–3: governance, identity, cross-domain pilots, SDKs, and hardware-in-the-loop validation. +## What it provides -What you’ll find: -- energi_bridge: Python package containing core schema definitions (LocalProblem, SharedVariables, PlanDelta, DualVariables, AuditLog, PrivacyBudget) -- energi_bridge/adapters: two toy adapters to simulate a solar-meter feed and a DER-aggregator -- tests/: pytest-based tests for schema validation and adapter interactions -- test.sh: script to run tests and build the package -- AGENTS.md: project-wide architectural and testing guidelines +- Canonical contract schemas for `LocalProblem`, `SharedVariables`, `PlanDelta`, `DualVariables`, `AuditLog`, `PrivacyBudget`, and `RegistryMetadata` +- A versioned `ContractRegistry` for Graph-of-Contracts style adapter/schema tracking +- A `VerifiableTelemOrchestrator` that collects adapter telemetry, signs plan deltas, and emits audit and registry metadata +- Two deterministic starter adapters: + - `SolarMeterAdapter` + - `DERAggregatorAdapter` +- Pytest coverage for schema validation, registry versioning, adapter replay behavior, and orchestration output -How to run -- Install dependencies and run tests: ./test.sh -- Build the package: python -m build +## Layout -This is a minimal, production-oriented scaffold intended to be extended in subsequent sprint tasks. +- `energi_bridge/`: core package +- `energi_bridge/adapters/`: starter adapters +- `tests/`: pytest suite +- `test.sh`: installs the package, runs tests, and builds the distribution + +## Run + +```bash +bash test.sh +``` + +```bash +python -m build +``` + +## Notes + +- The code is deterministic by default to support replayable tests and audit trails. +- The registry records contract version, adapter identity, schema name, and replay hints for each adapter contract. diff --git a/energi_bridge/__init__.py b/energi_bridge/__init__.py index 7027be9..bcedb2c 100644 --- a/energi_bridge/__init__.py +++ b/energi_bridge/__init__.py @@ -4,6 +4,7 @@ This package provides minimal schema definitions used for Phase 0 MVP and toy adapters that simulate telemetry and control planes. """ +from .contracts import ContractDefinition, ContractRegistry from .schemas import LocalProblem, SharedVariables, PlanDelta, DualVariables, AuditLog, PrivacyBudget, RegistryMetadata from .orchestrator import VerifiableTelemOrchestrator @@ -15,5 +16,7 @@ __all__ = [ "AuditLog", "PrivacyBudget", "RegistryMetadata", + "ContractDefinition", + "ContractRegistry", "VerifiableTelemOrchestrator", ] diff --git a/energi_bridge/adapters/der_aggregator.py b/energi_bridge/adapters/der_aggregator.py index 7498ad8..b1a7b68 100644 --- a/energi_bridge/adapters/der_aggregator.py +++ b/energi_bridge/adapters/der_aggregator.py @@ -1,18 +1,34 @@ -"""Toy DER Aggregator Adapter (Phase 0 MVP).""" -from dataclasses import dataclass -from typing import Dict, Any -import time +"""Toy DER Aggregator Adapter with deterministic replay support.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional @dataclass class DERAggregatorAdapter: aggregator_id: str + samples: Optional[List[Dict[str, Any]]] = None + _cursor: int = field(init=False, default=0) + + def __post_init__(self) -> None: + if self.samples is None: + self.samples = [] + def aggregate(self) -> Dict[str, Any]: - # Simple deterministic aggregation sample - ts = int(time.time()) + if self.samples: + sample_list = self.samples + if self._cursor >= len(sample_list): + raise IndexError("No more DER aggregation samples available") + reading = dict(sample_list[self._cursor]) + self._cursor += 1 + reading.setdefault("aggregator_id", self.aggregator_id) + reading.setdefault("timestamp", self._cursor) + reading.setdefault("status", "nominal") + return reading + return { "aggregator_id": self.aggregator_id, - "timestamp": ts, + "timestamp": self._cursor, "total_power_kw": 500.0, # fixed placeholder for deterministic behavior "status": "nominal", } diff --git a/energi_bridge/adapters/solar_meter_adapter.py b/energi_bridge/adapters/solar_meter_adapter.py index 94ba673..187c703 100644 --- a/energi_bridge/adapters/solar_meter_adapter.py +++ b/energi_bridge/adapters/solar_meter_adapter.py @@ -1,18 +1,34 @@ -"""Toy Solar Meter Adapter (Phase 0 MVP).""" -from dataclasses import dataclass -from typing import Dict, Any -import random -import time +"""Toy Solar Meter Adapter with deterministic replay support.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional @dataclass class SolarMeterAdapter: site_id: str + samples: Optional[List[Dict[str, Any]]] = None + _cursor: int = field(init=False, default=0) + + def __post_init__(self) -> None: + if self.samples is None: + self.samples = [] + def read_meter(self) -> Dict[str, Any]: - # Simulate solar meter reading + if self.samples: + sample_list = self.samples + if self._cursor >= len(sample_list): + raise IndexError("No more solar meter samples available") + reading = dict(sample_list[self._cursor]) + self._cursor += 1 + reading.setdefault("site_id", self.site_id) + reading.setdefault("timestamp", self._cursor) + reading.setdefault("status", "ok") + return reading + return { "site_id": self.site_id, - "timestamp": int(time.time()), - "solar_output_kw": round(random.uniform(0, 1000), 2), + "timestamp": self._cursor, + "solar_output_kw": 750.0, "status": "ok", } diff --git a/energi_bridge/contracts.py b/energi_bridge/contracts.py new file mode 100644 index 0000000..90eac19 --- /dev/null +++ b/energi_bridge/contracts.py @@ -0,0 +1,96 @@ +"""Graph-of-Contracts registry for versioned adapter/schema metadata.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime +from typing import Dict, Iterable, List, Optional, Tuple + +from energi_bridge.schemas import RegistryMetadata + + +def _parse_version(version: str) -> Tuple[int, int, int, str]: + core, _, suffix = version.partition("-") + parts = core.split(".") + while len(parts) < 3: + parts.append("0") + major, minor, patch = (int(parts[0]), int(parts[1]), int(parts[2])) + return major, minor, patch, suffix + + +@dataclass(frozen=True) +class ContractDefinition: + contract_id: str + version: str + adapter_id: str + schema_name: str + created_at: datetime = field(default_factory=datetime.utcnow) + replay_hint: Optional[str] = None + previous_version: Optional[str] = None + + def to_metadata(self) -> RegistryMetadata: + return RegistryMetadata( + contract_version=self.version, + adapter_id=self.adapter_id, + contract_id=self.contract_id, + schema_name=self.schema_name, + created_at=self.created_at, + replay_hint=self.replay_hint, + ) + + +class ContractRegistry: + """Maintain contract versions for schemas and adapters.""" + + def __init__(self) -> None: + self._contracts: Dict[str, Dict[str, ContractDefinition]] = {} + + def register( + self, + contract_id: str, + version: str, + adapter_id: str, + schema_name: str, + replay_hint: Optional[str] = None, + ) -> ContractDefinition: + versions = self._contracts.setdefault(contract_id, {}) + if version in versions: + raise ValueError(f"Contract {contract_id}@{version} already exists") + + previous_version = self.latest_version(contract_id) + definition = ContractDefinition( + contract_id=contract_id, + version=version, + adapter_id=adapter_id, + schema_name=schema_name, + replay_hint=replay_hint, + previous_version=previous_version, + ) + versions[version] = definition + return definition + + def latest_version(self, contract_id: str) -> Optional[str]: + versions = self._contracts.get(contract_id, {}) + if not versions: + return None + return max(versions, key=_parse_version) + + def resolve(self, contract_id: str, version: Optional[str] = None) -> ContractDefinition: + versions = self._contracts.get(contract_id, {}) + if not versions: + raise KeyError(contract_id) + selected = version or self.latest_version(contract_id) + if selected is None or selected not in versions: + raise KeyError(f"Unknown contract version: {contract_id}@{version}") + return versions[selected] + + def manifest(self) -> List[RegistryMetadata]: + metadata: List[RegistryMetadata] = [] + for contract_versions in self._contracts.values(): + for definition in contract_versions.values(): + metadata.append(definition.to_metadata()) + return sorted(metadata, key=lambda item: (item.contract_id or "", _parse_version(item.contract_version))) + + def items(self) -> Iterable[ContractDefinition]: + for contract_versions in self._contracts.values(): + yield from contract_versions.values() diff --git a/energi_bridge/orchestrator.py b/energi_bridge/orchestrator.py index c047893..8fdcc32 100644 --- a/energi_bridge/orchestrator.py +++ b/energi_bridge/orchestrator.py @@ -11,11 +11,12 @@ from __future__ import annotations from dataclasses import asdict from datetime import datetime -from typing import Any, Dict, List +from typing import Any, Callable, Dict, List, Optional import json import hmac import hashlib +from energi_bridge.contracts import ContractRegistry from energi_bridge.schemas import ( LocalProblem, SharedVariables, @@ -31,35 +32,72 @@ def _sign_payload(payload: str, key: str) -> str: return hmac.new(key.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest() +def _adapter_contract_id(adapter: Any) -> str: + if hasattr(adapter, "site_id"): + return f"solar_meter.{adapter.site_id}" + if hasattr(adapter, "aggregator_id"): + return f"der_aggregator.{adapter.aggregator_id}" + return adapter.__class__.__name__.lower() + + +def _adapter_schema_name(adapter: Any) -> str: + if hasattr(adapter, "read_meter"): + return "SolarMeterReading" + if hasattr(adapter, "aggregate"): + return "DERAggregationSnapshot" + return "TelemetryEnvelope" + + class VerifiableTelemOrchestrator: - def __init__(self, adapters: List[Any], signing_key: str = "secret") -> None: + def __init__( + self, + adapters: List[Any], + signing_key: str = "secret", + registry: Optional[ContractRegistry] = None, + clock: Optional[Callable[[], datetime]] = None, + ) -> None: self.adapters = adapters self.signing_key = signing_key + self.registry = registry or ContractRegistry() + self.clock = clock or datetime.utcnow def _collect_signals(self) -> Dict[str, Any]: signals: Dict[str, Any] = {} for adapter in self.adapters: if hasattr(adapter, "read_meter"): try: - signals["solar_meter"] = adapter.read_meter() + signals[_adapter_contract_id(adapter)] = adapter.read_meter() except Exception: - signals["solar_meter"] = {"error": "read_failed"} + signals[_adapter_contract_id(adapter)] = {"error": "read_failed"} if hasattr(adapter, "aggregate"): try: - signals["der_aggregator"] = adapter.aggregate() + signals[_adapter_contract_id(adapter)] = adapter.aggregate() except Exception: - signals["der_aggregator"] = {"error": "aggregate_failed"} + signals[_adapter_contract_id(adapter)] = {"error": "aggregate_failed"} return signals def run_once(self) -> Dict[str, Any]: - # Gather telemetry from adapters + now = self.clock() + + # Gather telemetry from adapters and register the contract graph. signals = self._collect_signals() + for adapter in self.adapters: + contract_id = _adapter_contract_id(adapter) + schema_name = _adapter_schema_name(adapter) + if self.registry.latest_version(contract_id) is None: + self.registry.register( + contract_id=contract_id, + version="0.1.0", + adapter_id=adapter.__class__.__name__, + schema_name=schema_name, + replay_hint="deterministic", + ) # Phase-0 MVP Hard-coded skeletons to demonstrate flow lp = LocalProblem( id="lp_mvp", domain="GridGuard", - assets=[getattr(a, "__class__", object).__name__ for a in self.adapters], + assets=[_adapter_contract_id(a) for a in self.adapters], constraints={"max_dispatch_kw": 5000, "min_dispatch_kw": 0}, objective={"type": "mesh_energy_balance"}, solver_hint=None, @@ -73,13 +111,13 @@ class VerifiableTelemOrchestrator: ) # Simple delta reflecting new signals; sign for verifiability - delta_payload = {"signals": signals, "ts": datetime.utcnow().isoformat()} + delta_payload = {"signals": signals, "ts": now.isoformat()} delta_json = json.dumps(delta_payload, sort_keys=True) signature = _sign_payload(delta_json, self.signing_key) pd = PlanDelta( delta=delta_payload, - timestamp=datetime.utcnow(), + timestamp=now, author="VerifiableTelemOrchestrator", contract_id="c0", signature=signature, @@ -87,9 +125,17 @@ class VerifiableTelemOrchestrator: ) dv = DualVariables() - al = AuditLog(entries=[{"event": "plan_delta_created", "ts": datetime.utcnow().isoformat()}], last_updated=datetime.utcnow()) + al = AuditLog() + al.record("plan_delta_created", ts=now.isoformat(), contract_id=pd.contract_id) pb = PrivacyBudget(budget=1.0, spent=0.0, policy={"phase": 0}) - rm = RegistryMetadata(contract_version="0.1.0", adapter_id="verifiable-telemetry-orchestrator", replay_hint=None) + rm = RegistryMetadata( + contract_version="0.1.0", + adapter_id="verifiable-telemetry-orchestrator", + contract_id="gridguard.orchestrator", + schema_name="VerifiableTelemetryCycle", + created_at=now, + replay_hint="deterministic", + ) # Return a dictionary with serializable payloads return { @@ -100,4 +146,5 @@ class VerifiableTelemOrchestrator: "audit_log": asdict(al), "privacy_budget": asdict(pb), "registry_metadata": asdict(rm), + "contract_registry": [asdict(item) for item in self.registry.manifest()], } diff --git a/energi_bridge/schemas.py b/energi_bridge/schemas.py index 23aa140..2d6de71 100644 --- a/energi_bridge/schemas.py +++ b/energi_bridge/schemas.py @@ -1,6 +1,6 @@ from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional from datetime import datetime +from typing import Any, Dict, List, Optional @dataclass @@ -12,6 +12,14 @@ class LocalProblem: objective: Dict[str, Any] = field(default_factory=dict) solver_hint: Optional[str] = None + def __post_init__(self) -> None: + if not self.id: + raise ValueError("LocalProblem.id is required") + if not self.domain: + raise ValueError("LocalProblem.domain is required") + if not self.assets: + raise ValueError("LocalProblem.assets must not be empty") + @dataclass class SharedVariables: @@ -20,6 +28,10 @@ class SharedVariables: privacy_tag: Optional[str] = None aggregation_method: str = "mean" + def __post_init__(self) -> None: + if self.version < 1: + raise ValueError("SharedVariables.version must be >= 1") + @dataclass class PlanDelta: @@ -30,6 +42,12 @@ class PlanDelta: signature: Optional[str] = None safety_tags: Optional[List[str]] = None + def __post_init__(self) -> None: + if not self.author: + raise ValueError("PlanDelta.author is required") + if not self.contract_id: + raise ValueError("PlanDelta.contract_id is required") + @dataclass class DualVariables: @@ -42,6 +60,10 @@ class AuditLog: entries: List[Dict[str, Any]] = field(default_factory=list) last_updated: Optional[datetime] = None + def record(self, event: str, **details: Any) -> None: + self.entries.append({"event": event, **details}) + self.last_updated = datetime.utcnow() + @dataclass class PrivacyBudget: @@ -49,9 +71,19 @@ class PrivacyBudget: spent: float = 0.0 policy: Dict[str, Any] = field(default_factory=dict) + def consume(self, amount: float) -> None: + if amount < 0: + raise ValueError("Privacy budget consumption must be positive") + if self.spent + amount > self.budget: + raise ValueError("Privacy budget exhausted") + self.spent += amount + @dataclass class RegistryMetadata: contract_version: str adapter_id: str + contract_id: Optional[str] = None + schema_name: Optional[str] = None + created_at: Optional[datetime] = None replay_hint: Optional[str] = None diff --git a/pyproject.toml b/pyproject.toml index 2641293..ebe5e2f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,6 @@ version = "0.1.0" description = "EnergiBridge-inspired canonical interop skeleton for GridGuard MVP" readme = "README.md" requires-python = ">=3.8" -license = { text = "MIT" } +license = "MIT" # Packaging configuration for setuptools can be discovered by the build tool. diff --git a/test.sh b/test.sh old mode 100644 new mode 100755 diff --git a/tests/test_contracts.py b/tests/test_contracts.py new file mode 100644 index 0000000..ff2d844 --- /dev/null +++ b/tests/test_contracts.py @@ -0,0 +1,41 @@ +from datetime import datetime + +import pytest + +from energi_bridge import AuditLog, ContractRegistry, LocalProblem, PrivacyBudget + + +def test_contract_registry_tracks_versions() -> None: + registry = ContractRegistry() + + first = registry.register( + contract_id="solar_meter.site-1", + version="0.1.0", + adapter_id="SolarMeterAdapter", + schema_name="SolarMeterReading", + ) + second = registry.register( + contract_id="solar_meter.site-1", + version="0.2.0", + adapter_id="SolarMeterAdapter", + schema_name="SolarMeterReading", + ) + + assert registry.latest_version("solar_meter.site-1") == "0.2.0" + assert registry.resolve("solar_meter.site-1").version == "0.2.0" + assert first.previous_version is None + assert second.previous_version == "0.1.0" + assert [item.contract_version for item in registry.manifest()] == ["0.1.0", "0.2.0"] + + +def test_schema_validations_and_governance_helpers() -> None: + with pytest.raises(ValueError): + LocalProblem(id="", domain="solar", assets=["panel-1"]) + + budget = PrivacyBudget(budget=1.0) + budget.consume(0.25) + assert budget.spent == 0.25 + + log = AuditLog() + log.record("cycle_completed", ts=datetime.utcnow().isoformat()) + assert log.entries[0]["event"] == "cycle_completed" diff --git a/tests/test_orchestrator.py b/tests/test_orchestrator.py new file mode 100644 index 0000000..71aa3c7 --- /dev/null +++ b/tests/test_orchestrator.py @@ -0,0 +1,39 @@ +import hashlib +import hmac +import json +from datetime import datetime + +from energi_bridge.adapters import DERAggregatorAdapter, SolarMeterAdapter +from energi_bridge.orchestrator import VerifiableTelemOrchestrator + + +def test_orchestrator_emits_registry_and_signed_plan() -> None: + adapters = [ + SolarMeterAdapter(site_id="site-a"), + DERAggregatorAdapter(aggregator_id="agg-a"), + ] + orchestrator = VerifiableTelemOrchestrator( + adapters=adapters, + signing_key="unit-test", + clock=lambda: datetime(2026, 4, 25, 12, 0, 0), + ) + + result = orchestrator.run_once() + + signals = result["shared_variables"]["signals"] + assert signals["solar_meter.site-a"]["solar_output_kw"] == 750.0 + assert signals["der_aggregator.agg-a"]["total_power_kw"] == 500.0 + + payload = { + "signals": signals, + "ts": "2026-04-25T12:00:00", + } + expected_signature = hmac.new( + b"unit-test", + json.dumps(payload, sort_keys=True).encode("utf-8"), + hashlib.sha256, + ).hexdigest() + assert result["plan_delta"]["signature"] == expected_signature + + contract_ids = [item["contract_id"] for item in result["contract_registry"]] + assert contract_ids == ["der_aggregator.agg-a", "solar_meter.site-a"]