451 lines
17 KiB
Python
451 lines
17 KiB
Python
#!/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)
|