当前位置： 首页 > news >正文

Python物联网实战：基于树莓派与DHT传感器打造微信QQ邮件智能温湿度监控系统

news 2026/7/28 5:32:23

1. 项目缘起：一个“懒人”的温度监控执念

不知道你有没有过这种经历：家里的老人或者宠物独自在家，你总担心室温是不是太冷或太热；又或者，你有一个精心搭建的服务器机柜或存放贵重设备的房间，需要时刻关注环境温度，防止过热损坏。传统的温度计需要你“在场”才能看到读数，而市面上一些智能温度计要么需要搭配专用的App，要么需要复杂的生态链设备，总让人觉得不够直接、不够“懒人友好”。

我的需求很简单：我想在任何地方，用我最常用的通讯工具——微信和QQ，随时问一句“现在温度多少？”，或者干脆让它定时给我发封邮件报告，我就能立刻知道某个地点的实时温度。听起来是不是有点像给温度计装上了“嘴巴”和“耳朵”？这就是“智控万物”系列里我想分享的第一个实战项目：用Python打造一个能通过微信、QQ和邮件汇报温度的智能温度计。

这个项目的核心思想是“通知触手可及”。我们不需要开发复杂的App，而是利用现有的、几乎人人都在用的通讯平台作为交互界面。微信和QQ用于主动查询或被动接收告警，邮件则用于接收格式更规范、信息更丰富的定时报告。实现这一切的“大脑”，是一段运行在树莓派、旧电脑甚至云服务器上的Python脚本。它通过一个廉价的温度传感器（比如DHT11/DHT22）采集数据，然后调用各个平台提供的API，将数据“说”出去。

接下来，我将从硬件选型、环境搭建、核心代码实现，到三大通知渠道（微信、QQ、邮件）的逐一接入，以及最终的系统集成与优化，完整地拆解这个项目。你会发现，只需一些基础的Python知识和动手能力，你就能让物理世界的数据主动找到你。

2. 硬件基石：传感器选型与电路连接

实现温度监控，第一步是让我们的“大脑”（Python程序）能感知到物理世界的温度。这就需要温度传感器。对于业余项目和家庭监控场景，DHT11和DHT22是两个经久不衰的选择。它们都能同时测量温度和湿度，价格低廉，使用简单。

2.1 DHT11 vs DHT22：如何选择？

虽然都是DHT系列，但两者在精度、量程和响应速度上有区别。选择哪一个，取决于你对数据精度的要求。

特性DHT11DHT22 (又名 AM2302)
温度量程0 ~ 50°C-40 ~ 80°C
温度精度±2°C±0.5°C
湿度量程20% ~ 90% RH0% ~ 100% RH
湿度精度±5% RH±2% RH
采样周期1秒2秒
价格更低稍高

选择建议：

  • DHT11：适用于室内常温环境的一般性监控，比如想知道房间大概多少度，精度要求不高，追求极致的低成本。
  • DHT22：如果你需要监控的环境温度可能低于0°C或高于50°C（比如车库、阁楼、北方阳台），或者你对温度精度要求较高（例如监控红酒柜、孵化器），那么DHT22是更好的选择。对于绝大多数项目，我推荐直接使用DHT22，多花几块钱获得更好的性能和更宽的量程是值得的。

注意：DHT传感器对读取时序要求比较严格，且一次读取失败后需要等待一段时间才能再次读取。在代码中必须做好异常处理。

2.2 电路连接：以树莓派为例

我们以最常见的树莓派（以40针GPIO的型号为例）作为主控平台。连接非常简单，只需要三根母对母杜邦线。

所需材料：

  1. 树莓派（任何型号均可，需启用GPIO）。
  2. DHT11或DHT22传感器模块（注意是模块，上面自带了一个上拉电阻，如果是单独的传感器芯片，需要自己连接一个4.7K-10KΩ的上拉电阻到数据线）。
  3. 三根母对母杜邦线。

连接方式：

  • VCC (或 ‘+’)-> 树莓派的3.3V引脚（例如物理引脚1或17）。切记接3.3V，不要接5V，否则可能损坏传感器或树莓派。
  • GND (或 ‘-’)-> 树莓派的GND引脚（例如物理引脚6、9、14、20、25、30、34、39等）。
  • DATA (或 ‘OUT’)-> 树莓派的任意一个GPIO引脚（例如GPIO4，对应物理引脚7）。

连接好后，硬件部分就完成了。你可以把传感器放在你想监控的位置，用一根延长线连接即可。

3. 软件环境搭建：Python与核心库安装

