STM32F4-Base/test/host/ftp_integrity_test.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""E12：CH395F SPI 缓存区 TSL 时序 A/B 的字节级校验工作负载。

背景（E12）：手册 §7.4 规定读写收发缓存区命令需在"长度与数据之间"留
TSL（读 0.4us / 写 0.2us）。本仓 ch395f_write_send_buf / ch395f_read_recv_buf
把命令头与数据打包进**同一 DMA 突发**（零间隔）。本脚本用 SHA-256 逐轮校验
"上传→回读"是否字节一致，为 A/B 提供灵敏判据：
  - STOR（PC→设备）走设备 read_recv_buf（TSL 读 0.4us）
  - RETR（设备→PC）走设备 write_send_buf（TSL 写 0.2us）
任一方向出现比特错位都会让 SHA 不匹配。

用法：
  python ftp_integrity_test.py --rounds 20 --size 65536
  python ftp_integrity_test.py --rounds 20 --size 65536 --tag A   # 结果标记

判据：全部轮次 SHA-256 一致 = PASS；出现不一致则打印首个差异偏移，
并自动二次回读以区分"存储副本错（写路径/上传）"与"回读瞬时错（读路径）"。
"""

import argparse
import hashlib
import io
import random
import sys
import time
from ftplib import FTP, error_perm

import bench_defaults  # 台架连接默认值单一真源

NAME = "e12a.bin"          # 8.3 短名（FatFs FF_USE_LFN=0）


def sha(b: bytes) -> str:
    return hashlib.sha256(b).hexdigest()


def first_diff(a: bytes, b: bytes):
    n = min(len(a), len(b))
    for i in range(n):
        if a[i] != b[i]:
            return i, a[i], b[i]
    if len(a) != len(b):
        return n, None, None
    return None


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default=bench_defaults.DEVICE_IP)
    ap.add_argument("--port", type=int, default=21)
    ap.add_argument("--user", default="admin")
    ap.add_argument("--pwd", default="admin")
    ap.add_argument("--rounds", type=int, default=20)
    ap.add_argument("--size", type=int, default=65536, help="每轮载荷字节数")
    ap.add_argument("--seed", type=int, default=20261013)
    ap.add_argument("--tag", default="", help="结果标记（A/B 分组）")
    args = ap.parse_args()

    rng = random.Random(args.seed)
    ftp = FTP()
    ftp.connect(args.host, args.port, timeout=20)
    ftp.login(args.user, args.pwd)
    ftp.set_pasv(True)
    print(f"[E12{args.tag}] 已登录 {args.host}:{args.port} "
          f"rounds={args.rounds} size={args.size}B", flush=True)

    bad = 0
    t_all = time.time()
    for i in range(1, args.rounds + 1):
        payload = rng.randbytes(args.size)
        want = sha(payload)

        t0 = time.time()
        ftp.storbinary(f"STOR {NAME}", io.BytesIO(payload))

        got = bytearray()
        ftp.retrbinary(f"RETR {NAME}", got.extend)
        got = bytes(got)
        dt = time.time() - t0

        if got == payload:
            print(f"  round {i:2d}/{args.rounds} OK   "
                  f"size={len(got)}B sha={want[:16]} {dt:.2f}s "
                  f"({args.size / 1024 / max(dt, 1e-3):.0f} KB/s)", flush=True)
            continue

        bad += 1
        fd = first_diff(got, payload)
        print(f"  round {i:2d}/{args.rounds} MISMATCH len={len(got)} want={len(payload)} "
              f"first_diff={fd}", flush=True)
        # 归因：二次回读
        got2 = bytearray()
        try:
            ftp.retrbinary(f"RETR {NAME}", got2.extend)
            got2 = bytes(got2)
            if got2 == payload:
                print("       → 二次回读一致：判为回读瞬时错（read 路径）", flush=True)
            else:
                print(f"       → 二次回读仍不一致（sha2={sha(got2)[:16]}）："
                      f"判为存储副本错（write/上传路径）", flush=True)
        except error_perm as e:
            print(f"       → 二次回读失败：{e}", flush=True)

    ftp.quit() if False else ftp.close()
    total = time.time() - t_all
    print(f"[E12{args.tag}] 汇总：{args.rounds - bad}/{args.rounds} 轮字节一致"
          f"（{bad} 轮不一致），总耗时 {total:.1f}s", flush=True)
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())