wangfq
b932dd7c6f
新增 _log_send() / _log_recv() 两个统一日志方法:
- 发送: 标签 + → topic + PASTE:{精简payload}
- 接收: ← topic + summary + RECV:{精简payload}
- payload >200/300 字符自动截断
覆盖所有收发路径:
- [模拟]/[协议]/[自定义]/[命令] 发送
- 响应/loop_data/event_report/heartbeat/Initialize 接收
- mqtt_client.py 自动订阅事件日志
145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
"""
|
|
MQTT 客户端封装
|
|
"""
|
|
import json
|
|
import time
|
|
import threading
|
|
from typing import Callable, Optional
|
|
from dataclasses import dataclass
|
|
|
|
import paho.mqtt.client as mqtt
|
|
|
|
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:
|
|
"""DLD960 IoT MQTT 客户端"""
|
|
|
|
def __init__(self):
|
|
self._client: Optional[mqtt.Client] = None
|
|
self._config = BrokerConfig()
|
|
self._connected = False
|
|
self._lock = threading.Lock()
|
|
self._message_callbacks: list[Callable[[str, str, dict], None]] = []
|
|
self._status_callbacks: list[Callable[[bool, str], None]] = []
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return self._connected
|
|
|
|
def on_status_change(self, callback: Callable[[bool, str], None]):
|
|
self._status_callbacks.append(callback)
|
|
|
|
def on_message(self, callback: Callable[[str, str, dict], None]):
|
|
"""回调参数: (topic, dev_serial, payload_dict)"""
|
|
self._message_callbacks.append(callback)
|
|
|
|
def _notify_status(self, connected: bool, msg: str):
|
|
for cb in self._status_callbacks:
|
|
try:
|
|
cb(connected, msg)
|
|
except Exception:
|
|
pass
|
|
|
|
def _notify_message(self, topic: str, payload: dict):
|
|
dev_serial = parse_topic_dev_serial(topic) or ""
|
|
for cb in self._message_callbacks:
|
|
try:
|
|
cb(topic, dev_serial, payload)
|
|
except Exception:
|
|
pass
|
|
|
|
# ---- MQTT callbacks ----
|
|
|
|
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._notify_status(True, "已连接到 Broker")
|
|
|
|
from .protocol import TOPIC_ALL_DEVICE_UP
|
|
print(f"[MqttClient] auto-subscribe: {TOPIC_ALL_DEVICE_UP}", flush=True)
|
|
client.subscribe(TOPIC_ALL_DEVICE_UP, qos=1)
|
|
else:
|
|
self._connected = False
|
|
self._notify_status(False, f"连接失败 (rc={rc})")
|
|
|
|
def _on_disconnect(self, client, userdata, *args):
|
|
self._connected = False
|
|
self._notify_status(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()}
|
|
self._notify_message(msg.topic, 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):
|
|
"""发送 JSON 消息到指定 topic"""
|
|
if not self._client or not self._connected:
|
|
raise ConnectionError("MQTT 未连接")
|
|
data = json.dumps(payload, ensure_ascii=False)
|
|
self._client.publish(topic, data, qos=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)
|
|
|
|
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"]
|