wangfq
2142674c3e
分析结论 (115200 vs 9600): - TX 阻塞: 4.3ms (曾 44.8ms) → 活动态 CPU 仅 +2.9% - 主循环延迟 < 半个测量窗, 不影响线圈检测/光耦/BLE - 从"临界可用"变为"完全充裕" 固件修复: - rs485_init: 硬编码 RS485_BAUD → 使用 g_storage_uart_baud (flash 可配) - 协议文档: Modbus 默认 115200 - DLD154Tool: 串口默认 115200
643 lines
24 KiB
Python
643 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
DLD154Tool — DLD154Pro 桌面串口调试工具
|
|
|
|
支持:
|
|
- 700BS 协议 (查询状态/频率/拨码/复位)
|
|
- Modbus RTU 协议 (读/写寄存器, 全数据块查询, 主动上报监控)
|
|
|
|
依赖: PySide6, pyserial
|
|
"""
|
|
import sys
|
|
import struct
|
|
import threading
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from PySide6.QtCore import Qt, QTimer, Signal, Slot, QThread
|
|
from PySide6.QtWidgets import (
|
|
QApplication, QMainWindow, QWidget, QTabWidget,
|
|
QVBoxLayout, QHBoxLayout, QFormLayout, QGroupBox,
|
|
QLabel, QLineEdit, QComboBox, QPushButton, QTextBrowser,
|
|
QSpinBox, QCheckBox, QSplitter, QGridLayout, QMessageBox,
|
|
)
|
|
|
|
from serial_client import SerialClient, SerialConfig
|
|
from protocol_700bs import (
|
|
encode_query_status, encode_query_freq, encode_query_dip, encode_reset_init,
|
|
parse_frame as parse_700bs,
|
|
)
|
|
from protocol_modbus import (
|
|
build_read_input_regs, build_read_holding_regs, build_read_discrete_inputs,
|
|
build_write_single_reg, build_write_multiple_regs, build_report_slave_id,
|
|
parse_response, is_active_report,
|
|
FullDataBlock,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 后台串口工作线程 (避免阻塞 GUI)
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
class SerialWorker(QThread):
|
|
"""串口数据接收线程 + 700BS 帧缓冲"""
|
|
|
|
data_received = Signal(bytes) # 原始字节
|
|
frame_700bs = Signal(object) # 700BS 解析后的帧
|
|
frame_modbus = Signal(bytes) # Modbus 完整帧 (含 CRC)
|
|
|
|
def __init__(self, serial_client: SerialClient):
|
|
super().__init__()
|
|
self._sc = serial_client
|
|
self._buf_700bs = bytearray()
|
|
self._running = True
|
|
|
|
def run(self):
|
|
self._sc.set_data_callback(self._on_data)
|
|
# 保持线程存活
|
|
while self._running:
|
|
time.sleep(0.1)
|
|
|
|
def stop(self):
|
|
self._running = False
|
|
|
|
def _on_data(self, data: bytes):
|
|
self.data_received.emit(data)
|
|
self._buf_700bs.extend(data)
|
|
|
|
# ── 700BS 帧解析 (依赖已知帧长) ──
|
|
while len(self._buf_700bs) > 0:
|
|
stx = self._buf_700bs[0]
|
|
if stx == 0xAA and len(self._buf_700bs) >= 9: # 心跳 9B
|
|
frame = bytes(self._buf_700bs[:9])
|
|
parsed = parse_700bs(frame)
|
|
if parsed:
|
|
self.frame_700bs.emit(parsed)
|
|
del self._buf_700bs[:9]
|
|
elif stx == 0xF5 and len(self._buf_700bs) >= 3: # 状态回复 3B
|
|
frame = bytes(self._buf_700bs[:3])
|
|
parsed = parse_700bs(frame)
|
|
if parsed:
|
|
self.frame_700bs.emit(parsed)
|
|
del self._buf_700bs[:3]
|
|
elif stx == 0x14 and len(self._buf_700bs) >= 6: # 频率回复 6B
|
|
frame = bytes(self._buf_700bs[:6])
|
|
parsed = parse_700bs(frame)
|
|
if parsed:
|
|
self.frame_700bs.emit(parsed)
|
|
del self._buf_700bs[:6]
|
|
elif stx == 0x32 and len(self._buf_700bs) >= 4: # 拨码回复 4B
|
|
frame = bytes(self._buf_700bs[:4])
|
|
parsed = parse_700bs(frame)
|
|
if parsed:
|
|
self.frame_700bs.emit(parsed)
|
|
del self._buf_700bs[:4]
|
|
else:
|
|
# 未知帧头, 丢弃 1 字节后重试
|
|
del self._buf_700bs[0]
|
|
|
|
# ── Modbus 帧解析 (被动: 从完整字节流中按 t3.5 超时提取) ──
|
|
# 简化: 不做超时, 基于 CRC 校验试切
|
|
buf = bytearray(data)
|
|
while len(buf) >= 4:
|
|
# 尝试不同长度
|
|
found = False
|
|
for flen in range(4, min(len(buf) + 1, 256)):
|
|
candidate = bytes(buf[:flen])
|
|
from protocol_modbus import crc16
|
|
if crc16(candidate[:-2]) == struct.unpack('<H', candidate[-2:])[0]:
|
|
self.frame_modbus.emit(candidate)
|
|
del buf[:flen]
|
|
found = True
|
|
break
|
|
if not found:
|
|
del buf[0]
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Tab 1: 串口设置
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
class SerialTab(QWidget):
|
|
connect_changed = Signal(bool)
|
|
|
|
def __init__(self, sc: SerialClient):
|
|
super().__init__()
|
|
self._sc = sc
|
|
self._connected = False
|
|
self._setup_ui()
|
|
self._refresh_ports()
|
|
|
|
def _setup_ui(self):
|
|
layout = QVBoxLayout(self)
|
|
|
|
# 端口选择
|
|
gb = QGroupBox("串口设置")
|
|
fl = QFormLayout(gb)
|
|
|
|
self._cmb_port = QComboBox()
|
|
self._cmb_port.setEditable(True)
|
|
btn_refresh = QPushButton("刷新")
|
|
btn_refresh.clicked.connect(self._refresh_ports)
|
|
|
|
h_port = QHBoxLayout()
|
|
h_port.addWidget(self._cmb_port)
|
|
h_port.addWidget(btn_refresh)
|
|
fl.addRow("端口:", h_port)
|
|
|
|
self._cmb_baud = QComboBox()
|
|
self._cmb_baud.addItems(["9600", "19200", "38400", "57600", "115200"])
|
|
self._cmb_baud.setCurrentText("115200")
|
|
fl.addRow("波特率:", self._cmb_baud)
|
|
|
|
self._spn_addr = QSpinBox()
|
|
self._spn_addr.setRange(1, 247)
|
|
self._spn_addr.setValue(1)
|
|
fl.addRow("从机地址:", self._spn_addr)
|
|
|
|
layout.addWidget(gb)
|
|
|
|
# 连接按钮
|
|
self._btn_conn = QPushButton("连接")
|
|
self._btn_conn.setStyleSheet("font-size: 14px; padding: 8px;")
|
|
self._btn_conn.clicked.connect(self._toggle_connect)
|
|
layout.addWidget(self._btn_conn)
|
|
|
|
self._lbl_status = QLabel("未连接")
|
|
self._lbl_status.setStyleSheet("color: gray;")
|
|
layout.addWidget(self._lbl_status)
|
|
|
|
layout.addStretch()
|
|
|
|
def _refresh_ports(self):
|
|
self._cmb_port.clear()
|
|
ports = SerialClient.list_ports()
|
|
if ports:
|
|
self._cmb_port.addItems(ports)
|
|
else:
|
|
self._cmb_port.addItem("(无可用端口)")
|
|
|
|
def _toggle_connect(self):
|
|
if self._connected:
|
|
self._sc.disconnect()
|
|
self._connected = False
|
|
self._btn_conn.setText("连接")
|
|
self._lbl_status.setText("未连接")
|
|
self._lbl_status.setStyleSheet("color: gray;")
|
|
else:
|
|
port = self._cmb_port.currentText()
|
|
if not port or port.startswith("("):
|
|
return
|
|
cfg = SerialConfig(
|
|
port=port,
|
|
baudrate=int(self._cmb_baud.currentText()),
|
|
)
|
|
try:
|
|
self._sc.connect(cfg)
|
|
self._connected = True
|
|
self._btn_conn.setText("断开")
|
|
self._lbl_status.setText(f"已连接 {port} @ {cfg.baudrate}")
|
|
self._lbl_status.setStyleSheet("color: green; font-weight: bold;")
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "连接失败", str(e))
|
|
return
|
|
self.connect_changed.emit(self._connected)
|
|
|
|
@property
|
|
def addr(self) -> int:
|
|
return self._spn_addr.value()
|
|
|
|
@property
|
|
def is_connected(self) -> bool:
|
|
return self._connected
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Tab 2: 700BS 协议
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
class Tab700BS(QWidget):
|
|
send_request = Signal(bytes)
|
|
log_signal = Signal(str)
|
|
|
|
def __init__(self, sc: SerialClient, addr_provider):
|
|
super().__init__()
|
|
self._sc = sc
|
|
self._addr_provider = addr_provider
|
|
self._auto_enabled = False
|
|
self._auto_timer = QTimer(self)
|
|
self._auto_timer.timeout.connect(self._auto_poll)
|
|
self._setup_ui()
|
|
|
|
def _setup_ui(self):
|
|
layout = QVBoxLayout(self)
|
|
|
|
# 命令按钮
|
|
gb_cmd = QGroupBox("命令")
|
|
gl = QGridLayout(gb_cmd)
|
|
|
|
btn_status = QPushButton("查询状态 (0xFA)")
|
|
btn_status.clicked.connect(lambda: self._send(encode_query_status))
|
|
gl.addWidget(btn_status, 0, 0)
|
|
|
|
btn_freq = QPushButton("查询频率 (0x1B)")
|
|
btn_freq.clicked.connect(lambda: self._send(encode_query_freq))
|
|
gl.addWidget(btn_freq, 0, 1)
|
|
|
|
btn_dip = QPushButton("查询拨码 (0x3D)")
|
|
btn_dip.clicked.connect(lambda: self._send(encode_query_dip))
|
|
gl.addWidget(btn_dip, 0, 2)
|
|
|
|
btn_reset = QPushButton("复位初始化 (0x2C)")
|
|
btn_reset.setStyleSheet("color: red;")
|
|
btn_reset.clicked.connect(self._send_reset)
|
|
gl.addWidget(btn_reset, 1, 0)
|
|
|
|
layout.addWidget(gb_cmd)
|
|
|
|
# 自动轮询
|
|
gb_auto = QGroupBox("自动轮询")
|
|
hl = QHBoxLayout(gb_auto)
|
|
self._chk_auto = QCheckBox("启用")
|
|
self._chk_auto.toggled.connect(self._toggle_auto)
|
|
hl.addWidget(self._chk_auto)
|
|
|
|
self._cmb_auto = QComboBox()
|
|
self._cmb_auto.addItems(["500ms", "1000ms", "2000ms", "5000ms"])
|
|
self._cmb_auto.setCurrentText("1000ms")
|
|
hl.addWidget(QLabel("间隔:"))
|
|
hl.addWidget(self._cmb_auto)
|
|
hl.addStretch()
|
|
layout.addWidget(gb_auto)
|
|
|
|
# 结果显示
|
|
gb_disp = QGroupBox("响应")
|
|
vl = QVBoxLayout(gb_disp)
|
|
self._txt_disp = QTextBrowser()
|
|
self._txt_disp.setReadOnly(True)
|
|
self._txt_disp.document().setMaximumBlockCount(200)
|
|
vl.addWidget(self._txt_disp)
|
|
layout.addWidget(gb_disp, 1)
|
|
|
|
def _send(self, encoder):
|
|
addr = self._addr_provider()
|
|
self.send_request.emit(encoder(addr))
|
|
|
|
def _send_reset(self):
|
|
addr = self._addr_provider()
|
|
if QMessageBox.question(self, "确认", f"发送 0x2C 复位命令到地址 {addr}?",
|
|
QMessageBox.Yes | QMessageBox.No) == QMessageBox.Yes:
|
|
self.send_request.emit(encode_reset_init(addr))
|
|
self.log_signal.emit("[700BS] 发送 0x2C 复位命令")
|
|
|
|
def _toggle_auto(self, checked: bool):
|
|
self._auto_enabled = checked
|
|
if checked:
|
|
interval = int(self._cmb_auto.currentText().replace("ms", ""))
|
|
self._auto_timer.start(interval)
|
|
self.log_signal.emit(f"[700BS] 自动轮询已启用, 间隔={interval}ms")
|
|
else:
|
|
self._auto_timer.stop()
|
|
self.log_signal.emit("[700BS] 自动轮询已停止")
|
|
|
|
def _auto_poll(self):
|
|
addr = self._addr_provider()
|
|
self.send_request.emit(encode_query_status(addr))
|
|
|
|
@Slot(object)
|
|
def on_frame(self, parsed):
|
|
if parsed:
|
|
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
|
self._txt_disp.append(f"[{ts}] {parsed.summary}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Tab 3: Modbus RTU
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
class TabModbus(QWidget):
|
|
send_request = Signal(bytes)
|
|
log_signal = Signal(str)
|
|
|
|
def __init__(self, sc: SerialClient, addr_provider):
|
|
super().__init__()
|
|
self._sc = sc
|
|
self._addr_provider = addr_provider
|
|
self._auto_report_enabled = False
|
|
self._setup_ui()
|
|
|
|
def _setup_ui(self):
|
|
layout = QVBoxLayout(self)
|
|
|
|
# ── 读操作 ──
|
|
gb_read = QGroupBox("读寄存器")
|
|
gl = QGridLayout(gb_read)
|
|
|
|
gl.addWidget(QLabel("起始地址 (Hex):"), 0, 0)
|
|
self._txt_start = QLineEdit("0100")
|
|
gl.addWidget(self._txt_start, 0, 1)
|
|
|
|
gl.addWidget(QLabel("数量:"), 0, 2)
|
|
self._spn_qty = QSpinBox()
|
|
self._spn_qty.setRange(1, 125)
|
|
self._spn_qty.setValue(19)
|
|
gl.addWidget(self._spn_qty, 0, 3)
|
|
|
|
btn_read_ir = QPushButton("读 IR (0x04)")
|
|
btn_read_ir.clicked.connect(self._read_ir)
|
|
gl.addWidget(btn_read_ir, 1, 0)
|
|
|
|
btn_read_hr = QPushButton("读 HR (0x03)")
|
|
btn_read_hr.clicked.connect(self._read_hr)
|
|
gl.addWidget(btn_read_hr, 1, 1)
|
|
|
|
btn_read_di = QPushButton("读 DI (0x02)")
|
|
btn_read_di.clicked.connect(self._read_di)
|
|
gl.addWidget(btn_read_di, 1, 2)
|
|
|
|
btn_slave_id = QPushButton("Report Slave ID (0x11)")
|
|
btn_slave_id.clicked.connect(self._report_slave_id)
|
|
gl.addWidget(btn_slave_id, 1, 3)
|
|
|
|
layout.addWidget(gb_read)
|
|
|
|
# ── 写操作 ──
|
|
gb_write = QGroupBox("写寄存器")
|
|
hl = QHBoxLayout(gb_write)
|
|
|
|
hl.addWidget(QLabel("地址 (Hex):"))
|
|
self._txt_wr_addr = QLineEdit("0010")
|
|
self._txt_wr_addr.setMaximumWidth(60)
|
|
hl.addWidget(self._txt_wr_addr)
|
|
|
|
hl.addWidget(QLabel("值 (Hex):"))
|
|
self._txt_wr_val = QLineEdit("0001")
|
|
self._txt_wr_val.setMaximumWidth(60)
|
|
hl.addWidget(self._txt_wr_val)
|
|
|
|
btn_write = QPushButton("写单寄存器 (0x06)")
|
|
btn_write.clicked.connect(self._write_single)
|
|
hl.addWidget(btn_write)
|
|
|
|
layout.addWidget(gb_write)
|
|
|
|
# ── 主动上报控制 ──
|
|
gb_ar = QGroupBox("主动上报")
|
|
hl_ar = QHBoxLayout(gb_ar)
|
|
self._chk_ar_enable = QCheckBox("使能 (HR 0x0010)")
|
|
self._chk_ar_enable.toggled.connect(self._toggle_auto_report)
|
|
hl_ar.addWidget(self._chk_ar_enable)
|
|
|
|
hl_ar.addWidget(QLabel("阈值:"))
|
|
self._spn_thr = QSpinBox()
|
|
self._spn_thr.setRange(0, 65535)
|
|
self._spn_thr.setValue(10)
|
|
hl_ar.addWidget(self._spn_thr)
|
|
|
|
btn_set_thr = QPushButton("设置 (HR 0x0011)")
|
|
btn_set_thr.clicked.connect(self._set_threshold)
|
|
hl_ar.addWidget(btn_set_thr)
|
|
|
|
hl_ar.addStretch()
|
|
layout.addWidget(gb_ar)
|
|
|
|
# ── 全数据块快速查询 ──
|
|
gb_fd = QGroupBox("全数据块 (0x0100)")
|
|
hl_fd = QHBoxLayout(gb_fd)
|
|
btn_fd = QPushButton("查询全数据块 (IR 0x0100, 19 regs)")
|
|
btn_fd.setStyleSheet("font-weight: bold; padding: 6px;")
|
|
btn_fd.clicked.connect(self._read_full_data)
|
|
hl_fd.addWidget(btn_fd)
|
|
layout.addWidget(gb_fd)
|
|
|
|
# ── 响应 / 全数据显示 ──
|
|
gb_disp = QGroupBox("响应 & 全数据")
|
|
vl = QVBoxLayout(gb_disp)
|
|
self._txt_modbus = QTextBrowser()
|
|
self._txt_modbus.setReadOnly(True)
|
|
self._txt_modbus.document().setMaximumBlockCount(500)
|
|
self._txt_modbus.setFontFamily("monospace")
|
|
vl.addWidget(self._txt_modbus)
|
|
layout.addWidget(gb_disp, 1)
|
|
|
|
# ── 发送 ──
|
|
|
|
def _addr(self) -> int:
|
|
return self._addr_provider()
|
|
|
|
def _read_ir(self):
|
|
start = int(self._txt_start.text(), 16)
|
|
qty = self._spn_qty.value()
|
|
self.send_request.emit(build_read_input_regs(self._addr(), start, qty))
|
|
self.log_signal.emit(f"[Modbus] FC 0x04 read IR 0x{start:04X} x{qty}")
|
|
|
|
def _read_hr(self):
|
|
start = int(self._txt_start.text(), 16)
|
|
qty = self._spn_qty.value()
|
|
self.send_request.emit(build_read_holding_regs(self._addr(), start, qty))
|
|
self.log_signal.emit(f"[Modbus] FC 0x03 read HR 0x{start:04X} x{qty}")
|
|
|
|
def _read_di(self):
|
|
start = int(self._txt_start.text(), 16)
|
|
qty = self._spn_qty.value()
|
|
self.send_request.emit(build_read_discrete_inputs(self._addr(), start, qty))
|
|
self.log_signal.emit(f"[Modbus] FC 0x02 read DI 0x{start:04X} x{qty}")
|
|
|
|
def _read_full_data(self):
|
|
self.send_request.emit(build_read_input_regs(self._addr(), 0x0100, 19))
|
|
self.log_signal.emit("[Modbus] FC 0x04 read IR 0x0100 x19 (全数据块)")
|
|
|
|
def _report_slave_id(self):
|
|
self.send_request.emit(build_report_slave_id(self._addr()))
|
|
self.log_signal.emit("[Modbus] FC 0x11 Report Slave ID")
|
|
|
|
def _write_single(self):
|
|
reg = int(self._txt_wr_addr.text(), 16)
|
|
val = int(self._txt_wr_val.text(), 16)
|
|
self.send_request.emit(build_write_single_reg(self._addr(), reg, val))
|
|
self.log_signal.emit(f"[Modbus] FC 0x06 write reg=0x{reg:04X} val=0x{val:04X}")
|
|
|
|
def _toggle_auto_report(self, checked: bool):
|
|
self._auto_report_enabled = checked
|
|
val = 1 if checked else 0
|
|
self.send_request.emit(build_write_single_reg(self._addr(), 0x0010, val))
|
|
self.log_signal.emit(f"[Modbus] 主动上报 {'使能' if checked else '关闭'} (HR 0x0010={val})")
|
|
|
|
def _set_threshold(self):
|
|
val = self._spn_thr.value()
|
|
self.send_request.emit(build_write_single_reg(self._addr(), 0x0011, val))
|
|
self.log_signal.emit(f"[Modbus] 设置活动阈值={val} (HR 0x0011)")
|
|
|
|
# ── 接收 ──
|
|
|
|
@Slot(object)
|
|
def on_frame(self, parsed):
|
|
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
|
if parsed is None:
|
|
self._txt_modbus.append(f"[{ts}] CRC 错误或无法解析")
|
|
return
|
|
|
|
tag = parsed.__class__.__name__
|
|
if tag == "ReadRegsResponse":
|
|
self._txt_modbus.append(f"[{ts}] {parsed.summary}")
|
|
# 如果是全数据块 (19 regs from 0x0100), 美化显示
|
|
if len(parsed.regs) == 19:
|
|
fd = FullDataBlock.from_regs(parsed.regs)
|
|
self._txt_modbus.append(fd.summary)
|
|
self._txt_modbus.append("-" * 50)
|
|
else:
|
|
hex_vals = " ".join(f"{r:04X}" for r in parsed.regs[:16])
|
|
if len(parsed.regs) > 16:
|
|
hex_vals += f" ... (+{len(parsed.regs)-16})"
|
|
self._txt_modbus.append(f" Hex: {hex_vals}")
|
|
elif tag == "ReadBitsResponse":
|
|
self._txt_modbus.append(f"[{ts}] {parsed.summary}")
|
|
bits = parsed.bits
|
|
names = ["有车", "线圈断", "稳定", "光耦", "上电初始", "继电器", "FreqLo", "FreqHi", "Init"]
|
|
st = ", ".join(f"{names[i]}={'1' if (bits>>i)&1 else '0'}" for i in range(min(9, len(names))))
|
|
self._txt_modbus.append(f" {st}")
|
|
elif tag == "WriteRegResponse":
|
|
self._txt_modbus.append(f"[{ts}] {parsed.summary} ✓")
|
|
elif tag == "SlaveIdResponse":
|
|
self._txt_modbus.append(f"[{ts}] {parsed.summary}")
|
|
elif tag == "ModbusException":
|
|
self._txt_modbus.append(f"[{ts}] ⚠ {parsed.summary}")
|
|
else:
|
|
self._txt_modbus.append(f"[{ts}] {parsed}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Tab 4: 日志
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
class LogTab(QWidget):
|
|
def __init__(self):
|
|
super().__init__()
|
|
layout = QVBoxLayout(self)
|
|
self._txt_log = QTextBrowser()
|
|
self._txt_log.setReadOnly(True)
|
|
self._txt_log.document().setMaximumBlockCount(1000)
|
|
self._txt_log.setFontFamily("monospace")
|
|
layout.addWidget(self._txt_log)
|
|
|
|
hl = QHBoxLayout()
|
|
btn_clear = QPushButton("清空")
|
|
btn_clear.clicked.connect(self._txt_log.clear)
|
|
hl.addStretch()
|
|
hl.addWidget(btn_clear)
|
|
layout.addLayout(hl)
|
|
|
|
@Slot(str)
|
|
def append(self, msg: str):
|
|
ts = datetime.now().strftime("%H:%M:%S.%f")[:-3]
|
|
self._txt_log.append(f"[{ts}] {msg}")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 主窗口
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
class MainWindow(QMainWindow):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.setWindowTitle("DLD154Tool — DLD154Pro 串口调试工具")
|
|
self.setMinimumSize(800, 650)
|
|
|
|
self._sc = SerialClient()
|
|
self._worker: Optional[SerialWorker] = None
|
|
|
|
# Tabs
|
|
self._serial_tab = SerialTab(self._sc)
|
|
self._tab_700bs = Tab700BS(self._sc, lambda: self._serial_tab.addr)
|
|
self._tab_modbus = TabModbus(self._sc, lambda: self._serial_tab.addr)
|
|
self._log_tab = LogTab()
|
|
|
|
tabs = QTabWidget()
|
|
tabs.addTab(self._serial_tab, "串口设置")
|
|
tabs.addTab(self._tab_700bs, "700BS")
|
|
tabs.addTab(self._tab_modbus, "Modbus RTU")
|
|
tabs.addTab(self._log_tab, "日志")
|
|
|
|
self.setCentralWidget(tabs)
|
|
|
|
# 信号连线
|
|
self._serial_tab.connect_changed.connect(self._on_connect_changed)
|
|
|
|
# 发送: GUI tab → 串口
|
|
self._tab_700bs.send_request.connect(self._send_raw)
|
|
self._tab_modbus.send_request.connect(self._send_raw)
|
|
|
|
# 日志
|
|
self._tab_700bs.log_signal.connect(self._log_tab.append)
|
|
self._tab_modbus.log_signal.connect(self._log_tab.append)
|
|
|
|
self._log_tab.append("DLD154Tool 启动")
|
|
|
|
def _on_connect_changed(self, connected: bool):
|
|
if connected:
|
|
self._start_worker()
|
|
self._log_tab.append("串口已连接")
|
|
else:
|
|
self._stop_worker()
|
|
self._log_tab.append("串口已断开")
|
|
|
|
def _start_worker(self):
|
|
self._stop_worker()
|
|
self._worker = SerialWorker(self._sc)
|
|
|
|
# 700BS 帧 → GUI
|
|
self._worker.frame_700bs.connect(self._tab_700bs.on_frame)
|
|
|
|
# Modbus 帧 → GUI
|
|
self._worker.frame_modbus.connect(self._on_modbus_frame)
|
|
|
|
# 原始数据 → 日志 (可选 hex dump)
|
|
self._worker.data_received.connect(self._on_raw_data)
|
|
|
|
self._worker.start()
|
|
|
|
def _stop_worker(self):
|
|
if self._worker:
|
|
self._worker.stop()
|
|
self._worker.wait(1000)
|
|
self._worker = None
|
|
|
|
@Slot(bytes)
|
|
def _send_raw(self, data: bytes):
|
|
if self._sc.is_connected:
|
|
self._sc.send(data)
|
|
self._log_tab.append(f"TX: {data.hex(' ').upper()}")
|
|
|
|
@Slot(bytes)
|
|
def _on_raw_data(self, data: bytes):
|
|
if len(data) <= 32:
|
|
self._log_tab.append(f"RX: {data.hex(' ').upper()}")
|
|
else:
|
|
self._log_tab.append(f"RX: {len(data)} bytes")
|
|
|
|
@Slot(bytes)
|
|
def _on_modbus_frame(self, frame: bytes):
|
|
parsed = parse_response(frame)
|
|
tag = "ActiveReport" if is_active_report(frame) else "Modbus"
|
|
if parsed:
|
|
self._tab_modbus.on_frame(parsed)
|
|
else:
|
|
self._log_tab.append(f"[{tag}] 无法解析: {frame.hex(' ').upper()[:60]}...")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 入口
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
def main():
|
|
app = QApplication(sys.argv)
|
|
app.setStyle("Fusion")
|
|
win = MainWindow()
|
|
win.show()
|
|
sys.exit(app.exec())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|