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的型号为例作为主控平台。连接非常简单只需要三根母对母杜邦线。所需材料树莓派任何型号均可需启用GPIO。DHT11或DHT22传感器模块注意是模块上面自带了一个上拉电阻如果是单独的传感器芯片需要自己连接一个4.7K-10KΩ的上拉电阻到数据线。三根母对母杜邦线。连接方式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_DHT4. 核心功能实现温度读取与数据格式化环境准备好后我们来编写最核心的温度读取代码。创建一个名为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第{i1}次读取: {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(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class SensorManager: def __init__(self, sensor_typeAdafruit_DHT.DHT22, gpio_pin4, max_retries3): 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, retries1, delay_seconds1) 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等都要求使用“授权码”代替密码。登录QQ邮箱点击“设置” - “账户”。找到“POP3/IMAP/SMTP/Exchange/CardDAV/CalDAV服务”部分。开启“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 pstrong报告时间/strong{sensor_reading[timestamp].strftime(%Y-%m-%d %H:%M:%S)}/p table border1 cellpadding10 styleborder-collapse: collapse; tr stylebackground-color: #f2f2f2; th指标/th th数值/th th状态/th /tr tr td️ 温度/td tdstrong{sensor_reading[temperature]} °C/strong/td td{self._get_temperature_status(sensor_reading[temperature])}/td /tr tr td 湿度/td tdstrong{sensor_reading[humidity]} %/strong/td td{self._get_humidity_status(sensor_reading[humidity])}/td /tr /table psmall此邮件由智能温度计系统自动发送。/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_numberqq.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版访问 Server酱官网 并登录可用GitHub账号。进入“SendKey”页面你会看到一个SENDKEY。点击“微信推送”按照指引扫码关注一个企业微信应用即可绑定微信。复制你的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 fhttps://sctapi.ftqq.com/{sendkey}.send def send_text_message(self, title, content, channelNone): 发送文本消息到微信 :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, datadata, timeout10) 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-cqhttpgo-cqhttp 是一个基于 Mirai 以及 MiraiGo 的 QQ 机器人框架它负责与QQ服务器通信。我们需要先部署它。下载 go-cqhttp从 GitHub 发布页下载对应你树莓派系统架构的版本通常是arm64或armv7。配置与登录解压后首次运行./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我 温度 | 湿度 | 环境, typeapplication, homepagehttps://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, indent2) 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}, ruleto_me(), priority5) humidity on_command(湿度, aliases{humi, humidity}, ruleto_me(), priority5) environment on_command(环境, aliases{env, status}, ruleto_me(), priority5) # 数据文件路径需要根据实际部署位置调整 # 假设机器人程序和主程序在同一台机器的相邻目录 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 配置与运行配置 NoneBot在qq_bot目录创建.env文件配置 go-cqhttp 的地址和访问令牌。HOST127.0.0.1 PORT8080 SUPERUSERS[你的QQ号] # 你的个人QQ用于管理 COMMAND_START[/, ] COMMAND_SEP[.] # OneBot 配置 ONEBOT_HTTP_URLhttp://127.0.0.1:5700/ # go-cqhttp的HTTP地址 ONEBOT_ACCESS_TOKENyour_access_token_here # 与go-cqhttp config.yml中一致启动系统在一个终端运行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( levellogging.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_serversmtp.qq.com, smtp_port465, sender_emailyour_emailqq.com, auth_codeyour_auth_code ) wechat WeChatSender(sendkeyyour_sendkey_here) # 收件人配置 RECIPIENT_EMAIL your_notification_emailexample.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(minutes5), idjob_read_sensor, name读取传感器数据, replace_existingTrue ) # 每天上午8点发送每日报告 scheduler.add_job( job_send_daily_report, CronTrigger(hour8, minute0), idjob_daily_report, name发送每日报告, replace_existingTrue ) # 每周一上午9点发送每周摘要 scheduler.add_job( job_send_weekly_summary, CronTrigger(day_of_weekmon, hour9, minute0), idjob_weekly_summary, name发送每周摘要, replace_existingTrue ) # 立即运行一次数据读取 job_read_sensor() logger.info(智能温度计调度系统已启动) print(调度器运行中按 CtrlC 退出。) 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] DescriptionSmart Thermometer Service Afternetwork.target [Service] Typesimple Userpi WorkingDirectory/home/pi/smart-thermometer EnvironmentPATH/home/pi/smart-thermometer/venv/bin ExecStart/home/pi/smart-thermometer/venv/bin/python /home/pi/smart-thermometer/main_scheduler.py Restarton-failure RestartSec10 [Install] WantedBymulti-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 扩展思路与优化建议一个基础系统搭建完成后可以考虑以下方向进行扩展和优化数据持久化目前数据只是临时保存在内存和文件中。可以集成轻量级数据库如 SQLite记录历史数据便于后续分析和绘制趋势图。Web控制面板使用 Flask 或 FastAPI 搭建一个简单的本地网页实时显示温湿度曲线并可以手动触发通知、修改阈值。多传感器支持扩展SensorManager类支持连接多个DHT传感器监控不同房间的温度。更多通知渠道集成钉钉、飞书、Telegram等平台的机器人API。能耗优化树莓派一直运行可能耗电。对于电池供电场景可以考虑使用 ESP8266/ESP32 等低功耗单片机采集数据然后通过MQTT协议发送到树莓派或直接发送到云平台。异常自恢复在服务脚本中增加更完善的健康检查比如网络断开重连、传感器永久失效报警等。这个项目就像一颗种子从最简单的温度读取开始通过接入微信、QQ、邮件这些熟悉的“水管”让数据流到你需要的地方。整个搭建过程涉及了硬件交互、网络通信、API调用、定时任务和系统服务等多个环节是一次非常完整的物联网IoT入门实践。最重要的是它解决了一个真实、具体的问题。当你躺在办公室手机“叮”一声收到家里温度正常的微信消息或者早上打开邮箱看到一份格式美观的环境报告时那种自己动手创造便利的成就感是无可替代的。
Python物联网实战：基于树莓派与DHT传感器打造微信QQ邮件智能温湿度监控系统
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的型号为例作为主控平台。连接非常简单只需要三根母对母杜邦线。所需材料树莓派任何型号均可需启用GPIO。DHT11或DHT22传感器模块注意是模块上面自带了一个上拉电阻如果是单独的传感器芯片需要自己连接一个4.7K-10KΩ的上拉电阻到数据线。三根母对母杜邦线。连接方式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_DHT4. 核心功能实现温度读取与数据格式化环境准备好后我们来编写最核心的温度读取代码。创建一个名为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第{i1}次读取: {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(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) class SensorManager: def __init__(self, sensor_typeAdafruit_DHT.DHT22, gpio_pin4, max_retries3): 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, retries1, delay_seconds1) 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等都要求使用“授权码”代替密码。登录QQ邮箱点击“设置” - “账户”。找到“POP3/IMAP/SMTP/Exchange/CardDAV/CalDAV服务”部分。开启“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 pstrong报告时间/strong{sensor_reading[timestamp].strftime(%Y-%m-%d %H:%M:%S)}/p table border1 cellpadding10 styleborder-collapse: collapse; tr stylebackground-color: #f2f2f2; th指标/th th数值/th th状态/th /tr tr td️ 温度/td tdstrong{sensor_reading[temperature]} °C/strong/td td{self._get_temperature_status(sensor_reading[temperature])}/td /tr tr td 湿度/td tdstrong{sensor_reading[humidity]} %/strong/td td{self._get_humidity_status(sensor_reading[humidity])}/td /tr /table psmall此邮件由智能温度计系统自动发送。/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_numberqq.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版访问 Server酱官网 并登录可用GitHub账号。进入“SendKey”页面你会看到一个SENDKEY。点击“微信推送”按照指引扫码关注一个企业微信应用即可绑定微信。复制你的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 fhttps://sctapi.ftqq.com/{sendkey}.send def send_text_message(self, title, content, channelNone): 发送文本消息到微信 :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, datadata, timeout10) 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-cqhttpgo-cqhttp 是一个基于 Mirai 以及 MiraiGo 的 QQ 机器人框架它负责与QQ服务器通信。我们需要先部署它。下载 go-cqhttp从 GitHub 发布页下载对应你树莓派系统架构的版本通常是arm64或armv7。配置与登录解压后首次运行./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我 温度 | 湿度 | 环境, typeapplication, homepagehttps://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, indent2) 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}, ruleto_me(), priority5) humidity on_command(湿度, aliases{humi, humidity}, ruleto_me(), priority5) environment on_command(环境, aliases{env, status}, ruleto_me(), priority5) # 数据文件路径需要根据实际部署位置调整 # 假设机器人程序和主程序在同一台机器的相邻目录 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 配置与运行配置 NoneBot在qq_bot目录创建.env文件配置 go-cqhttp 的地址和访问令牌。HOST127.0.0.1 PORT8080 SUPERUSERS[你的QQ号] # 你的个人QQ用于管理 COMMAND_START[/, ] COMMAND_SEP[.] # OneBot 配置 ONEBOT_HTTP_URLhttp://127.0.0.1:5700/ # go-cqhttp的HTTP地址 ONEBOT_ACCESS_TOKENyour_access_token_here # 与go-cqhttp config.yml中一致启动系统在一个终端运行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( levellogging.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_serversmtp.qq.com, smtp_port465, sender_emailyour_emailqq.com, auth_codeyour_auth_code ) wechat WeChatSender(sendkeyyour_sendkey_here) # 收件人配置 RECIPIENT_EMAIL your_notification_emailexample.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(minutes5), idjob_read_sensor, name读取传感器数据, replace_existingTrue ) # 每天上午8点发送每日报告 scheduler.add_job( job_send_daily_report, CronTrigger(hour8, minute0), idjob_daily_report, name发送每日报告, replace_existingTrue ) # 每周一上午9点发送每周摘要 scheduler.add_job( job_send_weekly_summary, CronTrigger(day_of_weekmon, hour9, minute0), idjob_weekly_summary, name发送每周摘要, replace_existingTrue ) # 立即运行一次数据读取 job_read_sensor() logger.info(智能温度计调度系统已启动) print(调度器运行中按 CtrlC 退出。) 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] DescriptionSmart Thermometer Service Afternetwork.target [Service] Typesimple Userpi WorkingDirectory/home/pi/smart-thermometer EnvironmentPATH/home/pi/smart-thermometer/venv/bin ExecStart/home/pi/smart-thermometer/venv/bin/python /home/pi/smart-thermometer/main_scheduler.py Restarton-failure RestartSec10 [Install] WantedBymulti-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 扩展思路与优化建议一个基础系统搭建完成后可以考虑以下方向进行扩展和优化数据持久化目前数据只是临时保存在内存和文件中。可以集成轻量级数据库如 SQLite记录历史数据便于后续分析和绘制趋势图。Web控制面板使用 Flask 或 FastAPI 搭建一个简单的本地网页实时显示温湿度曲线并可以手动触发通知、修改阈值。多传感器支持扩展SensorManager类支持连接多个DHT传感器监控不同房间的温度。更多通知渠道集成钉钉、飞书、Telegram等平台的机器人API。能耗优化树莓派一直运行可能耗电。对于电池供电场景可以考虑使用 ESP8266/ESP32 等低功耗单片机采集数据然后通过MQTT协议发送到树莓派或直接发送到云平台。异常自恢复在服务脚本中增加更完善的健康检查比如网络断开重连、传感器永久失效报警等。这个项目就像一颗种子从最简单的温度读取开始通过接入微信、QQ、邮件这些熟悉的“水管”让数据流到你需要的地方。整个搭建过程涉及了硬件交互、网络通信、API调用、定时任务和系统服务等多个环节是一次非常完整的物联网IoT入门实践。最重要的是它解决了一个真实、具体的问题。当你躺在办公室手机“叮”一声收到家里温度正常的微信消息或者早上打开邮箱看到一份格式美观的环境报告时那种自己动手创造便利的成就感是无可替代的。
