Introduction
Modern agriculture is data-driven. Soil moisture, ambient temperature, and light intensity are the three primary inputs to irrigation decisions. This tutorial shows you how to build a solar-powered edge station that reads these sensors, queries a weather API for forecast enrichment, triggers a valve actuator when moisture drops below a threshold, and deploys across a fleet of field stations.
Hardware Setup
Sensor Suite
- Soil Moisture: Capacitive sensor (e.g., V1.2) connected to ADC via GPIO.
- Ambient Temperature: DS18B20 1-Wire probe in a radiation shield.
- Light: BH1750 I2C lux sensor.
- Actuator: 12 V solenoid valve driven via relay HAT.
Solar Power Budget
A 20 W solar panel with a 12 Ah LiFePO4 battery powers the Pi 4 in duty-cycled mode. The agent wakes every 15 minutes, collects 60 seconds of samples, evaluates rules, transmits a summary, and sleeps.
Weather API Integration
Instead of shipping every raw reading to the cloud, we pull a local 3-hour rainfall forecast
from a weather API and use it to suppress unnecessary irrigation. The adapter follows the
duck-typing contract described in edge-custom-adapters.html.
import json
import urllib.request
from typing import Any, Dict, Optional
class WeatherHTTPAdapter:
"""Custom HTTP adapter that fetches rainfall forecast for irrigation decisions."""
def __init__(self, api_url: str, api_key: str, lat: float, lon: float) -> None:
self.api_url = api_url
self.api_key = api_key
self.lat = lat
self.lon = lon
def fetch_rainfall_mm(self, hours: int = 3) -> Optional[float]:
url = (
f"{self.api_url}/forecast?lat={self.lat}&lon={self.lon}"
f"&hours={hours}&apikey={self.api_key}"
)
try:
with urllib.request.urlopen(url, timeout=10) as resp:
data = json.loads(resp.read().decode("utf-8"))
return sum(p.get("rain_mm", 0.0) for p in data.get("periods", []))
except Exception as exc:
print(f"Weather fetch failed: {exc}")
return None
def generate_reading(self, sensor_name: str = "forecast_rain") -> Dict[str, Any]:
rainfall = self.fetch_rainfall_mm(hours=3)
return {
"sensor_name": sensor_name,
"timestamp": time.time(),
"value": rainfall if rainfall is not None else -1.0,
"unit": "mm",
}
Irrigation Control Rules
The core logic is simple: if soil moisture is below 30 % and no significant rain is forecast,
open the valve for 60 seconds. We express this as two RuleConfig objects in the
Pipeline.
from pyvorin_edge.pipeline import Pipeline, WindowConfig, RuleConfig
from pyvorin_edge.sensors import Sensor, SensorType, SensorReading
import time
pipeline = Pipeline(name="precision_agriculture")
pipeline.add_sensor(
Sensor(name="soil_moisture", sensor_type=SensorType.GENERIC, unit="%",
normal_range=(10.0, 60.0), location="Field_A")
)
pipeline.add_sensor(
Sensor(name="ambient_temp", sensor_type=SensorType.TEMPERATURE, unit="C",
normal_range=(5.0, 45.0), location="Field_A")
)
pipeline.add_sensor(
Sensor(name="light_lux", sensor_type=SensorType.GENERIC, unit="lux",
normal_range=(0.0, 100_000.0), location="Field_A")
)
# 15-minute rolling window for soil moisture trend
pipeline.add_window(
WindowConfig(duration_seconds=900.0, sensor_name="soil_moisture", window_type="rolling")
)
# Irrigation trigger rule
pipeline.add_rule(
RuleConfig(
name="irrigate",
condition_expr="ctx.value < 30.0",
severity="info",
cooldown_seconds=900.0,
action=lambda ctx: print(f"ACTUATE: Open valve for 60s (moisture={ctx.value}%)"),
)
)
# Frost protection rule
pipeline.add_rule(
RuleConfig(
name="frost_alert",
condition_expr="ctx.value < 2.0",
severity="critical",
cooldown_seconds=3600.0,
)
)
Solar Power Considerations
Continuous polling drains the battery overnight. Use duty cycling and the
SystemMetrics API in edge_runtime/pyv_edge_agent/health_monitor/metrics.py
to monitor thermal throttling under direct sun.
from pyv_edge_agent.health_monitor.metrics import SystemMetrics
metrics = SystemMetrics()
snapshot = metrics.snapshot()
print(f"CPU: {snapshot.cpu_percent:.1f}%")
print(f"RAM: {snapshot.ram_percent:.1f}%")
print(f"SoC temp: {snapshot.thermal_celsius:.1f}°C")
if snapshot.thermal_celsius and snapshot.thermal_celsius > 70.0:
print("Thermal throttling risk — reduce polling frequency.")
Fleet Deployment
When managing dozens of field stations, use BatchCostModel from
edge_sdk/pyvorin_edge/cost_model.py to estimate fleet-wide cloud costs.
from pyvorin_edge.cost_model import CostModel, TrafficModel, BatchCostModel
models = []
for station_id in range(1, 11):
traffic = TrafficModel(
properties=1,
sensors_per_property=3,
readings_per_sensor_per_day=96, # every 15 min
raw_payload_bytes=64,
edge_summaries_per_sensor_per_day=96,
edge_payload_bytes=128,
)
models.append(CostModel(traffic))
fleet = BatchCostModel(models)
print(f"Fleet size: {len(models)}")
print(f"Total raw cost: £{fleet.total_raw_cost():.2f}/month")
print(f"Total edge cost: £{fleet.total_edge_cost():.2f}/month")
print(f"Fleet savings: £{fleet.total_cost_savings():.2f}/month")
Complete Working Script
Save the following as precision_agriculture.py. It integrates the weather adapter,
pipeline, solar metrics check, and fleet cost model in one runnable file.
#!/usr/bin/env python3
"""Precision Agriculture — complete working example."""
import time
from pyvorin_edge.pipeline import Pipeline, WindowConfig, RuleConfig
from pyvorin_edge.sensors import Sensor, SensorType, SensorReading
from pyv_edge_agent.health_monitor.metrics import SystemMetrics
from pyvorin_edge.cost_model import CostModel, TrafficModel, BatchCostModel
class WeatherHTTPAdapter:
"""Minimal weather adapter for rainfall forecast."""
def __init__(self, api_url: str, api_key: str, lat: float, lon: float):
self.api_url = api_url
self.api_key = api_key
self.lat = lat
self.lon = lon
def fetch_rainfall_mm(self, hours: int = 3) -> float:
import json
import urllib.request
url = (
f"{self.api_url}/forecast?lat={self.lat}&lon={self.lon}"
f"&hours={hours}&apikey={self.api_key}"
)
try:
with urllib.request.urlopen(url, timeout=10) as resp:
data = json.loads(resp.read().decode("utf-8"))
return sum(p.get("rain_mm", 0.0) for p in data.get("periods", []))
except Exception as exc:
print(f"Weather fetch failed: {exc}")
return 0.0
def main():
pipeline = Pipeline(name="precision_agriculture")
pipeline.add_sensor(
Sensor(name="soil_moisture", sensor_type=SensorType.GENERIC, unit="%",
normal_range=(10.0, 60.0), location="Field_A")
)
pipeline.add_sensor(
Sensor(name="ambient_temp", sensor_type=SensorType.TEMPERATURE, unit="C",
normal_range=(5.0, 45.0), location="Field_A")
)
pipeline.add_sensor(
Sensor(name="light_lux", sensor_type=SensorType.GENERIC, unit="lux",
normal_range=(0.0, 100_000.0), location="Field_A")
)
pipeline.add_window(
WindowConfig(duration_seconds=900.0, sensor_name="soil_moisture", window_type="rolling")
)
pipeline.add_rule(
RuleConfig(
name="irrigate",
condition_expr="ctx.value < 30.0",
severity="info",
cooldown_seconds=900.0,
action=lambda ctx: print(f"ACTUATE: Open valve for 60s (moisture={ctx.value}%)"),
)
)
pipeline.add_rule(
RuleConfig(
name="frost_alert",
condition_expr="ctx.value < 2.0",
severity="critical",
cooldown_seconds=3600.0,
)
)
# Simulate readings
now = time.time()
readings = [
SensorReading(sensor_name="soil_moisture", timestamp=now, value=25.0, unit="%"),
SensorReading(sensor_name="ambient_temp", timestamp=now, value=8.0, unit="C"),
SensorReading(sensor_name="light_lux", timestamp=now, value=45000.0, unit="lux"),
]
result = pipeline.run(readings)
print(f"Events: {len(result.events)}")
for ev in result.events:
print(f" [{ev.severity}] {ev.rule_name}")
# Weather enrichment
weather = WeatherHTTPAdapter(
api_url="https://api.weather.example.com/v1",
api_key="demo_key",
lat=51.5,
lon=-0.1,
)
rain = weather.fetch_rainfall_mm(hours=3)
print(f"Forecast rain (next 3h): {rain} mm")
# System metrics
metrics = SystemMetrics()
snapshot = metrics.snapshot()
print(f"SoC temp: {snapshot.thermal_celsius}°C")
# Fleet cost model
models = []
for _ in range(10):
traffic = TrafficModel(
properties=1, sensors_per_property=3,
readings_per_sensor_per_day=96,
raw_payload_bytes=64,
edge_summaries_per_sensor_per_day=96,
edge_payload_bytes=128,
)
models.append(CostModel(traffic))
fleet = BatchCostModel(models)
print(f"Fleet savings: £{fleet.total_cost_savings():.2f}/month")
if __name__ == "__main__":
main()
Summary
You now have a precision-agriculture pipeline that reads soil, temperature, and light sensors; integrates an external weather API via a custom HTTP adapter; actuates irrigation valves; monitors thermal state under solar load; and models fleet-wide cloud costs. In production, seal the electronics in an IP67 enclosure, use a switched-mode power supply with LVD, and deploy the same bundle to every station using the fleet management tools.