community
/
idea170-agrimesh-federated-privacy
34
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

build(agent): c3po#b883b4 iteration

This commit is contained in:
agent-b883b4bc188823a2 2026-04-26 22:17:03 +02:00
parent a87998d1e8
commit a155f6a0ce
14 changed files with 789 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
.gitignore vendored Normal file
View File

@ -0,0 +1,21 @@
node_modules/
.npmrc
.env
.env.*
__tests__/
coverage/
.nyc_output/
dist/
build/
.cache/
*.log
.DS_Store
tmp/
.tmp/
__pycache__/
*.pyc
.venv/
venv/
*.egg-info/
.pytest_cache/
READY_TO_PUBLISH

30
AGENTS.md Normal file
View File

@ -0,0 +1,30 @@
# AGENTS.md
## Architecture
- Python 3.11 package under `src/idea170_agrimesh_federated_privacy`.
- SQLite-backed governance ledger for farms, adapters, contracts, deltas, and sync state.
- Pydantic models validate local problems, shared signals, plan deltas, and adapter contracts.
- `solver.py` contains the ADMM-lite allocator for cross-farm irrigation planning.
- `canonical.py` maps AgriMesh primitives into a CatOpt-style intermediate representation.
- `demo.py` assembles a drought scenario for 2 to 3 farms.
## Tech Stack
- Standard library: `sqlite3`, `json`, `hashlib`, `hmac`, `math`, `random`, `dataclasses`, `pathlib`.
- External dependency: `pydantic` for strict data validation.
- Tests: `pytest`.
## Rules
- Keep changes minimal and deterministic.
- Preserve SQLite schema compatibility unless a migration is explicitly added.
- Any new public data model should have validation and at least one test.
- Preserve canonical JSON ordering when hashing or signing payloads.
- Do not bypass the solver or ledger abstractions with ad hoc state.
## Testing
- `bash test.sh`
- `pytest -q`
- `python3 -m build`

47
README.md
View File

@ -1,3 +1,46 @@
# idea170-agrimesh-federated-privacy
# AgriMesh Federated Privacy
Source logic for Idea #170
AgriMesh is a Python package for privacy-preserving coordination across neighboring farms. It models local irrigation and resource constraints, exchanges aggregated signals, records tamper-evident plan deltas in SQLite, and produces coordinated irrigation plans for drought conditions.
## What It Includes
- Pydantic models for farms, local problems, shared signals, regional constraints, adapters, and plan deltas.
- A SQLite governance ledger for farms, contracts, adapters, events, and offline delta sync.
- An ADMM-lite irrigation allocator that respects farm budgets and regional water and energy caps.
- A canonical mapping layer that converts AgriMesh inputs into a CatOpt-style intermediate representation.
- A drought demo for 2 to 3 farms.
## Package Layout
- `src/idea170_agrimesh_federated_privacy/models.py` - validated data models.
- `src/idea170_agrimesh_federated_privacy/ledger.py` - SQLite governance ledger.
- `src/idea170_agrimesh_federated_privacy/solver.py` - cross-farm allocator.
- `src/idea170_agrimesh_federated_privacy/canonical.py` - CatOpt mapping.
- `src/idea170_agrimesh_federated_privacy/demo.py` - drought scenario builder.
## Quick Start
```python
from idea170_agrimesh_federated_privacy import build_drought_demo
result = build_drought_demo()
for delta in result.deltas:
print(delta.farm_id, delta.irrigation_liters)
```
```bash
bash test.sh
python3 -m idea170_agrimesh_federated_privacy
```
## Testing
The repository is wired for:
- `bash test.sh`
- `pytest -q`
- `python3 -m build`
## Packaging
This project is published as `idea170-agrimesh-federated-privacy` and uses `README.md` as its package description source.

22
pyproject.toml Normal file
View File

@ -0,0 +1,22 @@
[build-system]
requires = ["setuptools>=68", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "idea170-agrimesh-federated-privacy"
version = "0.1.0"
description = "AgriMesh federated, privacy-preserving cross-farm resource optimization platform"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"pydantic>=2.7,<3",
]
[project.optional-dependencies]
test = ["pytest>=8,<9"]
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]

28
src/idea170_agrimesh_federated_privacy/__init__.py Normal file
View File

