【盛思锐SEN66】物联网环境监测终端

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

项目介绍

准备工作：硬件连接、软件包安装、Docker部署、EMQX和HomeAssistant平台部署；
工程测试：加载官方 Demo 例程，实现 SEN66 传感器数据终端打印；
MQTT：结合 MQTT 协议，实现传感器数据上传至 EMQX 服务器平台，并远程读取；
Home Assistant：通过 MQTT 集成，实现 SEN66 传感器数据卡片自动获取，实现物联网环境气体监测。

准备工作

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

硬件连接

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

SEN66	Raspberry Pi	Note
SDA (green)	SDA (3)	Serial Data
SCL (yellow)	SCL (5)	Serial Clock
GND (black)	GND (6)	Ground
VCC (red)	3.3V (1)	Power

实物图

树莓派 40-Pin 接口定义

SEN66 引脚

引脚定义及描述

Pin	Color	Name	Description	Comments
1	red	VDD	Supply Voltage	3.3V ±5%
2	black	GND	Ground
3	green	SDA	I2C: Serial data input / output	TTL 5V compatible
4	yellow	SCL	I2C: Serial clock input	TTL 5V compatible
5		NC	Do not connect	Ground (Pins 2 and 5 are connected internally)
6		NC	Do not connect	Supply voltage (Pins 1 and 6 are connected internally)

详见：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

代码

终端执行 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 报文。

流程图

flowchart TD A([开始]) --> B[初始化] B --> C[扫描IIC设备] C --> D[复位 SEN66] D --> E[启动连续测量] E --> F[连接MQTT] F --> G[发布HA配置] G --> H{主循环} H --> I[读取9个参量] I --> J[提取数据] J --> K[封装JSON<br>发布] K --> L[打印日志] L --> H I -.->|异常| N[报错] H -->|Ctrl-C| N[停止测量<br>断开MQTT] N --> O([退出])

代码

终端执行 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 传感器，进入详情页面，获取设备信息；

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

效果

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

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

总结

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