DLD154Pro/DLD154Tool/protocol_700bs.py

"""
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