EEPW论坛

【树莓派5】环境气体监测

本文介绍了树莓派 5 结合 Sensirion SEN66 环境气体传感器套件，实现环境气体监测，并通过 MQTT 协议上传至 Home Assistant 智能家居平台，实现物联网远程环境监测终端的项目设计。

项目介绍

• 准备工作：硬件连接、软件包安装、Docker部署、EMQX和HomeAssistant平台部署；

• 工程测试：加载官方 Demo 例程，实现 SEN66 传感器数据终端打印；

• MQTT：结合 MQTT 协议，实现传感器数据上传至 EMQX 服务器平台，并远程读取；

• Home Assistant：通过 MQTT 集成，实现 SEN66 传感器数据卡片自动获取，实现物联网环境气体监测。

准备工作

包括硬件连接、软件包安装、Home Assistant 平台搭建等。

硬件连接

默认情况下，SEN66 示例程序设定传感器已连接到端口 /dev/i2c-1 ；

实物图

树莓派 40-Pin 接口定义

SEN66 引脚

引脚定义及描述

详见：https://github.com/Sensirion/raspberry-pi-i2c-sen66/

软件包安装

终端执行如下指令，安装 Sensirion Sen66 库

 pip install sensirion_i2c_sen66

详见：https://sensirion.github.io/python-i2c-sen66/index.html

搭建 Home Assistant 平台

树莓派安装 Docker 软件；

 sudo apt-get update && sudo apt-get upgrade
 curl -fsSL https://get.docker.com -o get-docker.sh
 sudo sh get-docker.sh
 docker version

拉取 EMQX 和 Home Assistant 最新镜像；

 sudo docker pull emqx/emqx:latest
 sudo docker pull homeassistant/home-assistant:latest

启动 EMQX 和 HA 容器；

 docker run -d --name emqx -p 1883:1883 -p 8083:8083 -p 8084:8084 -p 8883:8883 -p 18083:18083 emqx/emqx:latest
 
 docker run -d --restart always --name homeassistant -v /data/homeassistant/config:/config -e TZ=Asia/Shanghai -p 8123:8123 homeassistant/home-assistant:latest

MQTT 客户端

EMQX 创建 MQTT 客户端

• 进入 EMQX 主页，如 http://192.168.31.116:18083；

• 依次打开 访问控制 - 客户端认证 - 创建 - Password-Based - 内置数据库 - （默认配置）- 创建 ；

创建用户

• 用户管理 - 新建用户 - 自定义用户名和密码 - 保存.

HA 配置

• 浏览器输入网址 http://<IP>:8123，如 192.168.1.107:8123

• 进入 HA 主界面（首次打开需进行注册），输入用户名、密码等信息；

• 配置完成后进入 HA 概览 标签页；

工程测试

介绍了 C 语言和 Python 驱动 Sen66 传感器套件的工程测试方案。

编译

• 进入 SEN66 官方 GitHub 网页，下载或拉取最新版工程文件

git clone https://github.com/Sensirion/embedded-i2c-sen66.git

• 进入 example-usage 运行 make 编译工程，并执行启动程序

 cd embedded-i2c-sen66-1.2.0
 cd example-usage
 make
 ./sen66_i2c_example_usage

• 终端连续打印测量值

详见：https://github.com/Sensirion/embedded-i2c-sen66

Python 代码

