97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from idea170_agrimesh_federated_privacy import (
|
|
AgriMeshCanonicalMapper,
|
|
AdapterSpec,
|
|
FarmIdentity,
|
|
GovernanceLedger,
|
|
LocalProblem,
|
|
RegionalConstraint,
|
|
SharedSignals,
|
|
build_drought_demo,
|
|
solve_cross_farm_plan,
|
|
)
|
|
|
|
|
|
def test_solver_respects_quota_and_energy_caps() -> None:
|
|
farms = [
|
|
LocalProblem(
|
|
farm_id="farm-1",
|
|
current_soil_moisture_pct=15.0,
|
|
target_soil_moisture_pct=30.0,
|
|
irrigation_budget_liters=3000.0,
|
|
energy_budget_kwh=50.0,
|
|
plot_area_hectares=10.0,
|
|
fertilizer_cap_kg=10.0,
|
|
),
|
|
LocalProblem(
|
|
farm_id="farm-2",
|
|
current_soil_moisture_pct=20.0,
|
|
target_soil_moisture_pct=29.0,
|
|
irrigation_budget_liters=1800.0,
|
|
energy_budget_kwh=45.0,
|
|
plot_area_hectares=8.0,
|
|
fertilizer_cap_kg=9.0,
|
|
),
|
|
]
|
|
result = solve_cross_farm_plan(
|
|
farms,
|
|
RegionalConstraint(water_quota_liters=2500.0, drought_target_moisture_pct=28.0, energy_cap_kwh=40.0),
|
|
SharedSignals(mean_soil_moisture_pct=18.0, drought_pressure=0.9, forecast_rain_mm=0.0),
|
|
)
|
|
|
|
assert len(result.deltas) == 2
|
|
assert result.total_irrigation_liters <= 2500.0 + 1e-6
|
|
assert result.total_energy_kwh <= 40.0 + 1e-6
|
|
assert all(delta.irrigation_liters <= farm.irrigation_budget_liters for delta, farm in zip(result.deltas, farms, strict=True))
|
|
|
|
|
|
def test_ledger_persists_and_validates_chain(tmp_path: Path) -> None:
|
|
ledger = GovernanceLedger(tmp_path / "ledger.sqlite")
|
|
identity = FarmIdentity(farm_id="farm-a", name="Alpha Farm", public_key="public-key-123")
|
|
ledger.register_farm(identity)
|
|
|
|
adapter = AdapterSpec(
|
|
adapter_id="adapter-1",
|
|
farm_id="farm-a",
|
|
name="Drip Controller",
|
|
kind="irrigation_controller",
|
|
endpoint="mqtt://controller.local",
|
|
capabilities=["set_flow_rate"],
|
|
)
|
|
ledger.upsert_adapter(adapter)
|
|
ledger.record_contract("farm-a", "adapter-1", "controller-binding", {"protocol": "mqtt", "topic": "irrigation/set"})
|
|
|
|
delta = ledger.record_plan_delta(
|
|
build_drought_demo().deltas[0].model_copy(update={"farm_id": "farm-a", "previous_tag": ""}),
|
|
secret="shared-secret",
|
|
)
|
|
|
|
assert ledger.snapshot()["farms"] == 1
|
|
assert ledger.snapshot()["adapters"] == 1
|
|
assert ledger.snapshot()["contracts"] == 1
|
|
assert ledger.snapshot()["plan_deltas"] == 1
|
|
assert ledger.chain_is_valid("farm-a", secret="shared-secret")
|
|
assert delta.tag
|
|
ledger.close()
|
|
|
|
|
|
def test_canonical_mapper_and_demo_are_available() -> None:
|
|
mapper = AgriMeshCanonicalMapper()
|
|
local = LocalProblem(
|
|
farm_id="farm-x",
|
|
current_soil_moisture_pct=16.0,
|
|
target_soil_moisture_pct=30.0,
|
|
irrigation_budget_liters=1500.0,
|
|
energy_budget_kwh=25.0,
|
|
plot_area_hectares=7.0,
|
|
)
|
|
regional = RegionalConstraint(water_quota_liters=1200.0, drought_target_moisture_pct=29.0, energy_cap_kwh=20.0)
|
|
problem = mapper.to_catopt(local, regional)
|
|
|
|
assert problem.variables[0].name == "irrigation_liters"
|
|
assert problem.constraints[0].name == "water_quota"
|
|
assert build_drought_demo().deltas
|