60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
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]
|