农业的AI智能灌溉：从土壤传感器到模型驱动的水肥决策

农业的AI智能灌溉：从土壤传感器到模型驱动的水肥决策

一、场景痛点与技术挑战

传统农业灌溉依赖人工经验判断。灌溉时机、水量、施肥比例全靠直觉。数据缺失导致决策滞后和资源浪费。土壤湿度波动大，人工巡检频率不够。施肥过量引发土壤板结和地下水污染。施肥不足则作物产量直线下滑。

核心痛点有三个：一是数据采集维度单一。仅靠肉眼无法量化土壤氮磷钾含量。二是决策模型缺失。灌溉量和施肥比例缺乏科学依据。三是执行环节延迟。从发现问题到开阀灌溉耗时过长。

技术挑战更不容忽视。田间传感器网络需应对恶劣环境。高温、高湿、雷击都是硬件杀手。LoRa通信在农田远距离传输受限。边缘设备算力不足，模型推理慢。水肥一体化执行器精度要求高。阀门开度0.1%的偏差就影响产量。

二、核心原理与架构设计

智能灌溉系统分三层架构。

感知层负责数据采集。土壤湿度传感器、温度传感器、EC电导率传感器。pH传感器、氮磷钾离子选择性电极。气象站采集降雨量、风速、光照强度。所有传感器通过LoRaWAN上报边缘网关。

决策层是系统核心。边缘网关做数据清洗和特征提取。云端训练灌溉决策模型。模型输入包含土壤参数、气象预报、作物生长阶段。模型输出是灌溉量和施肥配方。

执行层负责落地。水肥一体化控制器接收决策指令。电磁阀调节灌溉管道流量。计量泵控制肥料注入比例。执行结果通过传感器反馈闭环。

灌溉决策模型采用多目标优化。目标函数最小化水资源消耗。约束条件保证土壤湿度在目标区间。施肥配方根据作物阶段动态调整。苗期偏氮肥，花期偏磷钾肥。

三、生产级代码实现

传感器数据采集与清洗

import numpy as np
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class SoilReading:
timestamp: datetime
moisture: float # 土壤湿度 %
temperature: float # 土壤温度 ℃
ec: float # 电导率 mS/cm
ph: float # pH值
nitrogen: float # 氮含量 mg/kg
phosphorus: float # 磷含量 mg/kg
potassium: float # 钾含量 mg/kg

class SoilDataCleaner:
"""土壤传感器数据清洗器"""

# 传感器物理量合理范围
RANGES = {
'moisture': (0.0, 100.0),
'temperature': (-10.0, 60.0),
'ec': (0.0, 5000.0),
'ph': (0.0, 14.0),
'nitrogen': (0.0, 500.0),
'phosphorus': (0.0, 200.0),
'potassium': (0.0, 800.0),
}

def __init__(self, window_size: int = 5):
self.window_size = window_size
self.history: dict[str, list[float]] = {}

def validate_range(self, field: str, value: float) -> bool:
low, high = self.RANGES[field]
return low <= value <= high

def median_filter(self, field: str, value: float) -> float:
"""中值滤波去除突发噪声"""
self.history.setdefault(field, []).append(value)
buf = self.history[field][-self.window_size:]
return float(np.median(buf))

def clean(self, reading: SoilReading) -> SoilReading | None:
"""清洗单条传感器数据"""
fields = {
'moisture': reading.moisture,
'temperature': reading.temperature,
'ec': reading.ec,
'ph': reading.ph,
'nitrogen': reading.nitrogen,
'phosphorus': reading.phosphorus,
'potassium': reading.potassium,
}

cleaned = {}
for field, raw_val in fields.items():
if not self.validate_range(field, raw_val):
# 超出物理范围，丢弃该条数据
return None
cleaned[field] = self.median_filter(field, raw_val)

return SoilReading(
timestamp=reading.timestamp,
moisture=cleaned['moisture'],
temperature=cleaned['temperature'],
ec=cleaned['ec'],
ph=cleaned['ph'],
nitrogen=cleaned['nitrogen'],
phosphorus=cleaned['phosphorus'],
potassium=cleaned['potassium'],
)