终端执行 touch sen66.py 新建文件并添加如下代码

 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 #
 # (c) Copyright 2025 Sensirion AG, Switzerland
 #
 #     THIS FILE IS AUTOMATICALLY GENERATED!
 #
 # Generator:     sensirion-driver-generator 1.1.2
 # Product:       sen66
 # Model-Version: 1.6.0
 #
 
 import argparse
 import time
 from sensirion_i2c_driver import LinuxI2cTransceiver, I2cConnection, CrcCalculator
 from sensirion_driver_adapters.i2c_adapter.i2c_channel import I2cChannel
 from sensirion_i2c_sen66.device import Sen66Device
 
 parser = argparse.ArgumentParser()
 parser.add_argument('--i2c-port', '-p', default='/dev/i2c-1')
 args = parser.parse_args()
 
 with LinuxI2cTransceiver(args.i2c_port) as i2c_transceiver:
     channel = I2cChannel(I2cConnection(i2c_transceiver),
                          slave_address=0x6B,
                          crc=CrcCalculator(8, 0x31, 0xff, 0x0))
     sensor = Sen66Device(channel)
     sensor.device_reset()
     time.sleep(1.2)
     serial_number = sensor.get_serial_number()
     print(f"serial_number: {serial_number}; "
           )
     sensor.start_continuous_measurement()
     for i in range(100):
         try:
             time.sleep(1.0)
             (mass_concentration_pm1p0, mass_concentration_pm2p5, mass_concentration_pm4p0, mass_concentration_pm10p0, humidity,
              temperature, voc_index, nox_index, co2
              ) = sensor.read_measured_values()
             print(f"mass_concentration_pm1p0: {mass_concentration_pm1p0}; "
                   f"mass_concentration_pm2p5: {mass_concentration_pm2p5}; "
                   f"mass_concentration_pm4p0: {mass_concentration_pm4p0}; "
                   f"mass_concentration_pm10p0: {mass_concentration_pm10p0}; "
                   f"humidity: {humidity}; "
                   f"temperature: {temperature}; "
                   f"voc_index: {voc_index}; "
                   f"nox_index: {nox_index}; "
                   f"co2: {co2}; "
                   )
         except BaseException:
             continue
     sensor.stop_measurement()

保存代码。

效果

终端执行 python sen66.py 运行程序；

打印设备序列号、环境参数（PM1.0、PM2.5、PM4.0、PM10.0、湿度、温度、VOC指数、NOx指数、二氧化碳浓度）

关键函数

read_measured_values ( )

返回测量值。

• 返回 mass_concentration_pm1p0：PM1.0 [µg/m³]

• 返回 mass_concentration_pm2p5：PM2.5 [µg/m³]

• 返回 mass_concentration_pm4p0：PM4.0 [µg/m³]

• 返回 mass_concentration_pm10p0：PM10.0 [µg/m³]

• 返回环境湿度：RH [%]

• 返回环境温度：T [°C]

• 返回 voc_index：VOC 指数

• 返回 nox_index：NOx 指数

• 返回二氧化碳：二氧化碳浓度 [ppm]

MQTT

通过 MQTT 协议实现传感器数据上传，并使用 EMQX 的 WebSocket 工具接收 JSON 报文。

流程图

代码

