wandering1
1634088bd7
屏板链路设计说明.md -> 屏板链路设计.md（去档名「说明」，与 ADC 节拍 / 运行与自检记录两篇口径一致）；控制台单写者日志设计说明.md 同改。 两档并存期间旧名引用 0 残留。 - 屏板链路 §3.2 台架日志口（RS232#1 / CN1 端子 / COM3）移入控制台档 §3.4：日志口的接线属 UART5 控制台通道，不属屏板链路；屏板档只留 「别去 CN1 找 USART1 端子」这句接线防错提示。 - 屏板链路 §7.1 由「根因账本 + 修复前后对比」改写为现状描述：延迟 应答机制表 + 耗时分解表 + 两个计数口径 + 已知边界，40 行压到 24 行。 顺带修正 §8 的错引：原指 `网络自愈架构设计.md` §8（实为「延后项」， 不含 uiTask 内容）改为 §4.1 `net_reconcile_tick`。 - §9.1 增补长稳行（串口日志 2026-11-13 18:31 -> 2026-11-15 21:43， 单次上电 51.2 h：rx=658262 ok=658262、crc/len/fmt/err 全 0、 drop=7 仅启动期、kerr/abor/bad=0、全程 off=0）；§9.2 销项三条 （仪器项、2 h 长稳、len=7 观察项），新增「按键通路」一条——51.2 h 日志 keys=0 kq=0 dfok=0 dfold=0，延迟应答在真机上从未执行过。 - 删除 sb_frame_b_build_logo() 及 SB_LOGO_MAGIC_* / SB_SCREEN_W/H 与对应单测（TC-SBPROTO-022/023 全组、024c/024d），组帧入口只剩 sb_frame_b_build_image()。sb_proto 单测 59 -> 45 条，45/45 通过。 - cfg_file.c 出厂模板三条中文字面量改 \uXXXX 转义：消除全量重编时的 4 条 #870-D invalid multibyte character sequence，运行期字节不变， 源码回到纯 ASCII（符合「字符串字面量必须纯 ASCII」）。 验证：keil-build 0 Error 0 Warning；sb_proto 45/45、sb_link 164/164； pre-commit 48 钩子全 Passed；check_doc_links --sections-strict 通过。
222 lines
8.7 KiB
Python
222 lines
8.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""RemoDispBus(7003) 拉帧压力客户端：「双通道并发回归」的网络侧激励。
|
||
|
||
做什么：
|
||
连设备 IP（默认见 bench_defaults.DEVICE_IP）:7003，先发 INIT 查询屏幕参数（应得 8 B：宽u16/高u16/帧字节数u32，
|
||
大端），随后按固定节拍持续发 LCDMEM(偏移 0) 拉整帧 3200 B 显存，逐帧校验
|
||
XOR 校验和与长度，统计请求数 / 应答数 / CRC 错 / 超时 / 往返延迟 min-avg-max。
|
||
可 --dump last_frame.pbm 存最后一帧（1bpp 布局与屏板帧 B 点阵段逐字节同构）。
|
||
|
||
为什么它能当 4-1 的激励：
|
||
7003 的每次 LCDMEM 往返在固件里是「uiTask 入队 + 等 netTask 回执」
|
||
（osMessageQueuePut(osWaitForever) + xTaskNotifyWait(portMAX_DELAY)，最坏 ≈4 s，
|
||
见 ui_net_sink.h 的 ⚠），即**每一拍拉帧都在真实占用 uiTask**——这正是
|
||
「7003 拉帧期间屏板不超时」要压的条件。屏板侧应答起在 USART1 空闲中断里，
|
||
理论上不被 uiTask 拖住；4-1 的判据就是拿两边的数交叉证明这件事：
|
||
网络侧（本脚本）：超时 ≈0、延迟分布正常；
|
||
固件侧（日志口 grep SBLINK）：tx 跟得上 rx（drop 无新增）、bad=0、kerr=0。
|
||
|
||
协议（Drivers/BSP/UI/ui_net_sink.c 实测口径）：
|
||
帧 = <TAG 1B> <CMD 1B> <LEN u16 大端> <payload LEN B> <XOR(payload) 1B>
|
||
客户端 TAG=0xAA，设备 TAG=0xBB；CMD：0=KEEP 1=INIT 2=KEY 3=LCDMEM。
|
||
LCDMEM 请求 payload = 偏移 u32 大端（仅支持 0）；应答 payload = 偏移 u32 + 3200 B 位图。
|
||
（设备对 CRC 只做异或累加——与屏板链路的 CRC-16 是两套，别混。）
|
||
|
||
用法（仓库根目录，先清代理再跑——本机有 HTTP 代理会劫持内网连接）：
|
||
& '.venv\\Scripts\\python.exe' test/host/remolcd_pull_test.py --seconds 30
|
||
--hz 10 拉帧节拍（默认 10 = 每秒 10 次，重压；1 次/秒 ≈ DTU-RemoteLCD 观感）
|
||
--dump x.pbm 退出前把最后一帧写成 PBM（160x160 1bpp，可直接图片查看）
|
||
|
||
退出码：0 = 全程无超时且 CRC 全对；1 = 出现超时/CRC 错/断连；2 = 用法错误。
|
||
|
||
设计依据：docs/子系统设计/屏板链路设计.md §9.1（双通道并发）。
|
||
创建日期：2026-10-30
|
||
"""
|
||
import argparse
|
||
import socket
|
||
import struct
|
||
import sys
|
||
import time
|
||
|
||
import bench_defaults # 台架连接默认值单一真源
|
||
|
||
DEFAULT_IP = bench_defaults.DEVICE_IP
|
||
DEFAULT_PORT = 7003
|
||
TAG_CLIENT = 0xAA
|
||
TAG_DEVICE = 0xBB
|
||
CMD_KEEP = 0
|
||
CMD_INIT = 1
|
||
CMD_LCDMEM = 3
|
||
FB_SIZE = 3200
|
||
SCREEN_W = 160
|
||
SCREEN_H = 160
|
||
|
||
|
||
def build_frame(cmd, payload=b""):
|
||
"""组一帧客户端请求：AA <cmd> <len16be> <payload> <xor>。"""
|
||
body = bytes([TAG_CLIENT, cmd, (len(payload) >> 8) & 0xFF, len(payload) & 0xFF])
|
||
xor = 0
|
||
for b in payload:
|
||
xor ^= b
|
||
return body + payload + bytes([xor])
|
||
|
||
|
||
class FrameReader:
|
||
"""设备应答流重组（逐字节状态机，TAG 定界——与固件侧同思路，噪声字节自动重同步）。"""
|
||
|
||
def __init__(self):
|
||
self.buf = bytearray()
|
||
|
||
def pop(self):
|
||
"""从缓冲里取一个完整 (cmd, payload)；没有则 None。坏帧丢弃继续找下一个 TAG。"""
|
||
while True:
|
||
i = self.buf.find(bytes([TAG_DEVICE]))
|
||
if i < 0:
|
||
del self.buf[:] # 整段都不是帧头：全弃
|
||
return None
|
||
if i > 0:
|
||
del self.buf[:i] # 丢帧头前的噪声
|
||
if len(self.buf) < 4:
|
||
return None
|
||
cmd = self.buf[1]
|
||
ln = (self.buf[2] << 8) | self.buf[3]
|
||
total = 4 + ln + 1
|
||
if len(self.buf) < total:
|
||
return None
|
||
payload = bytes(self.buf[4:4 + ln])
|
||
crc = self.buf[4 + ln]
|
||
del self.buf[:total]
|
||
xor = 0
|
||
for b in payload:
|
||
xor ^= b
|
||
if crc != xor:
|
||
return ("CRCERR", payload)
|
||
return (cmd, payload)
|
||
|
||
def feed(self, data):
|
||
self.buf += data
|
||
|
||
|
||
def run(args):
|
||
try:
|
||
s = socket.create_connection((args.ip, args.port), timeout=5.0)
|
||
except OSError as exc:
|
||
print("[ERR] 连 %s:%d 失败：%s" % (args.ip, args.port, exc))
|
||
print(" （若在内网却报超时/奇怪拒绝：先清 HTTP 代理再跑，见文件头用法行）")
|
||
return 1
|
||
s.settimeout(5.0)
|
||
rd = FrameReader()
|
||
|
||
def wait_frame(want_cmd):
|
||
"""收一个 cmd==want_cmd 的应答；CRC 错返回 ('CRCERR', None)；超时返回 None。"""
|
||
t0 = time.monotonic()
|
||
while time.monotonic() - t0 < 5.0:
|
||
f = rd.pop()
|
||
if f is not None:
|
||
cmd, payload = f
|
||
if cmd == "CRCERR":
|
||
return ("CRCERR", None)
|
||
if cmd == want_cmd:
|
||
return (cmd, payload)
|
||
# 其它命令的应答（如残留 KEEP 回显）：跳过继续找
|
||
continue
|
||
try:
|
||
data = s.recv(4096)
|
||
except socket.timeout:
|
||
return None
|
||
if not data:
|
||
return "CLOSED"
|
||
rd.feed(data)
|
||
return None
|
||
|
||
# ---- INIT：查询屏幕参数，顺带把几何钉死 ----
|
||
s.sendall(build_frame(CMD_INIT))
|
||
r = wait_frame(CMD_INIT)
|
||
if r is None or r == "CLOSED" or r[0] == "CRCERR":
|
||
print("[ERR] INIT 无有效应答（%s）" % (r,))
|
||
s.close()
|
||
return 1
|
||
_, payload = r
|
||
if len(payload) >= 8:
|
||
w, h, nbytes = struct.unpack(">HHI", payload[:8])
|
||
print("[INIT] %dx%d frame_bytes=%u" % (w, h, nbytes))
|
||
if w != SCREEN_W or h != SCREEN_H or nbytes != FB_SIZE:
|
||
print("[WARN] INIT 几何与预期 160x160/3200 不符——继续，以设备为准")
|
||
else:
|
||
print("[WARN] INIT 应答长度 %d < 8，跳过几何核对" % len(payload))
|
||
|
||
# ---- LCDMEM 主循环 ----
|
||
n_req = n_resp = n_crc = n_to = 0
|
||
lat_sum = lat_min = lat_max = 0.0
|
||
last_bmp = None
|
||
period = 1.0 / args.hz
|
||
t_end = time.monotonic() + args.seconds
|
||
next_at = time.monotonic()
|
||
while time.monotonic() < t_end:
|
||
now = time.monotonic()
|
||
if now < next_at:
|
||
time.sleep(min(next_at - now, 0.005))
|
||
continue
|
||
t0 = time.monotonic()
|
||
s.sendall(build_frame(CMD_LCDMEM, struct.pack(">I", 0)))
|
||
n_req += 1
|
||
r = wait_frame(CMD_LCDMEM)
|
||
if r == "CLOSED":
|
||
print("[ERR] 第 %d 次拉帧时对端断开" % n_req)
|
||
s.close()
|
||
break
|
||
if r is None:
|
||
n_to += 1
|
||
print("[WARN] #%d LCDMEM 超时（5 s）" % n_req)
|
||
else:
|
||
cmd_got, payload = r
|
||
if cmd_got == "CRCERR":
|
||
n_crc += 1
|
||
print("[WARN] #%d CRC 错" % n_req)
|
||
elif len(payload) != 4 + FB_SIZE:
|
||
n_crc += 1
|
||
print("[WARN] #%d 应答长度 %d != %d" % (n_req, len(payload), 4 + FB_SIZE))
|
||
else:
|
||
dt = time.monotonic() - t0
|
||
n_resp += 1
|
||
lat_sum += dt
|
||
lat_min = dt if lat_min == 0 else min(lat_min, dt)
|
||
lat_max = max(lat_max, dt)
|
||
last_bmp = payload[4:]
|
||
next_at += period
|
||
if next_at < time.monotonic() - 1.0: # 落后太多不追（避免突发风暴）
|
||
next_at = time.monotonic()
|
||
|
||
s.close()
|
||
avg = (lat_sum / n_resp * 1000.0) if n_resp else 0.0
|
||
print("[STAT] 请求 %d / 应答 %d / CRC错 %d / 超时 %d" % (n_req, n_resp, n_crc, n_to))
|
||
print("[STAT] 往返延迟 min %.1f / avg %.1f / max %.1f ms"
|
||
% (lat_min * 1000.0, avg, lat_max * 1000.0))
|
||
if args.dump and last_bmp:
|
||
with open(args.dump, "wb") as f:
|
||
f.write(b"P4\n%d %d\n" % (SCREEN_W, SCREEN_H))
|
||
f.write(last_bmp)
|
||
print("[DUMP] 最后一帧已写 %s（P4 PBM，与屏板帧 B 点阵段同构）" % args.dump)
|
||
ok = (n_to == 0) and (n_crc == 0) and (n_resp > 0)
|
||
print("[VERDICT] %s" % ("PASS" if ok else "FAIL"))
|
||
return 0 if ok else 1
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
||
ap.add_argument("--ip", default=DEFAULT_IP)
|
||
ap.add_argument("--port", type=int, default=DEFAULT_PORT)
|
||
ap.add_argument("--seconds", type=float, default=30.0)
|
||
ap.add_argument("--hz", type=float, default=10.0, help="拉帧节拍（次/秒，默认 10）")
|
||
ap.add_argument("--dump", default="", help="退出前把最后一帧写成的 PBM 路径")
|
||
args = ap.parse_args()
|
||
if args.seconds <= 0 or args.hz <= 0:
|
||
print("[ERR] --seconds / --hz 必须为正")
|
||
return 2
|
||
return run(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|