硬件就绪后，我们需要在树莓派上配置Python环境，并安装读取传感器和发送通知所需的库。

3.1 系统与Python准备

树莓派官方系统Raspbian/Raspberry Pi OS通常已预装Python 3。首先打开终端，更新系统并确认Python版本：

sudo apt update sudo apt upgrade -y python3 --version

确保输出是 Python 3.5+。本项目将完全使用 Python 3。

3.2 安装传感器驱动库

为了读取DHT传感器的数据，我们需要一个库来处理底层的GPIO通信和复杂的时序解析。Adafruit_DHT库是一个广泛使用的选择，但这里我推荐一个更现代、维护更好的替代品：adafruit-circuitpython-dht配合libgpiod。不过，为了最简化的入门，我们先使用经典的Adafruit_Python_DHT库。

安装依赖和库：

# 安装必要的系统依赖 sudo apt install -y python3-pip python3-dev python3-venv sudo apt install -y build-essential # 使用pip安装Adafruit_DHT库 pip3 install Adafruit_DHT

如果安装失败，可能需要先安装python3-dev包。这个库提供了一个简单的函数Adafruit_DHT.read_retry(sensor_type, gpio_pin)来读取数据，它会自动处理重试逻辑，非常方便。

3.3 创建项目目录与虚拟环境

良好的习惯是从一开始就使用虚拟环境，避免包版本冲突。

mkdir ~/smart-thermometer && cd ~/smart-thermometer python3 -m venv venv source venv/bin/activate

激活虚拟环境后，你的命令行提示符前会出现(venv)字样。之后所有pip install操作都只影响这个环境。

在虚拟环境中，再次安装我们需要的库：

pip install Adafruit_DHT

4. 核心功能实现：温度读取与数据格式化

环境准备好后，我们来编写最核心的温度读取代码。创建一个名为thermometer_core.py的文件。

4.1 编写基础读取函数

# thermometer_core.py import Adafruit_DHT import time from datetime import datetime # 设置传感器类型和GPIO引脚 # 如果是DHT11，使用 Adafruit_DHT.DHT11 # 如果是DHT22，使用 Adafruit_DHT.DHT22 SENSOR_TYPE = Adafruit_DHT.DHT22 GPIO_PIN = 4 # 我们之前连接到的GPIO4 def read_temperature_humidity(): """ 尝试读取温度和湿度数据。 返回一个字典：{'temperature': 温度值, 'humidity': 湿度值, 'success': 是否成功} 如果读取失败，温度湿度返回None。 """ # 使用 read_retry 函数，它会在失败时自动重试（默认重试15次，间隔2秒） humidity, temperature = Adafruit_DHT.read_retry(SENSOR_TYPE, GPIO_PIN) result = {'success': False, 'temperature': None, 'humidity': None, 'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')} if humidity is not None and temperature is not None: # 读取成功，DHT22返回的温度是摄氏度 result['success'] = True result['temperature'] = round(temperature, 1) # 保留一位小数 result['humidity'] = round(humidity, 1) print(f"[{result['timestamp']}] 读取成功: 温度={result['temperature']}°C, 湿度={result['humidity']}%") else: print(f"[{result['timestamp']}] 传感器读取失败！") return result if __name__ == '__main__': # 测试代码：连续读取5次，每次间隔3秒 for i in range(5): data = read_temperature_humidity() print(f"第{i+1}次读取: {data}") time.sleep(3)

运行这个脚本python3 thermometer_core.py，你应该能看到终端打印出温度和湿度数据。如果一直失败，请检查接线、GPIO引脚号是否正确，以及传感器是否完好。

4.2 数据格式化与容错处理

在实际应用中，我们需要更健壮的代码。传感器可能偶尔读取失败，我们需要记录日志，并可能对数据进行简单的平滑处理（比如取最近几次读取的平均值来避免单次跳变）。

创建一个新的文件sensor_manager.py来管理更复杂的逻辑：