灌溉决策模型

from enum import Enum
from typing import Tuple

class CropStage(Enum):
SEEDLING = "苗期" # 偏氮肥
GROWING = "生长期" # 偏氮磷
FLOWERING = "花期" # 偏磷钾
FRUITING = "果期" # 偏钾肥
HARVEST = "收获期" # 减量

class IrrigationModel:
"""基于多目标优化的灌溉决策模型"""

# 各阶段土壤湿度目标区间 (%)
MOISTURE_TARGETS = {
CropStage.SEEDLING: (40, 55),
CropStage.GROWING: (45, 65),
CropStage.FLOWERING: (50, 70),
CropStage.FRUITING: (55, 75),
CropStage.HARVEST: (40, 55),
}

# 各阶段水肥配方基准 (N:P:K 比例)
FERTILIZER_RATIOS = {
CropStage.SEEDLING: (3, 1, 1),
CropStage.GROWING: (2, 2, 1),
CropStage.FLOWERING: (1, 3, 3),
CropStage.FRUITING: (1, 1, 4),
CropStage.HARVEST: (1, 1, 1),
}

def __init__(self, field_area: float, crop_stage: CropStage):
self.field_area = field_area # 亩
self.crop_stage = crop_stage

def calc_irrigation_volume(
self,
current_moisture: float,
forecast_rain_mm: float,
) -> float:
"""计算灌溉水量 (升)"""
low, high = self.MOISTURE_TARGETS[self.crop_stage]
target = (low + high) / 2

if current_moisture >= low:
# 湿度充足，仅补充预报降雨缺口
deficit = max(0, target – current_moisture – forecast_rain_mm * 0.3)
else:
# 湿度不足，需要紧急灌溉
deficit = target – current_moisture – forecast_rain_mm * 0.3

# 每亩每1%湿度 deficit ≈ 6.67升水
volume_per_acre = deficit * 6.67
total_volume = volume_per_acre * self.field_area
return max(0.0, round(total_volume, 1))

def calc_fertilizer_dose(
self,
soil_n: float,
soil_p: float,
soil_k: float,
) -> Tuple[float, float, float]:
"""计算施肥量 (kg)"""
n_ratio, p_ratio, k_ratio = self.FERTILIZER_RATIOS[self.crop_stage]

# 目标土壤养分基准值
N_TARGET, P_TARGET, K_TARGET = 120.0, 40.0, 150.0

n_deficit = max(0, N_TARGET – soil_n)
p_deficit = max(0, P_TARGET – soil_p)
k_deficit = max(0, K_TARGET – soil_k)

# 按配方比例分配施肥量
total_deficit = n_deficit + p_deficit + k_deficit
if total_deficit <= 0:
return (0.0, 0.0, 0.0)

base_dose = total_deficit * 0.01 * self.field_area
n_dose = round(base_dose * n_ratio / (n_ratio + p_ratio + k_ratio), 2)
p_dose = round(base_dose * p_ratio / (n_ratio + p_ratio + k_ratio), 2)
k_dose = round(base_dose * k_ratio / (n_ratio + p_ratio + k_ratio), 2)

return (n_dose, p_dose, k_dose)

def decide(
self,
reading: SoilReading,
forecast_rain_mm: float,
) -> dict:
"""输出完整灌溉决策"""
volume = self.calc_irrigation_volume(
reading.moisture, forecast_rain_mm
)
n_dose, p_dose, k_dose = self.calc_fertilizer_dose(
reading.nitrogen, reading.phosphorus, reading.potassium
)
return {
"irrigation_liters": volume,
"fertilizer_kg": {"N": n_dose, "P": p_dose, "K": k_dose},
"stage": self.crop_stage.value,
"current_moisture": reading.moisture,
"target_moisture": self.MOISTURE_TARGETS[self.crop_stage],
}

执行器控制接口

import asyncio
import logging

logger = logging.getLogger("irrigation_executor")

