267 lines
11 KiB
Python
267 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
check_remo_sink.py — 模拟器 RemoDispBus（远端看图）服务端协议回归。
|
||
|
||
验证 simulator 端 ui_net_sink（Winsock TCP 监听，默认 7003）按 RemoDispBus 协议
|
||
正确提供当前 1bpp 屏幕图，供 DTU-RemoteLCD 等远端客户端取图。
|
||
|
||
守护两个历史缺陷（回归门禁，防止回退）：
|
||
1. INIT 污染 —— 连接后不得主动上报 INIT 帧，否则客户端首帧（LCDMEM 应答）
|
||
读取整体错位 → 花屏/无图。判据 A 断言首帧必须是 LCDMEM。
|
||
2. 取帧卡死 —— LCDMEM 取帧不得走 lv_snapshot 离屏重渲染（SDL 窗口 I1+PARTIAL
|
||
下会卡死 lv_draw_dispatch 派发循环 → 无应答超时），须走 SDL_RenderReadPixels
|
||
读回屏幕实际呈现像素。判据 C 断言位图非平凡（有真实内容、非全 0/全 1）。
|
||
|
||
流程：
|
||
1. 清理残留 lvgl_sim；以 CREATE_NEW_PROCESS_GROUP 启动窗口模拟器
|
||
（stdout/stderr→临时文件，断开管道继承；父进程立即返回，不阻塞 shell）。
|
||
注意：不能用 DETACHED_PROCESS（无控制台会让 SDL 建窗异常，进程活着但 7003 不绑定）。
|
||
2. 等 7003 端口就绪。
|
||
3. 连接并发 LCDMEM(offset=0)，读设备应答帧，断言 A/B/C（见上）。
|
||
4. 位图经 i1_to_png 渲染 PNG（默认 5x 最近邻）供目检。
|
||
5. （默认）另起无头参考（lvgl_sim --screenshot，同一初始主菜单）逐位比对，
|
||
<2% 差异判 MATCH —— 强证据：窗口取帧 == 已知正确渲染（连贯 UI，非噪声）。
|
||
6. 清理进程。
|
||
|
||
用法（仓库任意目录均可运行）：
|
||
python simulator/tests/check_remo_sink.py # 全量（含参考比对）
|
||
python simulator/tests/check_remo_sink.py --no-compare # 跳过参考比对
|
||
python simulator/tests/check_remo_sink.py --clean # 成功后删除临时产物
|
||
python simulator/tests/check_remo_sink.py --port 7003 --scale 6
|
||
|
||
前置：simulator 已构建（build/Debug/lvgl_sim.exe 存在）；Pillow 已装（.venv 已含）。
|
||
本测试依赖 Winsock，实际在 Windows 上运行。
|
||
退出码：0 全 PASS；1 协议/比对断言失败；2 运行/环境错误（exe 缺失、端口未就绪等）。
|
||
"""
|
||
import argparse
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
from pathlib import Path
|
||
|
||
# 本脚本位于 simulator/tests/；simulator 目录与 lvgl_sim.exe 据此定位（任意 cwd 可运行）
|
||
SIM_DIR = Path(__file__).resolve().parents[1]
|
||
EXE = SIM_DIR / "build" / "Debug" / "lvgl_sim.exe"
|
||
|
||
# i1_to_png.py 与本脚本同在 simulator/ 树，加入搜索路径后复用其 1bpp->PNG 转换
|
||
sys.path.insert(0, str(SIM_DIR))
|
||
from i1_to_png import i1_to_png # noqa: E402 （模块级仅导入，PIL 在其内部按需加载）
|
||
|
||
W = H = 160
|
||
STRIDE = W // 8 # I1：20 字节/行
|
||
FB = STRIDE * H # 3200 字节（160x160 1bpp）
|
||
|
||
TAG_C, TAG_D = 0xAA, 0xBB
|
||
CMD_KEEPLIVE, CMD_INIT, CMD_KEY, CMD_LCDMEM = 0, 1, 2, 3
|
||
|
||
NEWGRP = 0x200 # Windows CREATE_NEW_PROCESS_GROUP
|
||
REF_DIFF_TOL = 2.0 # 参考比对差异率阈值（%）
|
||
|
||
|
||
def _ensure_utf8_console():
|
||
"""Windows 控制台默认 GBK，中文会乱码；统一切 UTF-8（不支持则静默跳过）。"""
|
||
for stream in (sys.stdout, sys.stderr):
|
||
try:
|
||
stream.reconfigure(encoding="utf-8")
|
||
except (AttributeError, ValueError):
|
||
pass
|
||
|
||
|
||
def _kill_name(name):
|
||
if sys.platform == "win32":
|
||
subprocess.run(["taskkill", "/F", "/IM", name + ".exe"],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
else:
|
||
subprocess.run(["pkill", "-f", name],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||
|
||
|
||
def _start_sim(tmp):
|
||
"""启动窗口模拟器（stdout/stderr→文件，NEWGRP 断开管道继承，非阻塞）。"""
|
||
out = open(os.path.join(tmp, "sim_out.txt"), "wb")
|
||
err = open(os.path.join(tmp, "sim_err.txt"), "wb")
|
||
kwargs = {}
|
||
if sys.platform == "win32":
|
||
kwargs["creationflags"] = NEWGRP
|
||
p = subprocess.Popen([str(EXE)], stdout=out, stderr=err,
|
||
stdin=subprocess.DEVNULL, cwd=str(SIM_DIR), **kwargs)
|
||
return p
|
||
|
||
|
||
def _wait_port(port, timeout=25.0):
|
||
t0 = time.time()
|
||
while time.time() - t0 < timeout:
|
||
try:
|
||
s = socket.create_connection(("127.0.0.1", port), 0.5)
|
||
s.close()
|
||
return True
|
||
except OSError:
|
||
time.sleep(0.4)
|
||
return False
|
||
|
||
|
||
def _crc(d):
|
||
c = 0
|
||
for b in d:
|
||
c ^= b
|
||
return c
|
||
|
||
|
||
def _recv_exact(s, n, deadline):
|
||
buf = b""
|
||
while len(buf) < n:
|
||
if time.time() > deadline:
|
||
raise TimeoutError("recv 超时")
|
||
chunk = s.recv(n - len(buf))
|
||
if not chunk:
|
||
raise ConnectionError("连接被对端关闭")
|
||
buf += chunk
|
||
return buf
|
||
|
||
|
||
def _read_frame(s, deadline):
|
||
hdr = _recv_exact(s, 4, deadline)
|
||
tag, cmd = hdr[0], hdr[1]
|
||
n = (hdr[2] << 8) | hdr[3]
|
||
body = _recv_exact(s, n + 1, deadline)
|
||
data, crcb = body[:n], body[n]
|
||
ok = (tag == TAG_D) and (crcb == _crc(data))
|
||
return cmd, data, ok
|
||
|
||
|
||
def _bits(data):
|
||
return sum(bin(b).count("1") for b in data)
|
||
|
||
|
||
def _dump_err(path, tag):
|
||
if os.path.exists(path):
|
||
txt = open(path, encoding="utf-8", errors="replace").read().strip()
|
||
if txt:
|
||
print(" --- sim_err.txt（%s）---" % tag)
|
||
for ln in txt.splitlines()[-15:]:
|
||
print(" ", ln)
|
||
|
||
|
||
def main(argv=None):
|
||
_ensure_utf8_console()
|
||
ap = argparse.ArgumentParser(description="模拟器 RemoDispBus 服务端协议回归")
|
||
ap.add_argument("--port", type=int, default=7003, help="RemoDispBus 监听端口")
|
||
ap.add_argument("--no-compare", action="store_true", help="跳过无头参考逐位比对")
|
||
ap.add_argument("--scale", type=int, default=5, help="目检 PNG 放大倍数（最近邻）")
|
||
ap.add_argument("--clean", action="store_true", help="成功后删除临时产物（默认保留 PNG 供目检）")
|
||
args = ap.parse_args(argv)
|
||
|
||
if not EXE.exists():
|
||
print("FAIL: 找不到 %s（先构建 simulator：cmake --build simulator/build --config Debug）" % EXE)
|
||
return 2
|
||
|
||
tmp = tempfile.mkdtemp(prefix="dtu_remo_")
|
||
sim = None
|
||
try:
|
||
_kill_name("lvgl_sim")
|
||
time.sleep(1.0)
|
||
sim = _start_sim(tmp)
|
||
print("[1] 已启动模拟器 PID=%d（stdout->文件，不阻塞 shell）" % sim.pid)
|
||
|
||
if not _wait_port(args.port):
|
||
print("[2] FAIL: 端口 %d 未就绪" % args.port)
|
||
_dump_err(os.path.join(tmp, "sim_err.txt"), "port")
|
||
return 2
|
||
print("[2] 端口 %d 就绪" % args.port)
|
||
|
||
s = socket.create_connection(("127.0.0.1", args.port), 3)
|
||
s.settimeout(4)
|
||
deadline = time.time() + 6
|
||
|
||
# 客户端 -> 设备：LCDMEM(offset=0)
|
||
req = bytes([TAG_C, CMD_LCDMEM, 0, 4, 0, 0, 0, 0, _crc(b"\x00\x00\x00\x00")])
|
||
s.sendall(req)
|
||
try:
|
||
cmd, data, ok = _read_frame(s, deadline)
|
||
except (TimeoutError, ConnectionError) as e:
|
||
print(" FAIL: 设备无应答（%s）——取帧卡死/读回失败" % e)
|
||
s.close()
|
||
_dump_err(os.path.join(tmp, "sim_err.txt"), "noresp")
|
||
return 1
|
||
s.close()
|
||
|
||
fails = []
|
||
# 判据 A：首帧 = LCDMEM（INIT 污染已除）
|
||
if cmd != CMD_LCDMEM or not ok:
|
||
fails.append("首帧 cmd=0x%02X crc_ok=%s（期望 LCDMEM 0x%02X）" % (cmd, ok, CMD_LCDMEM))
|
||
bitmap = data[4:4 + FB]
|
||
# 判据 B：位图 3200 字节
|
||
if len(bitmap) != FB:
|
||
fails.append("位图长度 %d（期望 %d）" % (len(bitmap), FB))
|
||
bitmap = bitmap.ljust(FB, b"\x00")[:FB]
|
||
# 判据 C：非平凡
|
||
bits = _bits(bitmap)
|
||
total = W * H
|
||
if not (0 < bits < total):
|
||
fails.append("位图平凡（置位 %d/%d，全 0/全 1 或取帧失败）" % (bits, total))
|
||
|
||
if fails:
|
||
for m in fails:
|
||
print(" FAIL:", m)
|
||
_dump_err(os.path.join(tmp, "sim_err.txt"), "proto")
|
||
return 1
|
||
|
||
print(" 判据 A OK：首帧=LCDMEM（INIT 污染已除），CRC 通过")
|
||
print(" 判据 B OK：位图=%d 字节（addr=%s）" % (len(bitmap), data[:4].hex()))
|
||
print(" 判据 C OK：非平凡（置位 %d/%d，%.2f%%）" % (bits, total, bits / total * 100.0))
|
||
|
||
# 目检 PNG（复用 i1_to_png，最近邻放大；白=墨/黑=底）
|
||
bin_path = os.path.join(tmp, "proto_screen.bin")
|
||
png_path = os.path.join(tmp, "proto_screen.png")
|
||
with open(bin_path, "wb") as f:
|
||
f.write(bitmap)
|
||
i1_to_png(bin_path, output_path=png_path, width=W, height=H, scale=args.scale)
|
||
print(" PNG: %s" % png_path)
|
||
|
||
# 参考比对：窗口取帧 vs 无头参考（同一初始主菜单，独立进程，互不干扰）
|
||
if not args.no_compare:
|
||
ref_path = os.path.join(tmp, "ref_headless.bin")
|
||
r = subprocess.run([str(EXE), "--screenshot", ref_path],
|
||
cwd=str(SIM_DIR), capture_output=True, timeout=90)
|
||
if r.returncode != 0 or not os.path.exists(ref_path):
|
||
print(" 参考比对：跳过（无头参考生成失败 rc=%d）" % r.returncode)
|
||
else:
|
||
ref = open(ref_path, "rb").read()
|
||
n = min(len(ref), len(bitmap))
|
||
diff = sum(bin(ref[i] ^ bitmap[i]).count("1") for i in range(n))
|
||
pct = diff / total * 100.0
|
||
if pct < REF_DIFF_TOL:
|
||
print(" 参考比对 MATCH：与无头参考差异 %d/%d 位（%.3f%%）→ 连贯 UI 确认"
|
||
% (diff, total, pct))
|
||
else:
|
||
fails.append("参考比对 MISMATCH：差异 %.2f%%（>=%.0f%%）" % (pct, REF_DIFF_TOL))
|
||
print(" FAIL:", fails[-1])
|
||
|
||
if fails:
|
||
print("FAIL: %d 项未过" % len(fails))
|
||
return 1
|
||
|
||
print("RESULT: PASS（协议 + 取帧 + 参考比对全过；请目检 PNG 确认）")
|
||
if args.clean:
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
print("（--clean：已删除临时产物 %s）" % tmp)
|
||
else:
|
||
print("临时产物保留于：%s（PNG=%s）" % (tmp, png_path))
|
||
return 0
|
||
finally:
|
||
if sim is not None:
|
||
try:
|
||
sim.kill()
|
||
sim.wait(timeout=5)
|
||
except Exception:
|
||
_kill_name("lvgl_sim")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|