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

feat: 集成 DLD154Tool — PySide6 桌面串口调试工具

Browse Source 
7 文件 1331 行, 支持 DLD154Pro 双协议:
- 700BS: 状态/频率/拨码/复位查询 + 自动轮询
- Modbus RTU: FC 0x02/0x03/0x04/0x06/0x10/0x11
  + 全数据块(0x0100)美化解析 + 主动上报控制
- PySide6 GUI, 4 Tab, 后台串口线程

从独立仓库 wangfq/DLD154Tool 合并入 DLD154Pro monorepo
This commit is contained in:
wangfq
2026-07-30 16:06:58 +08:00
parent df6e38f118
commit e97388500b
7 changed files with 1331 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
DLD154Tool/protocol_modbus.py
+286
View File
@@ -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('<H', crc)
def build_read_input_regs(addr: int, start: int, qty: int) -> 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('<H', data[-2:])[0]:
return None
# 异常响应
if fc & 0x80:
return ModbusException(fc=fc & 0x7F, excode=data[2])
# 正常响应
if fc in (0x02, 0x01):
return ReadBitsResponse(fc=fc, bits=int.from_bytes(data[3:3+data[2]], 'big'))
elif fc in (0x03, 0x04):
bc = data[2]
regs = [int.from_bytes(data[3+i*2:3+i*2+2], 'big') for i in range(bc // 2)]
return ReadRegsResponse(fc=fc, regs=regs)
elif fc in (0x06, 0x10):
if fc == 0x06:
reg = (data[2] << 8) | data[3]
val = (data[4] << 8) | data[5]
else:
reg = (data[2] << 8) | data[3]
val = (data[4] << 8) | data[5]
return WriteRegResponse(fc=fc, reg=reg, value=val)
elif fc == 0x11:
bc = data[2]
model_end = data.index(0, 3, 3+bc) if 0 in data[3:3+bc] else 3+bc
model = data[3:model_end].decode('ascii', errors='replace')
status = data[3+bc-1] if bc > 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