feat: 新增 SerialMate 串口调试助手 v0.1.0
- 协议帧构成令牌序列可定制(包头/板ID/LEN口径/5种CRC x 3种覆盖), 默认兼容 DLPilot 光机协议
This commit is contained in:
@@ -0,0 +1,450 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
SerialAssistant 串口协议模块(纯标准库, 帧构成可定制)
|
||||
======================================================
|
||||
帧由"构成令牌序列"定义(默认与 DLPilot 光机 RS485 协议完全一致):
|
||||
HDR ID LEN PAYLOAD CRC -> 包头 + 板ID + LEN + payload + CRC
|
||||
LEN = payload 字节数(口径可选, 见 LEN_SEMS), 默认即"包长对应 payload 长度"
|
||||
令牌: HDR=包头 ID=板ID LEN=长度 PAYLOAD=载荷 CRC=校验(必须最后),
|
||||
其余 1~2 位十六进制(如 AA 55)为固定字面量字节(双字节同步头)。
|
||||
约束: LEN/PAYLOAD/CRC 恰好各一次; LEN 前至少一个 HDR 或字面量(同步定界用)。
|
||||
同一布局收发共用, 仅包头值分方向(TX=tx_hdr / RX=rx_hdr); ID 在 RX 侧
|
||||
按上游语义多为响应 TYPE 码, 解析时不校验、原样透出(Frame.code, 可为 None)。
|
||||
|
||||
CRC 算法(2 字节算法在线上的字节序: MODBUS 低位在前, 其余高位在前):
|
||||
SUM8 累加和取低 8 位(1 字节)
|
||||
CRC8 poly 0x07 init 0x00(1 字节, 标准校验值 check("123456789")=0xF4)
|
||||
CRC16_MODBUS poly 0x8005 反射 init 0xFFFF(2 字节, check=0x4B37)
|
||||
CRC16_XMODEM poly 0x1021 init 0x0000(2 字节, check=0x31C3)
|
||||
CRC16_CCITT poly 0x1021 init 0xFFFF(2 字节, check=0x29B1)
|
||||
|
||||
CRC 覆盖范围(CRC_SCOPES, 按帧内位置定义, 任意布局通用):
|
||||
with_header 包头(第1字节)起至 CRC 前
|
||||
no_header 第 2 字节起至 CRC 前(DLPilot 默认)
|
||||
payload_only 仅 payload
|
||||
|
||||
仅标准库; pyserial 只允许出现在 link.py。
|
||||
"""
|
||||
|
||||
import collections
|
||||
import re
|
||||
import struct
|
||||
|
||||
LEN_MAX = 255 # LEN 为单字节, payload 上限 255
|
||||
|
||||
TYPE_NAMES = { # DLPilot/上游 protocols.md 的响应 TYPE 码语义, 用于 RX 译码显示
|
||||
0xF0: '成功',
|
||||
0xF1: 'CRC 校验错',
|
||||
0xF2: '包头错',
|
||||
0xF3: '板类型错',
|
||||
0xF4: '格式/缺参数',
|
||||
0xFD: '参数越界/读回复核不一致',
|
||||
0xFE: '未知命令',
|
||||
0xFF: '通用错误(含就绪门禁拒绝)',
|
||||
0xFA: '开光机通信失败',
|
||||
0xFB: '关光机通信失败',
|
||||
0xFC: '读回复核通信失败',
|
||||
}
|
||||
|
||||
|
||||
# --- CRC 实现(查表, 供高波特率下增量解析用) ---------------------------------
|
||||
|
||||
def _table_crc8():
|
||||
tbl = []
|
||||
for b in range(256):
|
||||
c = b
|
||||
for _ in range(8):
|
||||
c = ((c << 1) ^ 0x07) & 0xFF if c & 0x80 else (c << 1) & 0xFF
|
||||
tbl.append(c)
|
||||
return tuple(tbl)
|
||||
|
||||
|
||||
def _table_crc16_msb():
|
||||
"""MSB-first 表(poly 0x1021, 用于 XMODEM/CCITT)"""
|
||||
tbl = []
|
||||
for b in range(256):
|
||||
c = (b << 8) & 0xFFFF
|
||||
for _ in range(8):
|
||||
c = ((c << 1) ^ 0x1021) & 0xFFFF if c & 0x8000 else (c << 1) & 0xFFFF
|
||||
tbl.append(c)
|
||||
return tuple(tbl)
|
||||
|
||||
|
||||
def _table_crc16_reflected():
|
||||
"""反射表(poly 0x8005 → 0xA001, 用于 MODBUS)"""
|
||||
tbl = []
|
||||
for b in range(256):
|
||||
c = b
|
||||
for _ in range(8):
|
||||
c = (c >> 1) ^ 0xA001 if c & 1 else c >> 1
|
||||
tbl.append(c)
|
||||
return tuple(tbl)
|
||||
|
||||
|
||||
_TAB_CRC8 = _table_crc8()
|
||||
_TAB_CRC16_MSB = _table_crc16_msb()
|
||||
_TAB_CRC16_REF = _table_crc16_reflected()
|
||||
|
||||
|
||||
def _sum8(data):
|
||||
return sum(data) & 0xFF
|
||||
|
||||
|
||||
def _crc8(data, _t=_TAB_CRC8):
|
||||
c = 0
|
||||
for b in data:
|
||||
c = _t[c ^ b]
|
||||
return c
|
||||
|
||||
|
||||
def _crc16_msb(data, init, _t=_TAB_CRC16_MSB):
|
||||
c = init
|
||||
for b in data:
|
||||
c = ((c << 8) & 0xFFFF) ^ _t[((c >> 8) ^ b) & 0xFF]
|
||||
return c
|
||||
|
||||
|
||||
def _crc16_xmodem(data):
|
||||
return _crc16_msb(data, 0x0000)
|
||||
|
||||
|
||||
def _crc16_ccitt(data):
|
||||
return _crc16_msb(data, 0xFFFF)
|
||||
|
||||
|
||||
def _crc16_modbus(data, _t=_TAB_CRC16_REF):
|
||||
c = 0xFFFF
|
||||
for b in data:
|
||||
c = (c >> 8) ^ _t[(c ^ b) & 0xFF]
|
||||
return c
|
||||
|
||||
|
||||
# name -> (CRC 字节数, 计算函数, 线上字节序: 'big'=高字节在前 / 'little'=低字节在前)
|
||||
CRC_ALGOS = {
|
||||
'SUM8': (1, _sum8, 'big'),
|
||||
'CRC8': (1, _crc8, 'big'),
|
||||
'CRC16_MODBUS': (2, _crc16_modbus, 'little'),
|
||||
'CRC16_XMODEM': (2, _crc16_xmodem, 'big'),
|
||||
'CRC16_CCITT': (2, _crc16_ccitt, 'big'),
|
||||
}
|
||||
CRC_ALGO_LABELS = { # 下拉框显示文本
|
||||
'SUM8': 'SUM8 累加和',
|
||||
'CRC8': 'CRC8 (poly 07)',
|
||||
'CRC16_MODBUS': 'CRC16 MODBUS',
|
||||
'CRC16_XMODEM': 'CRC16 XMODEM',
|
||||
'CRC16_CCITT': 'CRC16 CCITT-FALSE',
|
||||
}
|
||||
CRC_SCOPES = ('with_header', 'no_header', 'payload_only')
|
||||
CRC_SCOPE_LABELS = {
|
||||
'with_header': '含包头(整帧不含CRC)',
|
||||
'no_header': '不含包头(跳过第1字节, DLPilot)',
|
||||
'payload_only': '仅 payload',
|
||||
}
|
||||
LEN_SEMS = ('payload', 'body', 'total')
|
||||
LEN_SEM_LABELS = { # LEN 字节的计数口径
|
||||
'payload': 'payload 长度(默认)',
|
||||
'body': '到帧尾(payload+CRC)',
|
||||
'total': '整帧(含包头)',
|
||||
}
|
||||
|
||||
DEFAULT_LAYOUT = 'HDR ID LEN PAYLOAD CRC'
|
||||
_FRAME_TOKEN_RE = re.compile(r'^(?:0x)?([0-9A-Fa-f]{1,2})$')
|
||||
_KEYWORDS = {'HDR': 'hdr', 'ID': 'id', 'BOARD': 'id', 'LEN': 'len',
|
||||
'PAYLOAD': 'payload', 'DATA': 'payload', 'CRC': 'crc'}
|
||||
|
||||
|
||||
class FrameLayout:
|
||||
"""帧构成: 令牌序列(parse 文本而来), 供组帧与解析共用。
|
||||
|
||||
tokens 元素: 'hdr'/'id'/'len'/'payload'/'crc' 或 ('lit', 字节值)。
|
||||
约束(parse 内校验): LEN/PAYLOAD/CRC 恰好各一次; CRC 必须最后;
|
||||
LEN 之前至少一个 HDR 或字面量(否则无同步定界); ID 最多一次。
|
||||
payload 之前的令牌都是 1 字节(CRC 在最后), 故 payload 字节偏移=payload_pos。
|
||||
"""
|
||||
|
||||
def __init__(self, tokens):
|
||||
self.tokens = tuple(tokens)
|
||||
counts = collections.Counter(
|
||||
t if isinstance(t, str) else 'lit' for t in self.tokens)
|
||||
for key, cn in (('len', 'LEN'), ('payload', 'PAYLOAD'), ('crc', 'CRC')):
|
||||
if counts[key] != 1:
|
||||
raise ValueError(f'{cn} 必须恰好出现一次(当前 {counts[key]} 次)')
|
||||
if self.tokens[-1] != 'crc':
|
||||
raise ValueError('CRC 必须是最后一个令牌')
|
||||
if counts['id'] > 1:
|
||||
raise ValueError('ID 最多出现一次')
|
||||
self.len_pos = self.tokens.index('len')
|
||||
self.payload_pos = self.tokens.index('payload')
|
||||
id_positions = [i for i, t in enumerate(self.tokens) if t == 'id']
|
||||
self.id_pos = id_positions[0] if id_positions else None
|
||||
if not any(isinstance(t, tuple) or t == 'hdr'
|
||||
for t in self.tokens[:self.len_pos]):
|
||||
raise ValueError('LEN 之前需要至少一个 HDR 或十六进制字面量作同步头')
|
||||
|
||||
@classmethod
|
||||
def parse(cls, text):
|
||||
"""'AA 55 ID LEN PAYLOAD CRC' 之类 -> FrameLayout; 非法抛 ValueError"""
|
||||
parts = [p for p in re.split(r'[\s,，;；]+', (text or '').strip()) if p]
|
||||
if not parts:
|
||||
raise ValueError('帧构成为空')
|
||||
tokens = []
|
||||
for p in parts:
|
||||
kw = _KEYWORDS.get(p.upper())
|
||||
if kw:
|
||||
tokens.append(kw)
|
||||
continue
|
||||
m = _FRAME_TOKEN_RE.match(p)
|
||||
if not m:
|
||||
raise ValueError(f'未知令牌 "{p}"(可用: HDR ID LEN PAYLOAD CRC '
|
||||
f'或 0~FF 十六进制字面量)')
|
||||
tokens.append(('lit', int(m.group(1), 16)))
|
||||
return cls(tokens)
|
||||
|
||||
# --- 尺寸(依赖 CRC 字节数, 由 ProtoConfig 传参) ---
|
||||
|
||||
def rest_size(self, crc_size):
|
||||
"""payload 之后的字节数(ID/HDR/字面量 + CRC)"""
|
||||
return crc_size + len(self.tokens) - self.payload_pos - 2
|
||||
|
||||
def total_size(self, payload_len, crc_size):
|
||||
# payload 之前的令牌都是 1 字节, payload 字节偏移即 payload_pos
|
||||
return self.payload_pos + payload_len + self.rest_size(crc_size)
|
||||
|
||||
def len_extra(self, len_sem, crc_size):
|
||||
"""LEN 字节值 = payload_len + 该口径下的附加字节数"""
|
||||
if len_sem == 'body':
|
||||
return self.rest_size(crc_size)
|
||||
if len_sem == 'total':
|
||||
return self.payload_pos + self.rest_size(crc_size)
|
||||
return 0
|
||||
|
||||
|
||||
Frame = collections.namedtuple('Frame', 'code payload raw')
|
||||
"""code=ID 位置字节(RX 侧多为响应 TYPE 码; 布局无 ID 时为 None),
|
||||
payload=原始字节, raw=整帧含 CRC"""
|
||||
|
||||
|
||||
class ProtoConfig:
|
||||
"""协议配置: 帧构成(FrameLayout) + 包头(TX/RX)/板ID/LEN 口径/CRC 算法与覆盖"""
|
||||
|
||||
def __init__(self, tx_hdr=0xD5, rx_hdr=0xB5, board=0x01,
|
||||
crc_algo='SUM8', crc_scope='no_header',
|
||||
layout=None, len_sem='payload'):
|
||||
for name, v in (('tx_hdr', tx_hdr), ('rx_hdr', rx_hdr), ('board', board)):
|
||||
if not 0 <= int(v) <= 0xFF:
|
||||
raise ValueError(f'{name} 超出 0~255: {v}')
|
||||
if crc_algo not in CRC_ALGOS:
|
||||
raise ValueError(f'未知 CRC 算法: {crc_algo}')
|
||||
if crc_scope not in CRC_SCOPES:
|
||||
raise ValueError(f'未知 CRC 覆盖范围: {crc_scope}')
|
||||
if len_sem not in LEN_SEMS:
|
||||
raise ValueError(f'未知 LEN 口径: {len_sem}')
|
||||
self.tx_hdr, self.rx_hdr, self.board = int(tx_hdr), int(rx_hdr), int(board)
|
||||
self.crc_algo, self.crc_scope = crc_algo, crc_scope
|
||||
self.len_sem = len_sem
|
||||
self.layout = layout if layout is not None else FrameLayout.parse(DEFAULT_LAYOUT)
|
||||
|
||||
@property
|
||||
def crc_size(self):
|
||||
return CRC_ALGOS[self.crc_algo][0]
|
||||
|
||||
# --- CRC ---
|
||||
|
||||
def compute_crc(self, nocrc_frame, payload_len):
|
||||
"""对不含 CRC 的整帧按覆盖范围切片算 CRC 值(切片按位置定义, 任意布局通用)"""
|
||||
_size, fn, _order = CRC_ALGOS[self.crc_algo]
|
||||
if self.crc_scope == 'no_header':
|
||||
region = nocrc_frame[1:]
|
||||
elif self.crc_scope == 'payload_only':
|
||||
po = self.layout.payload_pos
|
||||
region = nocrc_frame[po:po + payload_len]
|
||||
else:
|
||||
region = nocrc_frame
|
||||
return fn(region)
|
||||
|
||||
def encode_crc(self, value):
|
||||
"""CRC 值 -> 线上字节(2 字节算法按算法约定字节序)"""
|
||||
size, _fn, order = CRC_ALGOS[self.crc_algo]
|
||||
if size == 1:
|
||||
return bytes((value & 0xFF,))
|
||||
hi, lo = (value >> 8) & 0xFF, value & 0xFF
|
||||
return bytes((hi, lo)) if order == 'big' else bytes((lo, hi))
|
||||
|
||||
# --- 组帧 ---
|
||||
|
||||
def build_frame(self, payload):
|
||||
"""payload bytes -> 完整帧 bytes; 空载荷允许(LEN=0), 超限抛 ValueError"""
|
||||
payload = bytes(payload)
|
||||
if len(payload) > LEN_MAX:
|
||||
raise ValueError(f'payload 超长(>{LEN_MAX} 字节)')
|
||||
cs = self.crc_size
|
||||
len_val = len(payload) + self.layout.len_extra(self.len_sem, cs)
|
||||
if len_val > LEN_MAX:
|
||||
raise ValueError(f'LEN 字节溢出(={len_val}>255), 检查帧构成与 LEN 口径')
|
||||
parts = []
|
||||
for tok in self.layout.tokens:
|
||||
if tok == 'crc':
|
||||
continue # CRC 最后, 跳过待拼
|
||||
if tok == 'hdr':
|
||||
parts.append(bytes((self.tx_hdr,)))
|
||||
elif tok == 'id':
|
||||
parts.append(bytes((self.board,)))
|
||||
elif tok == 'len':
|
||||
parts.append(bytes((len_val,)))
|
||||
elif tok == 'payload':
|
||||
parts.append(payload)
|
||||
else:
|
||||
parts.append(bytes((tok[1],)))
|
||||
nocrc = b''.join(parts)
|
||||
return nocrc + self.encode_crc(self.compute_crc(nocrc, len(payload)))
|
||||
|
||||
# --- 校验与拆帧 ---
|
||||
|
||||
def _plen_of(self, raw):
|
||||
"""从整帧 LEN 字节解出 payload 长度; 非法(负值)返回 None"""
|
||||
plen = raw[self.layout.len_pos] - self.layout.len_extra(self.len_sem, self.crc_size)
|
||||
return plen if plen >= 0 else None
|
||||
|
||||
def verify_frame(self, raw):
|
||||
"""整帧(含 CRC)校验: 长度与 LEN 口径自洽 + CRC 匹配"""
|
||||
cs = self.crc_size
|
||||
if len(raw) < self.layout.len_pos + 1 + cs:
|
||||
return False
|
||||
plen = self._plen_of(raw)
|
||||
if plen is None or len(raw) != self.layout.total_size(plen, cs):
|
||||
return False
|
||||
return self.encode_crc(self.compute_crc(raw[:-cs], plen)) == raw[-cs:]
|
||||
|
||||
def parse_frame(self, raw):
|
||||
"""整帧 -> Frame(不校验 CRC, 调用方先用 verify_frame); 长度不符抛 ValueError"""
|
||||
cs = self.crc_size
|
||||
plen = self._plen_of(raw)
|
||||
if plen is None or len(raw) != self.layout.total_size(plen, cs):
|
||||
raise ValueError('帧长与 LEN 口径不符')
|
||||
po = self.layout.payload_pos
|
||||
code = raw[self.layout.id_pos] if self.layout.id_pos is not None else None
|
||||
return Frame(code, raw[po:po + plen], bytes(raw))
|
||||
|
||||
|
||||
class FrameParser:
|
||||
"""增量字节流 -> 完整帧: 前缀令牌逐位对齐(帧头/字面量, ID 为通配) + LEN 定长
|
||||
+ CRC 校验; 任一处失配丢 1 字节重新同步。LEN 值经口径换算后即 payload 长度,
|
||||
负值视为伪帧头逐出 —— 伪帧头最多滞留一帧即被逐出重同步, 对调试工具可接受。
|
||||
"""
|
||||
|
||||
def __init__(self, cfg):
|
||||
self.cfg = cfg
|
||||
lay = cfg.layout
|
||||
cs = cfg.crc_size
|
||||
# LEN 之前的前缀令牌: (帧内位置, 期望字节|None=通配); hdr 按 RX 包头匹配
|
||||
self._prefix = []
|
||||
for pos, tok in enumerate(lay.tokens[:lay.len_pos]):
|
||||
exp = None
|
||||
if tok == 'hdr':
|
||||
exp = cfg.rx_hdr
|
||||
elif isinstance(tok, tuple):
|
||||
exp = tok[1]
|
||||
self._prefix.append((pos, exp))
|
||||
self._len_pos = lay.len_pos
|
||||
self._extra = lay.len_extra(cfg.len_sem, cs)
|
||||
self._buf = bytearray()
|
||||
|
||||
def clear(self):
|
||||
self._buf.clear()
|
||||
|
||||
def _aligned(self):
|
||||
"""把缓冲对齐到前缀匹配; 字节不足且已匹配部分则保持等待(返回 True),
|
||||
失配丢 1 字节重试, 缓冲耗尽返回 False"""
|
||||
buf = self._buf
|
||||
while True:
|
||||
for pos, exp in self._prefix:
|
||||
if pos >= len(buf):
|
||||
return True # 已匹配部分全对, 等更多字节
|
||||
if exp is not None and buf[pos] != exp:
|
||||
break # 失配 -> 丢 1 字节
|
||||
else:
|
||||
return True
|
||||
del buf[:1]
|
||||
if not buf:
|
||||
return False
|
||||
|
||||
def feed(self, data):
|
||||
"""喂入新收到的字节, 返回本次解析出的完整 Frame 列表"""
|
||||
self._buf += bytes(data)
|
||||
cfg, frames = self.cfg, []
|
||||
while self._aligned() and self._buf:
|
||||
if len(self._buf) < self._len_pos + 1:
|
||||
break # LEN 字节未到齐
|
||||
plen = self._buf[self._len_pos] - self._extra
|
||||
if plen < 0: # LEN 为负 -> 伪帧头, 逐出重同步
|
||||
del self._buf[:1]
|
||||
continue
|
||||
total = cfg.layout.total_size(plen, cfg.crc_size)
|
||||
if len(self._buf) < total:
|
||||
break # 整帧未收齐
|
||||
raw = bytes(self._buf[:total])
|
||||
if cfg.verify_frame(raw):
|
||||
frames.append(cfg.parse_frame(raw))
|
||||
del self._buf[:total]
|
||||
else:
|
||||
del self._buf[:1] # CRC 失配 -> 丢帧头重同步
|
||||
return frames
|
||||
|
||||
|
||||
# --- payload 数值提取(曲线通道用) --------------------------------------------
|
||||
|
||||
BIN_FORMATS = { # fmt -> (struct 格式, 字节数)
|
||||
'u8': ('B', 1), 's8': ('b', 1),
|
||||
'u16be': ('>H', 2), 'u16le': ('<H', 2),
|
||||
's16be': ('>h', 2), 's16le': ('<h', 2),
|
||||
'u32be': ('>I', 4), 'u32le': ('<I', 4),
|
||||
's32be': ('>i', 4), 's32le': ('<i', 4),
|
||||
'f32be': ('>f', 4), 'f32le': ('<f', 4),
|
||||
}
|
||||
NUM_RE = re.compile(r'-?\d+(?:\.\d+)?')
|
||||
|
||||
|
||||
def extract_value(payload, mode, *, offset=0, fmt='u16be', index=1,
|
||||
pattern=r'(?<=:)\d+', scale=1.0):
|
||||
"""从 payload 提取一个数值(已乘 scale); 提取失败返回 None。
|
||||
|
||||
bin 二进制定长提取: offset 起, 按 fmt(BIN_FORMATS 键)解
|
||||
num ASCII 文本第 index(1 基)个数字, 如 'okTL:246' -> 246
|
||||
re 正则第 1 捕获组(无组则整体), 如 '(?<=:)(\\d+)'
|
||||
"""
|
||||
payload = bytes(payload)
|
||||
try:
|
||||
if mode == 'bin':
|
||||
if fmt not in BIN_FORMATS:
|
||||
return None
|
||||
sfmt, size = BIN_FORMATS[fmt]
|
||||
if offset < 0 or len(payload) < offset + size:
|
||||
return None
|
||||
value = struct.unpack(sfmt, payload[offset:offset + size])[0]
|
||||
elif mode == 'num':
|
||||
nums = NUM_RE.findall(payload.decode('ascii', 'replace'))
|
||||
if not 1 <= index <= len(nums):
|
||||
return None
|
||||
value = float(nums[index - 1])
|
||||
elif mode == 're':
|
||||
m = re.search(pattern, payload.decode('ascii', 'replace'))
|
||||
if not m:
|
||||
return None
|
||||
value = float(m.group(1) if m.groups() else m.group(0))
|
||||
else:
|
||||
return None
|
||||
except (ValueError, re.error):
|
||||
return None
|
||||
return value * scale
|
||||
|
||||
|
||||
def hexs(data):
|
||||
"""bytes -> 'AA BB CC' 大写 hex, 日志统一格式"""
|
||||
return bytes(data).hex(' ').upper()
|
||||
|
||||
|
||||
def clean_text(s):
|
||||
"""控制字符 -> '·': 二进制按 ASCII 解出的 NUL 等会污染日志行,
|
||||
且 Windows 剪贴板遇 NUL 直接截断复制内容(DLPilot 同源教训)"""
|
||||
return ''.join(c if ord(c) >= 0x20 and ord(c) != 0x7F else '·' for c in s)
|
||||
Reference in New Issue
Block a user