#!/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(' 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()