STM32F4-Base/test/host/remolcd_pull_test.py

#!/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())