92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""链路自检(无硬件): pyserial loop:// 虚拟回环验证 组帧→发送→接收→解析→曲线提取
|
||
====================================================================
|
||
缺 pyserial 时跳过(退出码 0)。loop:// 回环口写即读, 发出的帧同样回到
|
||
接收侧, 但 TX 帧头(D5)不是 RX 帧头(B5), 会被解析器当噪声丢弃, 不影响判定。
|
||
|
||
用法: python selftest_loop.py (全部通过退出码 0)
|
||
"""
|
||
import sys
|
||
import time
|
||
|
||
try:
|
||
import serial # noqa: F401
|
||
except ImportError:
|
||
print('未安装 pyserial, 跳过回环自检')
|
||
sys.exit(0)
|
||
|
||
import protocol as d
|
||
from link import SerialLink
|
||
|
||
fails = []
|
||
|
||
|
||
def check(name, got, want):
|
||
ok = got == want
|
||
print(('PASS' if ok else 'FAIL'), name, '->', got, '' if ok else f'(期望 {want})')
|
||
if not ok:
|
||
fails.append(name)
|
||
|
||
|
||
def wait_for(pred, timeout=5.0):
|
||
deadline = time.time() + timeout
|
||
while time.time() < deadline:
|
||
if pred():
|
||
return True
|
||
time.sleep(0.05)
|
||
return False
|
||
|
||
|
||
got = {'status': [], 'frames': [], 'tx': [], 'raw': []}
|
||
cfg = d.ProtoConfig()
|
||
link = SerialLink(cfg, on_status=lambda s: got['status'].append(s))
|
||
link.on_frame = lambda fr: got['frames'].append(fr)
|
||
link.on_raw = lambda b: got['raw'].append(b)
|
||
link.on_tx = lambda lb, data: got['tx'].append((lb, data))
|
||
|
||
# 1) 打开虚拟回环口
|
||
link.open('loop://', 115200)
|
||
check('连接 loop://', wait_for(lambda: got['status'] and
|
||
got['status'][-1].startswith('已连接')), True)
|
||
|
||
# 2) 自动包装发送: TX 回显 + 回环字节
|
||
payload = 'M731S0T1000'
|
||
frame = cfg.build_frame(payload.encode('ascii'))
|
||
link.send_frame(frame, label=payload)
|
||
check('发送回显', wait_for(lambda: len(got['tx']) >= 1), True)
|
||
if got['tx']:
|
||
check('TX label', got['tx'][0][0], payload)
|
||
check('TX frame', got['tx'][0][1], frame)
|
||
|
||
# 3) 模拟设备应答: okTL:246(温度 24.6°C) → 解析 + 曲线提取
|
||
resp = bytes.fromhex('B5 F0 08 6F 6B 54 4C 3A 32 34 36 48') # sum(F0..36)=0x48
|
||
check('响应帧向量自校验', cfg.verify_frame(resp), True)
|
||
link.ser.write(resp)
|
||
check('应答解析', wait_for(lambda: len(got['frames']) >= 1), True)
|
||
if got['frames']:
|
||
fr = got['frames'][0]
|
||
check('TYPE 码', fr.code, 0xF0)
|
||
check('payload', fr.payload, b'okTL:246')
|
||
v = d.extract_value(fr.payload, 'num', index=1, scale=0.1)
|
||
check('曲线提取(×0.1)', v is not None and abs(v - 24.6) < 1e-9, True)
|
||
|
||
# 4) 透传模式切换: 关解析后收到原始字节
|
||
link.set_parse(False)
|
||
raw = bytes([0x11, 0x22, 0xB5, 0x00])
|
||
link.ser.write(raw)
|
||
check('透传原始字节', wait_for(lambda: len(got['raw']) >= 1), True)
|
||
if got['raw']:
|
||
check('透传内容', got['raw'][0], raw)
|
||
|
||
# 5) 切回解析模式, 缓冲已清空, 正常再收一帧
|
||
link.set_parse(True)
|
||
link.ser.write(bytes.fromhex('B5 F0 02 6F 6B CC'))
|
||
check('切回解析再收帧', wait_for(lambda: len(got['frames']) >= 2), True)
|
||
|
||
link.close()
|
||
check('断开', wait_for(lambda: got['status'] and got['status'][-1] == '已断开'), True)
|
||
|
||
print('RESULT:', 'FAIL:' + ','.join(fails) if fails else 'ALL PASS')
|
||
sys.exit(1 if fails else 0)
|