vd_960/DBNMQTTool/dbn_mqtt_tool/mqtt_client.py

"""
MQTT 客户端封装
"""
import json
import time
import threading
from typing import Callable, Optional
from dataclasses import dataclass

import paho.mqtt.client as mqtt
from PySide6.QtCore import QObject, Signal, Slot, Qt

from .protocol import parse_topic_dev_serial


@dataclass
class BrokerConfig:
    host: str = "121.37.20.199"
    port: int = 1883
    username: str = ""
    password: str = ""
    client_id: str = ""
    keepalive: int = 60


class MqttClient(QObject):
    """DLD960 IoT MQTT 客户端 — QObject 基类确保信号线程安全"""

    # 供外部监听
    _status_notify = Signal(bool, str)   # connected, message
    _msg_notify = Signal(str, str, object)  # topic, dev_serial, payload_dict

    # 内部线程安全的 publish 队列
    _publish_sig = Signal(str, str, int)  # topic, json_data, qos

    def __init__(self):
        super().__init__()
        self._client: Optional[mqtt.Client] = None
        self._config = BrokerConfig()
        self._connected = False
        self._lock = threading.Lock()

        # 信号→槽: publish 通过排队连接确保在主线程的 paho context 执行
        self._publish_sig.connect(self._do_publish, Qt.QueuedConnection)

    @property
    def connected(self) -> bool:
        return self._connected

    # ---- 公开回调 (兼容旧接口) ----

    def on_status_change(self, callback: Callable[[bool, str], None]):
        self._status_notify.connect(callback)

    def on_message(self, callback: Callable[[str, str, dict], None]):
        """回调参数: (topic, dev_serial, payload_dict)"""
        self._msg_notify.connect(callback)

    # ---- MQTT callbacks (paho 网络线程) ----

    def _on_connect(self, client, userdata, flags, reason_code, properties=None):
        rc = reason_code if isinstance(reason_code, int) else reason_code.value
        if rc == 0:
            self._connected = True
            self._status_notify.emit(True, "已连接到 Broker")

            from .protocol import TOPIC_ALL_DEVICE_UP
            client.subscribe(TOPIC_ALL_DEVICE_UP, qos=1)
        else:
            self._connected = False
            self._status_notify.emit(False, f"连接失败 (rc={rc})")

    def _on_disconnect(self, client, userdata, *args):
        self._connected = False
        self._status_notify.emit(False, "已断开连接")

    def _on_message(self, client, userdata, msg: mqtt.MQTTMessage):
        try:
            payload = json.loads(msg.payload.decode("utf-8"))
        except (json.JSONDecodeError, UnicodeDecodeError):
            payload = {"_raw": msg.payload.hex()}
        dev_serial = parse_topic_dev_serial(msg.topic) or ""
        self._msg_notify.emit(msg.topic, dev_serial, payload)

    # ---- Public API ----

    def connect(self, config: Optional[BrokerConfig] = None):
        if config:
            self._config = config

        if self._client:
            self.disconnect()

        client_id = self._config.client_id or f"dbn_mqtt_tool_{int(time.time())}"

        self._client = mqtt.Client(
            client_id=client_id,
            protocol=mqtt.MQTTv311,
            callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
        )
        if self._config.username:
            self._client.username_pw_set(self._config.username, self._config.password)

        self._client.on_connect = self._on_connect
        self._client.on_disconnect = self._on_disconnect
        self._client.on_message = self._on_message

        self._client.connect_async(self._config.host, self._config.port, self._config.keepalive)
        self._client.loop_start()

    def disconnect(self):
        if self._client:
            self._client.loop_stop()
            self._client.disconnect()
            self._client = None
        self._connected = False

    def publish(self, topic: str, payload: dict, qos: int = 1):
        """线程安全的发布 — 通过 Qt 信号排队到主线程执行"""
        if not self._client or not self._connected:
            raise ConnectionError("MQTT 未连接")
        data = json.dumps(payload, ensure_ascii=False)
        self._publish_sig.emit(topic, data, qos)

    def subscribe(self, topic: str, qos: int = 1):
        """订阅主题"""
        if self._client:
            self._client.subscribe(topic, qos=qos)

    def unsubscribe(self, topic: str):
        """取消订阅"""
        if self._client:
            self._client.unsubscribe(topic)

    @Slot(str, str, int)
    def _do_publish(self, topic: str, data: str, qos: int):
        """实际的 paho publish 调用 — 始终在主线程执行"""
        if self._client and self._connected:
            try:
                self._client.publish(topic, data, qos=qos)
            except Exception as e:
                print(f"[MqttClient] publish error: {e}")

    def send_command(self, dev_serial: str, cmd: str, data: Optional[dict] = None, qos: int = 1) -> int:
        """向设备发送命令，返回 msg_id"""
        from .protocol import build_request, get_topic_for_cmd
        msg = build_request(cmd, data)
        topic = get_topic_for_cmd(cmd, dev_serial)
        self.publish(topic, msg, qos=qos)
        return msg["msg_id"]