# sensor_manager.py import Adafruit_DHT from datetime import datetime import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class SensorManager: def __init__(self, sensor_type=Adafruit_DHT.DHT22, gpio_pin=4, max_retries=3): self.sensor_type = sensor_type self.gpio_pin = gpio_pin self.max_retries = max_retries self.reading_history = [] # 用于存储历史数据，可做平滑 self.history_size = 5 def read_with_retry(self): """带重试和日志的读取""" for attempt in range(1, self.max_retries + 1): humidity, temperature = Adafruit_DHT.read_retry(self.sensor_type, self.gpio_pin, retries=1, delay_seconds=1) if humidity is not None and temperature is not None: reading = { 'timestamp': datetime.now(), 'temperature': round(temperature, 1), 'humidity': round(humidity, 1) } # 更新历史记录 self.reading_history.append(reading) if len(self.reading_history) > self.history_size: self.reading_history.pop(0) logger.info(f"传感器读取成功 (尝试{attempt}次): {reading['temperature']}°C, {reading['humidity']}%") return reading else: logger.warning(f"传感器读取失败，第{attempt}次尝试。") time.sleep(0.5) # 短暂等待后重试 logger.error(f"传感器读取失败，已达最大重试次数{self.max_retries}。") return None def get_smoothed_reading(self): """获取平滑后的数据（最近几次的平均值）""" if not self.reading_history: return self.read_with_retry() valid_readings = [r for r in self.reading_history if r is not None] if not valid_readings: return None avg_temp = sum(r['temperature'] for r in valid_readings) / len(valid_readings) avg_humi = sum(r['humidity'] for r in valid_readings) / len(valid_readings) smoothed = { 'timestamp': datetime.now(), 'temperature': round(avg_temp, 1), 'humidity': round(avg_humi, 1), 'note': 'smoothed' } return smoothed def format_for_message(self, reading): """将读数格式化为友好的字符串，用于发送消息""" if not reading: return "【警告】无法获取传感器数据，请检查设备连接。" ts = reading['timestamp'].strftime('%m-%d %H:%M') return f"🌡️ 环境监测报告 [{ts}]\n温度: {reading['temperature']}°C\n湿度: {reading['humidity']}%"

这个SensorManager类提供了更好的封装：带重试的读取、简单的数据平滑、格式化输出以及详细的日志。这是我们后续所有通知功能的数据来源基础。

5. 通知渠道一：邮件定时报告

邮件是最传统、也最稳定的通知方式，适合接收格式完整的定时报告（如每日简报）。我们将使用Python内置的smtplib和email库来实现。

5.1 配置发件邮箱（以QQ邮箱为例）

重要：你不能直接用邮箱密码登录SMTP服务器了。主流邮箱服务商（QQ、163、Gmail等）都要求使用“授权码”代替密码。

  1. 登录QQ邮箱，点击“设置” -> “账户”。
  2. 找到“POP3/IMAP/SMTP/Exchange/CardDAV/CalDAV服务”部分。
  3. 开启“POP3/SMTP服务”或“IMAP/SMTP服务”。系统会提示你发送短信验证，完成后你会获得一个16位的授权码。务必复制保存好这个授权码，它只会显示一次！

5.2 编写邮件发送模块

创建mail_sender.py：

