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

refactor(DBNMQTTool): tkinter → PySide6

Browse Source 
- main.py: 完整重写为 PySide6 (Qt for Python)
- requirements.txt: 加 PySide6>=6.6.0
- 使用 Qt 信号/槽替代 tkinter 回调
- 界面更现代化，Fusion 风格
- 分组框 (QGroupBox) 布局，QSplitter 分栏
This commit is contained in:
wangfq
2026-07-06 21:19:55 +08:00
parent a2cfe4602c
commit e1bf3dcfca
1 changed files with 341 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DBNMQTTool/main.py
+339 -284
View File
@@ -1,283 +1,346 @@
#!/usr/bin/env python3
"""
DBN MQTT Tool — DLD960 IoT MQTT 设备管理工具
跨平台桌面软件 (tkinter)
跨平台桌面软件 (PySide6)
"""
import sys
import os
import json
import time
import tkinter as tk
from tkinter import ttk, messagebox, scrolledtext
from threading import Thread
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, DeviceInfo
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_LOOP_PARAM_SET, CMD_REPORT_CONFIG,
CMD_PWD_VERIFY, CMD_PWD_SET, CMD_FACTORY_RESET, CMD_DEVICE_RESET,
CMD_DEV_SERIAL_SET,
CMD_LOOP_DATA, CMD_EVENT_REPORT, CMD_HEARTBEAT,
ERROR_MSGS, FREQ_LEVELS, OUTPUT_MODES,
ERROR_MSGS,
data_ssc_net_set, data_iot_net_set, data_iot_topic_set,
data_loop_param_set, data_report_config,
data_pwd_verify, data_pwd_set, data_dev_serial_set,
topic_data_loop, topic_data_event, topic_status,
data_pwd_verify, data_pwd_set,
)
class DBNMqttTool(tk.Tk):
"""主窗口"""
class MainWindow(QMainWindow):
_mqtt_status = Signal(bool, str)
_mqtt_msg = Signal(str, str, object)
_devices_changed = Signal()
def __init__(self):
super().__init__()
self.title("DBN MQTT Tool — DLD960 IoT 设备管理")
self.geometry("1200x750")
self.minsize(900, 600)
self.setWindowTitle("DBN MQTT Tool — DLD960 IoT 设备管理")
self.resize(1280, 800)
self.setMinimumSize(960, 640)
self._mqtt = MqttClient()
self._devmgr = DeviceManager()
# MQTT 回调
self._mqtt.on_status_change(self._on_mqtt_status)
self._mqtt.on_message(self._on_mqtt_message)
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._devmgr.on_change(self._refresh_device_list)
self._pending: dict[int, str] = {} # msg_id → 命令名
self._log_lines: list[str] = []
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):
# -- 顶栏：连接 --
top_bar = ttk.Frame(self, padding=5)
top_bar.pack(fill=tk.X)
central = QWidget()
self.setCentralWidget(central)
root = QVBoxLayout(central)
root.setContentsMargins(8, 8, 8, 8)
root.setSpacing(6)
ttk.Label(top_bar, text="Broker:").pack(side=tk.LEFT, padx=(0, 5))
self._host_var = tk.StringVar(value="121.37.20.199")
ttk.Entry(top_bar, textvariable=self._host_var, width=16).pack(side=tk.LEFT, padx=2)
ttk.Label(top_bar, text="Port:").pack(side=tk.LEFT, padx=(5, 2))
self._port_var = tk.IntVar(value=1883)
ttk.Entry(top_bar, textvariable=self._port_var, width=6).pack(side=tk.LEFT, padx=2)
ttk.Label(top_bar, text="用户:").pack(side=tk.LEFT, padx=(5, 2))
self._user_var = tk.StringVar()
ttk.Entry(top_bar, textvariable=self._user_var, width=10).pack(side=tk.LEFT, padx=2)
ttk.Label(top_bar, text="密码:").pack(side=tk.LEFT, padx=(5, 2))
self._pass_var = tk.StringVar()
ttk.Entry(top_bar, textvariable=self._pass_var, width=10, show="*").pack(side=tk.LEFT, padx=2)
# -- 顶栏 --
root.addWidget(self._build_top_bar())
self._btn_connect = ttk.Button(top_bar, text="连接", command=self._toggle_connect)
self._btn_connect.pack(side=tk.LEFT, padx=10)
# -- 主体 --
splitter = QSplitter(Qt.Horizontal)
self._conn_status = ttk.Label(top_bar, text="● 未连接", foreground="red")
self._conn_status.pack(side=tk.LEFT, padx=10)
left = self._build_device_panel()
splitter.addWidget(left)
# -- 主体：左右分栏 --
main = ttk.PanedWindow(self, orient=tk.HORIZONTAL)
main.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
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)
# 左侧：设备列表
left = ttk.Frame(main, width=280)
main.add(left, weight=0)
splitter.setSizes([280, 960])
root.addWidget(splitter, 1)
ttk.Label(left, text="设备列表", font=("", 11, "bold")).pack(anchor=tk.W, pady=(0, 5))
self._dev_tree = ttk.Treeview(left, columns=("model",), show="tree headings", height=14)
self._dev_tree.heading("#0", text="设备序列码")
self._dev_tree.heading("model", text="型号")
self._dev_tree.column("#0", width=140)
self._dev_tree.column("model", width=80)
self._dev_tree.pack(fill=tk.BOTH, expand=True)
self._dev_tree.bind("<<TreeviewSelect>>", self._on_device_select)
def _build_top_bar(self) -> QWidget:
bar = QWidget()
layout = QHBoxLayout(bar)
layout.setContentsMargins(0, 0, 0, 0)
layout.setSpacing(6)
btn_frame = ttk.Frame(left)
btn_frame.pack(fill=tk.X, pady=5)
ttk.Button(btn_frame, text="查询设备信息", command=self._query_dev_info).pack(side=tk.LEFT, padx=2)
ttk.Button(btn_frame, text="查询全部", command=self._query_all).pack(side=tk.LEFT, padx=2)
layout.addWidget(QLabel("Broker:"))
self._host_edit = QLineEdit("121.37.20.199")
self._host_edit.setMaximumWidth(140)
layout.addWidget(self._host_edit)
# 右侧：标签页
right = ttk.Frame(main)
main.add(right, weight=1)
layout.addWidget(QLabel("Port:"))
self._port_edit = QLineEdit("1883")
self._port_edit.setMaximumWidth(60)
layout.addWidget(self._port_edit)
self._notebook = ttk.Notebook(right)
self._notebook.pack(fill=tk.BOTH, expand=True)
layout.addWidget(QLabel("用户:"))
self._user_edit = QLineEdit()
self._user_edit.setMaximumWidth(90)
layout.addWidget(self._user_edit)
self._tab_info = self._add_info_tab()
self._tab_config = self._add_config_tab()
self._tab_data = self._add_data_tab()
self._tab_log = self._add_log_tab()
layout.addWidget(QLabel("密码:"))
self._pass_edit = QLineEdit()
self._pass_edit.setMaximumWidth(90)
self._pass_edit.setEchoMode(QLineEdit.Password)
layout.addWidget(self._pass_edit)
def _add_info_tab(self):
f = ttk.Frame(self._notebook, padding=10)
self._notebook.add(f, text="设备信息")
self._btn_conn = QPushButton("连接")
self._btn_conn.setFixedWidth(70)
self._btn_conn.clicked.connect(self._toggle_connect)
layout.addWidget(self._btn_conn)
self._info_text = scrolledtext.ScrolledText(f, height=18, width=60, state=tk.DISABLED,
font=("Consolas", 10))
self._info_text.pack(fill=tk.BOTH, expand=True)
return f
self._status_label = QLabel("● 未连接")
self._status_label.setStyleSheet("color: red; font-weight: bold;")
layout.addWidget(self._status_label)
def _add_config_tab(self):
f = ttk.Frame(self._notebook, padding=10)
self._notebook.add(f, text="参数配置")
layout.addStretch()
return bar
row = 0
# -- 网口配置 --
ttk.Label(f, text="▸ SSC 网络配置", font=("", 10, "bold")).grid(row=row, column=0, columnspan=4, sticky=tk.W, pady=(0, 5))
row += 1
for label, var_name, col in [
("设备 IP", "ssc_ip", 0), ("子网掩码", "ssc_mask", 1),
("网关", "ssc_gw", 2), ("LSSC IP", "ssc_lssc", 3),
("DNS", "ssc_dns", 0), ("端口", "ssc_port", 1),
]:
ttk.Label(f, text=label + ":").grid(row=row, column=col * 2, sticky=tk.E, padx=2, pady=1)
var = tk.StringVar()
setattr(self, f"_var_{var_name}", var)
ttk.Entry(f, textvariable=var, width=15).grid(row=row, column=col * 2 + 1, sticky=tk.W, padx=2, pady=1)
if col == 3 or (col == 1 and var_name == "ssc_port"):
row += 1
def _build_device_panel(self) -> QWidget:
w = QWidget()
layout = QVBoxLayout(w)
layout.setContentsMargins(0, 0, 0, 0)
ttk.Button(f, text="查询 SSC 网络", command=self._query_ssc_net).grid(row=row, column=0, pady=5, padx=2, sticky=tk.W)
ttk.Button(f, text="设置 SSC 网络", command=self._set_ssc_net).grid(row=row, column=1, pady=5, padx=2, sticky=tk.W)
row += 1
layout.addWidget(QLabel("<b>设备列表</b>"))
# -- IoT 配置 --
ttk.Separator(f, orient=tk.HORIZONTAL).grid(row=row, column=0, columnspan=4, sticky=tk.EW, pady=10)
row += 1
ttk.Label(f, text="▸ IoT 网络配置", font=("", 10, "bold")).grid(row=row, column=0, columnspan=4, sticky=tk.W, pady=(0, 5))
row += 1
for label, var_name in [
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"),
("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),
]:
ttk.Label(f, text=label + ":").grid(row=row, column=0, sticky=tk.E, padx=2, pady=1)
var = tk.StringVar()
setattr(self, f"_var_{var_name}", var)
w = ttk.Entry(f, textvariable=var, width=30)
if "pass" in var_name.lower():
w.configure(show="*")
w.grid(row=row, column=1, columnspan=3, sticky=tk.W, padx=2, pady=1)
row += 1
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)
ttk.Button(f, text="查询 IoT 网络", command=self._query_iot_net).grid(row=row, column=0, pady=5, padx=2, sticky=tk.W)
ttk.Button(f, text="设置 IoT 网络", command=self._set_iot_net).grid(row=row, column=1, pady=5, padx=2, sticky=tk.W)
row += 1
layout.addStretch()
return w
# -- Topic 配置 --
ttk.Separator(f, orient=tk.HORIZONTAL).grid(row=row, column=0, columnspan=4, sticky=tk.EW, pady=10)
row += 1
ttk.Label(f, text="▸ IoT Topic 配置", font=("", 10, "bold")).grid(row=row, column=0, columnspan=4, sticky=tk.W, pady=(0, 5))
row += 1
self._var_cid_enable = tk.BooleanVar(value=True)
ttk.Checkbutton(f, text="Client ID 启用", variable=self._var_cid_enable).grid(row=row, column=0, columnspan=2, sticky=tk.W)
row += 1
for label, var_name in [("Topic 发布", "topic_pub"), ("Topic 订阅", "topic_sub")]:
ttk.Label(f, text=label + ":").grid(row=row, column=0, sticky=tk.E, padx=2, pady=1)
var = tk.StringVar()
setattr(self, f"_var_{var_name}", var)
ttk.Entry(f, textvariable=var, width=40).grid(row=row, column=1, columnspan=3, sticky=tk.W, padx=2, pady=1)
row += 1
def _build_data_tab(self) -> QWidget:
w = QWidget()
layout = QVBoxLayout(w)
ttk.Button(f, text="查询 Topic", command=self._query_iot_topic).grid(row=row, column=0, pady=5, padx=2, sticky=tk.W)
ttk.Button(f, text="设置 Topic", command=self._set_iot_topic).grid(row=row, column=1, pady=5, padx=2, sticky=tk.W)
row += 1
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)
# -- 控制区 --
ttk.Separator(f, orient=tk.HORIZONTAL).grid(row=row, column=0, columnspan=4, sticky=tk.EW, pady=10)
row += 1
ttk.Label(f, text="▸ 控制命令", font=("", 10, "bold")).grid(row=row, column=0, columnspan=4, sticky=tk.W, pady=(0, 5))
row += 1
self._var_pwd = tk.StringVar()
ttk.Label(f, text="密码:").grid(row=row, column=0, sticky=tk.E, padx=2)
ttk.Entry(f, textvariable=self._var_pwd, width=10, show="*").grid(row=row, column=1, sticky=tk.W, padx=2)
self._var_new_pwd = tk.StringVar()
ttk.Label(f, text="新密码:").grid(row=row, column=2, sticky=tk.E, padx=2)
ttk.Entry(f, textvariable=self._var_new_pwd, width=10, show="*").grid(row=row, column=3, sticky=tk.W, padx=2)
row += 1
ttk.Button(f, text="验证密码", command=self._do_pwd_verify).grid(row=row, column=0, pady=5, padx=2)
ttk.Button(f, text="设置密码", command=self._do_pwd_set).grid(row=row, column=1, pady=5, padx=2)
ttk.Button(f, text="出厂初始化", command=self._do_factory_reset).grid(row=row, column=2, pady=5, padx=2)
ttk.Button(f, text="设备复位", command=self._do_device_reset).grid(row=row, column=3, pady=5, padx=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
return f
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)
def _add_data_tab(self):
f = ttk.Frame(self._notebook, padding=10)
self._notebook.add(f, text="实时数据")
# 线圈数据
ttk.Label(f, text="▸ 线圈传感数据 (loop_data)", font=("", 10, "bold")).pack(anchor=tk.W)
self._loop_text = scrolledtext.ScrolledText(f, height=10, width=80, state=tk.DISABLED,
font=("Consolas", 10))
self._loop_text.pack(fill=tk.BOTH, expand=True, pady=5)
# 事件
ttk.Label(f, text="▸ 事件上报 (event_report)", font=("", 10, "bold")).pack(anchor=tk.W)
self._event_text = scrolledtext.ScrolledText(f, height=6, width=80, state=tk.DISABLED,
font=("Consolas", 10))
self._event_text.pack(fill=tk.BOTH, expand=True, pady=5)
return f
def _add_log_tab(self):
f = ttk.Frame(self._notebook, padding=10)
self._notebook.add(f, text="日志")
self._log_text = scrolledtext.ScrolledText(f, height=18, width=60, state=tk.DISABLED,
font=("Consolas", 9))
self._log_text.pack(fill=tk.BOTH, expand=True)
ttk.Button(f, text="清空", command=self._clear_log).pack(anchor=tk.E, pady=5)
return f
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 操作
# MQTT
# ================================================================
def _toggle_connect(self):
if self._mqtt.connected:
self._mqtt.disconnect()
else:
config = BrokerConfig(
host=self._host_var.get(),
port=self._port_var.get(),
username=self._user_var.get(),
password=self._pass_var.get(),
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(config)
self._log(f"正在连接 {config.host}:{config.port} ...")
self._mqtt.connect(cfg)
self._log(f"正在连接 {cfg.host}:{cfg.port} ...")
def _on_mqtt_status(self, connected: bool, msg: str):
self.after(0, lambda: self._update_conn_status(connected, msg))
def _update_conn_status(self, connected: bool, msg: str):
@Slot(bool, str)
def _on_status(self, connected: bool, msg: str):
if connected:
self._btn_connect.configure(text="断开")
self._conn_status.configure(text="● 已连接", foreground="green")
self._btn_conn.setText("断开")
self._status_label.setText("● 已连接")
self._status_label.setStyleSheet("color: green; font-weight: bold;")
else:
self._btn_connect.configure(text="连接")
self._conn_status.configure(text="● 未连接", foreground="red")
self._btn_conn.setText("连接")
self._status_label.setText("● 未连接")
self._status_label.setStyleSheet("color: red; font-weight: bold;")
self._log(msg)
def _on_mqtt_message(self, topic: str, dev_serial: str, payload: dict):
self.after(0, lambda: self._handle_message(topic, dev_serial, payload))
def _handle_message(self, topic: str, dev_serial: str, payload: dict):
@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")
@@ -289,26 +352,25 @@ class DBNMqttTool(tk.Tk):
if code == 0:
if cmd == CMD_DEV_INFO_QUERY:
self._devmgr.update_from_dev_info(dev_serial, data)
self._show_info(data)
elif cmd in (CMD_SSC_NET_QUERY, CMD_IOT_NET_QUERY, CMD_IOT_TOPIC_QUERY):
self._show_info({cmd: data})
elif cmd == CMD_LOOP_PARAM_QUERY:
self._show_info({cmd: 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_info({"error": f"code={code} {pmsg} ({ERROR_MSGS.get(code, '?')})"})
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)
self._log(f"[{ts}] {dev_serial} ← loop_data ({len(data.get('channels', []))} 通道)")
self._append_data(self._loop_text, json.dumps(payload, indent=2, ensure_ascii=False))
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)
events = data.get("events", [])
self._log(f"[{ts}] {dev_serial} ← event_report ({len(events)} 事件)")
self._append_data(self._event_text, json.dumps(payload, indent=2, ensure_ascii=False))
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", {})
@@ -320,18 +382,16 @@ class DBNMqttTool(tk.Tk):
# ================================================================
def _selected_dev(self) -> Optional[str]:
sel = self._dev_tree.selection()
if sel:
return self._dev_tree.item(sel[0], "text")
return None
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:
messagebox.showwarning("提示", "请先在左侧选择一个设备")
QMessageBox.warning(self, "提示", "请先选择一个设备")
return None
if not self._mqtt.connected:
messagebox.showwarning("提示", "请先连接 MQTT Broker")
QMessageBox.warning(self, "提示", "请先连接 MQTT Broker")
return None
return sn
@@ -340,10 +400,10 @@ class DBNMqttTool(tk.Tk):
if not sn:
return
try:
msg_id = self._mqtt.send_command(sn, cmd, data)
self._log(f"→ {sn} {cmd} (msg_id={msg_id})")
mid = self._mqtt.send_command(sn, cmd, data)
self._log(f"→ {sn} {cmd} (msg_id={mid})")
except Exception as e:
messagebox.showerror("发送失败", str(e))
QMessageBox.critical(self, "发送失败", str(e))
def _query_dev_info(self):
self._send_cmd(CMD_DEV_INFO_QUERY)
@@ -364,12 +424,12 @@ class DBNMqttTool(tk.Tk):
def _set_ssc_net(self):
self._send_cmd(CMD_SSC_NET_SET, data_ssc_net_set(
dev_ip=self._var_ssc_ip.get(),
subnet_mask=self._var_ssc_mask.get(),
route_ip=self._var_ssc_gw.get(),
lssc_ip=self._var_ssc_lssc.get(),
dns=self._var_ssc_dns.get(),
port=int(self._var_ssc_port.get() or "0"),
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):
@@ -377,11 +437,11 @@ class DBNMqttTool(tk.Tk):
def _set_iot_net(self):
self._send_cmd(CMD_IOT_NET_SET, data_iot_net_set(
host=self._var_iot_host.get(),
port=int(self._var_iot_port.get() or "1883"),
client_id=self._var_iot_cid.get(),
username=self._var_iot_user.get(),
password=self._var_iot_pass.get(),
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):
@@ -389,60 +449,67 @@ class DBNMqttTool(tk.Tk):
def _set_iot_topic(self):
self._send_cmd(CMD_IOT_TOPIC_SET, data_iot_topic_set(
client_id_enable=self._var_cid_enable.get(),
topic_pub=self._var_topic_pub.get(),
topic_sub=self._var_topic_sub.get(),
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._var_pwd.get()
pwd = self._edit_pwd.text()
if not pwd:
messagebox.showwarning("提示", "请输入密码")
QMessageBox.warning(self, "提示", "请输入密码")
return
self._send_cmd(CMD_PWD_VERIFY, data_pwd_verify(pwd))
def _do_pwd_set(self):
old = self._var_pwd.get()
new = self._var_new_pwd.get()
old = self._edit_pwd.text()
new = self._edit_new_pwd.text()
if not old or not new:
messagebox.showwarning("提示", "请输入旧密码和新密码")
QMessageBox.warning(self, "提示", "请输入旧密码和新密码")
return
self._send_cmd(CMD_PWD_SET, data_pwd_set(old, new))
def _do_factory_reset(self):
if not messagebox.askyesno("确认", "确定要恢复出厂设置吗？此操作不可撤销！"):
return
r = QMessageBox.question(self, "确认", "确定要恢复出厂设置吗？\n此操作不可撤销！",
QMessageBox.Yes | QMessageBox.No)
if r == QMessageBox.Yes:
self._send_cmd(CMD_FACTORY_RESET)
def _do_device_reset(self):
if not messagebox.askyesno("确认", "确定要复位设备吗？设备将重启。"):
return
r = QMessageBox.question(self, "确认", "确定要复位设备吗？设备将重启。",
QMessageBox.Yes | QMessageBox.No)
if r == QMessageBox.Yes:
self._send_cmd(CMD_DEVICE_RESET)
# ================================================================
# UI 更新
# ================================================================
def _refresh_device_list(self):
@Slot()
def _refresh_devices(self):
tree = self._dev_tree
current = set(tree.get_children(""))
devices = self._devmgr.devices
existing = {}
for i in range(tree.topLevelItemCount()):
item = tree.topLevelItem(i)
sn = item.text(0).lstrip("●○ ")
existing[sn] = item
for sn, dev in devices.items():
iid = sn
for sn, dev in self._devmgr.devices.items():
status = "●" if dev.online else "○"
display = f"{status} {sn}"
model = dev.model or "?"
if iid in current:
tree.item(iid, text=display, values=(model,))
current.discard(iid)
if sn in existing:
existing[sn].setText(0, f"{status} {sn}")
existing[sn].setText(1, model)
del existing[sn]
else:
tree.insert("", tk.END, iid=iid, text=display, values=(model,))
item = QTreeWidgetItem([f"{status} {sn}", model])
tree.addTopLevelItem(item)
for stale in current:
tree.delete(stale)
for sn, item in existing.items():
idx = tree.indexOfTopLevelItem(item)
tree.takeTopLevelItem(idx)
def _on_device_select(self, event):
def _on_dev_select(self):
sn = self._selected_dev()
if sn and sn in self._devmgr.devices:
dev = self._devmgr.devices[sn]
@@ -458,40 +525,28 @@ class DBNMqttTool(tk.Tk):
"online": dev.online,
"last_seen": dev.last_seen.strftime("%Y-%m-%d %H:%M:%S") if dev.last_seen else "—",
}
self._show_info(info)
self._show_json(info)
def _show_info(self, data: dict):
self._info_text.configure(state=tk.NORMAL)
self._info_text.delete("1.0", tk.END)
self._info_text.insert(tk.END, json.dumps(data, indent=2, ensure_ascii=False))
self._info_text.configure(state=tk.DISABLED)
def _show_json(self, data: dict):
self._info_text.setPlainText(json.dumps(data, indent=2, ensure_ascii=False))
def _append_data(self, widget: scrolledtext.ScrolledText, text: str):
widget.configure(state=tk.NORMAL)
def _append_text(self, widget: QTextEdit, text: str):
ts = datetime.now().strftime("%H:%M:%S")
widget.insert(tk.END, f"--- {ts} ---\n{text}\n\n")
widget.see(tk.END)
widget.configure(state=tk.DISABLED)
widget.append(f"--- {ts} ---\n{text}\n")
widget.moveCursor(QTextCursor.End)
def _log(self, msg: str):
ts = datetime.now().strftime("%H:%M:%S")
line = f"[{ts}] {msg}"
self._log_lines.append(line)
self._log_text.configure(state=tk.NORMAL)
self._log_text.insert(tk.END, line + "\n")
self._log_text.see(tk.END)
self._log_text.configure(state=tk.DISABLED)
def _clear_log(self):
self._log_text.configure(state=tk.NORMAL)
self._log_text.delete("1.0", tk.END)
self._log_text.configure(state=tk.DISABLED)
self._log_lines.clear()
self._log_text.append(f"[{ts}] {msg}")
self._log_text.moveCursor(QTextCursor.End)
def main():
app = DBNMqttTool()
app.mainloop()
app = QApplication(sys.argv)
app.setStyle("Fusion")
win = MainWindow()
win.show()
sys.exit(app.exec())
if __name__ == "__main__":