class ValveController:
"""电磁阀控制器"""

MIN_OPENING = 0.0 # 全关
MAX_OPENING = 100.0 # 全开

def __init__(self, valve_id: str, max_flow_lpm: float):
self.valve_id = valve_id
self.max_flow_lpm = max_flow_lpm # 最大流量 升/分钟
self.current_opening = 0.0

async def set_opening(self, percent: float) -> float:
"""设置阀门开度百分比"""
percent = max(self.MIN_OPENING, min(self.MAX_OPENING, percent))
self.current_opening = percent
logger.info(
f"阀门 {self.valve_id} 开度设为 {percent:.1f}%"
)
return percent

def estimate_flow(self) -> float:
"""估算当前流量"""
return self.max_flow_lpm * (self.current_opening / 100.0)

class FertilizerPump:
"""计量泵控制器"""

def __init__(self, pump_id: str, max_rate_mlpm: float):
self.pump_id = pump_id
self.max_rate_mlpm = max_rate_mlpm
self.current_rate = 0.0

async def inject(self, volume_ml: float, rate_mlpm: float) -> None:
"""注入指定体积肥料"""
rate_mlpm = min(rate_mlpm, self.max_rate_mlpm)
duration_min = volume_ml / rate_mlpm
self.current_rate = rate_mlpm
logger.info(
f"泵 {self.pump_id}: 注入 {volume_ml:.1f}ml, "
f"速率 {rate_mlpm:.1f}ml/min, "
f"耗时 {duration_min:.1f}min"
)
await asyncio.sleep(0.1) # 模拟执行延迟

class IrrigationExecutor:
"""灌溉执行器：协调阀门与计量泵"""

def __init__(self, valve: ValveController, pump: FertilizerPump):
self.valve = valve
self.pump = pump

async def execute(self, decision: dict) -> dict:
"""执行灌溉决策"""
liters = decision["irrigation_liters"]
fert = decision["fertilizer_kg"]

if liters <= 0 and all(v <= 0 for v in fert.values()):
logger.info("无需灌溉施肥，决策跳过")
return {"status": "skipped"}

# 计算阀门开度与灌溉时长
if liters > 0:
flow_lpm = self.valve.max_flow_lpm * 0.6 # 60%开度
opening = 60.0
duration_min = liters / flow_lpm
await self.valve.set_opening(opening)

# 施肥量转ml (1kg ≈ 1000ml, 密度近似)
total_ml = sum(fert.values()) * 1000
if total_ml > 0:
rate = self.pump.max_rate_mlpm * 0.5
await self.pump.inject(total_ml, rate)

return {
"status": "executed",
"valve_opening": self.valve.current_opening,
"flow_lpm": self.valve.estimate_flow(),
"duration_min": round(duration_min, 1) if liters > 0 else 0,
}

四、性能优化与工程实践

传感器采样频率需平衡精度和功耗。土壤湿度每30分钟采样一次即可。温度变化慢，每小时采集一次。EC和pH变化更慢，每2小时足够。降低采样频率直接减少LoRa传输次数。节点电池寿命从3个月延长到6个月。

边缘网关的数据清洗用中值滤波。比均值滤波更能抵抗突发噪声。窗口大小设5，兼顾平滑度和响应速度。超出物理范围的值直接丢弃不做修正。避免脏数据污染模型输入。

模型推理优化关注两个方面。一是模型轻量化。决策模型用规则引擎+线性回归组合。避免深度模型在边缘设备上的推理瓶颈。二是缓存策略。相同输入组合的决策结果缓存5分钟。传感器数据变化缓慢时避免重复计算。

水肥执行器安全机制必须完备。灌溉上限锁定，单次不超过亩均最大量。施肥上限锁定，防止过量注入。紧急停止接口，异常传感器数据触发停机。执行结果5分钟内通过传感器反馈验证。

日志和监控是运维基础。每条决策记录完整输入参数和输出结果。执行器状态实时上报云端Dashboard。异常告警：阀门未响应、泵超时、传感器断连。

五、总结与技术提炼