wangfq/vd_960
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

feat: MQTT 协议 V1.01 — Topic 压缩为双主题 (up/down)

Browse Source 
- 7个主题压缩为2个: dld960/{sn}/down + dld960/{sn}/up
- 消息类型由 JSON cmd 字段区分，不再依赖 topic 路径
- protocol.py: topic_up()/topic_down() 替代旧7个函数
- mqtt_client.py: 4个通配符订阅→1个 dld960/+/up
- main.py: 消息路由改为 cmd 匹配，协议Topic树简化
- 文档更新 V1.01
This commit is contained in:
wangfq
2026-07-07 09:18:29 +08:00
parent d29e16acc8
commit 213716033c
4 changed files with 86 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DBNMQTTool/dbn_mqtt_tool/mqtt_client.py
+2 -8
View File
@@ -68,14 +68,8 @@ class MqttClient:
self._connected = True
self._notify_status(True, "已连接到 Broker")
from .protocol import (
TOPIC_ALL_CONFIG_RESP, TOPIC_ALL_DATA_LOOP,
TOPIC_ALL_DATA_EVENT, TOPIC_ALL_STATUS
)
client.subscribe(TOPIC_ALL_CONFIG_RESP, qos=1)
client.subscribe(TOPIC_ALL_DATA_LOOP, qos=1)
client.subscribe(TOPIC_ALL_DATA_EVENT, qos=1)
client.subscribe(TOPIC_ALL_STATUS, qos=0)
from .protocol import TOPIC_ALL_DEVICE_UP
client.subscribe(TOPIC_ALL_DEVICE_UP, qos=1)
else:
self._connected = False
self._notify_status(False, f"连接失败 (rc={rc})")
DBNMQTTool/dbn_mqtt_tool/protocol.py
+21 -43
View File
@@ -1,7 +1,9 @@
"""
DLD960 IoT MQTT 协议定义
基于《DLD960_IoT_MQTT协议.md》V1.00
基于《DLD960_IoT_MQTT协议.md》V1.01
双主题结构：{sn}/down（服务器下发）、{sn}/up（设备上报）
消息类型由 JSON `cmd` 字段区分。
"""
import json
@@ -11,49 +13,32 @@ from dataclasses import dataclass, field, asdict
# ============================================================
# Topic 结构
# Topic 结构（V1.01 — 双主题）
# ============================================================
def topic_config_set(dev_serial: str) -> str:
"""服务器下发 — 设置配置"""
return f"dld960/{dev_serial}/srv/config/set"
def topic_down(dev_serial: str) -> str:
"""服务器下发 — 设备订阅此主题接收所有命令"""
return f"dld960/{dev_serial}/down"
def topic_config_query(dev_serial: str) -> str:
"""服务器下发 — 查询配置"""
return f"dld960/{dev_serial}/srv/config/query"
def topic_up(dev_serial: str) -> str:
"""设备上报 — 设备发布到此主题（数据、事件、心跳、响应）"""
return f"dld960/{dev_serial}/up"
def topic_ctrl(dev_serial: str) -> str:
"""服务器下发 — 控制命令"""
return f"dld960/{dev_serial}/srv/ctrl"
# 保持向后兼容的别名（4.7 iot_topic_set 仍使用 topic_pub / topic_sub）
def topic_dev_pub(dev_serial: str) -> str:
"""设备发布主题（= topic_up）"""
return topic_up(dev_serial)
def topic_config_resp(dev_serial: str) -> str:
"""设备上报 — 配置响应"""
return f"dld960/{dev_serial}/dev/config/resp"
def topic_dev_sub(dev_serial: str) -> str:
"""设备订阅主题（= topic_down）"""
return topic_down(dev_serial)
def topic_data_loop(dev_serial: str) -> str:
"""设备上报 — 线圈数据"""
return f"dld960/{dev_serial}/dev/data/loop"
def topic_data_event(dev_serial: str) -> str:
"""设备上报 — 事件"""
return f"dld960/{dev_serial}/dev/data/event"
def topic_status(dev_serial: str) -> str:
"""设备上报 — 状态/心跳"""
return f"dld960/{dev_serial}/dev/status"
# 订阅通配符（监听所有设备）
TOPIC_ALL_CONFIG_RESP = "dld960/+/dev/config/resp"
TOPIC_ALL_DATA_LOOP = "dld960/+/dev/data/loop"
TOPIC_ALL_DATA_EVENT = "dld960/+/dev/data/event"
TOPIC_ALL_STATUS = "dld960/+/dev/status"
# 服务器端通配符订阅（监听所有设备）
TOPIC_ALL_DEVICE_UP = "dld960/+/up"
# ============================================================
@@ -158,15 +143,8 @@ def build_response(cmd: str, msg_id: int, code: int = 0, msg_text: str = "succes
def get_topic_for_cmd(cmd: str, dev_serial: str) -> str:
"""根据命令获取对应的下发 topic"""
if cmd in CONFIG_SET_COMMANDS:
return topic_config_set(dev_serial)
elif cmd in CONFIG_QUERY_COMMANDS:
return topic_config_query(dev_serial)
elif cmd in CTRL_COMMANDS:
return topic_ctrl(dev_serial)
else:
return topic_config_set(dev_serial) # 默认
"""获取命令对应的下发 topic。所有命令统一使用 topic_down"""
return topic_down(dev_serial)
# ============================================================
DBNMQTTool/main.py
+29 -38
View File
@@ -35,8 +35,7 @@ from dbn_mqtt_tool.protocol import (
data_ssc_net_set, data_iot_net_set, data_iot_topic_set,
data_pwd_verify, data_pwd_set,
build_request, next_msg_id,
topic_config_set, topic_config_query, topic_ctrl,
topic_config_resp, topic_data_loop, topic_data_event, topic_status,
topic_down, topic_up,
)
class MainWindow(QMainWindow):
@@ -457,15 +456,15 @@ class MainWindow(QMainWindow):
def _sim_send_loop(self):
sn = self._sim_sn.text()
self._sim_publish(topic_data_loop(sn), self._sim_loop.toPlainText())
self._sim_publish(topic_up(sn), self._sim_loop.toPlainText())
def _sim_send_event(self):
sn = self._sim_sn.text()
self._sim_publish(topic_data_event(sn), self._sim_event.toPlainText())
self._sim_publish(topic_up(sn), self._sim_event.toPlainText())
def _sim_send_hb(self):
sn = self._sim_sn.text()
self._sim_publish(topic_status(sn), self._sim_hb.toPlainText())
self._sim_publish(topic_up(sn), self._sim_hb.toPlainText())
def _sim_toggle_loop(self, checked: bool):
if checked:
@@ -501,20 +500,20 @@ class MainWindow(QMainWindow):
layout.addLayout(row1)
# 服务器下发 topics
g1 = QGroupBox("服务器下发 Topic（可发布）")
g1 = QGroupBox("服务器下发 Topic → 设备订阅（可发布）")
self._proto_srv_tree = QTreeWidget()
self._proto_srv_tree.setHeaderLabels(["Topic", "用途", "示例命令"])
self._proto_srv_tree.setRootIsDecorated(True)
self._proto_srv_tree.setHeaderLabels(["Topic", "说明"])
self._proto_srv_tree.setRootIsDecorated(False)
self._proto_srv_tree.itemDoubleClicked.connect(self._proto_publish_item)
g1l = QVBoxLayout(g1)
g1l.addWidget(self._proto_srv_tree)
layout.addWidget(g1)
# 设备上报 topics
g2 = QGroupBox("设备上报 Topic（可订阅/发布）")
g2 = QGroupBox("设备上报 Topic → 服务器订阅（可发布/订阅）")
self._proto_dev_tree = QTreeWidget()
self._proto_dev_tree.setHeaderLabels(["Topic", "用途"])
self._proto_dev_tree.setRootIsDecorated(True)
self._proto_dev_tree.setHeaderLabels(["Topic", "说明"])
self._proto_dev_tree.setRootIsDecorated(False)
self._proto_dev_tree.itemDoubleClicked.connect(self._proto_publish_item)
g2l = QVBoxLayout(g2)
g2l.addWidget(self._proto_dev_tree)
@@ -545,38 +544,29 @@ class MainWindow(QMainWindow):
def _refresh_proto_topics(self):
sn = self._proto_sn.text().strip() or "{sn}"
# 服务器下发
# 服务器下发 — 单主题
srv = self._proto_srv_tree
srv.clear()
QTreeWidgetItem(srv, [topic_down(sn), "设备订阅此主题接收所有命令"])
def _add_srv(parent, name, tpl, desc, sample_cmd=""):
topic = tpl.format(sn=sn)
item = QTreeWidgetItem(parent or srv, [topic, desc, sample_cmd])
return item
root_set = _add_srv(None, "config/set", "dld960/{sn}/srv/config/set",
"设置配置参数", "dev_serial_set / iot_net_set / loop_param_set")
root_query = _add_srv(None, "config/query", "dld960/{sn}/srv/config/query",
"查询配置参数", "dev_info_query / ssc_net_query")
root_ctrl = _add_srv(None, "ctrl", "dld960/{sn}/srv/ctrl",
"控制命令", "pwd_verify / factory_reset / device_reset")
# 设备上报
# 设备上报 — 单主题
dev = self._proto_dev_tree
dev.clear()
for name, tpl, desc in [
("config/resp", "dld960/{sn}/dev/config/resp", "配置查询/设置响应"),
("data/loop", "dld960/{sn}/dev/data/loop", "线圈传感数据上报"),
("data/event", "dld960/{sn}/dev/data/event", "事件上报"),
("status", "dld960/{sn}/dev/status", "设备状态/心跳"),
item = QTreeWidgetItem(dev, [topic_up(sn), "设备发布：loop_data / event_report / heartbeat / 命令响应"])
# 子条目列出 cmd 类型
for cmd_name, desc in [
("loop_data", "线圈传感数据上报"),
("event_report", "事件上报（有车/无车/故障）"),
("heartbeat", "设备心跳"),
("响应", "各命令查询/设置响应（msg_id 匹配）"),
]:
QTreeWidgetItem(dev, [tpl.format(sn=sn), desc])
srv.expandAll()
dev.expandAll()
QTreeWidgetItem(item, [f" cmd={cmd_name}", desc])
def _proto_publish_item(self, item: QTreeWidgetItem, col: int):
# 如果点中的是子条目（无有效 topic），取父条目的 topic
topic = item.text(0)
if item.parent() and not topic.startswith("dld960/"):
topic = item.parent().text(0)
self._proto_pub_topic.setText(topic)
self._proto_payload.clear()
self._notebook.setCurrentWidget(self._proto_payload.parent().parent())
@@ -764,7 +754,8 @@ class MainWindow(QMainWindow):
cmd = payload.get("cmd", "")
ts = datetime.fromtimestamp(payload.get("ts", 0)).strftime("%H:%M:%S")
if "/dev/config/resp" in topic:
# 响应类消息（含 code 字段）
if "code" in payload:
code = payload.get("code", -1)
pmsg = payload.get("msg", "")
data = payload.get("data", {})
@@ -778,21 +769,21 @@ class MainWindow(QMainWindow):
else:
self._show_json({"error": f"code={code} {pmsg} ({ERROR_MSGS.get(code, '?')})"})
elif "/dev/data/loop" in topic:
elif cmd == CMD_LOOP_DATA:
data = payload.get("data", {})
self._devmgr.update_loop_data(dev_serial, data)
chs = len(data.get("channels", []))
self._log(f"[{ts}] {dev_serial} ← loop_data ({chs} 通道)")
self._append_text(self._loop_text, json.dumps(payload, indent=2, ensure_ascii=False))
elif "/dev/data/event" in topic:
elif cmd == CMD_EVENT_REPORT:
data = payload.get("data", {})
self._devmgr.update_event(dev_serial, data)
evt_count = len(data.get("events", []))
self._log(f"[{ts}] {dev_serial} ← event_report ({evt_count} 事件)")
self._append_text(self._event_text, json.dumps(payload, indent=2, ensure_ascii=False))
elif "/dev/status" in topic:
elif cmd == CMD_HEARTBEAT:
data = payload.get("data", {})
self._devmgr.update_heartbeat(dev_serial, data)
self._log(f"[{ts}] {dev_serial} ← heartbeat (uptime={data.get('uptime', 0)}s)")