1. 基于土壤传感器的精准灌溉决策系统

基于土壤传感器的精准灌溉决策系统 学习卡片

任务说明

构建一个精准灌溉决策系统:读取土壤湿度传感器数值,结合作物适宜湿度区间,自动生成灌溉控制指令(开启/关闭/延时),并将决策结果与节水效果记录到本地日志。系统需在无真实硬件条件下可完整运行与验证。

系统设计

系统由三个模块组成:传感器数据模拟器(生成可复现的土壤湿度序列)、灌溉决策引擎(基于阈值区间与滞回比较输出控制指令)、日志记录器(记录每次决策及累计节水量)。决策逻辑采用双阈值滞回控制:当湿度低于下限lower时开启灌溉;高于上限upper时停止灌溉;介于两者之间时保持上一状态,避免频繁开关。

土壤湿度模拟器 生成湿度序列 (可复现随机) 灌溉决策引擎 滞回比较控制 输出ON/OFF/HOLD 日志记录器 记录决策与节水 湿度值 控制指令 关键参数与数据流 • 土壤湿度区间:0~100%(体积含水量) • 作物适宜区间:lower=30%(开启灌溉),upper=60%(停止灌溉) • 采样周期:每10分钟一次;灌溉时长:单次开启持续30分钟 • 节水计算:当湿度≥lower且未灌溉时,按灌溉定额的30%计节水

代码实现


import random
import json
from datetime import datetime, timedelta

class SoilMoistureSimulator:
    """土壤湿度模拟器:生成可复现的湿度序列"""
    def __init__(self, seed=42, base=45.0, amplitude=20.0, noise=5.0):
        self.rng = random.Random(seed)
        self.base = base
        self.amplitude = amplitude
        self.noise = noise
        self.t = 0

    def read(self):
        """模拟一次湿度读取,返回0~100之间的浮点数"""
        # 模拟昼夜蒸发与随机波动
        daily_cycle = self.amplitude * (self.t % 24) / 24.0
        value = self.base - daily_cycle + self.rng.uniform(-self.noise, self.noise)
        self.t += 1
        return max(0.0, min(100.0, value))

class IrrigationController:
    """灌溉决策引擎:滞回比较控制"""
    def __init__(self, lower=30.0, upper=60.0, irrigation_duration=30):
        self.lower = lower
        self.upper = upper
        self.irrigation_duration = irrigation_duration  # 分钟
        self.state = "OFF"  # 当前灌溉状态
        self.remaining_time = 0  # 剩余灌溉分钟数

    def decide(self, moisture):
        """根据当前湿度返回控制指令: ON/OFF/HOLD"""
        if self.state == "OFF":
            if moisture < self.lower:
                self.state = "ON"
                self.remaining_time = self.irrigation_duration
                return "ON"
            else:
                return "OFF"
        else:  # state == "ON"
            if moisture >= self.upper or self.remaining_time <= 0:
                self.state = "OFF"
                self.remaining_time = 0
                return "OFF"
            else:
                self.remaining_time -= 10  # 每次决策间隔10分钟
                return "ON"

class IrrigationLogger:
    """日志记录器:记录决策与节水统计"""
    def __init__(self):
        self.records = []
        self.total_irrigation_minutes = 0
        self.total_water_saved = 0.0  # 单位:立方米(假设喷头流量0.5m³/10min)

    def log(self, timestamp, moisture, action, state):
        """记录一条决策日志"""
        record = {
            "timestamp": timestamp.strftime("%Y-%m-%d %H:%M"),
            "moisture": round(moisture, 2),
            "action": action,
            "state": state
        }
        self.records.append(record)

        # 统计灌溉时长与节水量
        if action == "ON":
            self.total_irrigation_minutes += 10
        elif action == "OFF" and state == "OFF" and moisture >= 30:
            # 湿度在适宜区间且未灌溉,视为节水
            self.total_water_saved += 0.5 * 0.3  # 0.5m³/10min * 30%节水率

    def save(self, filename="irrigation_log.json"):
        """保存日志到JSON文件"""
        output = {
            "total_irrigation_minutes": self.total_irrigation_minutes,
            "total_water_saved_m3": round(self.total_water_saved, 2),
            "records": self.records
        }
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(output, f, ensure_ascii=False, indent=2)
        return output

def run_simulation(hours=24, interval_minutes=10):
    """运行完整模拟:24小时,每10分钟采样一次"""
    sim = SoilMoistureSimulator(seed=42)
    controller = IrrigationController(lower=30, upper=60)
    logger = IrrigationLogger()

    start_time = datetime(2025, 1, 1, 0, 0)
    steps = int(hours * 60 / interval_minutes)

    for i in range(steps):
        moisture = sim.read()
        action = controller.decide(moisture)
        current_time = start_time + timedelta(minutes=i * interval_minutes)
        logger.log(current_time, moisture, action, controller.state)

    return logger.save()

if __name__ == "__main__":
    result = run_simulation()
    print(f"模拟完成:总灌溉时长 {result['total_irrigation_minutes']} 分钟")
    print(f"总节水量:{result['total_water_saved_m3']} 立方米")
    print(f"共记录 {len(result['records'])} 条决策日志")

运行与验证

运行方式:将上述代码保存为irrigation_system.py,在Python 3.6+环境执行python irrigation_system.py。程序将自动运行24小时模拟(每10分钟一次决策),生成irrigation_log.json日志文件。

验证方法:

  1. 检查控制台输出:应显示模拟完成信息,总灌溉时长与总节水量为数值结果;
  2. 打开生成的irrigation_log.json,确认包含144条记录(24小时×6次/小时);
  3. 抽查记录:当湿度值低于30时,对应action应为"ON";当湿度高于60时,action应为"OFF";介于30~60之间时,action保持上一状态(HOLD逻辑通过state字段体现);
  4. 验证节水统计合理性:总节水量应大于0(因为模拟数据中湿度经常处于30~60区间且未灌溉)。

判定标准:日志文件结构完整、决策逻辑与湿度值对应关系正确、节水统计为正数,即判定系统实现成功。