K1Guitar/tools/test_protocol_app_sim.py

# -*- coding: utf-8 -*-
"""
test_protocol_app_sim.py — 模拟手机 App 对吉他固件做全协议串口测试

协议来源: Doc/指令测试.docx (= 指令测试0907.docx)
通道:     UART4 (BLE 桥), 115200 8N1; 帧格式 F0 60 <cmd1> <cmd2> [data...] F7
          设备主动上报为 F0 51 ... F7 (测试时跳过, 不计为响应)

用法:
  python test_protocol_app_sim.py --port COM5                 # 跑全部安全用例
  python test_protocol_app_sim.py --port COM5 --allow-poweroff# 含 05 00 关机用例
  python test_protocol_app_sim.py --selftest                  # 无硬件自检(帧编解码)

依赖: pyserial (pip install pyserial)
退出码: 0 = 全部通过, 1 = 有 FAIL
"""
import argparse
import sys
import time

try:
    import serial
except ImportError:
    serial = None  # --selftest 不需要 pyserial

HEAD, TAIL, APP_ID, DEV_ID = 0xF0, 0xF7, 0x60, 0x51


# ---------------------------------------------------------------- frame codec
def build(*payload):
    """App -> 设备: F0 60 <payload...> F7"""
    return bytes([HEAD, APP_ID, *payload, TAIL])


class FrameParser:
    """字节流 -> 完整帧; 只收 F0...F7, 其余丢弃"""

    def __init__(self):
        self.buf = bytearray()

    def feed(self, data: bytes):
        """返回本轮解析出的完整帧列表"""
        out = []
        for b in data:
            if not self.buf:
                if b == HEAD:
                    self.buf.append(b)
                continue
            self.buf.append(b)
            if b == TAIL or len(self.buf) >= 64:
                if b == TAIL:
                    out.append(bytes(self.buf))
                self.buf.clear()
        return out


def hexs(b):
    return " ".join(f"{x:02X}" for x in b) if b else "(none)"


# ---------------------------------------------------------------- serial wrap
class AppSim:
    def __init__(self, port, baud=115200, timeout=0.15):
        self.ser = serial.Serial(port, baud, bytesize=8, parity="N",
                                 stopbits=1, timeout=timeout)
        self.parser = FrameParser()
        self.pending = []

    def close(self):
        self.ser.close()

    def _pump(self):
        data = self.ser.read(256)
        if data:
            self.pending.extend(self.parser.feed(data))

    def send(self, frame: bytes):
        self.ser.write(frame)
        self.ser.flush()

    def read_frame(self, timeout=1.0, want_dev=False):
        """读一帧; 默认跳过设备主动上报(0x51), want_dev=True 时只要 0x51"""
        deadline = time.monotonic() + timeout
        while True:
            self._pump()
            for i, f in enumerate(self.pending):
                is_dev = len(f) > 1 and f[1] == DEV_ID
                if is_dev == want_dev:
                    return self.pending.pop(i)
            self.pending.clear()  # 丢掉方向不符的帧
            if time.monotonic() >= deadline:
                return None
            time.sleep(0.01)

    def drain(self, quiet=0.2):
        while self.read_frame(timeout=quiet) is not None:
            pass

    def query(self, frame: bytes, timeout=1.0):
        self.drain()
        self.send(frame)
        return self.read_frame(timeout=timeout)


# ---------------------------------------------------------------- test engine
class Results:
    def __init__(self):
        self.rows = []

    def add(self, name, status, detail=""):
        self.rows.append((name, status, detail))
        print(f"[{status:>4}] {name}" + (f"  | {detail}" if detail else ""))

    def summary(self):
        n = {s: sum(1 for r in self.rows if r[1] == s)
             for s in ("PASS", "FAIL", "SKIP", "SENT")}
        print("\n===== 汇总 =====")
        for k in ("PASS", "FAIL", "SKIP", "SENT"):
            print(f"{k:>5}: {n[k]}")
        for name, status, detail in self.rows:
            if status == "FAIL":
                print(f"  FAIL: {name} | {detail}")
        return 1 if n["FAIL"] else 0


def expect(resp, prefix, length=None):
    """校验响应帧头/命令字/长度"""
    if resp is None:
        return False, "timeout, no reply"
    if len(resp) < 5 or resp[0] != HEAD or resp[-1] != TAIL:
        return False, f"bad frame: {hexs(resp)}"
    if tuple(resp[1:1 + len(prefix)]) != tuple(prefix):
        return False, f"prefix mismatch: {hexs(resp)}"
    if length is not None and len(resp) != length:
        return False, f"len {len(resp)} != {length}: {hexs(resp)}"
    return True, hexs(resp)


