尧图网站建设 
简介一份基于Python的蔬菜大棚管理系统设计源码面向农业物联网方向的开发者、相关专业学生及需要搭建智能监控与管理平台的个人。系统结合Django后端与Web前端覆盖数据管理、监控展示等模块能帮助使用者快速理解完整Web应用的工程化组织方式。压缩包共81个文件、约3.29MB包含20个Python脚本、14个SCSS与14个LESS样式文件、6个CSS样式文件、5个JavaScript脚本及多种字体图标资源同时附带依赖清单、版本控制配置和说明文档等工程文件目录结构清晰。目前已有117人学习下载适合作为课程设计、毕业设计或项目练手的参考源码。通过研读源码可掌握Django项目配置、MVT分层逻辑、MQTT接入、静态资源管理以及依赖部署等关键技能为后续二次开发或自主构建农业管理系统打下扎实基础。1. 基于Python的蔬菜大棚管理系统到底在管什么一栋连栋大棚里装了三四十个温湿度传感器十几个卷帘电机、水泵和风机的控制柜如果全靠人工巡检和手按开关夏天中午一个传感器读数失控就可能让棚内温度冲到40℃以上。蔬菜大棚管理系统要解决的就是把分散的传感器、控制器和数据记录收拢到一套可配置的软件里让“采集—判断—控制—告警”这条链路变成可编程的自动化流程。用Python做这件事的优势在于生态完整采集层有Modbus、MQTT、串口协议库处理层有NumPy、Pandas做曲线分析展示层有FastAPI/Flask加前端图表而且源码结构清晰适合后续按品种、按季节调整阈值。这套系统适合有单片机或PLC基础、想用一个统一语言把所有设备接起来的团队也适合农业物联网从业者拿来做原型验证。下面从数据模型、采集逻辑、调度部署、Web化交付四个层面讲清一套可复现的实现路线。2. 数据模型设计把大棚设备和传感器映射成Python对象2.1 为什么用Python做这类管理系统农业现场的设备协议五花八门同一栋大棚里可能既有Modbus RTU的温湿度变送器又有走TCP的卷帘控制器还有通过继电器开关控制的补光灯。传统组态软件能画界面但扩展规则死板纯单片机方案改一次需求就要重新烧录程序。Python作为胶水语言可以在驱动层面用pymodbus、paho-mqtt分别对接在业务层面用统一的对象模型屏蔽底层差异。管理系统的核心不是“读传感器”这个动作而是“把读到的值变成可查询、可追溯、可触发控制”的数据资产。所以第一件事是把物理世界抽象成几张关系表。常见做法是定义四个核心实体传感器点Sensor、设备控制点Device、采集数据SensorData、告警规则AlertRule。传感器点描述“棚1东侧1.5m高处有一个型号为SHT30的温湿度探头”设备控制点描述“棚1东侧风机对应的继电器通道”采集数据是时间序列告警规则则把阈值和关联设备绑定。用Python的SQLAlchemy ORM可以把这些关系直接写成类运行时通过迁移工具建表。2.2 核心数据表结构与字段说明下面是一份适合中小规模大棚500个测点以内的表结构设计我一般用MySQL的InnoDB引擎如果现场没有数据库服务也可以直接切换SQLite跑原型。表名关键字段用途说明sensorsid, name, location, sensor_type, unit, modbus_addr, enabled传感器点位信息modbus_addr存从站地址devicesid, name, device_type, gpio_channel, status, last_ctrl_time风机、卷帘、水泵等可控设备sensor_dataid, sensor_id, value, data_time, is_valid采集的历史数据is_valid标记异常值alert_rulesid, sensor_id, device_id, condition_type, threshold, action_type规则如温度高于35℃则启动风机alert_logsid, alert_rule_id, alert_time, trigger_value, handled告警事件记录用于追溯和统计在设计时需要留意几个容易踩的细节传感器读数和设备控制是两类数据不要混在一张表里否则后面做时序分析时索引会很乱data_time必须带时区最好统一存UTC展示时再转北京时区is_valid字段不是可选项室外传感器经常被鸟粪糊住或断线若没有异常标记统计均值会把0值当真值算进去。2.3 用SQLAlchemy定义ORM模型用SQLAlchemy定义上述模型非常直接通过relationship外键关联可以实现“查某条告警时带出传感器名称”的便捷操作from sqlalchemy import Column, Integer, Float, String, DateTime, Boolean, ForeignKey from sqlalchemy.orm import declarative_base, relationship from datetime import datetime Base declarative_base() class Sensor(Base): __tablename__ sensors id Column(Integer, primary_keyTrue) name Column(String(64), nullableFalse) # 传感器名称棚1东侧温湿度 location Column(String(128)) # 物理位置描述 sensor_type Column(String(32)) # 温度/湿度/光照/CO2 unit Column(String(16), default℃) modbus_addr Column(Integer, uniqueTrue) # Modbus从站地址避免重复 enabled Column(Boolean, defaultTrue) data_records relationship(SensorData, back_populatessensor) class Device(Base): __tablename__ devices id Column(Integer, primary_keyTrue) name Column(String(64)) device_type Column(String(32)) # 风机/卷帘/水泵/补光灯 gpio_channel Column(Integer) # 继电器或控制模块通道号 status Column(Boolean, defaultFalse) # 当前开关状态 last_ctrl_time Column(DateTime) trigger_rules relationship(AlertRule, back_populatesdevice) class SensorData(Base): __tablename__ sensor_data id Column(Integer, primary_keyTrue) sensor_id Column(Integer, ForeignKey(sensors.id)) value Column(Float) data_time Column(DateTime, defaultdatetime.utcnow) is_valid Column(Boolean, defaultTrue) sensor relationship(Sensor, back_populatesdata_records) class AlertRule(Base): __tablename__ alert_rules id Column(Integer, primary_keyTrue) sensor_id Column(Integer, ForeignKey(sensors.id)) device_id Column(Integer, ForeignKey(devices.id), nullableTrue) condition_type Column(String(20)) # gt/lt/between threshold Column(Float) threshold_high Column(Float, nullableTrue) action_type Column(String(20)) # notify/control_device代码里值得留意的是modbus_addr加了uniqueTrue这能防止两个传感器占用同一串口地址导致采集错乱。condition_type设计成字符串而不是硬编码多个布尔字段是为了后面扩展“湿度在30%到70%之间”这类区间规则。实际项目中我会额外加一个created_at字段给告警规则做审计这里为保持代码简洁省略了。3. 功能实现温湿度采集、设备联动与阈值告警源码3.1 用Modbus协议采集温湿度数据大棚现场最常见的传感器是通过Modbus RTU输出的变送器读取逻辑是给定从站地址、寄存器地址和字节顺序解析得到浮点值。可以使用pymodbus库也可以使用minimalmodbus前者支持设备多、后者更轻。下面以minimalmodbus为例实现一个可复用的采集函数import minimalmodbus import time def read_humiture_sensor(port: str, slave_addr: int, reg_temp: int 0, reg_hum: int 1): 读取温湿度传感器返回(temperature, humidity) instrument minimalmodbus.Instrument(port, slave_addr) instrument.serial.port port instrument.serial.baudrate 9600 instrument.serial.bytesize 8 instrument.serial.parity N instrument.serial.stopbits 1 instrument.serial.timeout 0.5 temperature instrument.read_register(reg_temp, number_of_decimals1, signedTrue) humidity instrument.read_register(reg_hum, number_of_decimals1, signedFalse) return temperature, humidity if __name__ __main__: # 示例读取地址为1的传感器需要先将USB转485插到电脑 try: temp, humi read_humiture_sensor(/dev/ttyUSB0, 1) print(ftemp{temp}℃, humidity{humi}%) except Exception as e: print(f[ERROR] 采集失败: {e})read_register方法里的number_of_decimals表示寄存器原始值要除以10还是100比如硬件返回260代表26.0℃就填1。signedTrue表示温度值可能是负数因为北方大棚冬天低温时寄存器会补码表示。实战中还会遇到字节序问题不同厂商的温湿度传感器寄存器高低字节顺序可能相反需要根据手册调整byteorder参数。3.2 设备联动当温度超过阈值自动启动风机有了数据后最简单也最稳妥的联动逻辑是“阈值比较直接控制继电器”。我在工程里会把规则判断独立成一个函数方便单元测试和后续接入规则引擎def check_rule_and_control(rule: AlertRule, sensor_value: float, device_ctl_func): 根据规则判断是否触发设备控制 rule.condition_type: gt(大于), lt(小于), between(区间) triggered False if rule.condition_type gt and sensor_value rule.threshold: triggered True elif rule.condition_type lt and sensor_value rule.threshold: triggered True elif rule.condition_type between and rule.threshold sensor_value rule.threshold_high: triggered True if triggered and rule.action_type control_device and rule.device_id: # 例如启动风机调用设备控制函数这里以写入GPIO高电平为例 device_ctl_func(rule.device_id, True) log_alert(rule.id, sensor_value, device_control) return True return False这段代码没有直接用if temperature 35写死在业务里而是把规则做成数据库记录这样非程序员也能在后台修改阈值。device_ctl_func是一个注入函数因为不同的控制器有不同的驱动方式。生产环境里要注意的是“死区”问题——如果温度在34.9℃和35.1℃之间反复波动风机会频繁启停容易烧坏接触器。常见的解决办法是在规则表里增加hysteresis字段比如当温度高于35℃开启风机只有低于32℃才关闭这样避免震荡。3.3 阈值告警发送钉钉机器人消息告警通知是管理系统最有感知价值的功能。如果只是采集数据而没人盯着大棚夜间升温慢一两小时可能不会造成大问题但突然断电后恢复送电时的温度骤变必须第一时间提醒。用钉钉机器人是最低成本的方案只需要一个Webhook URL和几十行代码import requests import json import time def send_dingtalk_alert(token: str, message: str): 通过钉钉机器人发送文本消息 url fhttps://oapi.dingtalk.com/robot/send?access_token{token} headers {Content-Type: application/json;charsetutf-8} payload { msgtype: text, text: { content: f[大棚告警] {message} } } try: resp requests.post(url, datajson.dumps(payload), headersheaders, timeout3) result resp.json() if result.get(errcode) ! 0: print(f发送失败: {result.get(errmsg)}) except Exception as e: print(f[ERROR] 钉钉消息发送异常: {e}) if __name__ __main__: send_dingtalk_alert( token你的access_token, messagef棚1温度38.5℃超过35℃报警值时间{time.strftime(%Y-%m-%d %H:%M:%S)} )注意钉钉机器人安全设置有“自定义关键词”和“加签”两种模式。如果使用加签方式需要在钉钉后台获取密钥并生成sign参数这里为了简洁只演示了不加签的明文模式。生产环境必须用加签因为Webhook一旦泄漏任何人都能向你的群里推送垃圾消息。requests.post发送后要检查返回的errcode0才表示成功但很多初学者只发不检查导致网站明明报错还一直重复发送。4. 从源码到可运行环境配置、数据库初始化与定时调度4.1 搭建Python环境与安装依赖无论项目源码是venv还是pipenv建议在部署时统一使用Python 3.10以上的虚拟环境因为新版本对pymodbus和sqlalchemy的兼容性更好。先创建虚拟环境再安装依赖依赖清单可以精简到以下几项python3 -m venv venv source venv/bin/activate pip install --upgrade pip pip install sqlalchemy2.0.23 pymysql minimalmodbus paho-mqtt apscheduler fastapi uvicorn requests pandas注意sqlalchemy不要安装在2.0以下版本新版ORM写法与1.4差异较大。pymysql是MySQL的纯Python驱动配合sqlalchemy使用无需额外装mysqlclient。如果现场没有MySQL可以把连接字符串改成sqlite:///greenhouse.db后续所有代码不用改就能跑通。4.2 初始化数据库并验证建表启动程序前需要先让ORM创建表并写入初始传感器和设备数据。下面这段脚本可以作为源码包里的init_db.pyfrom sqlalchemy import create_engine from models import Base, Sensor, Device def init_database(): # 按实际环境修改连接字符串 engine create_engine(mysqlpymysql://root:password127.0.0.1:3306/greenhouse?charsetutf8mb4) Base.metadata.create_all(engine) from sqlalchemy.orm import sessionmaker Session sessionmaker(bindengine) session Session() # 如果sensors表为空插入两个示例传感器 if session.query(Sensor).count() 0: session.add_all([ Sensor(name棚1东侧温湿度, location东侧1.5m, sensor_typeTH, unit℃/%, modbus_addr1, enabledTrue), Sensor(name棚1西侧温湿度, location西侧1.5m, sensor_typeTH, unit℃/%, modbus_addr2, enabledTrue), ]) session.commit() print([OK] 初始化传感器完成) if __name__ __main__: init_database()执行后可以先通过sqlalchemy的text查询确认表结构和数据如果忘记建库需要先执行CREATE DATABASE greenhouse CHARACTER SET utf8mb4。常见的坑是charset没设置成utf8mb4中文设备名写入时可能报“Incorrect string value”这一步很多新手会忽略。4.3 用APScheduler实现定时采集调度大棚管理系统的核心是无人值守的定时任务。我通常用APScheduler的BackgroundScheduler在独立进程里维护一个循环调度每30秒采集一次所有传感器每5分钟做一次规则检查。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger import time def collect_and_store(): print(f{time.strftime(%H:%M:%S)} 开始采集所有传感器) # 遍历所有enabled的传感器并读值写入sensor_data # 伪代码sensors session.query(Sensor).filter_by(enabledTrue).all() # 具体采集逻辑参考3.1节异常情况写入is_validFalse def check_all_rules(): print(f{time.strftime(%H:%M:%S)} 检查告警规则) # 从数据库查询最新一条sensor_data并按规则过滤 scheduler BackgroundScheduler(timezoneAsia/Shanghai) scheduler.add_job(collect_and_store, IntervalTrigger(seconds30), idcollect_job, max_instances1) scheduler.add_job(check_all_rules, IntervalTrigger(minutes5), idrule_job, max_instances1) scheduler.start() try: while True: time.sleep(10) except KeyboardInterrupt: scheduler.shutdown()在采集任务里加入max_instances1很关键如果前一次采集因串口卡住超过30秒还没完成调度器会跳过本次执行避免多个线程同时打开串口导致资源冲突。timezone参数要显式指定否则BackgroundScheduler默认用系统时区在部分云服务器上会因为UTC和北京时间偏差导致调度时间错乱。4.4 部署运行时的常见问题排查现象可能原因解决方式ModuleNotFoundError: No module named pymysql虚拟环境未激活或未安装依赖pip install pymysql确认用which python检查当前解释器ConnectionError: [Errno 111] Connection refusedMySQL未启动或端口不对service mysql status检查3306端口监听状态SerialException: Port is already open串口被其他进程占用或未释放排查是否有多个采集进程或使用lsof /dev/ttyUSB0查看占用采集数据全是0或固定值传感器地址错误或Modbus值不刷新使用Modbus工具直接读寄存器排除协议解析问题定时任务不触发时区设置错误或调度器被垃圾回收把scheduler.start()放到模块顶层或者在主进程持有引用5. 进阶技巧把源码快速Web化并提升数据可用性5.1 用FastAPI暴露REST接口如果只是后台采集和告警这套系统的价值还不完整。大棚管理员希望在手机上看实时数据、远程手动控制设备。常见做法是加一层FastAPI服务把数据模型映射成/api/current和/api/history接口from fastapi import FastAPI from sqlalchemy.orm import sessionmaker from datetime import datetime, timedelta app FastAPI(titleVegetable Greenhouse API) app.get(/api/current) def get_current_values(): 返回每个传感器最新的有效数据 db sessionmaker(bindengine)() rows db.execute( text( SELECT s.name, sd.value, sd.data_time FROM sensor_data sd JOIN sensors s ON s.id sd.sensor_id WHERE sd.is_valid TRUE AND sd.data_time (SELECT MAX(data_time) FROM sensor_data sd2 WHERE sd2.sensor_id s.id) ) ).fetchall() return [{name: r[0], value: r[1], time: r[2].isoformat()} for r in rows]这条SQL使用了子查询取每个传感器最新一条数据虽然在大数据量下性能一般但对于大棚几百条数据量完全够用。为了减少重复查询压力我还会给sensor_data表加联合索引(sensor_id, data_time)这是做时序数据最常用的索引优化。5.2 ECharts展示历史曲线的坑前端页面如果用ECharts画温湿度曲线后端返回的数据格式需要提前组织好。建议在接口里直接返回epoch_time和value两个数组避免前端做时间格式化。如果出现曲线横轴时间混乱多半是后端返回了字符串时间且未指定timezone统一用data_time.timestamp()转成毫秒时间戳传给前端最稳妥。5.3 数据清洗处理断线重连和异常跳变最后给一个很实用但容易被忽略的技巧阈值告警时要给传感器读数增加“变化率”判断。大棚里温度不可能在1秒内从30℃跳到60℃如果出现这种跳变大概率是传感器漂移或接线接触不良。在采集写入前加一层过滤def is_reasonable_change(sensor_id: str, new_value: float, max_delta: float 8.0): 判断与上一次值的差值是否在合理范围内 last_row session.query(SensorData).filter_by(sensor_idsensor_id)\ .order_by(SensorData.data_time.desc()).first() if last_row is None: return True return abs(new_value - last_row.value) max_delta把这个函数放在告警规则之前不合理的值直接标记is_validFalse而不是写入后再由规则误触发。系统运行一段时间后可以把这些异常值单独导出用于排查传感器质量问题。另外建议每天凌晨自动执行一次“数据完整性检查”统计各传感器当天记录数与理论记录数的比例低于90%的传感器说明存在断线丢包需要纳入运维工单。这套系统的可靠性就藏在这些小的过滤和自检逻辑里。本文还有配套的精品资源点击获取 基于Python的蔬菜大棚管理系统：传感器采集与设备联动控制实战
简介一份基于Python的蔬菜大棚管理系统设计源码面向农业物联网方向的开发者、相关专业学生及需要搭建智能监控与管理平台的个人。系统结合Django后端与Web前端覆盖数据管理、监控展示等模块能帮助使用者快速理解完整Web应用的工程化组织方式。压缩包共81个文件、约3.29MB包含20个Python脚本、14个SCSS与14个LESS样式文件、6个CSS样式文件、5个JavaScript脚本及多种字体图标资源同时附带依赖清单、版本控制配置和说明文档等工程文件目录结构清晰。目前已有117人学习下载适合作为课程设计、毕业设计或项目练手的参考源码。通过研读源码可掌握Django项目配置、MVT分层逻辑、MQTT接入、静态资源管理以及依赖部署等关键技能为后续二次开发或自主构建农业管理系统打下扎实基础。1. 基于Python的蔬菜大棚管理系统到底在管什么一栋连栋大棚里装了三四十个温湿度传感器十几个卷帘电机、水泵和风机的控制柜如果全靠人工巡检和手按开关夏天中午一个传感器读数失控就可能让棚内温度冲到40℃以上。蔬菜大棚管理系统要解决的就是把分散的传感器、控制器和数据记录收拢到一套可配置的软件里让“采集—判断—控制—告警”这条链路变成可编程的自动化流程。用Python做这件事的优势在于生态完整采集层有Modbus、MQTT、串口协议库处理层有NumPy、Pandas做曲线分析展示层有FastAPI/Flask加前端图表而且源码结构清晰适合后续按品种、按季节调整阈值。这套系统适合有单片机或PLC基础、想用一个统一语言把所有设备接起来的团队也适合农业物联网从业者拿来做原型验证。下面从数据模型、采集逻辑、调度部署、Web化交付四个层面讲清一套可复现的实现路线。2. 数据模型设计把大棚设备和传感器映射成Python对象2.1 为什么用Python做这类管理系统农业现场的设备协议五花八门同一栋大棚里可能既有Modbus RTU的温湿度变送器又有走TCP的卷帘控制器还有通过继电器开关控制的补光灯。传统组态软件能画界面但扩展规则死板纯单片机方案改一次需求就要重新烧录程序。Python作为胶水语言可以在驱动层面用pymodbus、paho-mqtt分别对接在业务层面用统一的对象模型屏蔽底层差异。管理系统的核心不是“读传感器”这个动作而是“把读到的值变成可查询、可追溯、可触发控制”的数据资产。所以第一件事是把物理世界抽象成几张关系表。常见做法是定义四个核心实体传感器点Sensor、设备控制点Device、采集数据SensorData、告警规则AlertRule。传感器点描述“棚1东侧1.5m高处有一个型号为SHT30的温湿度探头”设备控制点描述“棚1东侧风机对应的继电器通道”采集数据是时间序列告警规则则把阈值和关联设备绑定。用Python的SQLAlchemy ORM可以把这些关系直接写成类运行时通过迁移工具建表。2.2 核心数据表结构与字段说明下面是一份适合中小规模大棚500个测点以内的表结构设计我一般用MySQL的InnoDB引擎如果现场没有数据库服务也可以直接切换SQLite跑原型。表名关键字段用途说明sensorsid, name, location, sensor_type, unit, modbus_addr, enabled传感器点位信息modbus_addr存从站地址devicesid, name, device_type, gpio_channel, status, last_ctrl_time风机、卷帘、水泵等可控设备sensor_dataid, sensor_id, value, data_time, is_valid采集的历史数据is_valid标记异常值alert_rulesid, sensor_id, device_id, condition_type, threshold, action_type规则如温度高于35℃则启动风机alert_logsid, alert_rule_id, alert_time, trigger_value, handled告警事件记录用于追溯和统计在设计时需要留意几个容易踩的细节传感器读数和设备控制是两类数据不要混在一张表里否则后面做时序分析时索引会很乱data_time必须带时区最好统一存UTC展示时再转北京时区is_valid字段不是可选项室外传感器经常被鸟粪糊住或断线若没有异常标记统计均值会把0值当真值算进去。2.3 用SQLAlchemy定义ORM模型用SQLAlchemy定义上述模型非常直接通过relationship外键关联可以实现“查某条告警时带出传感器名称”的便捷操作from sqlalchemy import Column, Integer, Float, String, DateTime, Boolean, ForeignKey from sqlalchemy.orm import declarative_base, relationship from datetime import datetime Base declarative_base() class Sensor(Base): __tablename__ sensors id Column(Integer, primary_keyTrue) name Column(String(64), nullableFalse) # 传感器名称棚1东侧温湿度 location Column(String(128)) # 物理位置描述 sensor_type Column(String(32)) # 温度/湿度/光照/CO2 unit Column(String(16), default℃) modbus_addr Column(Integer, uniqueTrue) # Modbus从站地址避免重复 enabled Column(Boolean, defaultTrue) data_records relationship(SensorData, back_populatessensor) class Device(Base): __tablename__ devices id Column(Integer, primary_keyTrue) name Column(String(64)) device_type Column(String(32)) # 风机/卷帘/水泵/补光灯 gpio_channel Column(Integer) # 继电器或控制模块通道号 status Column(Boolean, defaultFalse) # 当前开关状态 last_ctrl_time Column(DateTime) trigger_rules relationship(AlertRule, back_populatesdevice) class SensorData(Base): __tablename__ sensor_data id Column(Integer, primary_keyTrue) sensor_id Column(Integer, ForeignKey(sensors.id)) value Column(Float) data_time Column(DateTime, defaultdatetime.utcnow) is_valid Column(Boolean, defaultTrue) sensor relationship(Sensor, back_populatesdata_records) class AlertRule(Base): __tablename__ alert_rules id Column(Integer, primary_keyTrue) sensor_id Column(Integer, ForeignKey(sensors.id)) device_id Column(Integer, ForeignKey(devices.id), nullableTrue) condition_type Column(String(20)) # gt/lt/between threshold Column(Float) threshold_high Column(Float, nullableTrue) action_type Column(String(20)) # notify/control_device代码里值得留意的是modbus_addr加了uniqueTrue这能防止两个传感器占用同一串口地址导致采集错乱。condition_type设计成字符串而不是硬编码多个布尔字段是为了后面扩展“湿度在30%到70%之间”这类区间规则。实际项目中我会额外加一个created_at字段给告警规则做审计这里为保持代码简洁省略了。3. 功能实现温湿度采集、设备联动与阈值告警源码3.1 用Modbus协议采集温湿度数据大棚现场最常见的传感器是通过Modbus RTU输出的变送器读取逻辑是给定从站地址、寄存器地址和字节顺序解析得到浮点值。可以使用pymodbus库也可以使用minimalmodbus前者支持设备多、后者更轻。下面以minimalmodbus为例实现一个可复用的采集函数import minimalmodbus import time def read_humiture_sensor(port: str, slave_addr: int, reg_temp: int 0, reg_hum: int 1): 读取温湿度传感器返回(temperature, humidity) instrument minimalmodbus.Instrument(port, slave_addr) instrument.serial.port port instrument.serial.baudrate 9600 instrument.serial.bytesize 8 instrument.serial.parity N instrument.serial.stopbits 1 instrument.serial.timeout 0.5 temperature instrument.read_register(reg_temp, number_of_decimals1, signedTrue) humidity instrument.read_register(reg_hum, number_of_decimals1, signedFalse) return temperature, humidity if __name__ __main__: # 示例读取地址为1的传感器需要先将USB转485插到电脑 try: temp, humi read_humiture_sensor(/dev/ttyUSB0, 1) print(ftemp{temp}℃, humidity{humi}%) except Exception as e: print(f[ERROR] 采集失败: {e})read_register方法里的number_of_decimals表示寄存器原始值要除以10还是100比如硬件返回260代表26.0℃就填1。signedTrue表示温度值可能是负数因为北方大棚冬天低温时寄存器会补码表示。实战中还会遇到字节序问题不同厂商的温湿度传感器寄存器高低字节顺序可能相反需要根据手册调整byteorder参数。3.2 设备联动当温度超过阈值自动启动风机有了数据后最简单也最稳妥的联动逻辑是“阈值比较直接控制继电器”。我在工程里会把规则判断独立成一个函数方便单元测试和后续接入规则引擎def check_rule_and_control(rule: AlertRule, sensor_value: float, device_ctl_func): 根据规则判断是否触发设备控制 rule.condition_type: gt(大于), lt(小于), between(区间) triggered False if rule.condition_type gt and sensor_value rule.threshold: triggered True elif rule.condition_type lt and sensor_value rule.threshold: triggered True elif rule.condition_type between and rule.threshold sensor_value rule.threshold_high: triggered True if triggered and rule.action_type control_device and rule.device_id: # 例如启动风机调用设备控制函数这里以写入GPIO高电平为例 device_ctl_func(rule.device_id, True) log_alert(rule.id, sensor_value, device_control) return True return False这段代码没有直接用if temperature 35写死在业务里而是把规则做成数据库记录这样非程序员也能在后台修改阈值。device_ctl_func是一个注入函数因为不同的控制器有不同的驱动方式。生产环境里要注意的是“死区”问题——如果温度在34.9℃和35.1℃之间反复波动风机会频繁启停容易烧坏接触器。常见的解决办法是在规则表里增加hysteresis字段比如当温度高于35℃开启风机只有低于32℃才关闭这样避免震荡。3.3 阈值告警发送钉钉机器人消息告警通知是管理系统最有感知价值的功能。如果只是采集数据而没人盯着大棚夜间升温慢一两小时可能不会造成大问题但突然断电后恢复送电时的温度骤变必须第一时间提醒。用钉钉机器人是最低成本的方案只需要一个Webhook URL和几十行代码import requests import json import time def send_dingtalk_alert(token: str, message: str): 通过钉钉机器人发送文本消息 url fhttps://oapi.dingtalk.com/robot/send?access_token{token} headers {Content-Type: application/json;charsetutf-8} payload { msgtype: text, text: { content: f[大棚告警] {message} } } try: resp requests.post(url, datajson.dumps(payload), headersheaders, timeout3) result resp.json() if result.get(errcode) ! 0: print(f发送失败: {result.get(errmsg)}) except Exception as e: print(f[ERROR] 钉钉消息发送异常: {e}) if __name__ __main__: send_dingtalk_alert( token你的access_token, messagef棚1温度38.5℃超过35℃报警值时间{time.strftime(%Y-%m-%d %H:%M:%S)} )注意钉钉机器人安全设置有“自定义关键词”和“加签”两种模式。如果使用加签方式需要在钉钉后台获取密钥并生成sign参数这里为了简洁只演示了不加签的明文模式。生产环境必须用加签因为Webhook一旦泄漏任何人都能向你的群里推送垃圾消息。requests.post发送后要检查返回的errcode0才表示成功但很多初学者只发不检查导致网站明明报错还一直重复发送。4. 从源码到可运行环境配置、数据库初始化与定时调度4.1 搭建Python环境与安装依赖无论项目源码是venv还是pipenv建议在部署时统一使用Python 3.10以上的虚拟环境因为新版本对pymodbus和sqlalchemy的兼容性更好。先创建虚拟环境再安装依赖依赖清单可以精简到以下几项python3 -m venv venv source venv/bin/activate pip install --upgrade pip pip install sqlalchemy2.0.23 pymysql minimalmodbus paho-mqtt apscheduler fastapi uvicorn requests pandas注意sqlalchemy不要安装在2.0以下版本新版ORM写法与1.4差异较大。pymysql是MySQL的纯Python驱动配合sqlalchemy使用无需额外装mysqlclient。如果现场没有MySQL可以把连接字符串改成sqlite:///greenhouse.db后续所有代码不用改就能跑通。4.2 初始化数据库并验证建表启动程序前需要先让ORM创建表并写入初始传感器和设备数据。下面这段脚本可以作为源码包里的init_db.pyfrom sqlalchemy import create_engine from models import Base, Sensor, Device def init_database(): # 按实际环境修改连接字符串 engine create_engine(mysqlpymysql://root:password127.0.0.1:3306/greenhouse?charsetutf8mb4) Base.metadata.create_all(engine) from sqlalchemy.orm import sessionmaker Session sessionmaker(bindengine) session Session() # 如果sensors表为空插入两个示例传感器 if session.query(Sensor).count() 0: session.add_all([ Sensor(name棚1东侧温湿度, location东侧1.5m, sensor_typeTH, unit℃/%, modbus_addr1, enabledTrue), Sensor(name棚1西侧温湿度, location西侧1.5m, sensor_typeTH, unit℃/%, modbus_addr2, enabledTrue), ]) session.commit() print([OK] 初始化传感器完成) if __name__ __main__: init_database()执行后可以先通过sqlalchemy的text查询确认表结构和数据如果忘记建库需要先执行CREATE DATABASE greenhouse CHARACTER SET utf8mb4。常见的坑是charset没设置成utf8mb4中文设备名写入时可能报“Incorrect string value”这一步很多新手会忽略。4.3 用APScheduler实现定时采集调度大棚管理系统的核心是无人值守的定时任务。我通常用APScheduler的BackgroundScheduler在独立进程里维护一个循环调度每30秒采集一次所有传感器每5分钟做一次规则检查。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger import time def collect_and_store(): print(f{time.strftime(%H:%M:%S)} 开始采集所有传感器) # 遍历所有enabled的传感器并读值写入sensor_data # 伪代码sensors session.query(Sensor).filter_by(enabledTrue).all() # 具体采集逻辑参考3.1节异常情况写入is_validFalse def check_all_rules(): print(f{time.strftime(%H:%M:%S)} 检查告警规则) # 从数据库查询最新一条sensor_data并按规则过滤 scheduler BackgroundScheduler(timezoneAsia/Shanghai) scheduler.add_job(collect_and_store, IntervalTrigger(seconds30), idcollect_job, max_instances1) scheduler.add_job(check_all_rules, IntervalTrigger(minutes5), idrule_job, max_instances1) scheduler.start() try: while True: time.sleep(10) except KeyboardInterrupt: scheduler.shutdown()在采集任务里加入max_instances1很关键如果前一次采集因串口卡住超过30秒还没完成调度器会跳过本次执行避免多个线程同时打开串口导致资源冲突。timezone参数要显式指定否则BackgroundScheduler默认用系统时区在部分云服务器上会因为UTC和北京时间偏差导致调度时间错乱。4.4 部署运行时的常见问题排查现象可能原因解决方式ModuleNotFoundError: No module named pymysql虚拟环境未激活或未安装依赖pip install pymysql确认用which python检查当前解释器ConnectionError: [Errno 111] Connection refusedMySQL未启动或端口不对service mysql status检查3306端口监听状态SerialException: Port is already open串口被其他进程占用或未释放排查是否有多个采集进程或使用lsof /dev/ttyUSB0查看占用采集数据全是0或固定值传感器地址错误或Modbus值不刷新使用Modbus工具直接读寄存器排除协议解析问题定时任务不触发时区设置错误或调度器被垃圾回收把scheduler.start()放到模块顶层或者在主进程持有引用5. 进阶技巧把源码快速Web化并提升数据可用性5.1 用FastAPI暴露REST接口如果只是后台采集和告警这套系统的价值还不完整。大棚管理员希望在手机上看实时数据、远程手动控制设备。常见做法是加一层FastAPI服务把数据模型映射成/api/current和/api/history接口from fastapi import FastAPI from sqlalchemy.orm import sessionmaker from datetime import datetime, timedelta app FastAPI(titleVegetable Greenhouse API) app.get(/api/current) def get_current_values(): 返回每个传感器最新的有效数据 db sessionmaker(bindengine)() rows db.execute( text( SELECT s.name, sd.value, sd.data_time FROM sensor_data sd JOIN sensors s ON s.id sd.sensor_id WHERE sd.is_valid TRUE AND sd.data_time (SELECT MAX(data_time) FROM sensor_data sd2 WHERE sd2.sensor_id s.id) ) ).fetchall() return [{name: r[0], value: r[1], time: r[2].isoformat()} for r in rows]这条SQL使用了子查询取每个传感器最新一条数据虽然在大数据量下性能一般但对于大棚几百条数据量完全够用。为了减少重复查询压力我还会给sensor_data表加联合索引(sensor_id, data_time)这是做时序数据最常用的索引优化。5.2 ECharts展示历史曲线的坑前端页面如果用ECharts画温湿度曲线后端返回的数据格式需要提前组织好。建议在接口里直接返回epoch_time和value两个数组避免前端做时间格式化。如果出现曲线横轴时间混乱多半是后端返回了字符串时间且未指定timezone统一用data_time.timestamp()转成毫秒时间戳传给前端最稳妥。5.3 数据清洗处理断线重连和异常跳变最后给一个很实用但容易被忽略的技巧阈值告警时要给传感器读数增加“变化率”判断。大棚里温度不可能在1秒内从30℃跳到60℃如果出现这种跳变大概率是传感器漂移或接线接触不良。在采集写入前加一层过滤def is_reasonable_change(sensor_id: str, new_value: float, max_delta: float 8.0): 判断与上一次值的差值是否在合理范围内 last_row session.query(SensorData).filter_by(sensor_idsensor_id)\ .order_by(SensorData.data_time.desc()).first() if last_row is None: return True return abs(new_value - last_row.value) max_delta把这个函数放在告警规则之前不合理的值直接标记is_validFalse而不是写入后再由规则误触发。系统运行一段时间后可以把这些异常值单独导出用于排查传感器质量问题。另外建议每天凌晨自动执行一次“数据完整性检查”统计各传感器当天记录数与理论记录数的比例低于90%的传感器说明存在断线丢包需要纳入运维工单。这套系统的可靠性就藏在这些小的过滤和自检逻辑里。本文还有配套的精品资源点击获取