# mail_sender.py import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from email.header import Header import logging logger = logging.getLogger(__name__) class MailSender: def __init__(self, smtp_server, smtp_port, sender_email, auth_code): """ 初始化邮件发送器 :param smtp_server: SMTP服务器地址，QQ邮箱是 'smtp.qq.com' :param smtp_port: 端口，QQ邮箱SSL端口是 465 :param sender_email: 发件人邮箱 :param auth_code: 邮箱授权码（不是密码！） """ self.smtp_server = smtp_server self.smtp_port = smtp_port self.sender_email = sender_email self.auth_code = auth_code def send_temperature_report(self, recipient_email, sensor_reading, subject_prefix="[智能温度计]"): """ 发送温度报告邮件 :param recipient_email: 收件人邮箱 :param sensor_reading: 来自 SensorManager 的读数字典 :param subject_prefix: 邮件主题前缀 """ if not sensor_reading: logger.error("无法发送邮件：传感器数据无效。") return False # 1. 构建邮件内容 subject = f"{subject_prefix} 环境报告 - {sensor_reading['timestamp'].strftime('%Y-%m-%d %H:%M')}" # 纯文本内容 text_content = sensor_reading.get('formatted', f"温度: {sensor_reading['temperature']}°C\n湿度: {sensor_reading['humidity']}%") # HTML内容，更美观 html_content = f""" <html> <body> <h2>🏠 智能环境监测报告</h2> <p><strong>报告时间：</strong>{sensor_reading['timestamp'].strftime('%Y-%m-%d %H:%M:%S')}</p> <table border="1" cellpadding="10" style="border-collapse: collapse;"> <tr style="background-color: #f2f2f2;"> <th>指标</th> <th>数值</th> <th>状态</th> </tr> <tr> <td>🌡️ 温度</td> <td><strong>{sensor_reading['temperature']} °C</strong></td> <td>{self._get_temperature_status(sensor_reading['temperature'])}</td> </tr> <tr> <td>💧 湿度</td> <td><strong>{sensor_reading['humidity']} %</strong></td> <td>{self._get_humidity_status(sensor_reading['humidity'])}</td> </tr> </table> <p><small>此邮件由智能温度计系统自动发送。</small></p> </body> </html> """ # 2. 创建MIMEMultipart对象，支持纯文本和HTML message = MIMEMultipart('alternative') message['From'] = Header(f"智能温度计 <{self.sender_email}>") message['To'] = Header(recipient_email) message['Subject'] = Header(subject, 'utf-8') # 附加纯文本和HTML版本 part1 = MIMEText(text_content, 'plain', 'utf-8') part2 = MIMEText(html_content, 'html', 'utf-8') message.attach(part1) message.attach(part2) # 3. 发送邮件 try: # 使用SSL连接 with smtplib.SMTP_SSL(self.smtp_server, self.smtp_port) as server: server.login(self.sender_email, self.auth_code) server.sendmail(self.sender_email, recipient_email, message.as_string()) logger.info(f"邮件成功发送至 {recipient_email}") return True except Exception as e: logger.error(f"邮件发送失败: {e}") return False def _get_temperature_status(self, temp): """简单的温度状态判断""" if temp < 10: return "❄️ 较低" elif 10 <= temp < 26: return "✅ 舒适" elif 26 <= temp < 32: return "⚠️ 较热" else: return "🔥 过热" def _get_humidity_status(self, humi): """简单的湿度状态判断""" if humi < 30: return "🏜️ 干燥" elif 30 <= humi < 70: return "✅ 适宜" else: return "💦 潮湿" # 配置示例 (将以下信息替换为你自己的) if __name__ == '__main__': # 你的QQ邮箱配置 SMTP_SERVER = 'smtp.qq.com' SMTP_PORT = 465 SENDER_EMAIL = 'your_qq_number@qq.com' # 替换为你的QQ邮箱 AUTH_CODE = 'your_16_char_auth_code' # 替换为你的授权码 mailer = MailSender(SMTP_SERVER, SMTP_PORT, SENDER_EMAIL, AUTH_CODE) # 模拟一个传感器读数 from datetime import datetime test_reading = { 'timestamp': datetime.now(), 'temperature': 23.5, 'humidity': 65.0 } # 发送测试邮件给自己 mailer.send_temperature_report(SENDER_EMAIL, test_reading)

运行这个测试脚本，检查你的收件箱（包括垃圾邮件箱）是否收到了格式美观的温度报告邮件。至此，邮件通知通道就打通了。

6. 通知渠道二：微信消息推送

微信是国内最普及的通讯工具，实现微信通知能极大提升便利性。我们无法直接给个人微信发消息，但可以通过“微信测试号”或“Server酱”等第三方服务，将消息推送到你的微信上。这里以免费、简单的Server酱 Turbo版（基于企业微信应用消息）为例。

6.1 获取微信推送通道（Server酱 Turbo版）

  1. 访问 Server酱官网 并登录（可用GitHub账号）。
  2. 进入“SendKey”页面，你会看到一个SENDKEY。点击“微信推送”，按照指引扫码关注一个企业微信应用，即可绑定微信。
  3. 复制你的SENDKEY。这个Key用于标识你的消息接收端。

6.2 编写微信推送模块

创建wechat_sender.py：

