diff --git a/README.md b/README.md index 8dc3b40..58518ef 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,8 @@ It coordinates solar, wind, and thermal storage sites through summarized signals - Canonical protocol primitives for local optimization, shared signals, dual variables, and plan deltas - A SQLite-backed graph-of-contracts registry for adapter versioning and schema conformance - An ADMM-lite solver for federated dispatch planning +- A federation bridge that packages solver rounds, audit records, and privacy-budget charges +- Deterministic starter adapters for solar inverter and thermal storage controllers - Signed message envelopes and DID-style identities for short-lived trust - Audit logging and privacy-budget accounting - Delta-sync journaling for islanded/offline operation @@ -18,6 +20,8 @@ It coordinates solar, wind, and thermal storage sites through summarized signals - `solfuse.identity`: identity generation, signing, and verification - `solfuse.registry`: adapter registry and schema conformance checks - `solfuse.solver`: federated consensus solver +- `solfuse.bridge`: round orchestration and governance-aware packaging +- `solfuse.adapters`: starter inverter and thermal controller adapters - `solfuse.delta_sync`: incremental plan journal and replay helpers - `solfuse.governance`: audit log and privacy budget ledger - `solfuse.transport`: canonical envelope serialization and TLS context helpers @@ -34,16 +38,15 @@ solfuse demo ## Minimal Example ```python -from solfuse.models import LocalProblem -from solfuse.solver import ADMMLiteSolver +from solfuse.bridge import FederationBridge +from solfuse.adapters import SolarInverterAdapter, ThermalStorageAdapter -solver = ADMMLiteSolver(rho=1.0, tolerance=1e-6, max_iterations=100) -result = solver.solve([ - LocalProblem(site_id="solar-a", preferred_dispatch=[4.0, 4.5], quadratic_weight=2.0), - LocalProblem(site_id="storage-b", preferred_dispatch=[1.0, 1.5], quadratic_weight=1.5), -]) +bridge = FederationBridge() +solar = SolarInverterAdapter().to_local_problem({"site_id": "solar-a", "forecast_kw": [4.0, 4.5]}, round_id=1) +thermal = ThermalStorageAdapter().to_local_problem({"site_id": "storage-b", "cooling_load_kw": [1.0, 1.5], "capacity_kw": 2.0}, round_id=1) +result = bridge.solve_round([solar, thermal], round_id=1) -print(result.consensus) +print(result.solver_result.consensus) ``` ## Governance Model diff --git a/src/solfuse/__init__.py b/src/solfuse/__init__.py index 0eb5bc1..4112158 100644 --- a/src/solfuse/__init__.py +++ b/src/solfuse/__init__.py @@ -1,12 +1,15 @@ """SolFuse canonical federation primitives.""" from .delta_sync import DeltaSyncJournal +from .bridge import FederationBridge, RoundSummary +from .adapters import AdapterRuntime, SolarInverterAdapter, ThermalStorageAdapter from .governance import AuditLedger, PrivacyBudgetLedger from .identity import AgentIdentity, generate_identity, sign_message, verify_message from .models import ( AdapterContract, AuditLogEntry, DualVariable, + FederationSnapshot, LocalProblem, PlanAction, PlanDelta, @@ -20,9 +23,12 @@ from .solver import ADMMLiteSolver __all__ = [ "ADMMLiteSolver", "AdapterContract", + "AdapterRuntime", "AgentIdentity", "AuditLedger", "AuditLogEntry", + "FederationBridge", + "FederationSnapshot", "DeltaSyncJournal", "DualVariable", "GraphOfContractsRegistry", @@ -31,8 +37,11 @@ __all__ = [ "PlanDelta", "PrivacyBudget", "PrivacyBudgetLedger", + "RoundSummary", "SharedVariable", "SolverResult", + "SolarInverterAdapter", + "ThermalStorageAdapter", "generate_identity", "sign_message", "verify_message", diff --git a/src/solfuse/adapters.py b/src/solfuse/adapters.py new file mode 100644 index 0000000..072a714 --- /dev/null +++ b/src/solfuse/adapters.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Any + +from pydantic import Field + +from .models import AdapterContract, LocalProblem, PlanAction, PlanDelta, SolfuseModel + + +class AdapterTelemetry(SolfuseModel): + site_id: str + round_id: int = Field(default=0, ge=0) + payload: dict[str, Any] + + +class AdapterRuntime(ABC): + contract: AdapterContract + + @abstractmethod + def to_local_problem(self, telemetry: dict[str, Any], *, round_id: int = 0) -> LocalProblem: + raise NotImplementedError + + def apply_plan(self, delta: PlanDelta) -> list[dict[str, Any]]: + return [action.model_dump(mode="json") for action in delta.actions] + + +class SolarInverterAdapter(AdapterRuntime): + contract = AdapterContract( + name="solar-inverter", + version="1.0.0", + adapter_type="inverter", + input_schema={ + "type": "object", + "properties": { + "site_id": {"type": "string"}, + "forecast_kw": {"type": "array", "items": {"type": "number"}, "minItems": 1}, + "upper_bound_kw": {"type": "array", "items": {"type": "number"}, "minItems": 1}, + "lower_bound_kw": {"type": "array", "items": {"type": "number"}, "minItems": 1}, + }, + "required": ["site_id", "forecast_kw"], + "additionalProperties": True, + }, + output_schema={ + "type": "object", + "properties": {"site_id": {"type": "string"}, "dispatch_kw": {"type": "array", "items": {"type": "number"}}}, + "required": ["site_id", "dispatch_kw"], + "additionalProperties": False, + }, + capabilities=["dispatch", "telemetry", "forecasting"], + ) + + def to_local_problem(self, telemetry: dict[str, Any], *, round_id: int = 0) -> LocalProblem: + forecast = [float(value) for value in telemetry["forecast_kw"]] + lower = telemetry.get("lower_bound_kw") + upper = telemetry.get("upper_bound_kw") + return LocalProblem( + site_id=str(telemetry["site_id"]), + preferred_dispatch=forecast, + lower_bounds=None if lower is None else [float(value) for value in lower], + upper_bounds=None if upper is None else [float(value) for value in upper], + quadratic_weight=float(telemetry.get("quadratic_weight", 1.0)), + round_id=round_id, + metadata={"adapter": self.contract.name, "source": "solar"}, + ) + + +class ThermalStorageAdapter(AdapterRuntime): + contract = AdapterContract( + name="thermal-storage", + version="1.0.0", + adapter_type="thermal", + input_schema={ + "type": "object", + "properties": { + "site_id": {"type": "string"}, + "cooling_load_kw": {"type": "array", "items": {"type": "number"}, "minItems": 1}, + "capacity_kw": {"type": "number"}, + "min_kw": {"type": "number"}, + "max_kw": {"type": "number"}, + }, + "required": ["site_id", "cooling_load_kw", "capacity_kw"], + "additionalProperties": True, + }, + output_schema={ + "type": "object", + "properties": {"site_id": {"type": "string"}, "dispatch_kw": {"type": "array", "items": {"type": "number"}}}, + "required": ["site_id", "dispatch_kw"], + "additionalProperties": False, + }, + capabilities=["dispatch", "thermal-storage", "demand-shaping"], + ) + + def to_local_problem(self, telemetry: dict[str, Any], *, round_id: int = 0) -> LocalProblem: + load = [float(value) for value in telemetry["cooling_load_kw"]] + capacity = float(telemetry["capacity_kw"]) + min_kw = float(telemetry.get("min_kw", 0.0)) + max_kw = float(telemetry.get("max_kw", capacity)) + preferred = [min(max(value, min_kw), max_kw) for value in load] + return LocalProblem( + site_id=str(telemetry["site_id"]), + preferred_dispatch=preferred, + lower_bounds=[min_kw] * len(preferred), + upper_bounds=[max_kw] * len(preferred), + quadratic_weight=float(telemetry.get("quadratic_weight", 1.0)), + round_id=round_id, + metadata={"adapter": self.contract.name, "source": "thermal"}, + ) diff --git a/src/solfuse/bridge.py b/src/solfuse/bridge.py new file mode 100644 index 0000000..0417e0b --- /dev/null +++ b/src/solfuse/bridge.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .governance import AuditLedger, PrivacyBudgetLedger +from .identity import AgentIdentity, sign_message +from .delta_sync import DeltaSyncJournal +from .models import DualVariable, FederationSnapshot, LocalProblem, SharedVariable +from .solver import ADMMLiteSolver + + +@dataclass(frozen=True) +class RoundSummary: + round_id: int + consensus: list[float] + delta_count: int + + +class FederationBridge: + def __init__( + self, + *, + solver: ADMMLiteSolver | None = None, + journal: DeltaSyncJournal | None = None, + audit_ledger: AuditLedger | None = None, + privacy_ledger: PrivacyBudgetLedger | None = None, + ) -> None: + self.solver = solver or ADMMLiteSolver() + self.journal = journal + self.audit_ledger = audit_ledger + self.privacy_ledger = privacy_ledger + + def solve_round( + self, + problems: list[LocalProblem], + *, + round_id: int, + shared_variables: list[SharedVariable] | None = None, + dual_variables: list[DualVariable] | None = None, + actor_id: str = "federation", + identity: AgentIdentity | None = None, + metadata: dict[str, Any] | None = None, + ) -> FederationSnapshot: + solver_result = self.solver.solve( + problems, + shared_variables=shared_variables, + dual_variables=dual_variables, + ) + snapshot = FederationSnapshot( + round_id=round_id, + local_problems=problems, + shared_variables=shared_variables or [], + dual_variables=dual_variables or [], + solver_result=solver_result, + metadata=metadata or {}, + ) + + if self.journal is not None: + self.journal.merge(solver_result.deltas) + + if self.privacy_ledger is not None and self.privacy_ledger.get_budget(actor_id) is not None: + epsilon = 0.01 * max(1, len(solver_result.deltas)) + delta = 1e-8 * max(1, sum(len(delta.actions) for delta in solver_result.deltas)) + self.privacy_ledger.charge(actor_id, epsilon, delta) + + if self.audit_ledger is not None: + payload = snapshot.model_dump(mode="json") + signature = sign_message(identity, payload) if identity is not None else None + self.audit_ledger.append( + self._build_audit_entry( + actor_id=actor_id, + payload=payload, + signature=signature, + ) + ) + + return snapshot + + def summarize_round(self, snapshot: FederationSnapshot) -> RoundSummary: + return RoundSummary( + round_id=snapshot.round_id, + consensus=snapshot.solver_result.consensus, + delta_count=len(snapshot.solver_result.deltas), + ) + + def _build_audit_entry(self, *, actor_id: str, payload: dict[str, Any], signature: str | None): + from .models import AuditLogEntry + + return AuditLogEntry( + actor_id=actor_id, + event_type="federation_round", + payload=payload, + signature=signature, + ) diff --git a/src/solfuse/cli.py b/src/solfuse/cli.py index 463c7b2..9aa138d 100644 --- a/src/solfuse/cli.py +++ b/src/solfuse/cli.py @@ -3,6 +3,8 @@ from __future__ import annotations import argparse import json +from .adapters import SolarInverterAdapter, ThermalStorageAdapter +from .bridge import FederationBridge from .models import LocalProblem from .solver import ADMMLiteSolver @@ -11,6 +13,7 @@ def main() -> None: parser = argparse.ArgumentParser(prog="solfuse") subcommands = parser.add_subparsers(dest="command", required=True) subcommands.add_parser("demo") + subcommands.add_parser("microgrid-demo") args = parser.parse_args() if args.command == "demo": @@ -22,3 +25,15 @@ def main() -> None: ] ) print(json.dumps(result.model_dump(mode="json"), indent=2)) + if args.command == "microgrid-demo": + bridge = FederationBridge(solver=ADMMLiteSolver()) + solar = SolarInverterAdapter().to_local_problem( + {"site_id": "solar-a", "forecast_kw": [4.0, 4.5], "upper_bound_kw": [5.0, 5.0]}, + round_id=1, + ) + thermal = ThermalStorageAdapter().to_local_problem( + {"site_id": "thermal-b", "cooling_load_kw": [1.0, 1.5], "capacity_kw": 2.0}, + round_id=1, + ) + snapshot = bridge.solve_round([solar, thermal], round_id=1) + print(json.dumps(snapshot.model_dump(mode="json"), indent=2)) diff --git a/src/solfuse/models.py b/src/solfuse/models.py index f6ce9ce..6532c60 100644 --- a/src/solfuse/models.py +++ b/src/solfuse/models.py @@ -130,3 +130,12 @@ class SolverResult(SolfuseModel): dual_residual: float deltas: list[PlanDelta] = Field(default_factory=list) metadata: dict[str, Any] = Field(default_factory=dict) + + +class FederationSnapshot(SolfuseModel): + round_id: int = Field(ge=0) + local_problems: list[LocalProblem] + shared_variables: list[SharedVariable] = Field(default_factory=list) + dual_variables: list[DualVariable] = Field(default_factory=list) + solver_result: SolverResult + metadata: dict[str, Any] = Field(default_factory=dict) diff --git a/src/solfuse/solver.py b/src/solfuse/solver.py index 29097bf..21aa632 100644 --- a/src/solfuse/solver.py +++ b/src/solfuse/solver.py @@ -5,7 +5,7 @@ import math import numpy as np -from .models import LocalProblem, PlanAction, PlanDelta, SolverResult +from .models import DualVariable, LocalProblem, PlanAction, PlanDelta, SharedVariable, SolverResult @dataclass(frozen=True) @@ -19,11 +19,19 @@ class ADMMLiteSolver: def __init__(self, *, rho: float = 1.0, tolerance: float = 1e-6, max_iterations: int = 100) -> None: if rho <= 0: raise ValueError("rho must be positive") + if max_iterations <= 0: + raise ValueError("max_iterations must be positive") self.rho = rho self.tolerance = tolerance self.max_iterations = max_iterations - def solve(self, problems: list[LocalProblem]) -> SolverResult: + def solve( + self, + problems: list[LocalProblem], + *, + shared_variables: list[SharedVariable] | None = None, + dual_variables: list[DualVariable] | None = None, + ) -> SolverResult: if not problems: raise ValueError("at least one local problem is required") @@ -39,11 +47,21 @@ class ADMMLiteSolver: x = preferred.copy() z = preferred.mean(axis=0) + if shared_variables: + candidate = np.asarray(shared_variables[0].values, dtype=float) + if candidate.shape == (horizon,): + z = candidate u = np.zeros_like(x) + if dual_variables: + candidate = np.asarray(dual_variables[0].values, dtype=float) + if candidate.shape == (horizon,): + u = np.broadcast_to(candidate, x.shape).copy() primal_residual = float("inf") dual_residual = float("inf") + last_iteration = 0 - for iteration in range(1, self.max_iterations + 1): + for current_iteration in range(1, self.max_iterations + 1): + last_iteration = current_iteration previous_z = z.copy() x = np.clip((weights * preferred + self.rho * (z - u)) / (weights + self.rho), lower, upper) z = np.mean(x + u, axis=0) @@ -64,9 +82,9 @@ class ADMMLiteSolver: PlanDelta( site_id=problem.site_id, adapter_id=f"adapter:{problem.site_id}", - revision=iteration, + revision=last_iteration, round_id=problem.round_id, - parent_revision=max(0, iteration - 1), + parent_revision=max(0, last_iteration - 1), actions=actions, metadata={"solver": "admm-lite", "round_id": problem.round_id}, ) @@ -74,7 +92,7 @@ class ADMMLiteSolver: return SolverResult( consensus=[float(value) for value in z], - iterations=iteration, + iterations=last_iteration, primal_residual=primal_residual, dual_residual=dual_residual, deltas=deltas, diff --git a/tests/test_bridge_and_adapters.py b/tests/test_bridge_and_adapters.py new file mode 100644 index 0000000..4bb1964 --- /dev/null +++ b/tests/test_bridge_and_adapters.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from solfuse import ( + AuditLedger, + DeltaSyncJournal, + FederationBridge, + PrivacyBudget, + PrivacyBudgetLedger, + DualVariable, + SolarInverterAdapter, + ThermalStorageAdapter, + SharedVariable, + generate_identity, +) + + +def test_bridge_records_rounds_and_respects_governance(): + journal = DeltaSyncJournal() + audit = AuditLedger() + budgets = PrivacyBudgetLedger() + budgets.set_budget("federation", PrivacyBudget(epsilon=1.0, delta=1e-5)) + bridge = FederationBridge(journal=journal, audit_ledger=audit, privacy_ledger=budgets) + + solar = SolarInverterAdapter().to_local_problem({"site_id": "solar-a", "forecast_kw": [4.0, 4.5]}, round_id=3) + thermal = ThermalStorageAdapter().to_local_problem( + {"site_id": "thermal-b", "cooling_load_kw": [1.0, 1.5], "capacity_kw": 2.0}, + round_id=3, + ) + shared = SharedVariable(name="forecast", version="1.0", values=[3.5, 3.5], round_id=3) + dual = DualVariable(name="dispatch-dual", version="1.0", values=[0.0, 0.0], round_id=3) + + snapshot = bridge.solve_round( + [solar, thermal], + round_id=3, + shared_variables=[shared], + dual_variables=[dual], + identity=generate_identity("federation"), + ) + + assert snapshot.round_id == 3 + assert snapshot.shared_variables[0].name == "forecast" + assert snapshot.dual_variables[0].name == "dispatch-dual" + assert snapshot.solver_result.deltas + assert journal.load() + assert audit.coverage() == 1.0 + updated_budget = budgets.get_budget("federation") + assert updated_budget is not None + assert updated_budget.spent_epsilon > 0 + + +def test_starter_adapters_generate_local_problems_and_plan_frames(): + solar = SolarInverterAdapter() + thermal = ThermalStorageAdapter() + + solar_problem = solar.to_local_problem({"site_id": "solar-a", "forecast_kw": [3.0, 3.5], "upper_bound_kw": [4.0, 4.0]}) + thermal_problem = thermal.to_local_problem({"site_id": "thermal-b", "cooling_load_kw": [1.0, 2.0], "capacity_kw": 2.5}) + + assert solar_problem.site_id == "solar-a" + assert thermal_problem.upper_bounds == [2.5, 2.5]