super_agent_bot
b0a6394aa1
- test/host/udp_queue_bench.py：Windows 上未 connect 的 UDP socket 调 getsockname() 直接抛 WinError 10022（请求一个未连接套接句的名称），脚本在第一条用例就崩。 源端口改为首次收到回显时打印——它本来就是「板端剥头/源地址判定对不对」的直接读数。 - 新增 test/host/udp_queue_bench_loopback.py：不用板子的 peer 自检，在 127.0.0.1 上放 一个按板端阶段 16 同样三条规矩说话的假板子（原样回显、tag 0x04 只回 64 B、>1KB 按 1024 拆片），先把步骤表 / 模式函数 / 分片预期验掉。2026-11-03 首跑 27/27 绿， 实测拆片形状 [1024, 1] 与 [1024, 448]。 - CH395F_Test_Guide.md §12.6：运行方法补「先跑自检」这一步；>1024 B 的 UDP 拆片从 「多收到一片就 FAIL」改成「实测观测量」（拼接后逐字节一致才算绿），并说清这是既有 NET_SEND_CHUNK_MAX 行为、不是本轮引入；脚本表补一行。 - 板子自 19:19 起未再上电（ping 不通、SWD No target、串口静默），阶段 16 真机那一趟 仍未跑，PR #74 的合并前提不变。
90 lines
3.6 KiB
Python
90 lines
3.6 KiB
Python
"""PC peer 自身的回路自检：不需要板子，用假板子把 udp_queue_bench.py 的步骤表跑一遍。
|
||
|
||
为什么要有这个文件：阶段 16 的真机台架是一次性的——板子通电、烧好台架镜像、PC 起对端，
|
||
对端脚本自己要是有 bug（取端口、步骤表、拆包预期），一趟台架时间就白费了。这里在
|
||
127.0.0.1 上放一个「按板端阶段 16 同样规矩说话」的假板子，把对端脚本的判定逻辑先验一遍。
|
||
|
||
假板子复刻的三条规矩（与 test/board/net_test_task.c 阶段 16 一致）：
|
||
1. 收到的数据报原样回显给发送方（tag 字节不动，所以源地址/剥头判据留在板端断言）；
|
||
2. tag 0x04 这一步板端故意只用 64 字节缓冲去读 200 字节的数据报，所以只回 64 字节；
|
||
3. 回显超过芯片发送缓冲（1KB）时按 1024 拆成多个数据报发出（NET_SEND_CHUNK_MAX）。
|
||
|
||
用法（在本仓库根目录）：
|
||
python test/host/udp_queue_bench_loopback.py
|
||
退出码：0 = 对端脚本判定逻辑全绿；1 = 有问题，先别上台架。
|
||
"""
|
||
|
||
import os
|
||
import socket
|
||
import sys
|
||
import threading
|
||
import time
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
if HERE not in sys.path:
|
||
sys.path.insert(0, HERE)
|
||
|
||
import udp_queue_bench as peer # noqa: E402
|
||
|
||
CHIP_SEND_BUF = 1024 # NET_SEND_CHUNK_MAX
|
||
BOARD_RX_CAP = 1600 # 板端 s_udp_buf 尺寸 UDP_BENCH_BUF_SIZE
|
||
TRUNCATE_TAG = 0x04 # 这一步板端只用 64 字节缓冲
|
||
TRUNCATE_CAP = 64
|
||
RUNOUT_S = 90.0
|
||
LISTEN_PORTS = (6001, 6002) # UDP_BENCH_PORT / UDP_BENCH_PORT2
|
||
|
||
|
||
def echo_one(sock, deadline, stop):
|
||
sock.settimeout(0.5)
|
||
while (not stop.is_set()) and (time.time() < deadline):
|
||
try:
|
||
data, addr = sock.recvfrom(4096)
|
||
except socket.timeout:
|
||
continue
|
||
except OSError:
|
||
return
|
||
if not data:
|
||
continue
|
||
tag = data[0]
|
||
cap = TRUNCATE_CAP if tag == TRUNCATE_TAG else BOARD_RX_CAP
|
||
echo = peer.payload(tag, min(len(data), cap))
|
||
print("[board] rx 0x{0:02X} n={1} on :{2} -> echo {3}B".format(
|
||
tag, len(data), sock.getsockname()[1], len(echo)), flush=True)
|
||
for off in range(0, len(echo), CHIP_SEND_BUF):
|
||
sock.sendto(echo[off:off + CHIP_SEND_BUF], addr)
|
||
|
||
|
||
def main() -> int:
|
||
if hasattr(sys.stdout, "reconfigure"):
|
||
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
||
deadline = time.time() + RUNOUT_S
|
||
stop = threading.Event()
|
||
for listen_port in LISTEN_PORTS:
|
||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
try:
|
||
s.bind(("127.0.0.1", listen_port))
|
||
except OSError as exc:
|
||
print("无法绑定 127.0.0.1:{0}：{1}（端口被占用？先腾出来再跑）".format(listen_port, exc),
|
||
file=sys.stderr)
|
||
return 2
|
||
threading.Thread(target=echo_one, args=(s, deadline, stop), daemon=True).start()
|
||
time.sleep(0.3)
|
||
|
||
# --late 必须落在板端 7000~25000 ms 判定窗内；这里取 9 秒，比默认 12 秒省一点台架外时间
|
||
sys.argv = ["udp_queue_bench.py", "--board", "127.0.0.1",
|
||
"--first-wait", "15", "--timeout", "6",
|
||
"--probe-silence", "2.0", "--late", "9.0"]
|
||
try:
|
||
rc = peer.main()
|
||
finally:
|
||
stop.set()
|
||
if rc == 0:
|
||
print("=== UDP Peer Loopback Self-Test: PASS（对端脚本判定逻辑可用，可上台架）===", flush=True)
|
||
else:
|
||
print("=== UDP Peer Loopback Self-Test: FAIL（先修对端脚本，别浪费台架时间）===", flush=True)
|
||
return rc
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|