# wechat_sender.py import requests import json import logging logger = logging.getLogger(__name__) class WeChatSender: def __init__(self, sendkey): """ 初始化微信推送器 (基于Server酱 Turbo版) :param sendkey: 从Server酱官网获取的SENDKEY """ self.sendkey = sendkey self.api_url = f"https://sctapi.ftqq.com/{sendkey}.send" def send_text_message(self, title, content, channel=None): """ 发送文本消息到微信 :param title: 消息标题 :param content: 消息内容 :param channel: 推送通道，默认为空（即默认推送方式） :return: 是否成功 """ data = { 'title': title, 'desp': content # Server酱的desp字段支持Markdown } if channel: data['channel'] = channel try: response = requests.post(self.api_url, data=data, timeout=10) result = response.json() if result.get('code') == 0: logger.info(f"微信消息推送成功: {result.get('message')}") return True else: logger.error(f"微信消息推送失败: {result}") return False except Exception as e: logger.error(f"微信消息推送请求异常: {e}") return False def send_temperature_alert(self, sensor_reading, thresholds={'temp_high': 30, 'temp_low': 10}): """ 根据阈值发送温度告警或常规报告 """ if not sensor_reading: self.send_text_message("❌ 传感器异常", "无法读取温度传感器数据，请检查设备。") return temp = sensor_reading['temperature'] humi = sensor_reading['humidity'] ts = sensor_reading['timestamp'].strftime('%H:%M') # 判断是否需要告警 if temp >= thresholds['temp_high']: title = f"🔥 高温告警 - {temp}°C" channel = '9' # 使用Server酱的“重要消息”通道，会有更强提醒 elif temp <= thresholds['temp_low']: title = f"❄️ 低温告警 - {temp}°C" channel = '9' else: title = f"🌡️ 温度正常 - {temp}°C" channel = None # 普通通道 # 构建Markdown格式的内容 content = f""" ## 环境状态报告 ({ts}) - **温度**: {temp} °C - **湿度**: {humi} % - **状态**: {'正常' if not channel else '告警'} > 来自您的智能温度计 """ self.send_text_message(title, content, channel) # 使用示例 if __name__ == '__main__': # 替换为你的 SENDKEY SENDKEY = "YOUR_SENDKEY_HERE" sender = WeChatSender(SENDKEY) from datetime import datetime # 测试告警 test_reading_alert = {'timestamp': datetime.now(), 'temperature': 35.0, 'humidity': 50.0} sender.send_temperature_alert(test_reading_alert) # 测试正常报告 test_reading_normal = {'timestamp': datetime.now(), 'temperature': 22.0, 'humidity': 60.0} sender.send_temperature_alert(test_reading_normal)

运行测试，你的微信就会收到消息了。Server酱的免费额度对于个人监控完全足够。通过这个通道，你可以实现异常告警即时推送和定时报告推送。

7. 通知渠道三：QQ机器人消息

QQ机器人提供了更交互式的可能，比如你可以主动在QQ群里@机器人并询问“温度多少？”，它会回复当前读数。这里我们使用go-cqhttp配合NoneBot2框架来实现，这是目前Python生态中最主流、稳定的QQ机器人方案。

7.1 搭建基础：go-cqhttp

go-cqhttp 是一个基于 Mirai 以及 MiraiGo 的 QQ 机器人框架，它负责与QQ服务器通信。我们需要先部署它。

  1. 下载 go-cqhttp：从 GitHub 发布页下载对应你树莓派系统架构的版本（通常是arm64或armv7）。
  2. 配置与登录：
    • 解压后，首次运行./go-cqhttp（或go-cqhttp.exe），会生成配置文件config.yml。
    • 编辑config.yml，主要修改：
      • account.uin: 你的机器人QQ号。
      • account.password: 机器人QQ密码（建议使用无重要信息的“小号”）。
      • servers.http部分，启用HTTP API，并设置host,port(如0.0.0.0:5700) 和access_token（设置一个复杂的字符串，用于验证）。
    • 再次运行 go-cqhttp，根据提示扫码或验证登录机器人QQ。

7.2 编写Python机器人逻辑（NoneBot2）

在项目目录下，我们为QQ机器人部分创建一个子目录。

cd ~/smart-thermometer mkdir qq_bot && cd qq_bot python3 -m venv venv source venv/bin/activate pip install nonebot2 nonebot-adapter-onebot

创建机器人入口文件bot.py：

# qq_bot/bot.py import nonebot from nonebot.adapters.onebot.v11 import Adapter as OneBotV11Adapter # 初始化 NoneBot nonebot.init() # 注册适配器 driver = nonebot.get_driver() driver.register_adapter(OneBotV11Adapter) # 加载插件（我们接下来会创建） nonebot.load_plugins("src/plugins") if __name__ == "__main__": nonebot.run()

创建插件目录和温度查询插件：

mkdir -p src/plugins/thermometer

创建src/plugins/thermometer/__init__.py：

# qq_bot/src/plugins/thermometer/__init__.py from nonebot.plugin import PluginMetadata from .query import * __plugin_meta__ = PluginMetadata( name="智能温度计", description="查询传感器温度湿度", usage="@我 温度 | 湿度 | 环境", type="application", homepage="https://github.com/yourname/smart-thermometer", supported_adapters={"~onebot.v11"}, )

创建核心插件逻辑文件src/plugins/thermometer/query.py。这里的关键是，我们的机器人需要能调用之前写好的SensorManager来获取数据。我们需要一种方式让机器人进程能访问到传感器数据。一个简单可靠的方法是让传感器读取程序将最新数据写入一个文件（如JSON），然后机器人插件去读取这个文件。

