4、传感器数据采集实战：使用Python与树莓派/Jetson Nano读取DHT22数据

好，咱们进入实战环节。

前面讲了那么多理论，什么PID啊、前馈啊，最终都得落到传感器数据上。数据不准，后面全是白搭。我个人习惯，做温室项目第一步，先把传感器调通，数据能稳定读上来，心里才有底。

4.1 为什么选DHT22？

市面上温湿度传感器很多，DHT11、DHT22、SHT30、BME280……我为什么推荐DHT22？

说白了，性价比。DHT11太粗糙，精度±2°C，做温室调控根本不够用。SHT30精度高，但贵，而且I2C通信对新手不太友好。DHT22精度±0.5°C，湿度±2%RH，单总线协议，一根线就能搞定。我在项目中遇到过用DHT11做育苗温室，结果温度偏差大，苗子全徒长了……后来全换成DHT22，再没出过类似问题。

参数 DHT11 DHT22
温度精度 ±2°C ±0.5°C
湿度精度 ±5%RH ±2%RH
采样周期 1秒 2秒
通信协议 单总线 单总线
价格 约5元 约15元
我的建议：温室项目别省这10块钱。DHT22多花的钱，后面调试省下的时间绝对值回来。

4.2 硬件接线

接线其实很简单。DHT22有四个引脚，但实际只用三个：VCC、DATA、GND。有个NC脚是空脚，不用管。

你想想看，树莓派和Jetson Nano的GPIO都是3.3V逻辑电平，而DHT22工作电压是3.3-5V。这里有个坑——数据引脚必须接上拉电阻。我曾经第一次调DHT22时，没接上拉电阻，数据死活读不出来，折腾了一下午才发现是这个问题。

具体接线：

  • VCC → 3.3V（树莓派Pin1，Jetson Nano Pin1）
  • DATA → GPIO4（树莓派Pin7，Jetson Nano Pin7）
  • GND → GND（树莓派Pin6，Jetson Nano Pin6）
  • DATA与VCC之间接一个4.7kΩ上拉电阻
注意：上拉电阻不能省！有些模块板载了上拉电阻，但大部分裸板没有。买模块时问清楚，或者干脆自己焊一个。

4.3 安装依赖库

Python读取DHT22，最常用的库是Adafruit CircuitPython DHT。为什么不用C语言？嗯，Python开发快，调试方便，原型验证阶段效率高。等产品定型了，再考虑用C重写底层驱动。

安装步骤：

# 更新系统包
sudo apt update
sudo apt upgrade -y

# 安装Python3和pip（一般树莓派系统自带）
sudo apt install python3-pip

# 安装Adafruit CircuitPython库
pip3 install adafruit-circuitpython-dht

# 如果使用树莓派，还需要安装libgpiod
sudo apt install libgpiod2

Jetson Nano上稍微有点不同。我记得第一次在Jetson Nano上跑DHT22，直接报错说找不到设备。后来发现需要手动安装Jetson GPIO库：

# Jetson Nano额外步骤
pip3 install Jetson.GPIO
sudo groupadd -f -r gpio
sudo usermod -a -G gpio $USER
# 然后重启

4.4 编写读取代码

好，硬件接好了，库装好了，咱们写代码。

核心逻辑其实就三步：初始化传感器、循环读取、处理异常。DHT22的采样周期是2秒，你读太快它会报错。我刚开始写的时候没注意这个，循环里设了0.5秒间隔，结果全是CRC校验错误。

import time
import board
import adafruit_dht

# 初始化DHT22，连接到GPIO4
dht_device = adafruit_dht.DHT22(board.D4)

while True:
    try:
        # 读取温湿度
        temperature = dht_device.temperature
        humidity = dht_device.humidity
        
        # 检查数据是否有效
        if temperature is not None and humidity is not None:
            print(f"温度: {temperature:.1f}°C, 湿度: {humidity:.1f}%")
        else:
            print("读取失败，重试中...")
            
    except RuntimeError as error:
        # DHT22偶尔会通信失败，这是正常的
        print(f"读取异常: {error.args[0]}")
        
    except Exception as error:
        # 其他异常，比如传感器断开
        print(f"严重错误: {error}")
        dht_device.exit()
        raise
        
    # 必须等待至少2秒
    time.sleep(2.0)
