wangfq/vd_960
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity
Files
5210051f199fe45266e57c9442fa69ff97577b75
vd_960/DBNMQTTool/main.py
T
wangfq e1bf3dcfca refactor(DBNMQTTool): tkinter → PySide6 
- main.py: 完整重写为 PySide6 (Qt for Python)
- requirements.txt: 加 PySide6>=6.6.0
- 使用 Qt 信号/槽替代 tkinter 回调
- 界面更现代化，Fusion 风格
- 分组框 (QGroupBox) 布局，QSplitter 分栏
2026-07-06 21:19:55 +08:00

554 lines
20 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
DBN MQTT Tool — DLD960 IoT MQTT 设备管理工具
跨平台桌面软件 (PySide6)
"""
import sys
import os
import json
from datetime import datetime
from typing import Optional
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from PySide6.QtWidgets import (
QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout,
QLabel, QLineEdit, QPushButton, QTreeWidget, QTreeWidgetItem,
QTabWidget, QTextEdit, QGroupBox, QGridLayout, QCheckBox,
QSplitter, QMessageBox, QHeaderView, QFrame,
)
from PySide6.QtCore import Qt, QTimer, Signal, Slot
from PySide6.QtGui import QFont, QColor, QTextCursor
from dbn_mqtt_tool.mqtt_client import MqttClient, BrokerConfig
from dbn_mqtt_tool.device_manager import DeviceManager
from dbn_mqtt_tool.protocol import (
CMD_DEV_INFO_QUERY, CMD_SSC_NET_QUERY, CMD_IOT_NET_QUERY,
CMD_IOT_TOPIC_QUERY, CMD_LOOP_PARAM_QUERY,
CMD_SSC_NET_SET, CMD_IOT_NET_SET, CMD_IOT_TOPIC_SET,
CMD_PWD_VERIFY, CMD_PWD_SET, CMD_FACTORY_RESET, CMD_DEVICE_RESET,
ERROR_MSGS,
data_ssc_net_set, data_iot_net_set, data_iot_topic_set,
data_pwd_verify, data_pwd_set,
)
class MainWindow(QMainWindow):
_mqtt_status = Signal(bool, str)
_mqtt_msg = Signal(str, str, object)
_devices_changed = Signal()
def __init__(self):
super().__init__()
self.setWindowTitle("DBN MQTT Tool — DLD960 IoT 设备管理")
self.resize(1280, 800)
self.setMinimumSize(960, 640)
self._mqtt = MqttClient()
self._devmgr = DeviceManager()
self._mqtt.on_status_change(lambda c, m: self._mqtt_status.emit(c, m))
self._mqtt.on_message(lambda t, s, p: self._mqtt_msg.emit(t, s, p))
self._devmgr.on_change(lambda: self._devices_changed.emit())
self._mqtt_status.connect(self._on_status)
self._mqtt_msg.connect(self._on_message)
self._devices_changed.connect(self._refresh_devices)
self._build_ui()
self._log("DBN MQTT Tool 已启动")
# ================================================================
# UI 构建
# ================================================================
def _build_ui(self):
central = QWidget()
self.setCentralWidget(central)
root = QVBoxLayout(central)
root.setContentsMargins(8, 8, 8, 8)
root.setSpacing(6)
# -- 顶栏 --
root.addWidget(self._build_top_bar())
# -- 主体 --
splitter = QSplitter(Qt.Horizontal)
left = self._build_device_panel()
splitter.addWidget(left)
right = QTabWidget()
right.addTab(self._build_info_tab(), "设备信息")
right.addTab(self._build_config_tab(), "参数配置")
right.addTab(self._build_data_tab(), "实时数据")
right.addTab(self._build_log_tab(), "日志")
splitter.addWidget(right)
splitter.setSizes([280, 960])
root.addWidget(splitter, 1)
def _build_top_bar(self) -> QWidget:
bar = QWidget()
layout = QHBoxLayout(bar)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(6)
layout.addWidget(QLabel("Broker:"))
self._host_edit = QLineEdit("121.37.20.199")
self._host_edit.setMaximumWidth(140)
layout.addWidget(self._host_edit)
layout.addWidget(QLabel("Port:"))
self._port_edit = QLineEdit("1883")
self._port_edit.setMaximumWidth(60)
layout.addWidget(self._port_edit)
layout.addWidget(QLabel("用户:"))
self._user_edit = QLineEdit()
self._user_edit.setMaximumWidth(90)
layout.addWidget(self._user_edit)
layout.addWidget(QLabel("密码:"))
self._pass_edit = QLineEdit()
self._pass_edit.setMaximumWidth(90)
self._pass_edit.setEchoMode(QLineEdit.Password)
layout.addWidget(self._pass_edit)
self._btn_conn = QPushButton("连接")
self._btn_conn.setFixedWidth(70)
self._btn_conn.clicked.connect(self._toggle_connect)
layout.addWidget(self._btn_conn)
self._status_label = QLabel("● 未连接")
self._status_label.setStyleSheet("color: red; font-weight: bold;")
layout.addWidget(self._status_label)
layout.addStretch()
return bar
def _build_device_panel(self) -> QWidget:
w = QWidget()
layout = QVBoxLayout(w)
layout.setContentsMargins(0, 0, 0, 0)
layout.addWidget(QLabel("<b>设备列表</b>"))
self._dev_tree = QTreeWidget()
self._dev_tree.setHeaderLabels(["设备序列码", "型号"])
self._dev_tree.setRootIsDecorated(False)
self._dev_tree.header().setStretchLastSection(True)
self._dev_tree.itemSelectionChanged.connect(self._on_dev_select)
layout.addWidget(self._dev_tree, 1)
btn_row = QHBoxLayout()
btn_query = QPushButton("查询设备信息")
btn_query.clicked.connect(self._query_dev_info)
btn_row.addWidget(btn_query)
btn_all = QPushButton("查询全部")
btn_all.clicked.connect(self._query_all)
btn_row.addWidget(btn_all)
layout.addLayout(btn_row)
return w
def _build_info_tab(self) -> QWidget:
w = QWidget()
layout = QVBoxLayout(w)
self._info_text = QTextEdit()
self._info_text.setReadOnly(True)
self._info_text.setFont(QFont("Consolas", 10))
layout.addWidget(self._info_text)
return w
def _build_config_tab(self) -> QWidget:
w = QWidget()
layout = QVBoxLayout(w)
# -- SSC 网络 --
g1 = QGroupBox("SSC 网络配置")
g1_layout = QGridLayout(g1)
fields_ssc = [
("设备 IP", "ssc_ip"), ("子网掩码", "ssc_mask"),
("网关", "ssc_gw"), ("LSSC IP", "ssc_lssc"),
("DNS", "ssc_dns"), ("端口", "ssc_port"),
]
for i, (label, name) in enumerate(fields_ssc):
g1_layout.addWidget(QLabel(label + ":"), i // 2, (i % 2) * 2)
edit = QLineEdit()
edit.setMaximumWidth(140)
g1_layout.addWidget(edit, i // 2, (i % 2) * 2 + 1)
setattr(self, f"_edit_{name}", edit)
btn_row = QHBoxLayout()
b1 = QPushButton("查询")
b1.clicked.connect(self._query_ssc_net)
b2 = QPushButton("设置")
b2.clicked.connect(self._set_ssc_net)
btn_row.addWidget(b1)
btn_row.addWidget(b2)
btn_row.addStretch()
g1_layout.addLayout(btn_row, 3, 0, 1, 4)
layout.addWidget(g1)
# -- IoT 网络 --
g2 = QGroupBox("IoT 网络配置")
g2_layout = QGridLayout(g2)
fields_iot = [
("MQTT Host", "iot_host"), ("MQTT Port", "iot_port"),
("Client ID", "iot_cid"), ("用户名", "iot_user"),
("密码", "iot_pass"),
]
for i, (label, name) in enumerate(fields_iot):
g2_layout.addWidget(QLabel(label + ":"), i, 0)
edit = QLineEdit()
edit.setMinimumWidth(200)
if "pass" in name:
edit.setEchoMode(QLineEdit.Password)
g2_layout.addWidget(edit, i, 1)
setattr(self, f"_edit_{name}", edit)
btn_row2 = QHBoxLayout()
b3 = QPushButton("查询")
b3.clicked.connect(self._query_iot_net)
b4 = QPushButton("设置")
b4.clicked.connect(self._set_iot_net)
btn_row2.addWidget(b3)
btn_row2.addWidget(b4)
btn_row2.addStretch()
g2_layout.addLayout(btn_row2, len(fields_iot), 0, 1, 2)
layout.addWidget(g2)
# -- Topic --
g3 = QGroupBox("IoT Topic 配置")
g3_layout = QGridLayout(g3)
self._chk_cid = QCheckBox("Client ID 启用")
self._chk_cid.setChecked(True)
g3_layout.addWidget(self._chk_cid, 0, 0, 1, 2)
g3_layout.addWidget(QLabel("Topic 发布:"), 1, 0)
self._edit_topic_pub = QLineEdit()
self._edit_topic_pub.setMinimumWidth(250)
g3_layout.addWidget(self._edit_topic_pub, 1, 1)
g3_layout.addWidget(QLabel("Topic 订阅:"), 2, 0)
self._edit_topic_sub = QLineEdit()
g3_layout.addWidget(self._edit_topic_sub, 2, 1)
btn_row3 = QHBoxLayout()
b5 = QPushButton("查询")
b5.clicked.connect(self._query_iot_topic)
b6 = QPushButton("设置")
b6.clicked.connect(self._set_iot_topic)
btn_row3.addWidget(b5)
btn_row3.addWidget(b6)
btn_row3.addStretch()
g3_layout.addLayout(btn_row3, 3, 0, 1, 2)
layout.addWidget(g3)
# -- 控制 --
g4 = QGroupBox("控制命令")
g4_layout = QGridLayout(g4)
g4_layout.addWidget(QLabel("密码:"), 0, 0)
self._edit_pwd = QLineEdit()
self._edit_pwd.setMaximumWidth(100)
self._edit_pwd.setEchoMode(QLineEdit.Password)
g4_layout.addWidget(self._edit_pwd, 0, 1)
g4_layout.addWidget(QLabel("新密码:"), 0, 2)
self._edit_new_pwd = QLineEdit()
self._edit_new_pwd.setMaximumWidth(100)
self._edit_new_pwd.setEchoMode(QLineEdit.Password)
g4_layout.addWidget(self._edit_new_pwd, 0, 3)
btn_row4 = QHBoxLayout()
for label, slot in [
("验证密码", self._do_pwd_verify), ("设置密码", self._do_pwd_set),
("出厂初始化", self._do_factory_reset), ("设备复位", self._do_device_reset),
]:
btn = QPushButton(label)
btn.clicked.connect(slot)
btn_row4.addWidget(btn)
btn_row4.addStretch()
g4_layout.addLayout(btn_row4, 1, 0, 1, 4)
layout.addWidget(g4)
layout.addStretch()
return w
def _build_data_tab(self) -> QWidget:
w = QWidget()
layout = QVBoxLayout(w)
layout.addWidget(QLabel("<b>线圈传感数据 (loop_data)</b>"))
self._loop_text = QTextEdit()
self._loop_text.setReadOnly(True)
self._loop_text.setFont(QFont("Consolas", 10))
layout.addWidget(self._loop_text, 2)
layout.addWidget(QLabel("<b>事件上报 (event_report)</b>"))
self._event_text = QTextEdit()
self._event_text.setReadOnly(True)
self._event_text.setFont(QFont("Consolas", 10))
layout.addWidget(self._event_text, 1)
return w
def _build_log_tab(self) -> QWidget:
w = QWidget()
layout = QVBoxLayout(w)
self._log_text = QTextEdit()
self._log_text.setReadOnly(True)
self._log_text.setFont(QFont("Consolas", 9))
layout.addWidget(self._log_text, 1)
btn_clear = QPushButton("清空")
btn_clear.clicked.connect(self._log_text.clear)
layout2 = QHBoxLayout()
layout2.addStretch()
layout2.addWidget(btn_clear)
layout.addLayout(layout2)
return w
# ================================================================
# MQTT
# ================================================================
def _toggle_connect(self):
if self._mqtt.connected:
self._mqtt.disconnect()
else:
cfg = BrokerConfig(
host=self._host_edit.text(),
port=int(self._port_edit.text() or "1883"),
username=self._user_edit.text(),
password=self._pass_edit.text(),
)
self._mqtt.connect(cfg)
self._log(f"正在连接 {cfg.host}:{cfg.port} ...")
@Slot(bool, str)
def _on_status(self, connected: bool, msg: str):
if connected:
self._btn_conn.setText("断开")
self._status_label.setText("● 已连接")
self._status_label.setStyleSheet("color: green; font-weight: bold;")
else:
self._btn_conn.setText("连接")
self._status_label.setText("● 未连接")
self._status_label.setStyleSheet("color: red; font-weight: bold;")
self._log(msg)
@Slot(str, str, object)
def _on_message(self, topic: str, dev_serial: str, payload: dict):
cmd = payload.get("cmd", "")
ts = datetime.fromtimestamp(payload.get("ts", 0)).strftime("%H:%M:%S")
if "/dev/config/resp" in topic:
code = payload.get("code", -1)
pmsg = payload.get("msg", "")
data = payload.get("data", {})
self._log(f"[{ts}] {dev_serial} ← {cmd} code={code} {pmsg}")
if code == 0:
if cmd == CMD_DEV_INFO_QUERY:
self._devmgr.update_from_dev_info(dev_serial, data)
self._show_json(data)
elif cmd in (CMD_SSC_NET_QUERY, CMD_IOT_NET_QUERY, CMD_IOT_TOPIC_QUERY, CMD_LOOP_PARAM_QUERY):
self._show_json({cmd: data})
else:
self._show_json({"error": f"code={code} {pmsg} ({ERROR_MSGS.get(code, '?')})"})
elif "/dev/data/loop" in topic:
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:
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:
data = payload.get("data", {})
self._devmgr.update_heartbeat(dev_serial, data)
self._log(f"[{ts}] {dev_serial} ← heartbeat (uptime={data.get('uptime', 0)}s)")
# ================================================================
# 设备操作
# ================================================================
def _selected_dev(self) -> Optional[str]:
items = self._dev_tree.selectedItems()
return items[0].text(0).lstrip("●○ ") if items else None
def _require_dev(self) -> Optional[str]:
sn = self._selected_dev()
if not sn:
QMessageBox.warning(self, "提示", "请先选择一个设备")
return None
if not self._mqtt.connected:
QMessageBox.warning(self, "提示", "请先连接 MQTT Broker")
return None
return sn
def _send_cmd(self, cmd: str, data: Optional[dict] = None):
sn = self._require_dev()
if not sn:
return
try:
mid = self._mqtt.send_command(sn, cmd, data)
self._log(f"→ {sn} {cmd} (msg_id={mid})")
except Exception as e:
QMessageBox.critical(self, "发送失败", str(e))
def _query_dev_info(self):
self._send_cmd(CMD_DEV_INFO_QUERY)
def _query_all(self):
sn = self._require_dev()
if not sn:
return
for cmd in [CMD_DEV_INFO_QUERY, CMD_SSC_NET_QUERY, CMD_IOT_NET_QUERY,
CMD_IOT_TOPIC_QUERY, CMD_LOOP_PARAM_QUERY]:
try:
self._mqtt.send_command(sn, cmd)
except Exception:
pass
def _query_ssc_net(self):
self._send_cmd(CMD_SSC_NET_QUERY)
def _set_ssc_net(self):
self._send_cmd(CMD_SSC_NET_SET, data_ssc_net_set(
dev_ip=self._edit_ssc_ip.text(),
subnet_mask=self._edit_ssc_mask.text(),
route_ip=self._edit_ssc_gw.text(),
lssc_ip=self._edit_ssc_lssc.text(),
dns=self._edit_ssc_dns.text(),
port=int(self._edit_ssc_port.text() or "0"),
))
def _query_iot_net(self):
self._send_cmd(CMD_IOT_NET_QUERY)
def _set_iot_net(self):
self._send_cmd(CMD_IOT_NET_SET, data_iot_net_set(
host=self._edit_iot_host.text(),
port=int(self._edit_iot_port.text() or "1883"),
client_id=self._edit_iot_cid.text(),
username=self._edit_iot_user.text(),
password=self._edit_iot_pass.text(),
))
def _query_iot_topic(self):
self._send_cmd(CMD_IOT_TOPIC_QUERY)
def _set_iot_topic(self):
self._send_cmd(CMD_IOT_TOPIC_SET, data_iot_topic_set(
client_id_enable=self._chk_cid.isChecked(),
topic_pub=self._edit_topic_pub.text(),
topic_sub=self._edit_topic_sub.text(),
))
def _do_pwd_verify(self):
pwd = self._edit_pwd.text()
if not pwd:
QMessageBox.warning(self, "提示", "请输入密码")
return
self._send_cmd(CMD_PWD_VERIFY, data_pwd_verify(pwd))
def _do_pwd_set(self):
old = self._edit_pwd.text()
new = self._edit_new_pwd.text()
if not old or not new:
QMessageBox.warning(self, "提示", "请输入旧密码和新密码")
return
self._send_cmd(CMD_PWD_SET, data_pwd_set(old, new))
def _do_factory_reset(self):
r = QMessageBox.question(self, "确认", "确定要恢复出厂设置吗？\n此操作不可撤销！",
QMessageBox.Yes | QMessageBox.No)
if r == QMessageBox.Yes:
self._send_cmd(CMD_FACTORY_RESET)
def _do_device_reset(self):
r = QMessageBox.question(self, "确认", "确定要复位设备吗？设备将重启。",
QMessageBox.Yes | QMessageBox.No)
if r == QMessageBox.Yes:
self._send_cmd(CMD_DEVICE_RESET)
# ================================================================
# UI 更新
# ================================================================
@Slot()
def _refresh_devices(self):
tree = self._dev_tree
existing = {}
for i in range(tree.topLevelItemCount()):
item = tree.topLevelItem(i)
sn = item.text(0).lstrip("●○ ")
existing[sn] = item
for sn, dev in self._devmgr.devices.items():
status = "●" if dev.online else "○"
model = dev.model or "?"
if sn in existing:
existing[sn].setText(0, f"{status} {sn}")
existing[sn].setText(1, model)
del existing[sn]
else:
item = QTreeWidgetItem([f"{status} {sn}", model])
tree.addTopLevelItem(item)
for sn, item in existing.items():
idx = tree.indexOfTopLevelItem(item)
tree.takeTopLevelItem(idx)
def _on_dev_select(self):
sn = self._selected_dev()
if sn and sn in self._devmgr.devices:
dev = self._devmgr.devices[sn]
info = {
"dev_serial": dev.dev_serial,
"model": dev.model,
"hard_ver": dev.hard_ver,
"soft_ver": dev.soft_ver,
"product_code": dev.product_code,
"net_enabled": dev.net_enabled,
"iot_enabled": dev.iot_enabled,
"bus": dev.bus,
"online": dev.online,
"last_seen": dev.last_seen.strftime("%Y-%m-%d %H:%M:%S") if dev.last_seen else "—",
}
self._show_json(info)
def _show_json(self, data: dict):
self._info_text.setPlainText(json.dumps(data, indent=2, ensure_ascii=False))
def _append_text(self, widget: QTextEdit, text: str):
ts = datetime.now().strftime("%H:%M:%S")
widget.append(f"--- {ts} ---\n{text}\n")
widget.moveCursor(QTextCursor.End)
def _log(self, msg: str):
ts = datetime.now().strftime("%H:%M:%S")
self._log_text.append(f"[{ts}] {msg}")
self._log_text.moveCursor(QTextCursor.End)
def main():
app = QApplication(sys.argv)
app.setStyle("Fusion")
win = MainWindow()
win.show()
sys.exit(app.exec())
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink