简介基于Python的智能花卉监控与控制系统源码包面向物联网开发者和智能农业爱好者解决花卉环境远程监控与自动控制问题。项目通过传感器读取温度、湿度、土壤湿度、光照强度和水位经GPIO控制灯光与浇水设备采用MQTT协议与服务器通信并支持定时任务和消息订阅是一套可落地的智能园艺方案。压缩包共54个文件以11个Python脚本为核心配34个txt说明文档、1个Markdown说明、4个PNG示意及1张JPG图片另有RAR附加包和Pyc编译文件整体仅876KB轻巧易用。资源还内置项目计划、QA检查、产品开发、配置管理等完整目录可帮助读者理解嵌入式项目的研发流程与文档规范目前已有70人浏览学习适合初次接触智能硬件的开发者参考。1. 智能花卉监控与控制系统拿到手先做什么节点、数据流与第一版形态一套基于 Python 的智能花卉监控与控制系统源码里往往有两块东西传感器轮询脚本和控制脚本。多数包能跑但改起来痛苦因为采集和决策写死在同一个 while True 里传感器一卡控制跟着停。我的处理习惯是先拆三条数据流传感器上报、控制决策、数据库与告警走旁路。拿到源码先读 drivers、mqtt、storage 三个目录。这套系统的价值在可观测的自动控制家庭阳台一路土壤湿度就能带动水泵花房环境再加温湿度、光照与继电器。适合想把树莓派零散脚本整理成工程的人也适合想用 Python 把传感器、消息队列、SQLite 串成 IoT 闭环的后端工程师。先跑通上报、下发、执行器回读再谈自动决策。2. 最小可运行底座Python 环境安装、源码目录与传感器驱动读取2.1 开发机与树莓派上的 Python 安装教程差异如果源码包里的代码要求跑在树莓派上先确认系统自带的 Python 版本。官方源里的 python3 版本可能低于依赖要求所以不要直接拿系统解释器裸跑。到设备上先做一次系统更新再装 venv 和 pipsudo apt update sudo apt install -y python3 python3-venv python3-pip python3 --version第一行命令是更新软件源索引第二行安装解释器、虚拟环境模块和包管理工具第三行确认版本。树莓派上我不太建议装 conda因为 ARM 下的预编译包不如 PyPI 全直接用系统 apt 提供的 python3-venv 更干净。装完后在项目根目录执行python3 -m venv .venv再激活虚拟环境装依赖。注意千万不要用sudo pip install否则后面用 systemd 开机启动时很容易出现“命令行能跑、服务起不来”的 ModuleNotFoundError。这里有一个非常常见的坑系统里同时存在/usr/bin/python3和.venv/bin/python写 systemd 服务时 ExecStart 直接写python app.py实际调用的是 PATH 里的旧全局解释器。我的建议是写绝对路径/opt/plantsense/.venv/bin/python从根源上避开解释器歧义。开发机上如果用 Windows装 Python 时记得勾选 Add to PATH并在终端用py -m venv .venv创建环境macOS 则先确认python3 --version拿到的是 Homebrew 版本而不是 Xcode 自带的老版本。2.2 源码目录怎么摆drivers、mqtt、storage 三层拿到一个压缩包源码我的第一件事是把目录结构改成按职责拆的方式。很多项目抱怨“能跑但不敢改”问题就在 config 散落各处、硬件操作和业务判断混在一起。推荐的最小结构是这样的plantsense/ ├── config/ │ └── settings.py # 阈值、GPIO、MQTT broker、数据库路径 ├── drivers/ │ ├── soil_moisture.py # 土壤湿度读取 │ └── dht11.py # 温湿度读取 ├── mqtt/ │ ├── publisher.py # 定时上报传感器数据 │ └── control_engine.py # 订阅数据、执行阈值决策 ├── storage/ │ └── database.py # SQLite 写入与查询 ├── web/ │ └── app.py # 局域网可视化 ├── tests/ │ └── test_thresholds.py └── requirements.txt这里每一层都有硬性约束drivers 目录只做硬件读取和单位换算不在文件里写“湿度低于 30 就开泵”mqtt 目录只处理消息发布、订阅和下发storage 目录不感知 GPIO。这样就算把 DHT11 换成 DHT20或者把 ADS1115 换成 ADS1015影响范围被限制在 drivers 单文件内。config/settings.py 单独拎出来是为了换环境时只改常量表不翻整棵树。tests/test_thresholds.py 是很多嵌入式源码包不会带的东西但它对这类系统价值很高。把传感器对象 Mock 成固定返回 25%验证控制引擎是否会发布“打开水泵”命令可以避免改动阈值时影响到真实硬件。先用这个结构把源码包重新摆一遍再往下看驱动比从主入口硬啃快得多。2.3 驱动土壤湿度与温湿度ADC 采样去抖与 DHT11 校验传感器接线是最容易反复返工的部分。以一个树莓派加 3 路传感器为例一个相对稳定的接法如下外设信号引脚地线说明土壤湿度模块 AOADS1115 AIN0GND模拟输出接 ADC不接 GPIO土壤湿度模块 DOGPIO 17GND数字输出板载电位器定阈值DHT11 DATAGPIO 4GND单总线数据线上拉到 3.3V继电器输入端GPIO 22/23GND分别控水泵与风机我一般不建议直接用土壤湿度模块的 DO 输出。板载电位器标定一次管不了一个月温湿度一变阈值就漂。读 AO 模拟量进 ADC然后把阈值放到 Python 配置里是更容易维护的做法。下面是一段 ADS1115 三通道读取并做中位数滤波的代码import time import board import busio import adafruit_ads1x15.ads1115 as ADS from adafruit_ads1x15.analog_in import AnalogIn i2c busio.I2C(board.SCL, board.SDA) ads ADS.ADS1115(i2c, gain1) # gain1 时量程约 4.096V channels [ AnalogIn(ads, ADS.P0), AnalogIn(ads, ADS.P1), AnalogIn(ads, ADS.P2), ] def read_soil_moisture(channel_index: int) - int: raw_samples [] for _ in range(5): # 采集 5 次丢弃尖峰 raw_samples.append(channels[channel_index].value) time.sleep(0.02) raw_samples.sort() raw raw_samples[len(raw_samples) // 2] percent max(0.0, min(100.0, 100.0 * (1.0 - raw / 26214.0))) return int(percent)逻辑说明ADS1115 是 16 位 ADCgain1 时满量程计数约 32767对应 4.096V。土壤湿度传感器的输出电压通常随湿度上升而下降所以用1 - raw / 26214换算成百分比其中 26214 是估算的干土计数值。这个数必须按你实际浸泡水和晾干土两组数据修正否则读出来可能长期在 0 或 100 附近。median filter 比平均值更抗尖峰代价是每次多约 0.1 秒土壤湿度这种慢变量完全可接受。DHT11 的坑主要在时序。使用 RPi.GPIO 做脉冲读取时通常要 root 权限而且不同驱动对 DHT11 的初始化时序要求不一致。我建议改用 Adafruit CircuitPython 的 DHT11 驱动它对时序做了封装读回后自己再校验一次校验和将湿度整数、湿度小数、温度整数、温度小数相加取低 8 位和校验字节比对。不一致就整包丢弃不参与控制。还要限定读取频率不低于 1 秒一次循环里 sleep(2) 是底线否则 DHT11 自身的电平位流会错乱。3. MQTT 解耦与阈值告警智能花卉监控与控制系统的控制核心3.1 为什么控制回路要从监控回路里拆出来实际跑起来后最容易出问题的不是传感器读数而是“读不到数据时整个流程卡住”。如果代码写成 while True 里依次做读传感器、写数据库、判断浇水、开继电器那 DHT11 偶尔卡 2 秒数据库、水泵和告警都会被拖住。正确做法是用 MQTT 把链路切断采集进程只管发布控制引擎单独订阅数据库和告警再从自己的订阅端消费各进程互不阻塞。MQTT 在智能花卉监控与控制系统里最重要的两个参数是 retain 和 will。retain 让 broker 保留该主题最后一条消息控制引擎重启后能立刻拿到设备最新状态。will 遗嘱消息则由 broker 在检测到发布端异常断开时代为发送让控制器知道“采集端掉线了”。这两项都在 paho-mqtt 客户端里配置不依赖 broker 的特殊插件是最容易复现的解耦方案。broker 端用 mosquitto 就足够资源占用低树莓派上一条命令装完。3.2 用 paho-mqtt 发布传感器数据的最小参数先看发布端代码。这个文件对应采集进程的 main loop每 10 秒或 30 秒发一次遥测数据import json import time import paho.mqtt.client as mqtt client mqtt.Client( client_idplantsense-publisher-001, clean_sessionTrue, ) client.username_pw_set(plantsense, change-me) client.reconnect_delay_set(1, 30) client.connect(127.0.0.1, 1883, keepalive15) def publish_telemetry(): payload { sensor: soil_a, value: 57, unit: percent, quality: 0.9, ts: int(time.time()), } result client.publish( plantsense/v1/telemetry/soil_a, json.dumps(payload), qos1, retainTrue, ) if result.rc mqtt.MQTT_ERR_SUCCESS: print(published, result.mid)参数说明client_id 在同一 broker 下必须唯一如果两个进程用同一 id 连接前者会被 broker 踢下线。clean_sessionTrue 表示订阅关系不持久重连后要重新订阅适合纯上报场景。qos1 保证消息至少到达一次但会有重复投递所以订阅端的处理逻辑要幂等。retainTrue 会让 broker 保留该主题最后一条消息新订阅者上线后立刻收到。keepalive15 是心跳间隔。publisher 的主循环如果只做 time.sleeploop 线程没有机会发 PING15 秒后 broker 会判定设备离线。所以不要调用阻塞式的 loop_forever而是在每个 sleep 周期里穿插client.loop(timeout1.0)或者用client.loop_start()启动后台网络线程。连接更高安全要求时MQTT 端口改为 8883配置 TLS 证书和用户名密码本机局域网跑测时 1883 就好但要保证 broker 默认只监听内网不要直接暴露到公网。3.3 订阅端做阈值判断和指令下发控制引擎是另一个进程订阅遥测主题并做决策import json import paho.mqtt.client as mqtt TOPIC_TELE plantsense/v1/telemetry/# TOPIC_CMD plantsense/v1/actuator/pump def on_tele(client, userdata, msg): payload json.loads(msg.payload) if payload.get(quality, 1.0) 0.5: return # 低质量数据不触发 if payload[sensor] soil_a and payload[value] 30: cmd {action: ON, duration: 6} client.publish(TOPIC_CMD, json.dumps(cmd), qos1)如果系统里的传感器只有三五个直接在 on_message 里写 if 也能接受。但规则变多以后我建议改成一张配置表把触发条件、执行对象、恢复条件分开维护。一张典型控制规则表可以这样设计维度触发条件执行对象恢复条件土壤湿度 soil_a低于 30%水泵开 6 秒高于 65%空气湿度 dht11低于 35%加湿器开 10 秒高于 60%温度 dht11高于 32℃风机开 15 秒低于 28℃光照 light低于 3000 lx补光灯开高于 8000 lx订阅端必须同时监听plantsense/v1/status/actuator/#接收继电器翻转后的回读状态。只发命令不读反馈等于失控。如果两轮执行器状态没有回来应该把该执行器标记为故障停止继续下发避免连续开泵造成溢水。命令消息也要设计成幂等比如duration表示开泵时长订阅端执行时先取消上一次同名命令的定时器再重新计时重复投递就不会叠加成 12 秒或 18 秒。4. SQLite 落库、批量写入与告警恢复智能花卉监控与控制系统的数据面4.1 为什么不用时序库单机传感器规模下 SQLite 更合适先把账算清楚8 个传感器每 30 秒一条一天 23040 条跑一年约 840 万行。这个量级对 SQLite 毫无压力前提是不要用默认的 journal_mode。默认的 DELETE 模式每写一次事务都要 fsync 一次树莓派 SD 卡既要磨损又要等延迟。切到 WAL 之后读不阻塞写写不阻塞读单机场景里比引入 InfluxDB 实惠得多。时序数据库适合的其实是多年历史保留、跨节点聚合、高基数维度查询。智能花卉监控与控制系统的数据量停留在“十万级到百万级”这个区间SQLite 单文件也能直接复制备份和源码打成 zip 一起交付很自然。运维上少一个数据库服务就少一个内存吃紧问题。唯一需要留意的是一次只能有一个写连接后面用写入队列来收敛并发即可。4.2 批量插入与缓存队列示例SQLite 连接初始化时先设置一组关键 PRAGMAimport sqlite3 conn sqlite3.connect(data/flower_monitor.db, timeout3.0) conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA synchronousNORMAL) conn.execute(PRAGMA busy_timeout3000) conn.execute(PRAGMA wal_autocheckpoint1000)参数含义journal_modeWAL 把写操作放到 WAL 文件里避免直接 rewrite 主库文件。synchronousNORMAL 在 WAL 模式下允许系统崩溃时丢最后一批事务对传感器数据来说可接受。timeout3.0 是 Python sqlite3 连接层的等待时间busy_timeout3000 是 SQLite 层的锁等待毫秒数两个都要设。wal_autocheckpoint1000 表示 WAL 日志超过 1000 页自动合并一次防止 -wal 文件持续膨胀。写入线程不要一条一条 insert。MQTT 订阅回调里把数据先塞进 queue.Queue独立线程每 5 秒或攒满 50 条批量落库buf [] while not shutting_down: try: item queue.get(timeout2.0) except queue.Empty: continue buf.append((item[sensor], item[value], item[ts])) if len(buf) 50: conn.executemany( INSERT INTO telemetry(sensor_id, value, ts) VALUES(?,?,?), buf, ) conn.commit() buf.clear()executemany 比循环 execute 快很多而且将写入频率从每秒几十次降到每 5 秒一次SD 卡写寿命压力也小。批量失败时不要立刻重试同一批数据传感器时间戳已经过期回插会把实时窗口拉偏。我一般会把失败批次先序列化到data/pending/目录下的 JSON 文件下一轮再合并写入这样至少不丢数据。4.3 告警阈值、恢复时间窗口与 SQL 查询示例单一数值越界立即告警会带来频繁抖动。土壤刚浇完水湿度会瞬间抬升两小时后可能又掉下来每次跨越边界都告警人很快会关掉通知。正确做法是同时定义触发阈值和恢复阈值并配合一个时间窗口做平均。下面是检查最近 5 分钟平均湿度的查询SELECT sensor_id, AVG(value) AS avg_value, COUNT(*) AS samples FROM telemetry WHERE ts strftime(%s, now) - 300 GROUP BY sensor_id HAVING avg_value 30;这条 SQL 的价值在于用均值而非单点判断。300 秒窗口适合土壤湿度这种变化慢的对象育苗盆要缩短到 60 秒室外大花箱可以拉到 600 秒。注意strftime(%s, now)得到的是 UTC epoch 秒插入数据时 ts 字段必须用相同基准否则窗口整体偏移。落库的表结构建议如下这组字段能同时服务展示和控制引擎字段含义示例sensor_id设备标识soil_balcony_01metric数据维度soil_moisture / temperaturevalue标定后的数值42quality数据可信度0.97tsepoch 秒1735880000最后是标定。土壤湿度传感器长期埋在土里会出现极化同一块土在不同深度读数差异很大。数据库里最好同时保存原始 ADC 值和标定后的 percent后续还可以通过percent raw * slope offset重算历史数据。我习惯每三个月重新做一次“干土、湿土”两点标定更新 settings.py 中的 offset 与 slope 系数不必更换传感器。5. 局域网可视化、告警推送与 zip 交付智能花卉监控与控制系统最后三件事5.1 用 Flask 在局域网内看实时数据不加前端框架树莓派上不要为了可视化直接上 Node 或重型 Web 框架一个 Flask 进程约 20MB 内存足够轻。接口通常是返回每个传感器最新一条记录供前端图表使用。一个最小可用的路由这样写app.route(/api/latest) def latest(): rows db.execute( SELECT t.sensor_id, t.metric, t.value, t.ts FROM telemetry t JOIN ( SELECT sensor_id, MAX(ts) AS max_ts FROM telemetry GROUP BY sensor_id ) m ON t.sensor_id m.sensor_id AND t.ts m.max_ts ).fetchall() return {data: [dict(row) for row in rows]}前端轮询这个接口每 3 秒一次完全够用。如果要多页面实时刷新用 Server-Sent Events 比 WebSocket 简单浏览器原生 EventSource后端只需在响应里不断写data: {...}\n\n不需要加第三方异步框架。5.2 Telegram 告警推送的参数与失败重试告警推送最容易被忽略的是超时和去重。下面这段代码适合作为控制引擎里的旁路通知函数import time import requests def send_alert(text: str, retries: int 3): url fhttps://api.telegram.org/bot{TOKEN}/sendMessage payload {chat_id: CHAT_ID, text: text, parse_mode: Markdown} for attempt in range(retries): try: resp requests.post(url, jsonpayload, timeout8) if resp.status_code 200: return True except requests.Timeout: pass time.sleep(2 ** attempt) return Falsetimeout8必须写。requests 不设超时会在网络异常时一直挂起告警线程被全部占满。重试退避用 1、2、4 秒三次失败后丢弃不再无限循环。相同事件加一个 15 分钟去重字典以 sensor_id 加事件类型作为 key避免阈值在边界抖动时把手机通知刷爆。提示永续告警导致的“狼来了”效应才是系统失效的开始告警宁可保守不能轰炸。5.3 工程收尾源码目录打成 zip 与 systemd 开机自启交付或备份时把项目目录打成 zip 很常见但要小心不要把虚拟环境和密钥一起打进包。推荐命令如下cd /opt/plantsense zip -r plantsense-release.zip . \ -x .venv/* .env data/*.db __pycache__/* .git/*-x指定排除路径.env里通常有 MQTT 密码绝不能进压缩包数据库文件单独用sqlite3 .backup备份不跟源码混在一起。最后把控制引擎注册成 systemd 服务开机自动拉起[Unit] DescriptionPlantsense MQTT Control Engine Afternetwork-online.target mosquitto.service [Service] WorkingDirectory/opt/plantsense ExecStart/opt/plantsense/.venv/bin/python -m mqtt.control_engine Restartalways RestartSec500Restartalways 表示进程异常退出后自动拉起RestartSec500 单位是毫秒也就是 0.5 秒后重启。MQTT 断线重连已由 paho 的 reconnect_delay_set 处理服务重启只是兜底。写完 unit 文件后用sudo systemctl daemon-reload重载再启用sudo systemctl enable --now plantsense.service。验证启动是否正常直接看日志journalctl -u plantsense.service -f如果看到 ModuleNotFoundError基本就是 ExecStart 里的 Python 路径没有指向虚拟环境。本文还有配套的精品资源点击获取