关键点：try-except不能省。DHT22的单总线协议对时序要求很严格，树莓派不是实时系统，偶尔会丢数据。我见过有人不加异常处理，程序跑半小时就崩了。

4.5 数据校验与滤波

原始数据读上来了，但直接用吗？不行。你想想看，温室里风机一开，温度传感器数值会剧烈跳动。直接拿这个值去控制设备，执行器会频繁启停，寿命大打折扣。

我常用的方法：滑动平均滤波。取最近5次数据求平均，能有效抑制噪声。

class SensorFilter:
    def __init__(self, window_size=5):
        self.window_size = window_size
        self.temp_buffer = []
        self.humid_buffer = []
    
    def add_reading(self, temp, humid):
        self.temp_buffer.append(temp)
        self.humid_buffer.append(humid)
        
        # 只保留最近N个数据
        if len(self.temp_buffer) > self.window_size:
            self.temp_buffer.pop(0)
        if len(self.humid_buffer) > self.window_size:
            self.humid_buffer.pop(0)
    
    def get_filtered(self):
        if len(self.temp_buffer) == 0:
            return None, None
        avg_temp = sum(self.temp_buffer) / len(self.temp_buffer)
        avg_humid = sum(self.humid_buffer) / len(self.humid_buffer)
        return avg_temp, avg_humid

# 使用示例
filter = SensorFilter(window_size=5)

while True:
    try:
        temp = dht_device.temperature
        humid = dht_device.humidity
        
        if temp is not None and humid is not None:
            filter.add_reading(temp, humid)
            f_temp, f_humid = filter.get_filtered()
            print(f"原始: {temp:.1f}°C, 滤波后: {f_temp:.1f}°C")
            
    except RuntimeError:
        pass
        
    time.sleep(2.0)

4.6 常见问题与避坑指南

做这个实验，你大概率会遇到下面几个问题。我一个个说：

  • 读不到数据：先检查接线，上拉电阻接了没？GPIO号对不对？我曾经犯过一个低级错误，把DATA线插到了3.3V上，传感器直接冒烟了……
  • CRC校验错误：这是DHT22的老毛病。单总线对时序敏感，树莓派后台跑的任务多了就会丢数据。解决方案：加重试机制，连续3次失败才报错。
  • 数值跳变：比如温度突然从25°C跳到50°C。这通常是电磁干扰。温室里变频器、水泵启动时会产生强干扰。解决办法：数据线用屏蔽线，或者加软件滤波。
  • 传感器发热：DHT22工作时会自发热，如果通风不好，测出来的温度会比实际高1-2°C。我建议传感器不要紧贴外壳，留点空隙。
我的经验：量产项目里，我会在PCB上预留两个传感器接口。一个DHT22做主传感器，一个DS18B20做温度备份。万一主传感器挂了，系统自动切换，不影响生产。

4.7 数据落盘与可视化

数据读出来了，总得存下来吧？调试阶段，我习惯把数据同时打印到终端和写入CSV文件。这样既能实时看，又能事后分析。

import csv
from datetime import datetime

csv_file = open('greenhouse_data.csv', 'a', newline='')
csv_writer = csv.writer(csv_file)
# 写入表头
csv_writer.writerow(['时间', '温度(°C)', '湿度(%)'])

while True:
    try:
        temp = dht_device.temperature
        humid = dht_device.humidity
        
        if temp is not None:
            now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            csv_writer.writerow([now, f"{temp:.1f}", f"{humid:.1f}"])
            csv_file.flush()  # 立即写入磁盘
            print(f"[{now}] {temp:.1f}°C, {humid:.1f}%")
            
    except RuntimeError:
        pass
        
    time.sleep(2.0)

嗯，这里要注意：flush()一定要加。Python的文件写入是有缓冲的，不加flush的话，程序崩溃时最后几条数据就丢了。我在现场调试时吃过这个亏，数据丢了半小时，找原因找了半天。

好了，DHT22的数据采集就讲到这里。下一章咱们会把这些数据送到数据库里，然后开始做真正的闭环控制。你先把这段代码跑通，有问题随时翻回来看看。

← 上一章 📖 返回目录 下一章 →