Python 中如何实现与 Zigbee 进行通信的接口

要跟Zigbee设备通信，通常得通过一个网关或者用支持Zigbee的USB适配器。Python这边，你得找个库来跟这个硬件打交道。最常用的方法是使用串口（UART）协议，或者用像MQTT这样的网络协议通过网关中转。

这里有个使用 pyserial 库通过串口与Zigbee协调器（比如CC2531 USB Dongle刷了Z-Stack固件）通信的基本例子。这假设你的协调器已经配置好，并在 /dev/ttyUSB0 上创建了一个虚拟串口。

import serial
import time

class ZigbeeSerialInterface:
    def __init__(self, port='/dev/ttyUSB0', baudrate=115200, timeout=1):
        """
        初始化串口连接。
        :param port: 串口设备路径 (Linux) 或 COM端口 (Windows， 如 'COM3')
        :param baudrate: 波特率，常见的有 115200, 9600
        :param timeout: 读超时时间
        """
        try:
            self.ser = serial.Serial(port, baudrate, timeout=timeout)
            print(f"成功连接到 {port}")
        except serial.SerialException as e:
            print(f"无法打开串口 {port}: {e}")
            self.ser = None

    def send_command(self, data_hex):
        """
        发送十六进制格式的命令。
        :param data_hex: 十六进制字符串，例如 'FE 02 01 02'
        """
        if not self.ser or not self.ser.is_open:
            print("串口未打开")
            return False
        try:
            # 将十六进制字符串转换为字节
            data_bytes = bytes.fromhex(data_hex)
            self.ser.write(data_bytes)
            print(f"发送: {data_hex}")
            return True
        except Exception as e:
            print(f"发送失败: {e}")
            return False

    def read_response(self, max_bytes=1024):
        """
        读取响应。
        :param max_bytes: 最大读取字节数
        :return: 响应的十六进制字符串
        """
        if not self.ser or not self.ser.is_open:
            print("串口未打开")
            return None
        try:
            response = self.ser.read(max_bytes)
            if response:
                hex_response = ' '.join([f'{b:02X}' for b in response])
                print(f"接收: {hex_response}")
                return hex_response
            else:
                print("未收到响应或超时")
                return None
        except Exception as e:
            print(f"读取失败: {e}")
            return None

    def close(self):
        """关闭串口连接"""
        if self.ser and self.ser.is_open:
            self.ser.close()
            print("串口连接已关闭")

# --- 使用示例 ---
if __name__ == "__main__":
    # 1. 初始化接口，根据你的系统修改端口
    zigbee = ZigbeeSerialInterface(port='/dev/ttyUSB0') # Linux
    # zigbee = ZigbeeSerialInterface(port='COM3') # Windows

    if zigbee.ser is not None:
        # 2. 发送一个示例命令（这个命令需要根据你的Zigbee协议来定）
        # 例如，一个简单的测试帧
        test_command = "FE 02 01 02" # 这只是一个示例，并非真实Zigbee指令
        zigbee.send_command(test_command)

        # 3. 等待并读取响应
        time.sleep(0.5) # 给设备一点响应时间
        response = zigbee.read_response()

        # 4. 关闭连接
        zigbee.close()

核心要点：

1. 硬件是基础：你需要一个Zigbee协调器（如CC2531/CC2652 USB Dongle）并刷好合适的固件（如Z-Stack），它才能在电脑上呈现为一个串口。
2. 协议是关键：上面的代码只处理了串口字节的收发。真正的通信逻辑取决于你使用的Zigbee应用层协议。你需要根据设备厂商提供的协议文档来构造正确的命令帧（替换掉示例中的 test_command）和解析响应帧。常见的家用协议有ZCL（Zigbee Cluster Library）。
3. 替代方案（更简单）：如果你用的是现成的Zigbee网关（如小米多模网关、Zigbee2MQTT网关），它们通常提供网络API（如HTTP REST、MQTT）。用Python的 requests 或 paho-mqtt 库去和网关通信会更方便，避免了直接处理底层的串口和Zigbee协议帧。
   • MQTT示例（连接Zigbee2MQTT网关）:

     import paho.mqtt.client as mqtt
     client = mqtt.Client()
     client.connect("你的网关IP", 1883)
     # 订阅设备状态
     client.subscribe("zigbee2mqtt/你的设备ID")
     # 发送控制命令
     client.publish("zigbee2mqtt/你的设备ID/set", '{"state": "ON"}')
     

总结：先确定你的硬件和协议，再用对应的Python库去连接。