292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""
|
||
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'')
|
||
|
||
|
||
def build_reset_mcu(addr: int) -> bytes:
|
||
"""FC 0x06 — 写 HR 0x0012=0x2C2C 复位 MCU"""
|
||
return build_write_single_reg(addr, 0x0012, 0x2C2C)
|
||
|
||
|
||
# ══════════════════════════════════════════════════
|
||
# 响应解析
|
||
# ══════════════════════════════════════════════════
|
||
|
||
@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
|