首先，修改主传感器程序，使其定期将数据写入文件。我们在项目根目录创建一个新的主程序main_daemon.py：

# ~/smart-thermometer/main_daemon.py import time import json from pathlib import Path from sensor_manager import SensorManager DATA_FILE = Path(__file__).parent / "latest_reading.json" def main(): sensor = SensorManager() print("智能温度计数据守护进程启动...") while True: try: # 读取传感器数据 reading = sensor.read_with_retry() # 将数据写入JSON文件 if reading: # 将datetime对象转换为字符串以便JSON序列化 reading_serializable = { 'timestamp': reading['timestamp'].isoformat(), 'temperature': reading['temperature'], 'humidity': reading['humidity'] } with open(DATA_FILE, 'w') as f: json.dump(reading_serializable, f, indent=2) print(f"数据已更新至文件: {reading_serializable}") else: print("读取失败，未更新数据文件。") except Exception as e: print(f"主循环发生错误: {e}") # 每10秒读取一次 time.sleep(10) if __name__ == '__main__': main()

然后，在QQ机器人插件中读取这个文件：

# qq_bot/src/plugins/thermometer/query.py import json from pathlib import Path from datetime import datetime from nonebot import on_command from nonebot.adapters.onebot.v11 import MessageEvent, GroupMessageEvent, PrivateMessageEvent from nonebot.rule import to_me # 命令触发器：当被@或私聊时，响应“温度”、“湿度”、“环境” temperature = on_command("温度", aliases={"temp", "temperature"}, rule=to_me(), priority=5) humidity = on_command("湿度", aliases={"humi", "humidity"}, rule=to_me(), priority=5) environment = on_command("环境", aliases={"env", "status"}, rule=to_me(), priority=5) # 数据文件路径，需要根据实际部署位置调整 # 假设机器人程序和主程序在同一台机器的相邻目录 DATA_FILE = Path(__file__).parent.parent.parent.parent.parent / "latest_reading.json" def get_latest_reading(): """从共享文件读取最新的传感器数据""" try: if DATA_FILE.exists(): with open(DATA_FILE, 'r') as f: data = json.load(f) # 将字符串时间转换回datetime对象（可选） data['timestamp'] = datetime.fromisoformat(data['timestamp']) return data else: return None except Exception as e: print(f"读取数据文件失败: {e}") return None @temperature.handle() async def handle_temperature(event: MessageEvent): reading = get_latest_reading() if reading: msg = f"当前温度：{reading['temperature']}°C (更新于 {reading['timestamp'].strftime('%H:%M:%S')})" else: msg = "抱歉，暂时无法获取温度数据。" await temperature.finish(msg) @humidity.handle() async def handle_humidity(event: MessageEvent): reading = get_latest_reading() if reading: msg = f"当前湿度：{reading['humidity']}% (更新于 {reading['timestamp'].strftime('%H:%M:%S')})" else: msg = "抱歉，暂时无法获取湿度数据。" await humidity.finish(msg) @environment.handle() async def handle_environment(event: MessageEvent): reading = get_latest_reading() if reading: ts = reading['timestamp'].strftime('%m-%d %H:%M:%S') msg = f"🌡️ 环境状态 [{ts}]\n温度：{reading['temperature']}°C\n湿度：{reading['humidity']}%" else: msg = "⚠️ 无法获取环境数据，请检查传感器。" await environment.finish(msg)

7.3 配置与运行

  1. 配置 NoneBot：在qq_bot目录创建.env文件，配置 go-cqhttp 的地址和访问令牌。
    HOST=127.0.0.1 PORT=8080 SUPERUSERS=["你的QQ号"] # 你的个人QQ，用于管理 COMMAND_START=["/", ""] COMMAND_SEP=["."] # OneBot 配置 ONEBOT_HTTP_URL=http://127.0.0.1:5700/ # go-cqhttp的HTTP地址 ONEBOT_ACCESS_TOKEN=your_access_token_here # 与go-cqhttp config.yml中一致
  2. 启动系统：
    • 在一个终端，运行python main_daemon.py，启动数据采集守护进程。
    • 在另一个终端，进入qq_bot目录，激活虚拟环境后运行python bot.py，启动QQ机器人。
    • 确保go-cqhttp也在运行。

现在，在你的QQ群中@机器人并发送“温度”或“环境”，它就会回复当前的温湿度数据了。私聊同样有效。

8. 系统集成与优化：让一切自动运行

现在，我们有了三个独立的功能模块：数据采集、邮件发送、微信/QQ通知。我们需要一个“指挥官”来协调它们，实现例如“每1小时采集一次数据，每6小时发送一次邮件报告，温度超过阈值时微信告警”这样的逻辑。