终端执行 touch sen66_mqtt_ha.py 指令新建程序文件，并添加如下代码

 #!/usr/bin/env python3
 """
 SEN66 → MQTT → Home Assistant
 """
 
 import argparse
 import time
 import json
 import socket
 import paho.mqtt.client as mqtt
 from sensirion_i2c_driver import LinuxI2cTransceiver, I2cConnection, CrcCalculator
 from sensirion_driver_adapters.i2c_adapter.i2c_channel import I2cChannel
 from sensirion_i2c_sen66.device import Sen66Device
 
 # ---------- 参数 ----------
 I2C_PORT         = '/dev/i2c-1'
 MQTT_BROKER      = '192.168.31.117'   # MQTT Broker IP
 MQTT_PORT        = 1883
 MQTT_USER        = 'xxx'
 MQTT_PASS        = 'xxx'
 MQTT_CLIENT_ID   = f'sen66_{socket.gethostname()}'
 BASE_TOPIC       = 'homeassistant/sensor/sen66'
 DEVICE_NAME      = 'SEN66'
 DEVICE_ID        = 'sen66_01'
 # -----------------------------
 
 # HA 自动发现的通用设备段
 DEVICE_INFO = {
     "name": DEVICE_NAME,
     "ids": [DEVICE_ID],
     "mdl": "SEN66",
     "mf": "Sensirion"
 }
 
 # 上传 9 个参量：topic_suffix, unit, icon, class
 QUANTITIES = [
     ("pm1",   "μg/m³", "mdi:blur",  "pm1"),
     ("pm25",  "μg/m³", "mdi:blur",  "pm25"),
     ("pm4",   "μg/m³", "mdi:blur",  "pm25"),
     ("pm10",  "μg/m³", "mdi:blur",  "pm10"),
     ("hum",   "%",     "mdi:water", "humidity"),
     ("temp",  "°C",    "mdi:thermometer", "temperature"),
     ("voc",   " ",     "mdi:cloud", None),
     ("nox",   " ",     "mdi:cloud", None),
     ("co2",   "ppm",   "mdi:molecule-co2", "carbon_dioxide"),
 ]
 
 def publish_discovery(client):
     """向 HA 发送自动发现配置，只需一次即可持久化"""
     for qty, unit, icon, dev_cla in QUANTITIES:
         topic = f"{BASE_TOPIC}/{qty}/config"
         payload = {
             "name": f"{DEVICE_NAME} {qty.upper()}",
             "state_topic": f"{BASE_TOPIC}/{qty}/state",
             "unit_of_measurement": unit,
             "icon": icon,
             "device": DEVICE_INFO,
             "unique_id": f"{DEVICE_ID}_{qty}",
         }
         if dev_cla:
             payload["device_class"] = dev_cla
         client.publish(topic, json.dumps(payload), retain=True)
 
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument('--i2c-port', '-p', default=I2C_PORT)
     args = parser.parse_args()
 
     # 连接 MQTT
     client = mqtt.Client(client_id=MQTT_CLIENT_ID)
     client.username_pw_set(MQTT_USER, MQTT_PASS)
     client.connect(MQTT_BROKER, MQTT_PORT, keepalive=60)
     client.loop_start()
 
     with LinuxI2cTransceiver(args.i2c_port) as i2c_transceiver:
         channel = I2cChannel(I2cConnection(i2c_transceiver),
                              slave_address=0x6B,
                              crc=CrcCalculator(8, 0x31, 0xff, 0x0))
         sensor = Sen66Device(channel)
         sensor.device_reset()
         time.sleep(1.2)
         print("SEN66 serial:", sensor.get_serial_number())
         sensor.start_continuous_measurement()
 
         # 发送一次自动发现配置
         publish_discovery(client)
 
         try:
             while True:
                 time.sleep(2)          # 2 s 上报一次
                 try:
                     vals = sensor.read_measured_values()
                     # vals 顺序与官方 demo 一致
                     pm1  = vals[0].value
                     pm25 = vals[1].value
                     pm4  = vals[2].value
                     pm10 = vals[3].value
                     hum  = vals[4].value
                     temp = vals[5].value
                     voc  = int(vals[6].value)
                     nox  = int(vals[7].value)
                     co2  = int(vals[8].value)
                     #----------------------------------
                     data = {
                         "pm1":  float(pm1),
                         "pm25": float(pm25),
                         "pm4":  float(pm4),
                         "pm10": float(pm10),
                         "hum":  float(hum),
                         "temp": float(temp),
                         "voc":  int(voc),
                         "nox":  int(nox),
                         "co2":  int(co2),
                     }
                     # 分开发，方便 HA 各自解析
                     for qty, val in data.items():
                         client.publish(f"{BASE_TOPIC}/{qty}/state", val)
                     print("published:", data)
                 except Exception as e:
                     print("read error:", e)
                     continue
         except KeyboardInterrupt:
             print("Ctrl-C 退出")
         finally:
             sensor.stop_measurement()
             client.loop_stop()
             client.disconnect()
 
 if __name__ == '__main__':
     main()

保存代码。

测试

• 终端执行 python sen66_mqtt_ha.py 指令运行程序；

• 连接 MQTT 服务器并上传数据；

• 终端打印 JSON 格式报文；

EMQX 测试

• 进入 EMQX 网页控制界面，打开侧边标签栏，选择 诊断工具 - WebSocket 客户端；

• 输入用户名和密码，连接 MQTT 服务器；

• 在订阅版块输入目标传感器主题，如 homeassistant/sensor/sen66/temp/state ，点击 订阅 按钮；

• 在已接收版块可看到传感器数据，两秒更新一次；

Home Assistant

通过 MQTT 协议上传 SEN66 传感器数据至 Home Assistant 平台，实现远程 IoT 数据实时监测。

参数配置

打开 Home Assistant 网页界面 http://192.168.31.118:8123 ，在 设置 - 设备与服务 标签页添加 MQTT 集成；

• 输入 EMQX 客户端信息，刷新页面，可看到新增的传感器；

• 点击 SEN66 传感器，进入详情页面，获取设备信息；

• 点击编辑按钮，配置所在区域等信息，并添加至仪表盘；

效果

进入 概览 页面，刷新网页，可获得实时传感器数据

• 点击目标传感器，可获得历史数据及演化曲线

总结

本文介绍了树莓派 5 结合盛思锐 SEN66 传感器套件实现环境气体监测并通过 MQTT 协议上传至 Home Assistant 智能家居平台，实现物联网环境监测终端的项目设计，为相关产品在物联网领域的快速开发和应用设计提供了参考。