树莓派4B传感器套件实战:从环境监测到物联网原型开发
1. 项目缘起:为什么是树莓派4B传感器套件?
如果你手头有一块树莓派4B,除了让它跑个服务器、做个媒体中心,或者当个桌面电脑用,有没有想过让它真正“感知”这个世界?这就是传感器套件存在的意义。它不是一个单一的硬件,而是一整套能让你的树莓派4B从“计算大脑”变成“感官系统”的组件集合。我最初接触这类套件,是因为想做一个能自动监测阳台植物土壤湿度、光照和温度的小系统,但又不想为了几个传感器去单独采购、焊接、调试,费时费力。一个整合好的传感器套件,就像一份为你准备好的“食材包”,让你能快速上手,专注于烹饪(也就是编程和项目实现),而不是满世界找食材。
树莓派4B作为目前性能最均衡的型号,其强大的处理能力、丰富的GPIO接口和稳定的供电,让它成为连接和控制各种传感器的绝佳平台。市面上的传感器套件五花八门,从几十元的基础包到几百元的豪华版都有。它们通常包含一些最常用、最经典的传感器模块,比如温湿度、光照、声音、红外、超声波、气体、触摸开关等等。对于初学者来说,这是一个零门槛接触物理计算和物联网的入口;对于有经验的开发者,它则是快速原型验证的利器。今天,我就以一个典型的“PI4B Sensor Kit”为例,带你从开箱到实战,完整走一遍传感器项目的开发流程,并分享一些我踩过的坑和总结的经验。
2. 套件开箱与核心组件深度解析
拿到一个传感器套件,第一步不是急着接线,而是先认识你的“士兵”。一个典型的套件通常包含以下几大类组件,理解它们的工作原理和电气特性,是后续稳定运行的基础。
2.1 传感器模块:世界的“感官”
套件里的传感器模块是核心。我们逐一拆解几个最常见的:
DHT11/DHT22温湿度传感器:这几乎是每个套件的标配。DHT11成本低,精度一般(温度±2°C,湿度±5%);DHT22精度更高(温度±0.5°C,湿度±2-5%),但价格也贵一些。它们都是单总线数字传感器,只需要一根数据线连接GPIO,使用特定的通信协议读取数据。这里有个关键点:它们对时序要求非常严格。在树莓派上,由于Linux不是实时操作系统,直接用Python的RPi.GPIO库进行位操作读取时,很容易因进程调度导致时序错乱而读取失败。我的经验是,优先使用经过优化的专用库,比如Adafruit_DHT,它底层使用了C语言扩展,时序更稳定。
HC-SR04超声波测距模块:通过发射和接收超声波来测量距离。它需要两个GPIO口,一个触发(Trig)发出脉冲,一个回声(Echo)接收返回信号。计算距离的公式是:距离 = (高电平时间 * 声速) / 2。这里最大的坑是电压匹配。HC-SR04的工作电压是5V,其Echo引脚输出也是5V电平。而树莓派GPIO的耐受电压是3.3V,直接连接有烧毁GPIO的风险!必须进行电平转换。最简单的办法是使用一个由两个电阻组成的分压电路,将5V降至3.3V左右。我常用的分压电阻组合是1kΩ和2kΩ。
光敏电阻模块:这其实是一个模拟传感器。它输出的电压值随光照强度变化。但树莓派GPIO没有模拟输入引脚(ADC),无法直接读取电压值。因此,套件里提供的“模块”实际上已经集成了一个比较器电路(比如LM393),并提供了一个数字输出引脚(DO)和一个模拟输出引脚(AO)。数字输出(DO)在光线超过/低于设定阈值时,输出高/低电平;模拟输出(AO)则需要外接一个ADC芯片(如ADS1115)才能被树莓派读取。对于初学者,先从DO口用起来,理解其开关特性。
声音传感器模块:原理和光敏电阻模块类似,通常也是基于LM393比较器,将麦克风接收到的声音信号幅度与一个可调阈值比较,输出数字信号。你可以通过旋钮调节灵敏度。它常用于制作声控开关或检测环境噪音是否超过阈值。
红外避障/循迹模块:发射红外光并接收反射光。黑色吸收红外线,白色反射红外线。因此,它可以用于循迹小车(检测地面黑线),或作为近距离避障传感器。注意,不同颜色的物体、不同材质的表面对红外线的反射率差异很大,需要在实际环境中校准阈值。
MQ-2烟雾/可燃气体传感器:这是一个模拟传感器,对液化气、丙烷、氢气、烟雾等敏感。它的输出电阻会随着气体浓度变化。同样,它需要搭配ADC才能使用。而且,这类传感器通常需要预热几十秒到几分钟才能稳定工作,读数前务必留足预热时间。
2.2 连接线与扩展板:系统的“神经网络”
套件里会有一把杜邦线(母对母、公对母、公对公)和一个面包板。面包板用于无焊接搭建电路,非常方便。但这里有一个进阶细节:线序和颜色管理。当你的项目用到5个以上传感器时,混乱的接线会成为调试的噩梦。我养成的一个习惯是,制定一个颜色规范:红色接5V/VCC,黑色或棕色接GND,黄色接信号线(如Trig),绿色接另一根信号线(如Echo),橙色接数据线(如DHT11的数据引脚)。并在代码注释或笔记本上记录每个传感器对应的GPIO引脚号。这能为你节省大量排查时间。
很多套件还会附带一个GPIO扩展板。它直接插在树莓派的40针GPIO排针上,将引脚以更友好的方式(标有名称和编号)引出来,方便插拔杜邦线。使用扩展板时,务必确认其引脚排列与树莓派原生GPIO的对应关系是否正确,有些廉价扩展板可能存在印刷错误。
2.3 其他可能组件
一些高级套件可能包含舵机(SG90)、步进电机(28BYJ-48及驱动板)、OLED显示屏(I2C接口)、继电器模块等。这些已经超出了基础传感器的范畴,进入了“执行器”和控制领域,能让你的项目从“感知”升级到“行动”。
3. 开发环境搭建与基础代码框架
硬件认识了,接下来是让它们“活”起来的软件环境。这一步的稳定性直接决定了后续开发的效率。
3.1 操作系统与基础配置
首先,为你的树莓派4B安装最新的Raspberry Pi OS(原Raspbian)。建议使用64位版本以充分发挥4B的性能。安装完成后,第一件事是启用接口:
sudo raspi-config在配置工具中,依次启用:
- I2C:用于连接OLED屏、ADC转换芯片等I2C设备。
- SPI:某些高级传感器或显示屏可能需要。
- Serial Port:但关闭串口控制台登录功能,仅保留硬件串口使能,以备连接GPS、某些串口传感器等。
- 1-Wire:如果你后续使用DS18B20这类单总线温度传感器。
启用后,重启树莓派。你可以使用命令检查I2C设备是否被识别:
sudo i2cdetect -y 1如果看到设备地址列表,说明I2C总线工作正常。
3.2 Python库的安装与管理
树莓派上最常用的传感器编程语言是Python。我们需要安装几个核心库:
sudo apt update sudo apt install python3-pip python3-venv我强烈建议为每个项目创建独立的虚拟环境,避免库版本冲突:
cd ~/your_project_folder python3 -m venv venv source venv/bin/activate在虚拟环境中安装常用库:
pip install RPi.GPIO # 基础GPIO控制 pip install smbus2 # I2C通信(Python 3) pip install adafruit-circuitpython-dht # DHT传感器(替代旧的Adafruit_DHT,更易用) pip install Pillow # 图像处理,如果用到OLED显示图片 pip install adafruit-circuitpython-ssd1306 # OLED驱动 pip install adafruit-circuitpython-ads1x15 # ADS1115/ADS1015 ADC驱动注意:
Adafruit_DHT库比较老旧,对Python 3和新系统支持有时有问题。adafruit-circuitpython-dht是Adafruit维护的新库,支持更好,但用法略有不同,需要根据其文档调整代码。
3.3 基础代码框架与异常处理
在开始为每个传感器编写代码前,建立一个健壮的基础框架至关重要。这个框架要处理GPIO的初始化、数据的读取、异常的处理以及资源的清理。
#!/usr/bin/env python3 import time import signal import sys import logging from typing import Optional, Dict, Any # 配置日志,方便调试 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class SensorBase: """传感器基类,定义通用接口和异常处理""" def __init__(self, name: str): self.name = name self._is_active = False def setup(self) -> bool: """初始化传感器,返回是否成功""" try: # 具体的初始化逻辑由子类实现 self._setup_hardware() self._is_active = True logger.info(f"Sensor {self.name} initialized successfully.") return True except Exception as e: logger.error(f"Failed to initialize sensor {self.name}: {e}") self._is_active = False return False def read(self) -> Optional[Dict[str, Any]]: """读取传感器数据,返回字典或None(如果失败)""" if not self._is_active: logger.warning(f"Sensor {self.name} is not active.") return None try: data = self._read_data() logger.debug(f"Sensor {self.name} read: {data}") return data except Exception as e: logger.error(f"Failed to read from sensor {self.name}: {e}") # 可选:尝试一次重新初始化 # self.setup() return None def cleanup(self): """清理传感器资源""" self._is_active = False self._cleanup_hardware() logger.info(f"Sensor {self.name} cleaned up.") # 以下方法需要子类重写 def _setup_hardware(self): raise NotImplementedError def _read_data(self) -> Dict[str, Any]: raise NotImplementedError def _cleanup_hardware(self): pass # 大部分传感器不需要特殊清理 def signal_handler(sig, frame): """捕获Ctrl+C信号,优雅退出""" logger.info("Interrupt received, shutting down...") # 这里可以调用所有传感器的cleanup方法 sys.exit(0) if __name__ == "__main__": signal.signal(signal.SIGINT, signal_handler) # 注册信号处理 # 你的主程序逻辑从这里开始 logger.info("Sensor application started.") # ... 初始化并运行你的传感器 # 保持主循环 try: while True: time.sleep(1) except KeyboardInterrupt: pass finally: logger.info("Application terminated.")这个框架的好处是:
- 结构化:每个传感器都是一个类,职责清晰。
- 健壮性:所有操作都有
try-except包裹,避免单个传感器故障导致整个程序崩溃。 - 可维护性:日志系统让你能清晰看到每个传感器的状态和数据。
- 优雅退出:通过信号处理,确保在程序被强制终止时能尽可能清理资源(虽然Python的GPIO库在程序退出时会尝试清理,但显式调用更可靠)。
4. 典型传感器驱动实现与避坑指南
有了框架,我们来为几个典型传感器实现具体的驱动类。这里会包含详细的接线图(文字描述)、代码和最重要的——避坑点。
4.1 DHT11/DHT22温湿度传感器实现
接线:
- VCC -> 树莓派 5V 或 3.3V(建议3.3V,与GPIO电平匹配)
- GND -> 树莓派 GND
- DATA -> 树莓派 GPIO4(或其他任意GPIO,这里以GPIO4为例)
代码实现(使用adafruit-circuitpython-dht):
import adafruit_dht import board from SensorBase import SensorBase # 假设我们继承了上面的基类 class DHT11Sensor(SensorBase): def __init__(self, pin=board.D4, sensor_type=adafruit_dht.DHT11): super().__init__("DHT11") self.dht_device = sensor_type(pin, use_pulseio=False) # use_pulseio=False 在非Linux SBC上使用,树莓派上True/False均可,False更稳定 def _setup_hardware(self): # Adafruit库的初始化在实例化时已完成,这里可以做一些预热 time.sleep(2) # 传感器首次上电需要1-2秒稳定 def _read_data(self): # 尝试读取,可能会因时序问题失败 for _ in range(3): # 重试3次 try: temperature_c = self.dht_device.temperature humidity = self.dht_device.humidity if temperature_c is not None and humidity is not None: return { "temperature_c": round(temperature_c, 1), "temperature_f": round(temperature_c * 9.0 / 5.0 + 32.0, 1), "humidity": round(humidity, 1) } except RuntimeError as e: logger.debug(f"DHT read attempt failed: {e}") time.sleep(0.5) # 失败后稍等再试 # 如果三次都失败,抛出异常,让基类处理 raise RuntimeError("Failed to read from DHT sensor after 3 attempts") def _cleanup_hardware(self): self.dht_device.exit()避坑指南:
- 读取失败率高:这是DHT系列传感器最常见的问题。除了使用高质量的库,硬件上也要注意:DATA引脚需要接一个4.7kΩ - 10kΩ的上拉电阻到VCC(3.3V)。很多模块已经内置了这个电阻,但如果你用的是裸传感器,必须自己加。没有上拉电阻,信号线无法稳定在高电平,通信必然失败。
- 读取间隔:DHT11两次读取间隔至少需要1秒,DHT22至少需要2秒。频繁读取会导致传感器无响应。在代码中必须加入延时。
- 电源噪声:如果从树莓派GPIO取电,当GPIO上有大电流负载(如舵机)突然启动时,可能会引起电源电压瞬间跌落,导致DHT传感器复位或读取出错。对于要求高的应用,建议为传感器提供独立的、经过滤波的3.3V电源。
4.2 HC-SR04超声波传感器实现(带电平转换)
接线(使用分压电路):
- VCC -> 树莓派 5V
- GND -> 树莓派 GND
- Trig -> 树莓派 GPIO23(无需电平转换,输出3.3V足够触发)
- Echo -> 分压电路 -> 树莓派 GPIO24
- 分压电路:Echo引脚接一个1kΩ电阻,电阻另一端接GPIO24,同时从GPIO24和GND之间接一个2kΩ电阻。这样5V输入会被分压到约3.33V。
代码实现:
import RPi.GPIO as GPIO from SensorBase import SensorBase class HCSR04Sensor(SensorBase): def __init__(self, trig_pin=23, echo_pin=24): super().__init__("HC-SR04") self.TRIG = trig_pin self.ECHO = echo_pin self.speed_of_sound = 34300 # 厘米/秒 (在20°C干燥空气中) def _setup_hardware(self): GPIO.setmode(GPIO.BCM) GPIO.setup(self.TRIG, GPIO.OUT) GPIO.setup(self.ECHO, GPIO.IN) GPIO.output(self.TRIG, False) time.sleep(0.5) # 让传感器稳定 def _read_data(self): # 发送10us的高电平脉冲触发 GPIO.output(self.TRIG, True) time.sleep(0.00001) # 10微秒 GPIO.output(self.TRIG, False) # 等待回声引脚变高(开始计时) pulse_start = time.time() timeout = pulse_start + 0.04 # 最大测量距离约7米,对应约40ms回波时间 while GPIO.input(self.ECHO) == 0 and time.time() < timeout: pulse_start = time.time() if time.time() >= timeout: raise RuntimeError("Echo start timeout") # 等待回声引脚变低(结束计时) pulse_end = time.time() while GPIO.input(self.ECHO) == 1 and time.time() < timeout: pulse_end = time.time() if time.time() >= timeout: raise RuntimeError("Echo end timeout") # 计算高电平持续时间 pulse_duration = pulse_end - pulse_start # 计算距离 (距离 = 时间 * 声速 / 2) distance = pulse_duration * self.speed_of_sound / 2.0 # 有效距离通常在2cm到400cm之间,过滤异常值 if 2 < distance < 400: return {"distance_cm": round(distance, 2)} else: raise RuntimeError(f"Distance out of range: {distance}cm") def _cleanup_hardware(self): # 注意:GPIO.cleanup()会清理所有GPIO设置,如果多个传感器共用GPIO,应由主程序统一调用 # 这里我们只标记本传感器不再使用,实际清理放在主程序 pass避坑指南:
- 电平转换是必须的:重申一次,5V的Echo信号直接接3.3V的GPIO,长期使用必烧引脚。分压电路是最廉价可靠的方案。
- 测量盲区:HC-SR04在传感器前方2cm以内无法准确测量,因为此时回波和发射波可能重叠。如果你的应用需要检测非常近的物体,需要考虑其他传感器(如红外接近开关)。
- 环境干扰:超声波在空气中传播受温度、湿度影响很大。上述代码使用了固定的声速(20°C干燥空气)。对于精度要求高的应用,需要加入温度补偿公式:
声速 = 331.3 + 0.606 * 温度(°C)m/s。你可以结合DHT11测得的温度来动态计算声速。 - 多传感器干扰:如果同时使用多个HC-SR04,它们的超声波可能会互相干扰。解决方案是分时工作,或者使用不同频率的超声波传感器(但HC-SR04频率固定为40kHz)。
4.3 光敏与声音传感器(数字模式)实现
这两个模块用法类似,都是读取数字输出引脚的高低电平。
接线(以光敏传感器为例):
- VCC -> 树莓派 3.3V
- GND -> 树莓派 GND
- DO -> 树莓派 GPIO17
- (AO悬空,暂时不用)
代码实现(通用数字传感器类):
import RPi.GPIO as GPIO class DigitalSensor(SensorBase): """通用数字传感器类,用于光敏、声音、红外避障等""" def __init__(self, name, data_pin, active_low=False): """ :param name: 传感器名称 :param data_pin: 数据引脚 (BCM编号) :param active_low: 是否为低电平有效。例如,光敏传感器常亮时输出低电平,遮光时高电平。 """ super().__init__(name) self.DATA_PIN = data_pin self.ACTIVE_LOW = active_low self._last_state = None def _setup_hardware(self): GPIO.setmode(GPIO.BCM) GPIO.setup(self.DATA_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP) # 启用内部上拉,防止引脚悬空 def _read_data(self): current_state = GPIO.input(self.DATA_PIN) # 如果定义了低电平有效,则反转逻辑 if self.ACTIVE_LOW: current_state = not current_state # 检测状态变化(可选,用于事件触发) state_changed = (self._last_state is not None and current_state != self._last_state) self._last_state = current_state return { "value": current_state, # 1或0 "state_changed": state_changed, "description": "Detected" if current_state else "Not detected" } # 使用示例 light_sensor = DigitalSensor("LightSensor", data_pin=17, active_low=True) sound_sensor = DigitalSensor("SoundSensor", data_pin=27, active_low=False) # 声音触发时通常输出高电平避坑指南:
- 引脚悬空:数字输入引脚在不连接时处于“悬空”状态,会随机读取到高或低电平,产生误触发。务必启用内部上拉或下拉电阻。
GPIO.PUD_UP(上拉到3.3V)或GPIO.PUD_DOWN(下拉到GND)的选择取决于你的传感器模块的默认输出电平。如果不确定,可以先不接传感器,读取引脚电平,接上传感器后再读取,根据变化判断。 - 阈值调节:模块上的蓝色电位器是用来调节灵敏度的。用一把小螺丝刀,在传感器处于临界状态时(比如光线明暗交界处)缓慢旋转,直到输出状态发生变化。调好后点一点热熔胶固定,防止震动导致阈值漂移。
- 开关抖动:当检测到状态变化(如由暗变亮)时,信号可能在短时间内快速抖动多次。如果直接用这个变化去触发某个动作(如开灯),可能会导致动作重复执行。解决方法是在软件中加入“防抖”逻辑,即检测到变化后,等待几十毫秒再读取一次,确认状态是否稳定。
5. 数据融合与项目实战:构建一个环境监测站
单独读取传感器数据只是第一步。真正的价值在于将多个传感器的数据融合起来,做出判断或呈现。我们用一个综合项目来串联以上知识:构建一个带显示和报警功能的环境监测站。
项目目标:实时监测室内温度、湿度、光照和噪音水平,将数据显示在OLED屏幕上,当温度超过阈值或持续检测到噪音时,通过命令行输出报警信息,并可将数据记录到文件。
硬件清单:
- 树莓派4B
- DHT22传感器(更高精度)
- 光敏电阻模块(数字输出)
- 声音传感器模块(数字输出)
- 0.96寸OLED显示屏(I2C接口,SSD1306驱动)
- 面包板和杜邦线
接线概要:
- DHT22: VCC->3.3V, GND->GND, DATA->GPIO4
- 光敏模块: VCC->3.3V, GND->GND, DO->GPIO17
- 声音模块: VCC->3.3V, GND->GND, DO->GPIO27
- OLED: VCC->3.3V, GND->GND, SCL->GPIO3 (BCM2, I2C SCL), SDA->GPIO2 (BCM3, I2C SDA)
软件架构: 我们将使用多线程。主线程负责协调和显示,每个传感器一个独立的读取线程,并将数据放入一个线程安全的队列中。主线程定时从队列中取出最新数据,更新显示并检查报警条件。
import threading import queue import json from datetime import datetime from pathlib import Path # 假设我们已经有了DHT22Sensor, DigitalSensor类 # 假设我们有一个OledDisplay类(使用adafruit-circuitpython-ssd1306) class EnvironmentalMonitor: def __init__(self, config): self.config = config self.data_queue = queue.Queue() self.sensors = {} self.display = None self.running = False self.data_log = [] # 报警阈值 self.temp_high_threshold = 30.0 # 摄氏度 self.noise_duration_threshold = 5.0 # 持续秒数 self.noise_start_time = None def setup(self): logger.info("Setting up Environmental Monitor...") # 1. 初始化传感器 self.sensors['dht22'] = DHT22Sensor(pin=board.D4) self.sensors['light'] = DigitalSensor("Light", 17, active_low=True) self.sensors['sound'] = DigitalSensor("Sound", 27, active_low=False) for name, sensor in self.sensors.items(): if not sensor.setup(): logger.error(f"Failed to setup {name}, exiting.") return False # 2. 初始化显示屏 try: self.display = OledDisplay() # 需要实现这个类 self.display.setup() self.display.show_text("System Booting...", 0, 0) except Exception as e: logger.error(f"Failed to setup display: {e}") # 显示屏失败不影响核心功能,继续运行 self.display = None # 3. 启动传感器读取线程 self.running = True for name, sensor in self.sensors.items(): thread = threading.Thread(target=self._sensor_reading_worker, args=(name, sensor), daemon=True) thread.start() logger.info("Environmental Monitor setup complete.") return True def _sensor_reading_worker(self, name, sensor): """传感器读取工作线程""" read_interval = { 'dht22': 2.5, # DHT22需要至少2秒间隔 'light': 0.1, 'sound': 0.05 }.get(name, 1.0) while self.running: try: data = sensor.read() if data is not None: data['sensor'] = name data['timestamp'] = datetime.now().isoformat() self.data_queue.put(data) except Exception as e: logger.error(f"Error in {name} worker: {e}") time.sleep(read_interval) def _process_data(self): """主循环中调用,处理队列中的数据""" current_data = {'temperature': None, 'humidity': None, 'light': None, 'sound': None} while not self.data_queue.empty(): try: item = self.data_queue.get_nowait() sensor_name = item.pop('sensor') if sensor_name == 'dht22': current_data['temperature'] = item.get('temperature_c') current_data['humidity'] = item.get('humidity') elif sensor_name == 'light': current_data['light'] = item.get('value') elif sensor_name == 'sound': current_data['sound'] = item.get('value') # 可以在这里记录原始数据到日志文件 self._log_raw_data(item) except queue.Empty: break # 更新显示 if self.display: self._update_display(current_data) # 检查报警条件 self._check_alarms(current_data) # 保存当前状态(例如每分钟保存一次) self.data_log.append({**current_data, 'timestamp': datetime.now().isoformat()}) if len(self.data_log) > 3600: # 保留一小时的数据(假设每秒一次) self.data_log.pop(0) def _update_display(self, data): """在OLED上更新数据""" lines = [] if data['temperature'] is not None: lines.append(f"Temp: {data['temperature']:.1f}C") if data['humidity'] is not None: lines.append(f"Hum: {data['humidity']:.1f}%") lines.append(f"Light: {'ON ' if data['light'] else 'OFF'}") lines.append(f"Sound: {'YES' if data['sound'] else 'NO '}") # 调用OLED驱动显示这4行文字 if self.display: self.display.show_multiline(lines) def _check_alarms(self, data): """检查并触发报警""" # 高温报警 if data['temperature'] is not None and data['temperature'] > self.temp_high_threshold: logger.warning(f"High temperature alert: {data['temperature']:.1f}C") if self.display: self.display.show_alert("HIGH TEMP!") # 持续噪音报警 if data['sound']: # 检测到声音 if self.noise_start_time is None: self.noise_start_time = time.time() else: duration = time.time() - self.noise_start_time if duration > self.noise_duration_threshold: logger.warning(f"Continuous noise alert for {duration:.1f}s") if self.display: self.display.show_alert("NOISE!") else: self.noise_start_time = None # 噪音停止,重置计时器 def _log_raw_data(self, data): """将原始数据记录到文件(可选)""" log_file = Path.home() / "sensor_log.jsonl" try: with open(log_file, 'a') as f: f.write(json.dumps(data) + "\n") except Exception as e: logger.error(f"Failed to log data: {e}") def run(self): """主运行循环""" logger.info("Starting main loop...") try: while self.running: self._process_data() time.sleep(0.5) # 主循环频率 except KeyboardInterrupt: logger.info("Main loop interrupted.") finally: self.shutdown() def shutdown(self): """清理资源""" logger.info("Shutting down...") self.running = False time.sleep(1.1) # 等待所有工作线程结束循环 for sensor in self.sensors.values(): sensor.cleanup() if self.display: self.display.cleanup() GPIO.cleanup() # 最后统一清理GPIO logger.info("Shutdown complete.") if __name__ == "__main__": monitor = EnvironmentalMonitor({}) if monitor.setup(): monitor.run()项目避坑与优化:
- 线程安全:我们使用了
queue.Queue,它是线程安全的,确保传感器工作线程和主线程之间安全地传递数据。 - 不同的读取频率:不同传感器对读取间隔要求不同。DHT22需要2秒以上,而声音传感器可能需要毫秒级的检测频率。为每个传感器工作线程设置独立的
sleep间隔是合理的。 - 显示刷新优化:OLED屏幕刷新整个画面相对较慢。如果每次只更新变化的数据行,而不是重刷全部文本,可以提升效率并避免屏幕闪烁。这需要在
OledDisplay类中实现更精细的控制。 - 数据持久化:上述代码使用JSON Lines格式记录原始数据,每行一个JSON对象。这种格式易于追加和后续分析(可以用
jq命令行工具或Pandas处理)。对于长期运行,需要考虑日志轮转,避免单个文件过大。 - 报警防抖:对于噪音报警,我们实现了简单的持续时间判断。对于温度报警,可以加入“迟滞”功能,例如温度超过30°C报警,但需要降到28°C以下才解除报警,避免在阈值附近频繁开关报警。
这个环境监测站项目,从硬件连接到软件架构,覆盖了传感器套件应用的绝大多数核心环节。你可以在此基础上扩展,比如加入网页界面(用Flask)、数据上传到云端、或者控制继电器打开风扇降温。树莓派和传感器套件的魅力就在于,它为你提供了一个无限可能的起点,剩下的就交给你的想象力了。