@ -0,0 +1,28 @@
"""AgriMesh federated optimization primitives."""
from .canonical import AgriMeshCanonicalMapper, CatOptProblem
from .demo import build_drought_demo
from .ledger import GovernanceLedger
from .models import (
AdapterSpec,
FarmIdentity,
LocalProblem,
PlanDelta,
RegionalConstraint,
SharedSignals,
)
from .solver import solve_cross_farm_plan
__all__ = [
"AgriMeshCanonicalMapper",
"AdapterSpec",
"CatOptProblem",
"FarmIdentity",
"GovernanceLedger",
"LocalProblem",
"PlanDelta",
"RegionalConstraint",
"SharedSignals",
"build_drought_demo",
"solve_cross_farm_plan",
]

13
src/idea170_agrimesh_federated_privacy/__main__.py Normal file
View File

@ -0,0 +1,13 @@
from .demo import build_drought_demo
def main() -> None:
result = build_drought_demo()
for delta in result.deltas:
print(delta.model_dump_json())
print(f"total_irrigation_liters={result.total_irrigation_liters}")
print(f"total_energy_kwh={result.total_energy_kwh}")
if __name__ == "__main__":
main()

44
src/idea170_agrimesh_federated_privacy/canonical.py Normal file
View File

@ -0,0 +1,44 @@
from __future__ import annotations
from .models import CatOptConstraint, CatOptProblem, CatOptVariable, LocalProblem, PlanDelta, RegionalConstraint
class AgriMeshCanonicalMapper:
def to_catopt(self, local_problem: LocalProblem, regional_constraint: RegionalConstraint) -> CatOptProblem:
need = max(local_problem.target_soil_moisture_pct - local_problem.current_soil_moisture_pct, 0.0)
return CatOptProblem(
problem_id=f"{local_problem.farm_id}:{regional_constraint.regional_water_quota_id}",
variables=[
CatOptVariable(
name="irrigation_liters",
lower=0.0,
upper=local_problem.irrigation_budget_liters,
target=min(local_problem.irrigation_budget_liters, need * local_problem.plot_area_hectares * 10.0),
)
],
constraints=[
CatOptConstraint(
name="water_quota",
expression=f"irrigation_liters <= {regional_constraint.water_quota_liters}",
),
CatOptConstraint(
name="energy_cap",
expression=f"energy_kwh <= {regional_constraint.energy_cap_kwh}",
),
],
objective="minimize squared deviation from target soil moisture",
metadata={
"farm_id": local_problem.farm_id,
"drought_target_moisture_pct": regional_constraint.drought_target_moisture_pct,
"plot_area_hectares": local_problem.plot_area_hectares,
},
)
def from_plan_delta(self, delta: PlanDelta) -> dict[str, float | str]:
return {
"farm_id": delta.farm_id,
"irrigation_liters": delta.irrigation_liters,
"fertilizer_kg": delta.fertilizer_kg,
"energy_kwh": delta.energy_kwh,
"tag": delta.tag,
}

42
src/idea170_agrimesh_federated_privacy/demo.py Normal file
View File

@ -0,0 +1,42 @@
from __future__ import annotations
from .models import LocalProblem, RegionalConstraint, SharedSignals
from .solver import SolveResult, solve_cross_farm_plan
def build_drought_demo() -> SolveResult:
farms = [
LocalProblem(
farm_id="farm-a",
current_soil_moisture_pct=18.0,
target_soil_moisture_pct=31.0,
irrigation_budget_liters=3200.0,
energy_budget_kwh=90.0,
plot_area_hectares=18.0,
crop_priority=1.0,
fertilizer_cap_kg=30.0,
),
LocalProblem(
farm_id="farm-b",
current_soil_moisture_pct=22.0,
target_soil_moisture_pct=30.0,
irrigation_budget_liters=2800.0,
energy_budget_kwh=80.0,
plot_area_hectares=14.0,
crop_priority=0.9,
fertilizer_cap_kg=25.0,
),
LocalProblem(
farm_id="farm-c",
current_soil_moisture_pct=26.0,
target_soil_moisture_pct=32.0,
irrigation_budget_liters=2400.0,
energy_budget_kwh=75.0,
plot_area_hectares=12.0,
crop_priority=0.8,
fertilizer_cap_kg=20.0,
),
]
regional = RegionalConstraint(water_quota_liters=5200.0, drought_target_moisture_pct=30.0, energy_cap_kwh=170.0)
signals = SharedSignals(mean_soil_moisture_pct=22.0, drought_pressure=0.78, forecast_rain_mm=1.5, local_dp_epsilon=1.0)
return solve_cross_farm_plan(farms, regional, signals)

