智能花盆土壤湿度监测系统
一、实际应用场景描述
随着都市生活节奏加快，越来越多的家庭选择在阳台或室内种植绿植花卉。然而，忙碌的工作和生活常常让我们忘记给植物浇水，导致植物因缺水而枯萎。传统的人工检查方式不仅费时费力，还容易因为疏忽造成植物死亡。
本系统基于智能仪器课程中的传感器技术、嵌入式开发和物联网理念，设计一款适合家庭使用的智能花盆监测系统。系统通过土壤湿度传感器实时监测花盆土壤湿度，当湿度低于30%时自动发出"该浇水"提示，帮助用户及时照料植物。
二、引入痛点
1. 忘记浇水：现代人工作繁忙，经常忘记给植物浇水
2. 过度浇水：担心植物缺水而频繁浇水，反而导致根部腐烂
3. 无法量化判断：凭经验判断土壤干湿程度不准确
4. 缺乏提醒机制：没有有效的自动化提醒手段
5. 多盆管理困难：家中多盆植物难以逐一关注
三、核心逻辑讲解
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ 传感器采集 │────▶│ 数据处理 │────▶│ 阈值判断 │
│ 土壤湿度数据 │ │ 转换为百分比 │ │ 是否<30% │
└─────────────────┘ └─────────────────┘ └─────────────────┘
│
┌─────────────────┐ │
│ 用户提示 │◀─────────────────┘
│ "该浇水" │
└─────────────────┘
核心逻辑流程：
1. 初始化土壤湿度传感器连接
2. 定时读取传感器原始数据
3. 将原始数据转换为0-100%的湿度百分比
4. 判断湿度值是否低于30%阈值
5. 若低于阈值，触发声音/视觉/消息提醒
6. 记录日志并持续监测
四、代码模块化
项目结构
smart_flowerpot/
├── main.py # 主程序入口
├── config.py # 配置文件
├── sensor/
│ ├── __init__.py
│ └── moisture_sensor.py # 湿度传感器模块
├── alert/
│ ├── __init__.py
│ └── notifier.py # 提醒模块
├── utils/
│ ├── __init__.py
│ └── logger.py # 日志工具
├── data/
│ └── humidity_log.csv # 湿度数据记录
├── requirements.txt
└── README.md
1. config.py - 配置文件
"""
配置文件模块
存储系统运行所需的配置参数
"""
# 传感器配置
SENSOR_CONFIG = {
'pin': 17, # GPIO引脚号（树莓派）
'min_raw': 0, # 传感器最小原始值
'max_raw': 1023, # 传感器最大原始值
'threshold': 30 # 湿度阈值（%）
}
# 提醒配置
ALERT_CONFIG = {
'enable_sound': True, # 启用声音提醒
'enable_visual': True, # 启用视觉提醒
'check_interval': 300 # 检测间隔（秒）
}
# 日志配置
LOG_CONFIG = {
'log_file': 'data/humidity_log.csv',
'log_interval': 3600 # 日志记录间隔（秒）
}
2. sensor/moisture_sensor.py - 湿度传感器模块
"""
湿度传感器模块
负责与土壤湿度传感器通信并读取数据
"""
import time
from config import SENSOR_CONFIG
class MoistureSensor:
"""
土壤湿度传感器类
该类封装了土壤湿度传感器的所有操作，包括：
- 传感器初始化
- 原始数据读取
- 数据转换处理
"""
def __init__(self, pin=None):
"""
初始化传感器
Args:
pin: GPIO引脚号，默认从配置文件读取
"""
self.pin = pin or SENSOR_CONFIG['pin']
self.min_raw = SENSOR_CONFIG['min_raw']
self.max_raw = SENSOR_CONFIG['max_raw']
self._setup_sensor()
def _setup_sensor(self):
"""
设置传感器硬件连接
在实际硬件环境中，这里会初始化GPIO接口
"""
print(f"[传感器] 正在初始化传感器，GPIO引脚: {self.pin}")
# 模拟硬件初始化（实际项目中替换为真实GPIO操作）
# import RPi.GPIO as GPIO
# GPIO.setmode(GPIO.BCM)
# GPIO.setup(self.pin, GPIO.IN)
time.sleep(1) # 模拟初始化延迟
print("[传感器] 初始化完成")
def read_raw_value(self):
"""
读取传感器原始数据
Returns:
int: 传感器原始读数 (0-1023)
Note:
模拟环境中返回随机测试数据
实际环境中应读取ADC转换值
"""
# 模拟读取过程（实际项目中替换为真实传感器读取）
# raw_value = ADC.read(self.pin)
import random
raw_value = random.randint(self.min_raw, self.max_raw)
return raw_value
def convert_to_percentage(self, raw_value):
"""
将原始数据转换为湿度百分比
Args:
raw_value: 传感器原始读数
Returns:
float: 湿度百分比 (0.0-100.0)
Formula:
湿度(%) = ((max_raw - raw_value) / (max_raw - min_raw)) * 100
"""
# 计算湿度百分比
humidity_range = self.max_raw - self.min_raw
humidity_percent = ((self.max_raw - raw_value) / humidity_range) * 100
# 确保结果在有效范围内
humidity_percent = max(0.0, min(100.0, humidity_percent))
return round(humidity_percent, 2)
def get_humidity(self):
"""
获取当前土壤湿度百分比
Returns:
dict: 包含原始值和湿度百分比的字典
{
'raw_value': int,
'humidity_percent': float,
'timestamp': str
}
"""
raw_value = self.read_raw_value()
humidity_percent = self.convert_to_percentage(raw_value)
from datetime import datetime
result = {
'raw_value': raw_value,
'humidity_percent': humidity_percent,
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
return result
def cleanup(self):
"""清理传感器资源"""
print("[传感器] 正在清理资源...")
# 实际项目中关闭GPIO连接
# GPIO.cleanup(self.pin)
print("[传感器] 资源清理完成")
if __name__ == "__main__":
# 测试传感器模块
sensor = MoistureSensor()
for i in range(5):
data = sensor.get_humidity()
print(f"读取 #{i+1}: 原始值={data['raw_value']}, "
f"湿度={data['humidity_percent']}%")
time.sleep(1)
sensor.cleanup()
3. alert/notifier.py - 提醒模块
"""
提醒模块
负责在湿度过低时向用户发送提醒
"""
import time
from config import ALERT_CONFIG, SENSOR_CONFIG
class Notifier:
"""
提醒器类
支持多种提醒方式：
- 控制台打印
- 声音提醒（蜂鸣器）
- 视觉提醒（LED灯）
- 桌面通知
"""
def __init__(self):
"""初始化提醒器"""
self.threshold = SENSOR_CONFIG['threshold']
self.alert_count = 0
print(f"[提醒器] 初始化完成，湿度阈值: {self.threshold}%")
def check_and_alert(self, humidity_data):
"""
检查湿度并在需要时发送提醒
Args:
humidity_data: 湿度数据字典，来自MoistureSensor.get_humidity()
Returns:
bool: 是否触发了提醒
"""
humidity = humidity_data['humidity_percent']
timestamp = humidity_data['timestamp']
if humidity < self.threshold:
self._send_alert(humidity, timestamp)
self.alert_count += 1
return True
else:
print(f"[{timestamp}] 当前湿度: {humidity}%，状态正常 ✅")
return False
def _send_alert(self, humidity, timestamp):
"""
发送提醒（内部方法）
Args:
humidity: 当前湿度百分比
timestamp: 时间戳
"""
message = f"⚠️ 该浇水！当前土壤湿度: {humidity}%（低于{self.threshold}%）"
# 控制台提醒
print("=" * 50)
print(f"[{timestamp}] {message}")
print("=" * 50)
# 声音提醒
if ALERT_CONFIG['enable_sound']:
self._sound_alert()
# 视觉提醒
if ALERT_CONFIG['enable_visual']:
self._visual_alert()
def _sound_alert(self):
"""
声音提醒
实际环境中可连接蜂鸣器
"""
print("🔊 [声音提醒] 滴-滴-滴！该浇水啦！")
# 实际项目中：GPIO控制蜂鸣器发声
# import RPi.GPIO as GPIO
# GPIO.output(BUZZER_PIN, GPIO.HIGH)
# time.sleep(0.5)
# GPIO.output(BUZZER_PIN, GPIO.LOW)
def _visual_alert(self):
"""
视觉提醒
实际环境中可控制LED灯闪烁
"""
print("💡 [视觉提醒] LED灯闪烁中...")
# 实际项目中：GPIO控制LED闪烁
# import RPi.GPIO as GPIO
# for _ in range(5):
# GPIO.output(LED_PIN, GPIO.HIGH)
# time.sleep(0.3)
# GPIO.output(LED_PIN, GPIO.LOW)
# time.sleep(0.3)
def send_daily_summary(self, log_data):
"""
发送每日摘要
Args:
log_data: 日志数据列表
"""
if not log_data:
return
avg_humidity = sum(d['humidity'] for d in log_data) / len(log_data)
min_humidity = min(d['humidity'] for d in log_data)
max_humidity = max(d['humidity'] for d in log_data)
print("\n" + "=" * 50)
print("📊 每日湿度报告")
print("=" * 50)
print(f"平均湿度: {avg_humidity:.2f}%")
print(f"最低湿度: {min_humidity:.2f}%")
print(f"最高湿度: {max_humidity:.2f}%")
print(f"今日提醒次数: {self.alert_count}次")
print("=" * 50 + "\n")
def reset_alert_count(self):
"""重置提醒计数"""
self.alert_count = 0
if __name__ == "__main__":
# 测试提醒模块
notifier = Notifier()
# 模拟低湿度情况
test_data_low = {
'raw_value': 800,
'humidity_percent': 19.6,
'timestamp': '2026-03-21 10:00:00'
}
# 模拟正常湿度情况
test_data_normal = {
'raw_value': 400,
'humidity_percent': 58.8,
'timestamp': '2026-03-21 10:05:00'
}
notifier.check_and_alert(test_data_normal)
notifier.check_and_alert(test_data_low)
notifier.check_and_alert(test_data_low)
4. utils/logger.py - 日志工具
"""
日志工具模块
负责记录湿度数据和系统运行日志
"""
import csv
import os
from datetime import datetime
from config import LOG_CONFIG
class HumidityLogger:
"""
湿度日志记录器类
功能：
- 记录每次湿度检测数据
- 生成CSV格式日志文件
- 提供数据查询接口
"""
def __init__(self, log_file=None):
"""
初始化日志记录器
Args:
log_file: 日志文件路径
"""
self.log_file = log_file or LOG_CONFIG['log_file']
self._ensure_log_file_exists()
print(f"[日志] 日志文件: {self.log_file}")
def _ensure_log_file_exists(self):
"""确保日志文件存在，不存在则创建并写入表头"""
log_dir = os.path.dirname(self.log_file)
if log_dir and not os.path.exists(log_dir):
os.makedirs(log_dir)
if not os.path.exists(self.log_file):
with open(self.log_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(['timestamp', 'raw_value', 'humidity_percent'])
def log(self, humidity_data):
"""
记录单条湿度数据
Args:
humidity_data: 湿度数据字典
"""
with open(self.log_file, 'a', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow([
humidity_data['timestamp'],
humidity_data['raw_value'],
humidity_data['humidity_percent']
])
def get_today_records(self):
"""
获取今日所有记录
Returns:
list: 今日记录列表
"""
today = datetime.now().strftime('%Y-%m-%d')
records = []
try:
with open(self.log_file, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
if row['timestamp'].startswith(today):
records.append({
'timestamp': row['timestamp'],
'raw_value': int(row['raw_value']),
'humidity': float(row['humidity_percent'])
})
except FileNotFoundError:
pass
return records
def get_statistics(self, days=7):
"""
获取最近N天的统计信息
Args:
days: 天数
Returns:
dict: 统计信息
"""
from datetime import timedelta
end_date = datetime.now()
start_date = end_date - timedelta(days=days)
all_records = []
try:
with open(self.log_file, 'r', encoding='utf-8') as f:
reader = csv.DictReader(f)
for row in reader:
record_date = datetime.strptime(
row['timestamp'], '%Y-%m-%d %H:%M:%S'
)
if start_date <= record_date <= end_date:
all_records.append(float(row['humidity_percent']))
except FileNotFoundError:
pass
if not all_records:
return None
return {
'count': len(all_records),
'average': sum(all_records) / len(all_records),
'minimum': min(all_records),
'maximum': max(all_records)
}
def export_to_text(self, output_file):
"""
导出日志为文本格式
Args:
output_file: 输出文件路径
"""
with open(self.log_file, 'r', encoding='utf-8') as f_in, \
open(output_file, 'w', encoding='utf-8') as f_out:
f_out.write("时间戳\t\t\t\t原始值\t湿度(%)\n")
f_out.write("-" * 50 + "\n")
for line in f_in:
f_out.write(line)
if __name__ == "__main__":
# 测试日志模块
logger = HumidityLogger('data/test_log.csv')
# 模拟记录数据
test_data = [
{'timestamp': '2026-03-21 08:00:00', 'raw_value': 500, 'humidity_percent': 49.0},
{'timestamp': '2026-03-21 09:00:00', 'raw_value': 600, 'humidity_percent': 39.2},
{'timestamp': '2026-03-21 10:00:00', 'raw_value': 750, 'humidity_percent': 24.5},
]
for data in test_data:
logger.log(data)
print(f"已记录: {data['timestamp']} - {data['humidity_percent']}%")
# 读取今日记录
print("\n今日记录:")
for record in logger.get_today_records():
print(f" {record['timestamp']}: {record['humidity']}%")
# 获取统计信息
stats = logger.get_statistics(days=7)
if stats:
print(f"\n7天统计: 平均={stats['average']:.2f}%, "
f"最小={stats['minimum']:.2f}%, 最大={stats['maximum']:.2f}%")
5. main.py - 主程序入口
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
智能花盆土壤湿度监测系统 - 主程序
功能：
- 实时监测土壤湿度
- 低于阈值时自动提醒
- 记录湿度数据日志
- 支持命令行参数控制
作者：全栈开发工程师
版本：1.0.0
日期：2026-03-21
"""
import time
import signal
import sys
from datetime import datetime
from sensor.moisture_sensor import MoistureSensor
from alert.notifier import Notifier
from utils.logger import HumidityLogger
from config import ALERT_CONFIG, LOG_CONFIG
class SmartFlowerpotSystem:
"""
智能花盆系统主类
整合所有模块，提供完整的系统功能
"""
def __init__(self):
"""初始化系统"""
print("=" * 60)
print("🌱 智能花盆土壤湿度监测系统 v1.0.0")
print("=" * 60)
# 初始化各模块
self.sensor = MoistureSensor()
self.notifier = Notifier()
self.logger = HumidityLogger(LOG_CONFIG['log_file'])
# 系统状态
self.is_running = True
self.check_count = 0
# 设置信号处理器（优雅退出）
signal.signal(signal.SIGINT, self._signal_handler)
signal.signal(signal.SIGTERM, self._signal_handler)
print("系统初始化完成，开始监测...\n")
def _signal_handler(self, signum, frame):
"""
信号处理函数
用于优雅地停止系统
Args:
signum: 信号编号
frame: 当前栈帧
"""
print(f"\n收到信号 {signum}，正在停止系统...")
self.is_running = False
def run(self):
"""
运行主循环
持续监测土壤湿度并作出相应处理
"""
check_interval = ALERT_CONFIG['check_interval']
while self.is_running:
try:
# 读取湿度数据
humidity_data = self.sensor.get_humidity()
# 记录日志
self.logger.log(humidity_data)
# 检查是否需要提醒
self.notifier.check_and_alert(humidity_data)
self.check_count += 1
# 显示运行状态
self._show_status(check_interval)
# 等待下次检测
self._sleep_with_interrupt_check(check_interval)
except Exception as e:
print(f"[错误] 发生异常: {e}")
time.sleep(10) # 出错后等待较长时间再重试
def _show_status(self, next_check_seconds):
"""
显示系统运行状态
Args:
next_check_seconds: 下次检测的等待秒数
"""
current_time = datetime.now().strftime('%H:%M:%S')
progress_bar = self._create_progress_bar(next_check_seconds)
print(f"\r[{current_time}] "
f"检测次数: {self.check_count} | "
f"{progress_bar}", end='')
def _create_progress_bar(self, total_seconds, bar_length=20):
"""
创建进度条
Args:
total_seconds: 总秒数
bar_length: 进度条长度
Returns:
str: 进度条字符串
"""
# 简单模拟进度（实际应用中需要更精确的时间跟踪）
import math
filled = int(bar_length * 0.3) # 模拟30%进度
empty = bar_length - filled
return f"[{'█' * filled}{'░' * empty}]"
def _sleep_with_interrupt_check(self, seconds):
"""
带中断检查的睡眠
允许系统在睡眠期间响应停止信号
Args:
seconds: 睡眠秒数
"""
sleep_interval = 1 # 每1秒检查一次
slept = 0
while slept < seconds and self.is_running:
time.sleep(sleep_interval)
slept += sleep_interval
def show_daily_report(self):
"""
显示每日报告
"""
print("\n" + "=" * 50)
print("📋 生成每日报告...")
print("=" * 50)
# 获取今日记录
today_records = self.logger.get_today_records()
if not today_records:
print("今日暂无记录")
return
# 显示统计信息
print(f"今日共检测 {len(today_records)} 次")
humidities = [r['humidity'] for r in today_records]
print(f"平均湿度: {sum(humidities)/len(humidities):.2f}%")
print(f"最低湿度: {min(humidities):.2f}%")
print(f"最高湿度: {max(humidities):.2f}%")
# 显示最近5条记录
print("\n最近5条记录:")
for record in today_records[-5:]:
print(f" {record['timestamp']}: {record['humidity']}%")
# 显示7天统计
stats = self.logger.get_statistics(days=7)
if stats:
print(f"\n近7天统计: 平均={stats['average']:.2f}%, "
f"最小={stats['minimum']:.2f}%, 最大={stats['maximum']:.2f}%")
def cleanup(self):
"""
清理系统资源
"""
print("\n正在清理资源...")
self.sensor.cleanup()
print("系统已安全停止")
print("=" * 60)
def main():
"""
主函数
程序入口点
"""
system = SmartFlowerpotSystem()
try:
system.run()
except KeyboardInterrupt:
pass
finally:
system.show_daily_report()
system.cleanup()
if __name__ == "__main__":
main()
6. requirements.txt
# 智能花盆系统依赖
# 注：模拟环境不需要特殊依赖
# 实际部署到树莓派时可能需要：
# RPi.GPIO>=0.7.0
# adafruit-circuitpython-ads1x15>=1.4.0
五、README文件
# 智能花盆土壤湿度监测系统
[](https://www.python.org/)
> 基于智能仪器课程技术，为家庭养花打造的智能监测解决方案
## 📖 项目简介
本系统通过土壤湿度传感器实时监测花盆土壤湿度，当湿度低于30%时自动提醒"该浇水"，帮助您科学养护植物，避免因忘记浇水导致的植物枯萎。
## ✨ 主要功能
- 🌡️ **实时监测**: 每5分钟自动检测土壤湿度
- 🚨 **智能提醒**: 湿度低于30%时多渠道提醒
- 📝 **数据记录**: 自动保存湿度历史数据
- 📊 **统计分析**: 提供日/周湿度统计报告
- ⏹️ **优雅退出**: 支持Ctrl+C安全停止
## 🛠️ 硬件要求
| 组件 | 型号/规格 | 数量 |
|------|----------|------|
| 微控制器 | 树莓派3B+/4B | 1 |
| 土壤湿度传感器 | 电容式/电阻式 | 1 |
| 杜邦线 | 母对母 | 3 |
| 面包板 | 830孔 | 1 |
## 💻 软件环境
- Python 3.6+
- RPi.GPIO (树莓派GPIO控制)
- 可选: Adafruit_ADS1x15 (ADC转换)
## 🚀 安装与部署
### 1. 克隆项目
bash
git clone "https://github.com/your-username/smart-flowerpot.git" (https://github.com/your-username/smart-flowerpot.git)
cd smart-flowerpot
### 2. 安装依赖
bash
pip install -r requirements.txt
### 3. 硬件连接
土壤湿度传感器 → 树莓派
VCC → 3.3V/5V
GND → GND
DO/AO → GPIO 17
### 4. 运行程序
bash
python main.py
## 📁 项目结构
smart_flowerpot/
├── main.py # 主程序入口
├── config.py # 系统配置
├── sensor/ # 传感器模块
│ └── moisture_sensor.py
├── alert/ # 提醒模块
│ └── notifier.py
├── utils/ # 工具模块
│ └── logger.py
├── data/ # 数据目录
└── README.md
## ⚙️ 配置说明
编辑 `config.py` 自定义系统参数：
python
修改检测间隔（秒）
ALERT_CONFIG = {
'check_interval': 300 # 5分钟
}
修改湿度阈值（%）
SENSOR_CONFIG = {
'threshold': 30 # 低于此值提醒
}
## 🎯 使用场景
- 家庭阳台/室内盆栽
- 办公室绿植养护
- 多盆植物集中管理
- 植物养护新手学习
## 📊 运行效果
============================================================
🌱 智能花盆土壤湿度监测系统 v1.0.0
[传感器] 正在初始化传感器，GPIO引脚: 17
[传感器] 初始化完成
[提醒器] 初始化完成，湿度阈值: 30%
[日志] 日志文件: data/humidity_log.csv
系统初始化完成，开始监测...
[10:00:00] 检测次数: 1 | [████░░░░░░░░░░░░░░░░]
[10:05:00] 当前湿度: 45.2%，状态正常 ✅
[10:10:00] 当前湿度: 28.5%，状态正常 ✅
[10:15:00] ==============================================
利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！
]—东方仙盟)