8.1 使用调度器：APScheduler

我们将使用APScheduler这个强大的Python库来管理定时任务。

首先在项目根目录的虚拟环境中安装它：

pip install apscheduler

创建我们的主调度程序main_scheduler.py：

# main_scheduler.py import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.cron import CronTrigger from datetime import datetime import logging from sensor_manager import SensorManager from mail_sender import MailSender from wechat_sender import WeChatSender # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("thermometer_system.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # 初始化各个模块 sensor = SensorManager() mailer = MailSender( smtp_server='smtp.qq.com', smtp_port=465, sender_email='your_email@qq.com', auth_code='your_auth_code' ) wechat = WeChatSender(sendkey='your_sendkey_here') # 收件人配置 RECIPIENT_EMAIL = 'your_notification_email@example.com' # 温度告警阈值 TEMP_ALERT_HIGH = 30.0 # 高温阈值 TEMP_ALERT_LOW = 10.0 # 低温阈值 def job_read_sensor(): """定时任务：读取传感器并更新数据文件""" logger.info("执行任务：读取传感器数据") reading = sensor.read_with_retry() # 数据已由SensorManager写入历史记录，主守护进程会负责写文件 # 这里我们主要进行阈值判断和告警 if reading and reading['temperature'] is not None: temp = reading['temperature'] # 微信告警 if temp >= TEMP_ALERT_HIGH or temp <= TEMP_ALERT_LOW: logger.warning(f"温度超出阈值！当前温度: {temp}°C，发送微信告警。") wechat.send_temperature_alert(reading, thresholds={'temp_high': TEMP_ALERT_HIGH, 'temp_low': TEMP_ALERT_LOW}) # 你也可以在这里添加其他即时通知逻辑，比如调用其他API def job_send_daily_report(): """定时任务：发送每日邮件报告（例如每天上午8点）""" logger.info("执行任务：发送每日邮件报告") reading = sensor.get_smoothed_reading() # 使用平滑后的数据 if reading: subject = f"[智能温度计] 每日环境报告 - {datetime.now().strftime('%Y-%m-%d')}" mailer.send_temperature_report(RECIPIENT_EMAIL, reading, subject_prefix="[每日报告]") else: logger.error("无法获取传感器数据，邮件报告发送取消。") def job_send_weekly_summary(): """定时任务：发送每周摘要邮件（例如每周一早上9点）""" logger.info("执行任务：发送每周摘要邮件") # 这里可以扩展，从数据库或日志文件中汇总一周的数据 # 目前我们先发送当前状态 reading = sensor.get_smoothed_reading() if reading: mailer.send_temperature_report(RECIPIENT_EMAIL, reading, subject_prefix="[每周摘要]") # 未来可以在这里添加数据分析，比如最高/最低温度，平均湿度等。 def main(): scheduler = BlockingScheduler() # 添加任务 # 每5分钟读取一次传感器（并检查告警） scheduler.add_job( job_read_sensor, IntervalTrigger(minutes=5), id='job_read_sensor', name='读取传感器数据', replace_existing=True ) # 每天上午8点发送每日报告 scheduler.add_job( job_send_daily_report, CronTrigger(hour=8, minute=0), id='job_daily_report', name='发送每日报告', replace_existing=True ) # 每周一上午9点发送每周摘要 scheduler.add_job( job_send_weekly_summary, CronTrigger(day_of_week='mon', hour=9, minute=0), id='job_weekly_summary', name='发送每周摘要', replace_existing=True ) # 立即运行一次数据读取 job_read_sensor() logger.info("智能温度计调度系统已启动！") print("调度器运行中，按 Ctrl+C 退出。") try: scheduler.start() except (KeyboardInterrupt, SystemExit): logger.info("收到停止信号，正在关闭调度器...") scheduler.shutdown() logger.info("调度器已关闭。") if __name__ == '__main__': main()

这个调度系统是项目的“大脑”。它协调了数据采集的频率、告警的判断以及定期报告的发送。你可以根据需要轻松修改时间间隔和任务逻辑。

8.2 系统守护与开机自启

为了让项目在树莓派上稳定运行，我们需要将其设置为系统服务。

创建一个服务文件/etc/systemd/system/smart-thermometer.service：

[Unit] Description=Smart Thermometer Service After=network.target [Service] Type=simple User=pi WorkingDirectory=/home/pi/smart-thermometer Environment="PATH=/home/pi/smart-thermometer/venv/bin" ExecStart=/home/pi/smart-thermometer/venv/bin/python /home/pi/smart-thermometer/main_scheduler.py Restart=on-failure RestartSec=10 [Install] WantedBy=multi-user.target

