ZigBee设备如何对接开源物联网平台ThingsPanel
发布日期：
ZigBee设备通过ZigBee2MQTT网关，可以转换ZigBee协议为MQTT，通过MQTT协议接入ThingsPanel，实现对ZigBee设备的智能管理。
本次介绍的方案主要通过Zigbee2MQTT接入，这个方案有如下特性：
- 设备支持广泛：目前支持 4272 个设备，来自 489 个不同的供应商。这种广泛的设备兼容性意味着它能够连接和管理多种类型、多个品牌的 Zigbee 设备，用户可以方便地将各种不同的智能设备集成到一个统一的智能家居系统中，而无需担心设备兼容性问题。
- 设备类型丰富：涵盖了多种功能类型的设备，如智能调光器、智能能源监测器、温度传感器、湿度传感器、空气质量监测器、植物浇水传感器等。这表明 Zigbee2MQTT 网关可以满足不同用户在智能家居环境中的多样化需求，无论是对环境监测、能源管理还是设备控制等方面，都能提供相应的支持。
本文主要讲解ZigBee设备到ThingsPanel的主要过程。其他问题可以加入我们的QQ群进一步交流（QQ群：371794256）。
配置流程描述
1. 涂鸦温湿度传感器 → ZigBee2MQTT硬件网关
● 连接方式：ZigBee无线协议
● 数据流：传感器采集温湿度数据，通过ZigBee无线网络传输
● 配置方式：手动打开允许所有设备接入，在温湿度传感器上用针插小孔5秒配对
2. ZigBee2MQTT硬件网关 → ZigBee2MQTT软件
● 连接方式：USB接口连接
● 硬件网关：通常为CC2531/CC2652R等ZigBee协调器，要注意接口和芯片型号
● 数据流：协调器接收ZigBee数据帧并通过USB传输给主机
3. ZigBee2MQTT软件 → 本地Mosquitto MQTT Broker
● 连接方式：软件内部MQTT客户端发布
● MQTT服务器：本地运行的Mosquitto服务器（127.0.0.1:1883）
● 数据格式：JSON格式，包含设备状态和属性
● 配置方式：配置为系统服务
4. 本地Mosquitto → Python MQTT桥接程序
● 连接方式：MQTT订阅
● 订阅主题：zigbee2mqtt/#（订阅所有ZigBee设备的消息）
● 配置方式：配置为系统服务
5. Python MQTT桥接程序 → ThingsPanel MQTT Broker
● 连接方式：MQTT发布
● 配置方式：配置为系统服务
● 远程服务器：ThingsPanel的MQTT服务器（47.115.210.16:1883）
● 认证信息：使用client_id、username和password进行认证
● 发布主题：devices/telemetry
● 数据发送策略：实时：设备数据更新时立即发送
○ 定期：即使没有新数据也每60秒发送一次最新状态
准备设备
我使用的设备是涂鸦温湿度传感器。
硬件网关使用的是ZigBee2MQTT网关。
主机可使用树莓派，电脑也可以。
硬件网关具备自动扫描、识别、添加设备的能力。
需要将这些设备通电并链接。
最终的结果
配置与代码
安装Mosquitto MQTT broker并配置
Mosquitto是最靠近ZigBee2MQTT的MQTT服务器，用于给MQTT客户端提供订阅和发送服务，有点像快递站。
sudo apt install -y mosquitto mosquitto-clients 安装
cat /etc/mosquitto/mosquitto.conf
listener 1883
allow_anonymous true
安装zigbee2mqtt
zigbee2mqtt是和硬件打交道的，管理ZigBee网关设备和子设备。
cd /opt
sudo git clone https://github.com/Koenkk/zigbee2mqtt.git
sudo chown -R $USER:$USER zigbee2mqtt
cd zigbee2mqtt
npm install
具体按照zigbee2mqtt的官网教程配置。
Zigbee2mqtt配置
cat /opt/zigbee2mqtt/data/configuration.yaml
homeassistant:
enabled: true
mqtt:
base_topic: zigbee2mqtt
server: mqtt://localhost:1883
serial:
port: /dev/ttyUSB0
adapter: zstack
frontend:
enabled: true
port: 8060
advanced:
log_level: debug
version: 4
devices:
'0x086bd7fffec2fd1e':
friendly_name: 涂鸦温湿度传感器
需要注意如上的
port: /dev/ttyUSB0 如果你接入到了USB，那就是这个
adapter: zstack 这个要核对芯片
devices:
'0x086bd7fffec2fd1e':
friendly_name: 涂鸦温湿度传感器
Zigbee2MQTT的服务
这个服务的目的是让ZigBee2MQTT能够方便的被管理，要不启动会比较不规范标准，也不方便。
cat /etc/systemd/system/zigbee2mqtt.service
[Unit]
Description=Zigbee2MQTT
After=network.target
[Service]
ExecStart=/usr/bin/npm start
WorkingDirectory=/opt/zigbee2mqtt
StandardOutput=inherit
StandardError=inherit
Restart=always
User=pi
[Install]
WantedBy=multi-user.target
保存后
sudo systemctl daemon-reload
sudo systemctl enable zigbee2mqtt
sudo systemctl start zigbee2mqtt
sudo journalctl -u zigbee2mqtt -f按照配置，访问的端口是8086，打开后允许装置加入，拿个小针去捅传感器的小针孔5秒。这里就识别到了。
桥接代码
/opt/mqtt-bridge.py
这个桥接代码的作用是，从快递站取到东西，送到ThingsPanel。
import json
import paho.mqtt.client as mqtt
import time
# 配置
local_broker = "127.0.0.1"
local_port = 1883
local_topic = "zigbee2mqtt/#"
remote_broker = "47.115.210.16"
remote_port = 1883
remote_client_id = "mqtt_bd6045da-d55"
remote_username = "3641c829-af0b-6b0d-72a"
remote_password = "719f457"
remote_topic = "devices/telemetry"
# 存储最新的传感器数据
last_sensor_data = {}
# 上次发送时间
last_send_time = 0
send_interval = 60 # 即使没有新数据，也每60秒发送一次
# 连接回调
def on_connect_local(client, userdata, flags, rc):
print(f"已连接到本地MQTT服务器，状态码: {rc}")
client.subscribe(local_topic)
print(f"已订阅主题: {local_topic}")
def on_connect_remote(client, userdata, flags, rc):
print(f"已连接到远程MQTT服务器，状态码: {rc}")
# 消息回调
def on_message(client, userdata, msg):
global last_sensor_data, last_send_time
try:
topic = msg.topic
print(f"收到消息: {topic}")
# 跳过桥接消息
if "bridge" in topic:
return
# 获取设备ID
device_id = topic.split("/")[-1]
payload = json.loads(msg.payload.decode())
# 只处理传感器数据 (温度、湿度等)
if isinstance(payload, dict):
filtered_payload = {}
for key, value in payload.items():
# 只保留数值和布尔值字段
if isinstance(value, (int, float, bool)) and key not in ["linkquality"]:
filtered_payload[key] = value
if filtered_payload:
# 更新存储的数据
last_sensor_data[device_id] = filtered_payload
# 发送到ThingsPanel
print(f"发送数据: {json.dumps(filtered_payload)} 到主题: {remote_topic}")
remote_client.publish(remote_topic, json.dumps(filtered_payload))
last_send_time = time.time()
except Exception as e:
print(f"处理消息时出错: {e}")
# 设置本地MQTT客户端
local_client = mqtt.Client()
local_client.on_connect = on_connect_local
local_client.on_message = on_message
try:
print("尝试连接到本地MQTT服务器...")
local_client.connect(local_broker, local_port, 60)
except Exception as e:
print(f"连接本地MQTT服务器失败: {e}")
# 设置远程MQTT客户端
remote_client = mqtt.Client(remote_client_id)
remote_client.on_connect = on_connect_remote
remote_client.username_pw_set(remote_username, remote_password)
try:
print("尝试连接到远程MQTT服务器...")
remote_client.connect(remote_broker, remote_port, 60)
except Exception as e:
print(f"连接远程MQTT服务器失败: {e}")
# 启动循环
local_client.loop_start()
remote_client.loop_start()
print("MQTT桥接已启动...")
try:
while True:
# 定期发送数据，即使没有新的消息
current_time = time.time()
if last_sensor_data and (current_time - last_send_time) > send_interval:
# 发送所有存储的传感器数据
for device_id, data in last_sensor_data.items():
print(f"定期发送 {device_id} 数据: {json.dumps(data)}")
remote_client.publish(remote_topic, json.dumps(data))
last_send_time = current_time
time.sleep(1)
except KeyboardInterrupt:
print("脚本已停止")
local_client.loop_stop()
remote_client.loop_stop()
桥接代码的服务
为了方便的使用桥接代码，规范并简化使用
cat /etc/systemd/system/mqtt-bridge.service
[Unit]
Description=MQTT Bridge for ThingsPanel
After=network.target
[Service]
ExecStart=/usr/bin/python3 /opt/mqtt-bridge.py
Restart=always
User=pi
[Install]
WantedBy=multi-user.target
编辑后启动服务
sudo systemctl daemon-reload
sudo systemctl enable mqtt-bridge
sudo systemctl restart mqtt-bridge
sudo systemctl status mqtt-bridge
解决上报电压为3000的问题
没有如下的Lua脚本，你看到的电压就是3000伏，而不是正常的3伏
Lua脚本
-- 导入JSON库用于处理数据
local json = require("json")
function encodeInp(msg, topic)
-- 说明：该函数为编码函数，将输入的消息编码为平台可识别的消息格式
-- 入参：输入的msg，可以是任意数据类型的字符串
-- 出参：返回值为编码后的消息，需要是json字符串形式
-- 首先打印原始消息以便调试
print("原始消息: " .. tostring(msg))
-- 尝试解析JSON数据
local jsonData = {}
local status, err = pcall(function()
jsonData = json.decode(msg)
end)
-- 如果解析失败，记录错误并返回原消息
if not status or type(jsonData) ~= "table" then
print("JSON解析错误或结果不是表: " .. tostring(err))
return msg
end
-- 输出解析后的jsonData以检查结构
print("JSON解析结果: " .. json.encode(jsonData))
-- 检查voltage字段是否存在
if jsonData.voltage ~= nil then
-- 将毫伏转换为伏特，保留一位小数
jsonData.voltage = jsonData.voltage / 1000
print("电压转换: " .. jsonData.voltage .. "V")
-- 基于电压值估算电池状态
if jsonData.voltage >= 3.0 then
jsonData.battery_status = "good"
elseif jsonData.voltage >= 2.7 then
jsonData.battery_status = "medium"
else
jsonData.battery_status = "low"
end
print("电池状态: " .. jsonData.battery_status)
else
print("未找到voltage字段")
end
-- 处理温度数据 - 如果存在
if jsonData.temperature ~= nil then
-- 确保温度数据精确到一位小数
jsonData.temperature = math.floor(jsonData.temperature * 10) / 10
print("温度数据: " .. jsonData.temperature .. "°C")
else
print("未找到temperature字段")
end
-- 处理湿度数据 - 如果存在
if jsonData.humidity ~= nil then
-- 湿度通常是整数百分比
jsonData.humidity = math.floor(jsonData.humidity)
print("湿度数据: " .. jsonData.humidity .. "%")
else
print("未找到humidity字段")
end
-- 将处理后的数据转换回JSON字符串
local newJsonString = json.encode(jsonData)
print("输出消息: " .. newJsonString)
return newJsonString
end