201
src/idea170_agrimesh_federated_privacy/ledger.py Normal file
View File

@ -0,0 +1,201 @@
from __future__ import annotations
import sqlite3
from pathlib import Path
from typing import Any, Iterable
from .models import AdapterSpec, FarmIdentity, GovernanceEvent, PlanDelta
from .utils import canonical_json, sha256_hex
class GovernanceLedger:
def __init__(self, path: str | Path = ":memory:") -> None:
self.path = str(path)
self._conn = sqlite3.connect(self.path)
self._conn.row_factory = sqlite3.Row
self._ensure_schema()
def close(self) -> None:
self._conn.close()
def _ensure_schema(self) -> None:
self._conn.executescript(
"""
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS farms (
farm_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
public_key TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS adapters (
adapter_id TEXT PRIMARY KEY,
farm_id TEXT NOT NULL,
name TEXT NOT NULL,
kind TEXT NOT NULL,
endpoint TEXT NOT NULL,
capabilities_json TEXT NOT NULL,
metadata_json TEXT NOT NULL,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
FOREIGN KEY(farm_id) REFERENCES farms(farm_id)
);
CREATE TABLE IF NOT EXISTS events (
event_id TEXT PRIMARY KEY,
event_type TEXT NOT NULL,
payload_json TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS plan_deltas (
delta_id TEXT PRIMARY KEY,
farm_id TEXT NOT NULL,
step_index INTEGER NOT NULL,
payload_json TEXT NOT NULL,
previous_tag TEXT NOT NULL,
tag TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(farm_id) REFERENCES farms(farm_id)
);
CREATE TABLE IF NOT EXISTS sync_outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
delta_id TEXT NOT NULL,
payload_json TEXT NOT NULL,
created_at TEXT NOT NULL,
delivered_at TEXT,
FOREIGN KEY(delta_id) REFERENCES plan_deltas(delta_id)
);
CREATE TABLE IF NOT EXISTS contracts (
contract_id TEXT PRIMARY KEY,
farm_id TEXT NOT NULL,
adapter_id TEXT NOT NULL,
contract_type TEXT NOT NULL,
payload_json TEXT NOT NULL,
content_hash TEXT NOT NULL,
created_at TEXT NOT NULL,
FOREIGN KEY(farm_id) REFERENCES farms(farm_id),
FOREIGN KEY(adapter_id) REFERENCES adapters(adapter_id)
);
"""
)
self._conn.commit()
def register_farm(self, identity: FarmIdentity) -> None:
self._conn.execute(
"INSERT OR REPLACE INTO farms (farm_id, name, public_key, created_at) VALUES (?, ?, ?, ?)",
(identity.farm_id, identity.name, identity.public_key, identity.created_at.isoformat()),
)
self._conn.commit()
def upsert_adapter(self, adapter: AdapterSpec) -> None:
self._conn.execute(
"""
INSERT OR REPLACE INTO adapters
(adapter_id, farm_id, name, kind, endpoint, capabilities_json, metadata_json, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
adapter.adapter_id,
adapter.farm_id,
adapter.name,
adapter.kind,
adapter.endpoint,
canonical_json(adapter.capabilities),
canonical_json(adapter.metadata),
adapter.created_at.isoformat(),
adapter.updated_at.isoformat(),
),
)
self._conn.commit()
def record_event(self, event: GovernanceEvent) -> None:
self._conn.execute(
"INSERT OR REPLACE INTO events (event_id, event_type, payload_json, created_at) VALUES (?, ?, ?, ?)",
(str(event.event_id), event.event_type, canonical_json(event.payload), event.created_at.isoformat()),
)
self._conn.commit()
def record_plan_delta(self, delta: PlanDelta, secret: str) -> PlanDelta:
payload = delta.model_dump(mode="json", exclude={"tag"})
tag = delta.tag or self._compute_tag(delta.previous_tag, payload, secret)
tagged = delta.model_copy(update={"tag": tag})
self._conn.execute(
"""
INSERT OR REPLACE INTO plan_deltas
(delta_id, farm_id, step_index, payload_json, previous_tag, tag, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(
str(tagged.delta_id),
tagged.farm_id,
tagged.step_index,
canonical_json(tagged.model_dump(mode="json")),
tagged.previous_tag,
tagged.tag,
tagged.created_at.isoformat(),
),
)
self._conn.execute(
"INSERT INTO sync_outbox (delta_id, payload_json, created_at) VALUES (?, ?, ?)",
(str(tagged.delta_id), canonical_json(tagged.model_dump(mode="json")), tagged.created_at.isoformat()),
)
self._conn.commit()
return tagged
def queue_delta(self, delta: PlanDelta) -> None:
self._conn.execute(
"INSERT INTO sync_outbox (delta_id, payload_json, created_at) VALUES (?, ?, ?)",
(str(delta.delta_id), canonical_json(delta.model_dump(mode="json")), delta.created_at.isoformat()),
)
self._conn.commit()
def pending_deltas(self) -> list[dict[str, Any]]:
rows = self._conn.execute(
"SELECT delta_id, payload_json, created_at FROM sync_outbox WHERE delivered_at IS NULL ORDER BY id ASC"
).fetchall()
return [dict(row) for row in rows]
def mark_delivered(self, delta_id: str) -> None:
self._conn.execute(
"UPDATE sync_outbox SET delivered_at = CURRENT_TIMESTAMP WHERE delta_id = ? AND delivered_at IS NULL",
(delta_id,),
)
self._conn.commit()
def record_contract(self, farm_id: str, adapter_id: str, contract_type: str, payload: dict[str, Any]) -> str:
contract_id = sha256_hex({"farm_id": farm_id, "adapter_id": adapter_id, "type": contract_type, "payload": payload})
self._conn.execute(
"""
INSERT OR REPLACE INTO contracts
(contract_id, farm_id, adapter_id, contract_type, payload_json, content_hash, created_at)
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
""",
(contract_id, farm_id, adapter_id, contract_type, canonical_json(payload), contract_id),
)
self._conn.commit()
return contract_id
def list_contracts(self) -> list[dict[str, Any]]:
rows = self._conn.execute("SELECT * FROM contracts ORDER BY created_at ASC").fetchall()
return [dict(row) for row in rows]
def chain_is_valid(self, farm_id: str, secret: str) -> bool:
rows = self._conn.execute(
"SELECT payload_json, previous_tag, tag FROM plan_deltas WHERE farm_id = ? ORDER BY created_at ASC", (farm_id,)
).fetchall()
previous = ""
for row in rows:
payload = dict(__import__("json").loads(row["payload_json"]))
payload.pop("tag", None)
expected = self._compute_tag(previous, payload, secret)
if row["previous_tag"] != previous or row["tag"] != expected:
return False
previous = row["tag"]
return True
def snapshot(self) -> dict[str, int]:
tables = ["farms", "adapters", "events", "plan_deltas", "sync_outbox", "contracts"]
return {name: int(self._conn.execute(f"SELECT COUNT(*) FROM {name}").fetchone()[0]) for name in tables}
@staticmethod
def _compute_tag(previous_tag: str, payload: dict[str, Any], secret: str) -> str:
return sha256_hex({"previous_tag": previous_tag, "payload": payload, "secret": secret})

105
src/idea170_agrimesh_federated_privacy/models.py Normal file
View File

@ -0,0 +1,105 @@
from __future__ import annotations
from datetime import datetime, timezone
from typing import Any, Literal
from uuid import UUID, uuid4
from pydantic import BaseModel, Field, field_validator
def utc_now() -> datetime:
return datetime.now(timezone.utc)
class FarmIdentity(BaseModel):
farm_id: str = Field(min_length=1)
name: str = Field(min_length=1)
public_key: str = Field(min_length=8)
created_at: datetime = Field(default_factory=utc_now)
class LocalProblem(BaseModel):
farm_id: str = Field(min_length=1)
current_soil_moisture_pct: float = Field(ge=0.0, le=100.0)
target_soil_moisture_pct: float = Field(ge=0.0, le=100.0)
irrigation_budget_liters: float = Field(gt=0.0)
energy_budget_kwh: float = Field(ge=0.0)
plot_area_hectares: float = Field(gt=0.0)
crop_priority: float = Field(default=1.0, ge=0.0)
fertilizer_cap_kg: float = Field(default=0.0, ge=0.0)
@field_validator("target_soil_moisture_pct")
@classmethod
def validate_target(cls, value: float, info):
current = info.data.get("current_soil_moisture_pct")
if current is not None and value < current * 0.5:
raise ValueError("target moisture is implausibly low relative to current moisture")
return value
class SharedSignals(BaseModel):
timestamp: datetime = Field(default_factory=utc_now)
mean_soil_moisture_pct: float = Field(ge=0.0, le=100.0)
drought_pressure: float = Field(ge=0.0, le=1.0)
forecast_rain_mm: float = Field(ge=0.0)
secure_aggregation_version: Literal["v1"] = "v1"
local_dp_epsilon: float | None = Field(default=None, gt=0.0)
class RegionalConstraint(BaseModel):
water_quota_liters: float = Field(gt=0.0)
drought_target_moisture_pct: float = Field(ge=0.0, le=100.0)
energy_cap_kwh: float = Field(ge=0.0)
regional_water_quota_id: str = Field(default_factory=lambda: str(uuid4()))
class PlanDelta(BaseModel):
delta_id: UUID = Field(default_factory=uuid4)
farm_id: str = Field(min_length=1)
step_index: int = Field(ge=0)
irrigation_liters: float = Field(ge=0.0)
fertilizer_kg: float = Field(ge=0.0)
energy_kwh: float = Field(ge=0.0)
rationale: str = Field(min_length=1)
previous_tag: str = Field(default="")
tag: str = Field(default="")
created_at: datetime = Field(default_factory=utc_now)
class AdapterSpec(BaseModel):
adapter_id: str = Field(min_length=1)
farm_id: str = Field(min_length=1)
name: str = Field(min_length=1)
kind: Literal["irrigation_controller", "soil_sensor_gateway", "weather_feed", "farm_management_system"]
endpoint: str = Field(min_length=1)
capabilities: list[str] = Field(default_factory=list)
metadata: dict[str, Any] = Field(default_factory=dict)
created_at: datetime = Field(default_factory=utc_now)
updated_at: datetime = Field(default_factory=utc_now)
class GovernanceEvent(BaseModel):
event_id: UUID = Field(default_factory=uuid4)
event_type: str = Field(min_length=1)
payload: dict[str, Any] = Field(default_factory=dict)
created_at: datetime = Field(default_factory=utc_now)
class CatOptVariable(BaseModel):
name: str
lower: float
upper: float
target: float
class CatOptConstraint(BaseModel):
name: str
expression: str
class CatOptProblem(BaseModel):
problem_id: str
variables: list[CatOptVariable]
constraints: list[CatOptConstraint]
objective: str
metadata: dict[str, Any] = Field(default_factory=dict)

103
src/idea170_agrimesh_federated_privacy/solver.py Normal file
View File

@ -0,0 +1,103 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Sequence
from .models import LocalProblem, PlanDelta, RegionalConstraint, SharedSignals
from .utils import clamp, laplace_noise
@dataclass(frozen=True)
class SolveResult:
deltas: list[PlanDelta]
total_irrigation_liters: float
total_energy_kwh: float
def _project_capped_simplex(values: Sequence[float], quota: float, caps: Sequence[float]) -> list[float]:
clipped = [clamp(v, 0.0, cap) for v, cap in zip(values, caps, strict=True)]
total = sum(clipped)
if total <= quota:
return clipped
lo = min(v - cap for v, cap in zip(clipped, caps, strict=True)) - quota
hi = max(clipped)
for _ in range(80):
mid = (lo + hi) / 2.0
projected = [clamp(v - mid, 0.0, cap) for v, cap in zip(clipped, caps, strict=True)]
current = sum(projected)
if current > quota:
lo = mid
else:
hi = mid
return [clamp(v - hi, 0.0, cap) for v, cap in zip(clipped, caps, strict=True)]
def solve_cross_farm_plan(
local_problems: Sequence[LocalProblem],
regional_constraint: RegionalConstraint,
shared_signals: SharedSignals,
*,
farm_secret: str = "agrimesh-secret",
iterations: int = 12,
) -> SolveResult:
if not local_problems:
return SolveResult([], 0.0, 0.0)
needs = []
caps = []
for problem in local_problems:
moisture_gap = max(problem.target_soil_moisture_pct - problem.current_soil_moisture_pct, 0.0)
weather_adjustment = max(0.2, 1.0 - shared_signals.forecast_rain_mm / 50.0)
drought_adjustment = 1.0 + shared_signals.drought_pressure * 0.35
need = moisture_gap * problem.plot_area_hectares * 10.0 * weather_adjustment * drought_adjustment
needs.append(need)
caps.append(min(problem.irrigation_budget_liters, regional_constraint.water_quota_liters))
rho = 1.2
x = [0.0 for _ in local_problems]
z = [0.0 for _ in local_problems]
u = [0.0 for _ in local_problems]
for _ in range(iterations):
x = [clamp((need + rho * (zi - ui)) / (1.0 + rho), 0.0, cap) for need, zi, ui, cap in zip(needs, z, u, caps, strict=True)]
z = _project_capped_simplex([xi + ui for xi, ui in zip(x, u, strict=True)], regional_constraint.water_quota_liters, caps)
u = [ui + xi - zi for xi, zi, ui in zip(x, z, u, strict=True)]
total_irrigation = sum(z)
if shared_signals.local_dp_epsilon is not None:
total_irrigation = max(0.0, total_irrigation + laplace_noise(scale=1.0 / shared_signals.local_dp_epsilon, seed=7))
deltas: list[PlanDelta] = []
total_energy = 0.0
for index, (problem, irrigation) in enumerate(zip(local_problems, z, strict=True)):
fertilizer = min(problem.fertilizer_cap_kg, irrigation / 200.0)
energy = min(problem.energy_budget_kwh, irrigation * 0.02)
total_energy += energy
deltas.append(
PlanDelta(
farm_id=problem.farm_id,
step_index=index,
irrigation_liters=round(irrigation, 3),
fertilizer_kg=round(fertilizer, 3),
energy_kwh=round(energy, 3),
rationale=f"Consensus allocation under water quota {regional_constraint.water_quota_liters:.1f}",
)
)
if total_energy > regional_constraint.energy_cap_kwh and total_energy > 0:
scale = regional_constraint.energy_cap_kwh / total_energy
deltas = [
delta.model_copy(
update={
"irrigation_liters": round(delta.irrigation_liters * scale, 3),
"fertilizer_kg": round(delta.fertilizer_kg * scale, 3),
"energy_kwh": round(delta.energy_kwh * scale, 3),
}
)
for delta in deltas
]
total_energy = regional_constraint.energy_cap_kwh
total_irrigation = sum(delta.irrigation_liters for delta in deltas)
return SolveResult(deltas=deltas, total_irrigation_liters=round(total_irrigation, 3), total_energy_kwh=round(total_energy, 3))

32
src/idea170_agrimesh_federated_privacy/utils.py Normal file
View File

@ -0,0 +1,32 @@
from __future__ import annotations
import hashlib
import hmac
import json
import math
import random
from typing import Any
def canonical_json(value: Any) -> str:
return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
def sha256_hex(value: Any) -> str:
return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
def hmac_hex(secret: str, value: Any) -> str:
return hmac.new(secret.encode("utf-8"), canonical_json(value).encode("utf-8"), hashlib.sha256).hexdigest()
def laplace_noise(scale: float, seed: int | None = None) -> float:
if scale <= 0:
return 0.0
rng = random.Random(seed)
u = rng.random() - 0.5
return -scale * math.copysign(math.log1p(-2 * abs(u)), u)
def clamp(value: float, lower: float, upper: float) -> float:
return min(max(value, lower), upper)

7
test.sh Normal file
View File

@ -0,0 +1,7 @@
#!/usr/bin/env bash
set -euo pipefail
python3 -m pip install --upgrade pip >/dev/null
python3 -m pip install -e .[test] build >/dev/null
pytest -q
python3 -m build

96
tests/test_agrimesh.py Normal file
View File

@ -0,0 +1,96 @@
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