"Tutorial: Precision Agriculture and Smart Irrigation"

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.