wangfq
a2cfe4602c
基于《DLD960_IoT_MQTT协议.md》V1.00 实现 功能： - MQTT Broker 连接管理 - 设备自动发现（订阅 dld960/+/dev/# 通配符） - 设备信息查询（dev_info_query） - 网络配置（SSC/IoT TCP）+ Topic 配置 - 实时线圈数据监控（loop_data）+ 事件上报（event_report） - 心跳监控（heartbeat） - 控制命令：密码验证/设置、出厂初始化、设备复位 项目结构： - main.py 主窗口 (tkinter GUI) - dbn_mqtt_tool/ - protocol.py DLD960 IoT MQTT 协议定义 - mqtt_client.py MQTT 客户端封装 (paho-mqtt) - device_manager.py 设备发现与状态管理
499 lines
20 KiB
Python
499 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
DBN MQTT Tool — DLD960 IoT MQTT 设备管理工具
|
|
|
|
跨平台桌面软件 (tkinter)
|
|
"""
|
|
|
|
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 dbn_mqtt_tool.mqtt_client import MqttClient, BrokerConfig
|
|
from dbn_mqtt_tool.device_manager import DeviceManager, DeviceInfo
|
|
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,
|
|
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,
|
|
)
|
|
|
|
|
|
class DBNMqttTool(tk.Tk):
|
|
"""主窗口"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
self.title("DBN MQTT Tool — DLD960 IoT 设备管理")
|
|
self.geometry("1200x750")
|
|
self.minsize(900, 600)
|
|
|
|
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._devmgr.on_change(self._refresh_device_list)
|
|
|
|
self._pending: dict[int, str] = {} # msg_id → 命令名
|
|
self._log_lines: list[str] = []
|
|
|
|
self._build_ui()
|
|
|
|
# ================================================================
|
|
# UI 构建
|
|
# ================================================================
|
|
|
|
def _build_ui(self):
|
|
# -- 顶栏：连接 --
|
|
top_bar = ttk.Frame(self, padding=5)
|
|
top_bar.pack(fill=tk.X)
|
|
|
|
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)
|
|
|
|
self._btn_connect = ttk.Button(top_bar, text="连接", command=self._toggle_connect)
|
|
self._btn_connect.pack(side=tk.LEFT, padx=10)
|
|
|
|
self._conn_status = ttk.Label(top_bar, text="● 未连接", foreground="red")
|
|
self._conn_status.pack(side=tk.LEFT, padx=10)
|
|
|
|
# -- 主体：左右分栏 --
|
|
main = ttk.PanedWindow(self, orient=tk.HORIZONTAL)
|
|
main.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
|
|
|
|
# 左侧：设备列表
|
|
left = ttk.Frame(main, width=280)
|
|
main.add(left, weight=0)
|
|
|
|
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)
|
|
|
|
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)
|
|
|
|
# 右侧：标签页
|
|
right = ttk.Frame(main)
|
|
main.add(right, weight=1)
|
|
|
|
self._notebook = ttk.Notebook(right)
|
|
self._notebook.pack(fill=tk.BOTH, expand=True)
|
|
|
|
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()
|
|
|
|
def _add_info_tab(self):
|
|
f = ttk.Frame(self._notebook, padding=10)
|
|
self._notebook.add(f, text="设备信息")
|
|
|
|
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
|
|
|
|
def _add_config_tab(self):
|
|
f = ttk.Frame(self._notebook, padding=10)
|
|
self._notebook.add(f, text="参数配置")
|
|
|
|
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
|
|
|
|
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
|
|
|
|
# -- 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 [
|
|
("MQTT Host", "iot_host"), ("MQTT Port", "iot_port"),
|
|
("Client ID", "iot_cid"), ("用户名", "iot_user"), ("密码", "iot_pass"),
|
|
]:
|
|
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
|
|
|
|
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
|
|
|
|
# -- 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
|
|
|
|
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
|
|
|
|
# -- 控制区 --
|
|
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)
|
|
|
|
return f
|
|
|
|
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
|
|
|
|
# ================================================================
|
|
# 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(),
|
|
)
|
|
self._mqtt.connect(config)
|
|
self._log(f"正在连接 {config.host}:{config.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):
|
|
if connected:
|
|
self._btn_connect.configure(text="断开")
|
|
self._conn_status.configure(text="● 已连接", foreground="green")
|
|
else:
|
|
self._btn_connect.configure(text="连接")
|
|
self._conn_status.configure(text="● 未连接", foreground="red")
|
|
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):
|
|
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_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})
|
|
else:
|
|
self._show_info({"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))
|
|
|
|
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))
|
|
|
|
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]:
|
|
sel = self._dev_tree.selection()
|
|
if sel:
|
|
return self._dev_tree.item(sel[0], "text")
|
|
return None
|
|
|
|
def _require_dev(self) -> Optional[str]:
|
|
sn = self._selected_dev()
|
|
if not sn:
|
|
messagebox.showwarning("提示", "请先在左侧选择一个设备")
|
|
return None
|
|
if not self._mqtt.connected:
|
|
messagebox.showwarning("提示", "请先连接 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:
|
|
msg_id = self._mqtt.send_command(sn, cmd, data)
|
|
self._log(f"→ {sn} {cmd} (msg_id={msg_id})")
|
|
except Exception as e:
|
|
messagebox.showerror("发送失败", 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._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"),
|
|
))
|
|
|
|
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._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(),
|
|
))
|
|
|
|
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._var_cid_enable.get(),
|
|
topic_pub=self._var_topic_pub.get(),
|
|
topic_sub=self._var_topic_sub.get(),
|
|
))
|
|
|
|
def _do_pwd_verify(self):
|
|
pwd = self._var_pwd.get()
|
|
if not pwd:
|
|
messagebox.showwarning("提示", "请输入密码")
|
|
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()
|
|
if not old or not new:
|
|
messagebox.showwarning("提示", "请输入旧密码和新密码")
|
|
return
|
|
self._send_cmd(CMD_PWD_SET, data_pwd_set(old, new))
|
|
|
|
def _do_factory_reset(self):
|
|
if not messagebox.askyesno("确认", "确定要恢复出厂设置吗？此操作不可撤销！"):
|
|
return
|
|
self._send_cmd(CMD_FACTORY_RESET)
|
|
|
|
def _do_device_reset(self):
|
|
if not messagebox.askyesno("确认", "确定要复位设备吗？设备将重启。"):
|
|
return
|
|
self._send_cmd(CMD_DEVICE_RESET)
|
|
|
|
# ================================================================
|
|
# UI 更新
|
|
# ================================================================
|
|
|
|
def _refresh_device_list(self):
|
|
tree = self._dev_tree
|
|
current = set(tree.get_children(""))
|
|
devices = self._devmgr.devices
|
|
|
|
for sn, dev in devices.items():
|
|
iid = sn
|
|
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)
|
|
else:
|
|
tree.insert("", tk.END, iid=iid, text=display, values=(model,))
|
|
|
|
for stale in current:
|
|
tree.delete(stale)
|
|
|
|
def _on_device_select(self, event):
|
|
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_info(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 _append_data(self, widget: scrolledtext.ScrolledText, text: str):
|
|
widget.configure(state=tk.NORMAL)
|
|
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)
|
|
|
|
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()
|
|
|
|
|
|
def main():
|
|
app = DBNMqttTool()
|
|
app.mainloop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|