50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
"""模拟真机 M2S100 场景: 设备 100ms 周期上报 30 帧, 验证帧按帧粒度
|
|
到达 GUI 回调(间隔≈100ms, 到达→消费延迟<300ms), 而非攒批后一次涌入。
|
|
"""
|
|
import os
|
|
import sys
|
|
import time
|
|
import threading
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from link import SerialLink
|
|
import protocol as d
|
|
|
|
got = []
|
|
cfg = d.ProtoConfig()
|
|
link = SerialLink(cfg, on_status=lambda s, ts: None)
|
|
link.on_frame = lambda fr, ts: got.append((time.time(), ts))
|
|
|
|
link.open('loop://', 115200)
|
|
deadline = time.time() + 5
|
|
while time.time() < deadline and not link.is_connected():
|
|
time.sleep(0.05)
|
|
assert link.is_connected(), 'loop:// 未连接'
|
|
|
|
resp = bytes.fromhex('B5 F0 08 6F 6B 54 4C 3A 32 34 36 48')
|
|
|
|
def feeder():
|
|
for _ in range(30):
|
|
link.ser.write(resp)
|
|
time.sleep(0.1)
|
|
|
|
t = threading.Thread(target=feeder)
|
|
t.start()
|
|
t.join()
|
|
time.sleep(0.5) # 等最后一帧走完读线程
|
|
|
|
n = len(got)
|
|
print('收到帧数:', n, '(期望 30)')
|
|
if n >= 2:
|
|
gaps = [b[0] - a[0] for a, b in zip(got, got[1:])]
|
|
lag = [cb - ts for cb, ts in got]
|
|
print('回调间隔 min/max/avg(ms): %d / %d / %d'
|
|
% (min(gaps) * 1000, max(gaps) * 1000, sum(gaps) / len(gaps) * 1000))
|
|
print('到达→消费延迟 max(ms):', round(max(lag) * 1000))
|
|
ok = n == 30 and max(gaps) < 0.5 and max(lag) < 0.3
|
|
else:
|
|
ok = False
|
|
print('RESULT:', 'PASS' if ok else 'FAIL')
|
|
link.close()
|
|
sys.exit(0 if ok else 1)
|