def run_tests(sim, allow_poweroff=False):
    R = Results()

    # ---------------- 1. 设备信息 0x01 ----------------
    r = sim.query(build(0x01, 0x01))
    ok, d = expect(r, (0x60, 0x01, 0x01), 6)
    R.add("01 01 连接设备", "PASS" if ok else "FAIL", d)

    r = sim.query(build(0x01, 0x02))
    ok, d = expect(r, (0x60, 0x01, 0x02))
    if ok:
        name = bytes(r[4:-1])
        ok = all(0x20 <= c < 0x7F for c in name)
        d += f"  name={name!r}"
    R.add("01 02 设备名", "PASS" if ok else "FAIL", d)

    for sub, name in ((0x03, "固件主版本"), (0x04, "音源版本"),
                      (0x05, "UI版本"), (0x0C, "用户固件版本")):
        r = sim.query(build(0x01, sub))
        ok, d = expect(r, (0x60, 0x01, sub), 9)
        R.add(f"01 {sub:02X} {name}", "PASS" if ok else "FAIL", d)

    r = sim.query(build(0x01, 0x06))
    ok, d = expect(r, (0x60, 0x01, 0x06), 13)
    R.add("01 06 其他信息", "PASS" if ok else "FAIL", d)

    r = sim.query(build(0x01, 0x07))
    ok, d = expect(r, (0x60, 0x01, 0x07), 17)
    R.add("01 07 设备编码(UID)", "PASS" if ok else "FAIL", d)

    r = sim.query(build(0x01, 0xFF))
    ok, d = expect(r, (0x60, 0x01, 0xFF))
    R.add("01 FF 固件编码", "PASS" if ok else "FAIL", d)

    # 自动关机: 读 -> 设15 -> 读验证 -> 恢复原值
    r0 = sim.query(build(0x01, 0x0A))
    ok, d = expect(r0, (0x60, 0x01, 0x0A), 6)
    R.add("01 0A 自动关机(读)", "PASS" if ok else "FAIL", d)
    if ok:
        orig = r0[4]
        r = sim.query(build(0x01, 0x11, 15))
        ok, d = expect(r, (0x60, 0x01, 0x11, 15), 6)
        R.add("01 11 自动关机(设15)", "PASS" if ok else "FAIL", d)
        r = sim.query(build(0x01, 0x0A))
        ok, d = expect(r, (0x60, 0x01, 0x0A, 15), 6)
        R.add("01 0A 回读=15", "PASS" if ok else "FAIL", d)
        sim.query(build(0x01, 0x11, orig))  # restore

    r = sim.query(build(0x01, 0x00))
    ok, d = expect(r, (0x60, 0x01, 0x00), 6)
    R.add("01 00 断开设备", "PASS" if ok else "FAIL", d)
    sim.query(build(0x01, 0x01))  # 重新连接, 便于后续用例

    # ---------------- 2. 和弦映射 0x02 ----------------
    r = sim.query(build(0x02, 0x01))
    ok, d = expect(r, (0x60, 0x02, 0x01), 47)
    R.add("02 01 读和弦映射表", "PASS" if ok else "FAIL", d)
    cur_map = bytes(r[4:-1]) if ok else None  # 21B pitch + 21B chord

    if cur_map:
        key = 1
        orig_pitch, orig_chord = cur_map[key - 1], cur_map[21 + key - 1]

        sim.send(build(0x02, 0x02, key - 1, 0x02))   # pitch: 升位
        time.sleep(0.1)
        r = sim.query(build(0x02, 0x01))
        ok = r is not None and len(r) == 47 and r[3] == 0x01 and r[4 + key - 1] == 0x02
        R.add("02 02 Pitch偏移(写+读回)", "PASS" if ok else "FAIL", hexs(r))
        sim.send(build(0x02, 0x02, key - 1, orig_pitch))  # restore

        sim.send(build(0x02, 0x03, key - 1, 0x08))   # chord: 小三
        time.sleep(0.1)
        r = sim.query(build(0x02, 0x01))
        ok = r is not None and len(r) == 47 and r[4 + 21 + key - 1] == 0x08
        R.add("02 03 Chord偏移(写+读回)", "PASS" if ok else "FAIL", hexs(r))
        sim.send(build(0x02, 0x03, key - 1, orig_chord))  # restore
        time.sleep(0.1)

        # 02 04 整表写入(回写当前表) -> 设备应回 02 01 格式整表
        r = sim.query(build(0x02, 0x04, *cur_map), timeout=2.0)
        ok, d = expect(r, (0x60, 0x02, 0x01), 47)
        R.add("02 04 整表写入(回读帧)", "PASS" if ok else "FAIL", d)
        r2 = sim.query(build(0x02, 0x01))
        ok = r2 is not None and bytes(r2[4:-1]) == cur_map
        R.add("02 04 写入后整表一致", "PASS" if ok else "FAIL", hexs(r2))
    else:
        for n in ("02 02 Pitch偏移", "02 03 Chord偏移", "02 04 整表写入"):
            R.add(n, "SKIP", "读映射表失败, 级联跳过")

    # ---------------- 3. 吉他参数 0x03 ----------------
    r = sim.query(build(0x03, 0x04, 0x00))
    ok, d = expect(r, (0x60, 0x03, 0x04, 0x00), 9)
    R.add("03 04 读当前节奏风格", "PASS" if ok else "FAIL", d)
    saved_style = bytes(r[5:8]) if ok else None

    # 开机特例: F0 60 03 04 01 00 00 00 F7 -> F0 60 03 04 01 <bpm> F7 + F0 51 03 <hi> <lo> F7
    r = sim.query(build(0x03, 0x04, 0x01, 0x00, 0x00, 0x00))
    ok, d = expect(r, (0x60, 0x03, 0x04, 0x01), 7)
    R.add("03 04 开机特例(读风格#0)", "PASS" if ok else "FAIL", d)
    r51 = sim.read_frame(timeout=1.0, want_dev=True)
    ok51 = (r51 is not None and len(r51) == 6
            and tuple(r51[:3]) == (HEAD, DEV_ID, 0x03))
    R.add("03 04 开机特例(F0 51 BPM上报)",
          "PASS" if ok51 else "FAIL", hexs(r51))

    r = sim.query(build(0x03, 0x04, 0x01, 0x01, 0x00, 0x00))
    ok, d = expect(r, (0x60, 0x03, 0x04, 0x01), 7)
    R.add("03 04 设用户风格#0", "PASS" if ok else "FAIL", d)

    # 用户风格分页: 回复 F0 60 03 04 02 <cntHi> <cntLo> [<idHi> <idLo> <code4B> ...] F7
    r = sim.query(build(0x03, 0x04, 0x02, 0x00, 0x00, 0x00, 0x07))
    ok, d = expect(r, (0x60, 0x03, 0x04, 0x02))
    if ok:
        cnt = r[5] * 128 + r[6]
        ok = len(r) == 7 + cnt * 6 + 1
        d += f"  cnt={cnt}"
    R.add("03 04 用户风格分页", "PASS" if ok else "FAIL", d)

    # 用户风格总数: 回复 F0 60 03 04 00 <total 3B BE> F7
    r = sim.query(build(0x03, 0x04, 0x03))
    ok, d = expect(r, (0x60, 0x03, 0x04, 0x00), 9)
    if ok:
        d += f"  total={(r[5] << 16) | (r[6] << 8) | r[7]}"
    R.add("03 04 用户风格总数", "PASS" if ok else "FAIL", d)

    if saved_style:
        sim.query(build(0x03, 0x04, 0x01, *saved_style))  # restore style

    # 弦音色: 读 -> 写 -> 读 -> 恢复
    r0 = sim.query(build(0x03, 0x05, 0x00))
    ok, d = expect(r0, (0x60, 0x03, 0x05, 0x00), 8)
    R.add("03 05 读弦音色", "PASS" if ok else "FAIL", d)
    if ok:
        orig = r0[6]
        new = (orig + 1) % 5
        sim.send(build(0x03, 0x05, 0x01, 0x00, new))
        time.sleep(0.1)
        r = sim.query(build(0x03, 0x05, 0x00))
        ok = r is not None and len(r) == 8 and r[6] == new
        R.add("03 05 写弦音色(写+读回)", "PASS" if ok else "FAIL", hexs(r))
        sim.send(build(0x03, 0x05, 0x01, 0x00, orig))  # restore

    # BPM: 读 -> 写120 -> 读 -> 恢复
    r0 = sim.query(build(0x03, 0x06, 0x00))
    ok, d = expect(r0, (0x60, 0x03, 0x06, 0x00), 8)
    R.add("03 06 读BPM", "PASS" if ok else "FAIL", d)
    if ok:
        orig = r0[5] * 128 + r0[6]
        sim.send(build(0x03, 0x06, 0x01, 120 // 128, 120 % 128))
        time.sleep(0.1)
        r = sim.query(build(0x03, 0x06, 0x00))
        ok = r is not None and len(r) == 8 and (r[5] * 128 + r[6]) == 120
        R.add("03 06 写BPM=120(写+读回)", "PASS" if ok else "FAIL", hexs(r))
        sim.send(build(0x03, 0x06, 0x01, orig // 128, orig % 128))  # restore

    # 移调: 读 -> 写 -> 读 -> 恢复 (doc: F0 60 03 07 00 <val> F7, 7 bytes)
    r0 = sim.query(build(0x03, 0x07, 0x00))
    ok, d = expect(r0, (0x60, 0x03, 0x07, 0x00), 7)
    R.add("03 07 读移调", "PASS" if ok else "FAIL", d)
    if ok:
        orig = r0[5]
        new = (orig + 1) % 12
        sim.send(build(0x03, 0x07, 0x01, new))
        time.sleep(0.1)
        r = sim.query(build(0x03, 0x07, 0x00))
        ok = r is not None and len(r) == 7 and r[5] == new
        R.add("03 07 写移调(写+读回)", "PASS" if ok else "FAIL", hexs(r))
        sim.send(build(0x03, 0x07, 0x01, orig))  # restore

    # ---------------- 4. LED / 播放控制 0x04 (无应答, 仅下发) ----------------
    for i in range(7):
        sim.drain()
        sim.send(build(0x04, i, 0x02, 0x02))
        r = sim.read_frame(timeout=0.3)
        R.add(f"04 0{i} 点亮LED{i + 1}", "SENT" if r is None else "PASS",
              "" if r is None else f"unexpected reply {hexs(r)}")
        time.sleep(0.05)

    sim.drain()
    sim.send(build(0x04, 0x07, 0x00, 0x00))
    R.add("04 07 End/结束播放", "SENT", "无应答属正常")

    # ---------------- 6. 段落跳转 0x06 (无应答, 仅下发) ----------------
    for sub, name in ((0x01, "前奏"), (0x02, "间奏"), (0x03, "尾奏"),
                      (0x05, "A段"), (0x06, "B段"), (0x07, "C段"), (0x08, "D段")):
        sim.drain()
        sim.send(build(0x06, sub, 0x00))
        R.add(f"06 {sub:02X} {name}", "SENT", "无应答属正常")
        time.sleep(0.05)

    # ---------------- 5. 复位/关机 0x05 ----------------
    if allow_poweroff:
        r = sim.query(build(0x05, 0x00), timeout=2.0)
        ok, d = expect(r, (0x60, 0x05, 0x00), 6)
        R.add("05 00 复位/关机", "PASS" if ok else "FAIL", d + " (设备将软关机)")
    else:
        R.add("05 00 复位/关机", "SKIP", "需 --allow-poweroff")

    return R


# ---------------------------------------------------------------- selftest
def selftest():
    R = Results()
    f = build(0x03, 0x06, 0x01, 0x00, 0x40)
    R.add("build frame", "PASS" if f == bytes.fromhex("F060030601004 0F7".replace(" ", "")) else "FAIL", hexs(f))

    p = FrameParser()
    out = p.feed(bytes.fromhex("AA F0 60 01 01")) + p.feed(bytes.fromhex("01 F7"))
    R.add("parser chunked", "PASS" if out == [bytes.fromhex("F0600101 01F7".replace(" ", ""))] else "FAIL",
          str([hexs(x) for x in out]))

    p = FrameParser()
    out = p.feed(bytes.fromhex("F0 51 05 0A F7 F0 60 01 01 01 F7".replace(" ", "")))
    R.add("parser two frames", "PASS" if len(out) == 2 else "FAIL", str(len(out)))
    return R.summary()


def main():
    ap = argparse.ArgumentParser(description="模拟手机App的吉他协议全量测试")
    ap.add_argument("--port", help="串口, 如 COM5 (UART4 115200 8N1)")
    ap.add_argument("--baud", type=int, default=115200)
    ap.add_argument("--allow-poweroff", action="store_true",
                    help="允许执行 05 00 关机用例")
    ap.add_argument("--selftest", action="store_true", help="无硬件自检")
    args = ap.parse_args()

    if args.selftest:
        sys.exit(selftest())
    if not args.port:
        ap.error("需要 --port (或用 --selftest)")
    if serial is None:
        sys.exit("缺少 pyserial: pip install pyserial")

    sim = AppSim(args.port, args.baud)
    print(f"=== 协议测试开始 {args.port}@{args.baud} ===")
    try:
        code = run_tests(sim, allow_poweroff=args.allow_poweroff).summary()
    finally:
        sim.close()
    sys.exit(code)


if __name__ == "__main__":
    main()