hookvision-rk/hookvision/mqtt_client.py

import paho.mqtt.client as mqtt
import json
import logging
from typing import Callable

class MQTTClient:
    def __init__(self, config, on_message_callback: Callable[[str, dict], None]):
        self.config = config
        self.on_message_callback = on_message_callback
        self.client = mqtt.Client()
        self.client.on_connect = self._on_connect
        self.client.on_message = self._on_message
        self.connected = False

    def connect(self):
        if not self.config.enabled:
            return
        try:
            self.client.connect(self.config.broker, self.config.port, 60)
            self.client.loop_start()
            self.connected = True
        except Exception as e:
            logging.error(f"MQTT连接失败: {e}")
            self.connected = False

    def disconnect(self):
        if self.connected:
            self.client.loop_stop()
            self.client.disconnect()

    def subscribe(self, topics):
        if not self.connected:
            return
        for topic in topics:
            self.client.subscribe(topic)

    def _on_connect(self, client, userdata, flags, rc):
        if rc == 0:
            self.connected = True
            for topic in self.config.topics:
                client.subscribe(topic)
        else:
            logging.warning(f"MQTT连接失败，返回码: {rc}")

    def _on_message(self, client, userdata, msg):
        try:
            payload_str = msg.payload.decode('utf-8')
            if not payload_str.strip():
                return
            payload = json.loads(payload_str)
            if payload is not None:
                self.on_message_callback(msg.topic, payload)
            else:
                logging.debug("收到 null payload，忽略")
        except Exception as e:
            logging.exception(f"MQTT消息解析错误")  # 打印堆栈