From e97388500bb8dd055e37e81024cb8b6ec3335faf Mon Sep 17 00:00:00 2001 From: wangfq Date: Thu, 30 Jul 2026 16:06:58 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=9B=86=E6=88=90=20DLD154Tool=20?= =?UTF-8?q?=E2=80=94=20PySide6=20=E6=A1=8C=E9=9D=A2=E4=B8=B2=E5=8F=A3?= =?UTF-8?q?=E8=B0=83=E8=AF=95=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 7 文件 1331 行, 支持 DLD154Pro 双协议: - 700BS: 状态/频率/拨码/复位查询 + 自动轮询 - Modbus RTU: FC 0x02/0x03/0x04/0x06/0x10/0x11 + 全数据块(0x0100)美化解析 + 主动上报控制 - PySide6 GUI, 4 Tab, 后台串口线程 从独立仓库 wangfq/DLD154Tool 合并入 DLD154Pro monorepo --- DLD154Tool/.gitignore | 4 + DLD154Tool/README.md | 90 +++++ DLD154Tool/main.py | 642 ++++++++++++++++++++++++++++++++++ DLD154Tool/protocol_700bs.py | 207 +++++++++++ DLD154Tool/protocol_modbus.py | 286 +++++++++++++++ DLD154Tool/requirements.txt | 2 + DLD154Tool/serial_client.py | 100 ++++++ 7 files changed, 1331 insertions(+) create mode 100644 DLD154Tool/.gitignore create mode 100644 DLD154Tool/README.md create mode 100644 DLD154Tool/main.py create mode 100644 DLD154Tool/protocol_700bs.py create mode 100644 DLD154Tool/protocol_modbus.py create mode 100644 DLD154Tool/requirements.txt create mode 100644 DLD154Tool/serial_client.py diff --git a/DLD154Tool/.gitignore b/DLD154Tool/.gitignore new file mode 100644 index 0000000..bdd06af --- /dev/null +++ b/DLD154Tool/.gitignore @@ -0,0 +1,4 @@ +venv/ +__pycache__/ +*.pyc +.DS_Store diff --git a/DLD154Tool/README.md b/DLD154Tool/README.md new file mode 100644 index 0000000..259c102 --- /dev/null +++ b/DLD154Tool/README.md @@ -0,0 +1,90 @@ +# DLD154Tool — DLD154Pro 桌面串口调试工具 + +基于 PySide6 + pyserial 的 RS485 串口调试工具，支持 DLD154Pro 的两种 RS485 协议。 + +## 协议支持 + +| 协议 | Tab | 功能 | +|------|-----|------| +| **700BS** | `700BS` | 查询状态 (0xFA)、查询频率 (0x1B)、查询拨码 (0x3D)、复位初始化 (0x2C)、自动轮询 | +| **Modbus RTU** | `Modbus RTU` | FC 0x02/0x03/0x04/0x06/0x10/0x11、全数据块查询 (0x0100)、主动上报使能/阈值设置 | + +## 快速开始 + +```bash +cd DLD154Tool +python3 -m venv venv +source venv/bin/activate # Windows: venv\\Scripts\\activate +pip install -r requirements.txt +python main.py +``` + +## 界面布局 + +``` +┌────────────────────────────────────────────┐ +│ Tab: 串口设置 │ 700BS │ Modbus RTU │ 日志 │ +├────────────────────────────────────────────┤ +│ │ +│ 串口设置: │ +│ [端口▼] [刷新] 波特率: [9600▼] │ +│ 从机地址: [1] │ +│ [连接] │ +│ │ +├────────────────────────────────────────────┤ +│ 700BS: │ +│ [查询状态] [查询频率] [查询拨码] │ +│ [复位初始化] │ +│ 自动轮询: ☑启用 间隔: [1000ms▼] │ +│ 响应显示区 │ +├────────────────────────────────────────────┤ +│ Modbus RTU: │ +│ 读: 起始 [0100] 数量 [19] │ +│ [读IR] [读HR] [读DI] [SlaveID] │ +│ 写: 地址 [0010] 值 [0001] [写单寄存器] │ +│ 主动上报: ☑使能 阈值 [10] [设置] │ +│ [查询全数据块] 响应 & 全数据显示 │ +├────────────────────────────────────────────┤ +│ 日志: TX/RX hex dump + 事件 │ +└────────────────────────────────────────────┘ +``` + +## 项目结构 + +``` +DLD154Tool/ +├── main.py # PySide6 GUI (4 tabs + 后台串口线程) +├── serial_client.py # 串口抽象层 (扫描/连接/收发) +├── protocol_700bs.py # 700BS 协议编解码 +├── protocol_modbus.py # Modbus RTU 协议编解码 + CRC16 +├── requirements.txt # pyside6, pyserial +├── .gitignore +└── README.md +``` + +## RS485 接线 + +``` +USB-to-RS485 模块 DLD154Pro +───────────────── ───────── +A (D+) ───────────────── A (PA15 / PA14) +B (D−) ───────────────── B (PA15 / PA14) +GND ───────────────── GND +``` + +> DLD154Pro RS485 为半双工，UART0 通过 PA13 自动切换收发方向。 + +## 串口参数 + +| 参数 | 默认值 | 范围 | +|------|--------|------| +| 波特率 | 9600 | 9600/19200/38400/57600/115200 | +| 数据位 | 8 | — | +| 校验 | 无 | — | +| 停止位 | 1 | — | +| 从机地址 | 1 | 1–247 | + +## 协议文档 + +- 700BS: `docs/DLD-700BS-V1.15-使用说明书.md` +- Modbus: `docs/rs485-modbus-protocol.md` (V1.6) diff --git a/DLD154Tool/main.py b/DLD154Tool/main.py new file mode 100644 index 0000000..1a3f04a --- /dev/null +++ b/DLD154Tool/main.py @@ -0,0 +1,642 @@ +#!/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, QTextEdit, + 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 = QTextEdit() + self._txt_disp.setReadOnly(True) + self._txt_disp.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 = QTextEdit() + self._txt_modbus.setReadOnly(True) + self._txt_modbus.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 = QTextEdit() + self._txt_log.setReadOnly(True) + self._txt_log.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() diff --git a/DLD154Tool/protocol_700bs.py b/DLD154Tool/protocol_700bs.py new file mode 100644 index 0000000..ec28308 --- /dev/null +++ b/DLD154Tool/protocol_700bs.py @@ -0,0 +1,207 @@ +""" +protocol_700bs.py — DLD154Pro 700BS 串口协议编解码 + +帧格式: + 主机→从机: STX(1) + ADDR(1) + XOR(1) = 3 字节 + 从机→主机: 可变长, XOR 校验 + +命令码: + 0xFA 查询有车状态 + 线圈好坏 → RSP 0xF5 (3B) + 0x1B 查询线圈频率 → RSP 0x14 (6B, 含 3B Freq + ADDR + XOR) + 0x2C 复位初始化线圈 → 无响应 + 0x3D 查询拨码 DIP1-5 → RSP 0x32 (4B) + 0xAA 心跳 (从机主动上报) → 9B 帧 + +RS485 地址: 0–31 (5-bit, 帧中 ADDR 低 5 位) +""" +from dataclasses import dataclass, field +from typing import Optional +import struct + + +# ══════════════════════════════════════════════════ +# 协议常量 +# ══════════════════════════════════════════════════ + +CMD_QUERY_STATUS = 0xFA # 查询有车状态 +CMD_QUERY_FREQ = 0x1B # 查询频率 +CMD_RESET_INIT = 0x2C # 复位初始化 +CMD_QUERY_DIP = 0x3D # 查询拨码 + +RSP_STATUS = 0xF5 # 状态回复 +RSP_FREQ = 0x14 # 频率回复 +RSP_DIP = 0x32 # 拨码回复 +HB_PREAMBLE = 0xAA # 心跳帧头 + + +# ══════════════════════════════════════════════════ +# 数据类 +# ══════════════════════════════════════════════════ + +@dataclass +class StatusResponse: + addr: int = 0 # bit4-0 + car_present: bool = False # bit7 + loop_fault: bool = False # bit5 + + @property + def summary(self) -> str: + car = "有车" if self.car_present else "无车" + coil = "故障" if self.loop_fault else "正常" + return f"地址={self.addr} {car} 线圈={coil}" + + +@dataclass +class FreqResponse: + freq_hz: int = 0 # 3 字节频率 (Hz) + addr: int = 0 + + @property + def summary(self) -> str: + return f"频率={self.freq_hz} Hz (地址={self.addr})" + + +@dataclass +class DipResponse: + dip: int = 0 # DIP1-5 位图 + addr: int = 0 + + @property + def summary(self) -> str: + sens = self.dip & 0x03 + pulse = "脉冲" if (self.dip & 0x04) else "存在" + perm = "永久" if (self.dip & 0x08) else "有限" + delay = "有延时" if (self.dip & 0x10) else "无延时" + return f"灵敏度={sens} {pulse} {perm} {delay} (地址={self.addr})" + + +@dataclass +class Heartbeat: + init_flag: bool = False # 首次心跳 (A1) + car_present: bool = False # bit7 + loop_fault: bool = False # bit5 + freq_hz: int = 0 # 3 字节 + addr: int = 0 + + @property + def summary(self) -> str: + init = "首次" if self.init_flag else "周期" + car = "有车" if self.car_present else "无车" + coil = "故障" if self.loop_fault else "正常" + return f"[{init}] 地址={self.addr} {car} 线圈={coil} 频率={self.freq_hz} Hz" + + +# ══════════════════════════════════════════════════ +# 编码 (主机→从机) +# ══════════════════════════════════════════════════ + +def encode_query_status(addr: int) -> bytes: + """0xFA — 查询有车状态""" + stx = CMD_QUERY_STATUS + adr = addr & 0x1F + xorr = stx ^ adr + return bytes([stx, adr, xorr]) + + +def encode_query_freq(addr: int) -> bytes: + """0x1B — 查询频率""" + stx = CMD_QUERY_FREQ + adr = addr & 0x1F + xorr = stx ^ adr + return bytes([stx, adr, xorr]) + + +def encode_reset_init(addr: int) -> bytes: + """0x2C — 复位初始化 (全地址响应)""" + stx = CMD_RESET_INIT + adr = addr & 0x1F + xorr = stx ^ adr + return bytes([stx, adr, xorr]) + + +def encode_query_dip(addr: int) -> bytes: + """0x3D — 查询拨码""" + stx = CMD_QUERY_DIP + adr = addr & 0x1F + xorr = stx ^ adr + return bytes([stx, adr, xorr]) + + +# ══════════════════════════════════════════════════ +# 解码 (从机→主机) +# ══════════════════════════════════════════════════ + +def xor_checksum(data: bytes) -> int: + """XOR 累加校验""" + result = 0 + for b in data: + result ^= b + return result & 0xFF + + +def parse_status(data: bytes) -> Optional[StatusResponse]: + """解析 0xF5 状态回复 (3 字节: STX DATA XOR)""" + if len(data) < 3 or data[0] != RSP_STATUS: + return None + if xor_checksum(data[:2]) != data[2]: + return None + d = data[1] + return StatusResponse( + addr=d & 0x1F, + car_present=bool(d & 0x80), + loop_fault=bool(d & 0x20), + ) + + +def parse_freq(data: bytes) -> Optional[FreqResponse]: + """解析 0x14 频率回复 (6 字节: STX FREQ_H FREQ_M FREQ_L ADDR XOR)""" + if len(data) < 6 or data[0] != RSP_FREQ: + return None + if xor_checksum(data[:5]) != data[5]: + return None + freq = (data[1] << 16) | (data[2] << 8) | data[3] + return FreqResponse(freq_hz=freq, addr=data[4] & 0x1F) + + +def parse_dip(data: bytes) -> Optional[DipResponse]: + """解析 0x32 拨码回复 (4 字节: STX DIP ADDR XOR)""" + if len(data) < 4 or data[0] != RSP_DIP: + return None + if xor_checksum(data[:3]) != data[3]: + return None + return DipResponse(dip=data[1], addr=data[2] & 0x1F) + + +def parse_heartbeat(data: bytes) -> Optional[Heartbeat]: + """解析 0xAA 心跳 (9 字节)""" + if len(data) < 9 or data[0] != HB_PREAMBLE: + return None + if xor_checksum(data[:8]) != data[8]: + return None + init = data[1] # 0xA1=首次, 0xA0=正常 + st = data[2] + freq = (data[4] << 16) | (data[5] << 8) | data[6] + return Heartbeat( + init_flag=(init == 0xA1), + car_present=bool(st & 0x80), + loop_fault=bool(st & 0x20), + freq_hz=freq, + addr=data[7] & 0x1F, + ) + + +def parse_frame(data: bytes) -> Optional[object]: + """自动检测帧类型并解析""" + if not data: + return None + stx = data[0] + parsers = { + RSP_STATUS: parse_status, + RSP_FREQ: parse_freq, + RSP_DIP: parse_dip, + HB_PREAMBLE: parse_heartbeat, + } + parser = parsers.get(stx) + if parser: + return parser(data) + return None diff --git a/DLD154Tool/protocol_modbus.py b/DLD154Tool/protocol_modbus.py new file mode 100644 index 0000000..c50a580 --- /dev/null +++ b/DLD154Tool/protocol_modbus.py @@ -0,0 +1,286 @@ +""" +protocol_modbus.py — DLD154Pro Modbus RTU 协议编解码 + +帧格式: ADDR(1) + FC(1) + DATA(N) + CRC16(2) + +支持的功能码: + 0x02 Read Discrete Inputs + 0x03 Read Holding Registers + 0x04 Read Input Registers + 0x06 Write Single Register + 0x10 Write Multiple Registers + 0x11 Report Slave ID + +寄存器映射参见 docs/rs485-modbus-protocol.md V1.6 +""" +from dataclasses import dataclass, field +from typing import Optional, Union +import struct + + +# ══════════════════════════════════════════════════ +# CRC16 查表 +# ══════════════════════════════════════════════════ + +_CRC16_TABLE = [ + 0x0000,0xC0C1,0xC181,0x0140,0xC301,0x03C0,0x0280,0xC241, + 0xC601,0x06C0,0x0780,0xC741,0x0500,0xC5C1,0xC481,0x0440, + 0xCC01,0x0CC0,0x0D80,0xCD41,0x0F00,0xCFC1,0xCE81,0x0E40, + 0x0A00,0xCAC1,0xCB81,0x0B40,0xC901,0x09C0,0x0880,0xC841, + 0xD801,0x18C0,0x1980,0xD941,0x1B00,0xDBC1,0xDA81,0x1A40, + 0x1E00,0xDEC1,0xDF81,0x1F40,0xDD01,0x1DC0,0x1C80,0xDC41, + 0x1400,0xD4C1,0xD581,0x1540,0xD701,0x17C0,0x1680,0xD641, + 0xD201,0x12C0,0x1380,0xD341,0x1100,0xD1C1,0xD081,0x1040, + 0xF001,0x30C0,0x3180,0xF141,0x3300,0xF3C1,0xF281,0x3240, + 0x3600,0xF6C1,0xF781,0x3740,0xF501,0x35C0,0x3480,0xF441, + 0x3C00,0xFCC1,0xFD81,0x3D40,0xFF01,0x3FC0,0x3E80,0xFE41, + 0xFA01,0x3AC0,0x3B80,0xFB41,0x3900,0xF9C1,0xF881,0x3840, + 0x2800,0xE8C1,0xE981,0x2940,0xEB01,0x2BC0,0x2A80,0xEA41, + 0xEE01,0x2EC0,0x2F80,0xEF41,0x2D00,0xEDC1,0xEC81,0x2C40, + 0xE401,0x24C0,0x2580,0xE541,0x2700,0xE7C1,0xE681,0x2640, + 0x2200,0xE2C1,0xE381,0x2340,0xE101,0x21C0,0x2080,0xE041, + 0xA001,0x60C0,0x6180,0xA141,0x6300,0xA3C1,0xA281,0x6240, + 0x6600,0xA6C1,0xA781,0x6740,0xA501,0x65C0,0x6480,0xA441, + 0x6C00,0xACC1,0xAD81,0x6D40,0xAF01,0x6FC0,0x6E80,0xAE41, + 0xAA01,0x6AC0,0x6B80,0xAB41,0x6900,0xA9C1,0xA881,0x6840, + 0x7800,0xB8C1,0xB981,0x7940,0xBB01,0x7BC0,0x7A80,0xBA41, + 0xBE01,0x7EC0,0x7F80,0xBF41,0x7D00,0xBDC1,0xBC81,0x7C40, + 0xB401,0x74C0,0x7580,0xB541,0x7700,0xB7C1,0xB681,0x7640, + 0x7200,0xB2C1,0xB381,0x7340,0xB101,0x71C0,0x7080,0xB041, + 0x5000,0x90C1,0x9181,0x5140,0x9301,0x53C0,0x5280,0x9241, + 0x9601,0x56C0,0x5780,0x9741,0x5500,0x95C1,0x9481,0x5440, + 0x9C01,0x5CC0,0x5D80,0x9D41,0x5F00,0x9FC1,0x9E81,0x5E40, + 0x5A00,0x9AC1,0x9B81,0x5B40,0x9901,0x59C0,0x5880,0x9841, + 0x8801,0x48C0,0x4980,0x8941,0x4B00,0x8BC1,0x8A81,0x4A40, + 0x4E00,0x8EC1,0x8F81,0x4F40,0x8D01,0x4DC0,0x4C80,0x8C41, + 0x4400,0x84C1,0x8581,0x4540,0x8701,0x47C0,0x4680,0x8641, + 0x8201,0x42C0,0x4380,0x8341,0x4100,0x81C1,0x8081,0x4040, +] + + +def crc16(data: bytes) -> int: + crc = 0xFFFF + for b in data: + crc = (crc >> 8) ^ _CRC16_TABLE[(crc ^ b) & 0xFF] + return crc + + +# ══════════════════════════════════════════════════ +# 帧编解码 +# ══════════════════════════════════════════════════ + +def build_request(addr: int, fc: int, payload: bytes) -> bytes: + """构建 Modbus RTU 请求帧""" + frame = bytes([addr & 0xFF, fc]) + payload + crc = crc16(frame) + return frame + struct.pack(' bytes: + """FC 0x04 — 读输入寄存器""" + return build_request(addr, 0x04, struct.pack('>HH', start, qty)) + + +def build_read_holding_regs(addr: int, start: int, qty: int) -> bytes: + """FC 0x03 — 读保持寄存器""" + return build_request(addr, 0x03, struct.pack('>HH', start, qty)) + + +def build_read_discrete_inputs(addr: int, start: int, qty: int) -> bytes: + """FC 0x02 — 读离散输入""" + return build_request(addr, 0x02, struct.pack('>HH', start, qty)) + + +def build_write_single_reg(addr: int, reg: int, value: int) -> bytes: + """FC 0x06 — 写单寄存器""" + return build_request(addr, 0x06, struct.pack('>HH', reg, value)) + + +def build_write_multiple_regs(addr: int, start: int, values: list[int]) -> bytes: + """FC 0x10 — 批量写寄存器""" + qty = len(values) + bc = qty * 2 + payload = struct.pack('>HHB', start, qty, bc) + for v in values: + payload += struct.pack('>H', v) + return build_request(addr, 0x10, payload) + + +def build_report_slave_id(addr: int) -> bytes: + """FC 0x11 — Report Slave ID""" + return build_request(addr, 0x11, b'') + + +# ══════════════════════════════════════════════════ +# 响应解析 +# ══════════════════════════════════════════════════ + +@dataclass +class ModbusException: + fc: int + excode: int + + @property + def summary(self) -> str: + names = {0x01: "Illegal Function", 0x02: "Illegal Data Address", + 0x03: "Illegal Data Value", 0x04: "Slave Failure"} + return f"异常 FC=0x{self.fc:02X} {names.get(self.excode, f'code={self.excode}')}" + + +@dataclass +class ReadBitsResponse: + fc: int + bits: int # 最多 16 位 + + @property + def summary(self) -> str: + return f"FC=0x{self.fc:02X} bits=0x{self.bits:04X}" + + +@dataclass +class ReadRegsResponse: + fc: int + regs: list[int] # 16-bit 寄存器值列表 + + @property + def summary(self) -> str: + return f"FC=0x{self.fc:02X} {len(self.regs)} registers" + + +@dataclass +class WriteRegResponse: + fc: int + reg: int + value: int + + @property + def summary(self) -> str: + return f"FC=0x{self.fc:02X} reg=0x{self.reg:04X} val={self.value}" + + +@dataclass +class SlaveIdResponse: + model: str = "" + status: int = 0 + + @property + def summary(self) -> str: + st = "RUN" if self.status == 0 else "STOP" + return f"型号={self.model} 状态={st}" + + +@dataclass +class FullDataBlock: + """全数据块 0x0100–0x0112 (19 regs) 解析结果""" + sensitivity: int = 0 # 0x0100 + freq_level: int = 0 # 0x0101 0=高频 1=低频 + fusion_mode: int = 0 # 0x0102 0-6 + car_state: int = 0 # 0x0103 0=无车 1=有车 + loop_disconnected: bool = False # 0x0104 bit0 + loop_stable: bool = False # 0x0104 bit1 + opto_active: bool = False # 0x0105 + freq_hz: int = 0 # 0x0106-07 + capvd: int = 0 # 0x0108-09 + variation: int = 0 # 0x010A-0B (signed) + cut_count: int = 0 # 0x010C-0D + relay_count: int = 0 # 0x010E-0F + time_type: int = 0xFFFF # 0x0110 0=间隙 1=通过 0xFFFF=无效 + misc_time_10ms: int = 0 # 0x0111-12 + + FUSION_NAMES = {0: "或", 1: "混行防砸", 2: "ETC", 3: "且", 4: "仅地感", 5: "仅光耦", 6: "复位"} + + @classmethod + def from_regs(cls, regs: list[int]) -> "FullDataBlock": + if len(regs) < 19: + return cls() + fd = cls() + fd.sensitivity = regs[0] + fd.freq_level = regs[1] + fd.fusion_mode = regs[2] + fd.car_state = regs[3] + fd.loop_disconnected = bool(regs[4] & 0x01) + fd.loop_stable = bool(regs[4] & 0x02) + fd.opto_active = bool(regs[5]) + fd.freq_hz = (regs[6] << 16) | regs[7] + fd.capvd = (regs[8] << 16) | regs[9] + fd.variation = _to_int32((regs[10] << 16) | regs[11]) + fd.cut_count = (regs[12] << 16) | regs[13] + fd.relay_count = (regs[14] << 16) | regs[15] + fd.time_type = regs[16] + fd.misc_time_10ms = (regs[17] << 16) | regs[18] + return fd + + @property + def summary(self) -> str: + fusion = self.FUSION_NAMES.get(self.fusion_mode, f"?{self.fusion_mode}") + car = "有车" if self.car_state else "无车" + coil = "断开" if self.loop_disconnected else ("稳定" if self.loop_stable else "收敛中") + opto = "有效" if self.opto_active else "无效" + var_str = f"{self.variation:+d}" + gap = f"间隙={self.misc_time_10ms*10}ms" if self.time_type == 0 else "" + pas = f"通过={self.misc_time_10ms*10}ms" if self.time_type == 1 else "" + tm = gap or pas or "" + lines = [ + f"灵敏度={self.sensitivity} 频率档={'低' if self.freq_level else '高'} 融合={fusion}", + f"{car} 线圈={coil} 光耦={opto}", + f"频率={self.freq_hz} Hz CAPVD={self.capvd} Variation={var_str}", + f"断开次数={self.cut_count} 继电器次数={self.relay_count} {tm}", + ] + return "\n".join(lines) + + +def _to_int32(u32: int) -> int: + if u32 & 0x80000000: + return u32 - 0x100000000 + return u32 + + +def parse_response(data: bytes) -> Optional[Union[ + ReadBitsResponse, ReadRegsResponse, WriteRegResponse, + SlaveIdResponse, ModbusException +]]: + """解析 Modbus RTU 响应帧""" + if len(data) < 4: + return None + addr, fc = data[0], data[1] + + # CRC 校验 + if crc16(data[:-2]) != struct.unpack(' 0 else 0xFF + return SlaveIdResponse(model=model, status=status) + + return None + + +# ══════════════════════════════════════════════════ +# 主动上报帧识别 +# ══════════════════════════════════════════════════ + +def is_active_report(data: bytes) -> bool: + """检测是否为主动上报帧 (FC 0x04 + byte_cnt=38)""" + return len(data) >= 5 and data[1] == 0x04 and data[2] == 38 diff --git a/DLD154Tool/requirements.txt b/DLD154Tool/requirements.txt new file mode 100644 index 0000000..fd61e5b --- /dev/null +++ b/DLD154Tool/requirements.txt @@ -0,0 +1,2 @@ +PySide6>=6.6 +pyserial>=3.5 diff --git a/DLD154Tool/serial_client.py b/DLD154Tool/serial_client.py new file mode 100644 index 0000000..9f36c6b --- /dev/null +++ b/DLD154Tool/serial_client.py @@ -0,0 +1,100 @@ +""" +serial_client.py — RS485/串口抽象层 + +封装 pyserial: 端口扫描/打开/关闭/收发, 线程安全。 +""" +import threading +import time +from dataclasses import dataclass +from typing import Optional, Callable + +import serial +import serial.tools.list_ports + + +@dataclass +class SerialConfig: + port: str = "" + baudrate: int = 9600 + bytesize: int = serial.EIGHTBITS + parity: str = serial.PARITY_NONE + stopbits: int = serial.STOPBITS_ONE + timeout: float = 0.05 # 读超时 + + +class SerialClient: + """串口客户端 — 后台读线程 + 回调通知""" + + def __init__(self): + self._ser: Optional[serial.Serial] = None + self._thread: Optional[threading.Thread] = None + self._running = False + self._lock = threading.Lock() + self._on_data: Optional[Callable[[bytes], None]] = None + + # ── 静态工具 ────────────────────────────────── + + @staticmethod + def list_ports() -> list[str]: + return [p.device for p in serial.tools.list_ports.comports()] + + # ── 连接管理 ────────────────────────────────── + + def connect(self, cfg: SerialConfig) -> None: + self.disconnect() + self._ser = serial.Serial( + port=cfg.port, + baudrate=cfg.baudrate, + bytesize=cfg.bytesize, + parity=cfg.parity, + stopbits=cfg.stopbits, + timeout=cfg.timeout, + ) + self._running = True + self._thread = threading.Thread(target=self._read_loop, daemon=True) + self._thread.start() + + def disconnect(self) -> None: + self._running = False + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=1.0) + with self._lock: + if self._ser and self._ser.is_open: + self._ser.close() + self._ser = None + + @property + def is_connected(self) -> bool: + with self._lock: + return self._ser is not None and self._ser.is_open + + # ── 发送 ────────────────────────────────────── + + def send(self, data: bytes) -> None: + with self._lock: + if self._ser and self._ser.is_open: + self._ser.write(data) + + # ── 接收回调 ────────────────────────────────── + + def set_data_callback(self, cb: Optional[Callable[[bytes], None]]) -> None: + self._on_data = cb + + # ── 内部读线程 ──────────────────────────────── + + def _read_loop(self) -> None: + while self._running: + try: + with self._lock: + if not self._ser or not self._ser.is_open: + break + waiting = self._ser.in_waiting + if waiting > 0: + with self._lock: + data = self._ser.read(waiting) + if data and self._on_data: + self._on_data(data) + else: + time.sleep(0.01) + except (serial.SerialException, OSError): + time.sleep(0.1)