然后启用并启动服务：

sudo systemctl daemon-reload sudo systemctl enable smart-thermometer.service sudo systemctl start smart-thermometer.service

你可以用sudo systemctl status smart-thermometer.service查看运行状态。这样，即使树莓派重启，你的智能温度计也会自动运行。

8.3 扩展思路与优化建议

一个基础系统搭建完成后，可以考虑以下方向进行扩展和优化：

  1. 数据持久化：目前数据只是临时保存在内存和文件中。可以集成轻量级数据库（如 SQLite），记录历史数据，便于后续分析和绘制趋势图。
  2. Web控制面板：使用 Flask 或 FastAPI 搭建一个简单的本地网页，实时显示温湿度曲线，并可以手动触发通知、修改阈值。
  3. 多传感器支持：扩展SensorManager类，支持连接多个DHT传感器，监控不同房间的温度。
  4. 更多通知渠道：集成钉钉、飞书、Telegram等平台的机器人API。
  5. 能耗优化：树莓派一直运行可能耗电。对于电池供电场景，可以考虑使用 ESP8266/ESP32 等低功耗单片机采集数据，然后通过MQTT协议发送到树莓派或直接发送到云平台。
  6. 异常自恢复：在服务脚本中增加更完善的健康检查，比如网络断开重连、传感器永久失效报警等。

这个项目就像一颗种子，从最简单的温度读取开始，通过接入微信、QQ、邮件这些熟悉的“水管”，让数据流到你需要的地方。整个搭建过程涉及了硬件交互、网络通信、API调用、定时任务和系统服务等多个环节，是一次非常完整的物联网（IoT）入门实践。最重要的是，它解决了一个真实、具体的问题。当你躺在办公室，手机“叮”一声收到家里温度正常的微信消息，或者早上打开邮箱看到一份格式美观的环境报告时，那种自己动手创造便利的成就感，是无可替代的。

http://www.cnnetsun.cn/news/3700012.html

相关文章：

  • 如何为iOS Dev Directory贡献你的技术博客？完整流程解析
  • SmolForge自定义皮肤与动画功能实测：从原理到项目落地指南
  • 大模型组合提示技术：原理、实践与优化策略
  • 开源租赁小程序全栈开发实战：从部署到二次开发完整指南
  • Sign-Language-Interpreter-using-Deep-Learning代码解析：final.py如何实现手势捕捉与实时翻译
  • 从猴子吃桃问题解析算法思维：逆向推导、循环递归与工程实践
  • Zoplicate高级技巧：导入导出非重复条目设置，多设备同步更便捷
  • 实战指南：如何高效配置OBS虚拟摄像头实现4路视频同时分发
  • 多目标人工蜂鸟算法在移动机器人路径规划中的应用
  • 7-Zip如何成为你电脑中不可或缺的压缩工具？
  • Jellium Desktop媒体标签设置教程：配置标签的完整指南
  • Jellium Desktop启动脚本编辑器：创建与编辑启动脚本的完整指南
  • Jellium Desktop播放进度同步设置教程：配置同步
  • OpenAI API集成实战：从调用限制到稳定集成的解决方案
  • C#编程实现Windows静态IP自动配置：WMI与netsh方案详解
  • 狼群算法在柔性车间调度中的Matlab实现与应用
  • Java+Vue声纹识别门禁系统开发实践
  • C++ this指针：从隐式参数到对象模型核心机制详解
  • 影刀RPA完全指南：RPA流程系统测试规范与发布SOP完整手册
  • goimports-reviser vs goimports：为什么这款工具能提升你30%的开发效率？
  • 电竞显示器优化与《龙珠Z》主题定制指南
  • 基于3D打印机改造的自动冰球机器人：视觉识别与运动控制实践
  • 如何使用Backslash Powered Scanner发现JSON注入与服务器端请求伪造漏洞
  • C语言printf打印double输出0.000000：类型不匹配的底层原理与解决方案
  • Transformer自注意力机制原理与工程实践详解
  • 10分钟上手py-junos-eznc：从安装到执行第一个网络自动化任务
  • 基于Matlab的智能停车位识别系统设计与实现
  • mutation-summary性能优化：提升DOM监控效率的10个技巧
  • Linux软件管理与内核升级实战：从rpm/yum到编译安装的深度解析
  • 深入理解C++11内存模型：原子操作、内存序与无锁编程实战