151 lines
5.3 KiB
Python
151 lines
5.3 KiB
Python
"""Verifiable Telemetry Orchestrator (Phase 0 MVP).
|
|
|
|
This module provides a minimal, production-oriented skeleton that:
|
|
- Collects data from toy adapters (solar meter and DER aggregator)
|
|
- Builds LocalProblem/SharedVariables/PlanDelta structures from the data
|
|
- Signs the PlanDelta in a deterministic, test-friendly way
|
|
- Produces an AuditLog and PrivacyBudget scaffold for governance
|
|
- Exposes a simple run_once() to simulate a reconciliation cycle
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from datetime import datetime
|
|
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,
|
|
PlanDelta,
|
|
DualVariables,
|
|
AuditLog,
|
|
PrivacyBudget,
|
|
RegistryMetadata,
|
|
)
|
|
|
|
|
|
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",
|
|
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[_adapter_contract_id(adapter)] = adapter.read_meter()
|
|
except Exception:
|
|
signals[_adapter_contract_id(adapter)] = {"error": "read_failed"}
|
|
if hasattr(adapter, "aggregate"):
|
|
try:
|
|
signals[_adapter_contract_id(adapter)] = adapter.aggregate()
|
|
except Exception:
|
|
signals[_adapter_contract_id(adapter)] = {"error": "aggregate_failed"}
|
|
return signals
|
|
|
|
def run_once(self) -> Dict[str, Any]:
|
|
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=[_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,
|
|
)
|
|
|
|
sv = SharedVariables(
|
|
version=1,
|
|
signals=signals,
|
|
privacy_tag="phase0",
|
|
aggregation_method="mean",
|
|
)
|
|
|
|
# Simple delta reflecting new signals; sign for verifiability
|
|
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=now,
|
|
author="VerifiableTelemOrchestrator",
|
|
contract_id="c0",
|
|
signature=signature,
|
|
safety_tags=["mesh_balancing"],
|
|
)
|
|
|
|
dv = DualVariables()
|
|
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",
|
|
contract_id="gridguard.orchestrator",
|
|
schema_name="VerifiableTelemetryCycle",
|
|
created_at=now,
|
|
replay_hint="deterministic",
|
|
)
|
|
|
|
# Return a dictionary with serializable payloads
|
|
return {
|
|
"local_problem": asdict(lp),
|
|
"shared_variables": asdict(sv),
|
|
"plan_delta": asdict(pd),
|
|
"dual_variables": asdict(dv),
|
|
"audit_log": asdict(al),
|
|
"privacy_budget": asdict(pb),
|
|
"registry_metadata": asdict(rm),
|
|
"contract_registry": [asdict(item) for item in self.registry.manifest()],
|
|
}
|