448 lines
17 KiB
Python
448 lines
17 KiB
Python
# -*- 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
|
||
# BLE-MIDI 粘包时可能再次出现 F0；重新同步，避免 F0 F0 60…
|
||
if b == HEAD:
|
||
self.buf = bytearray([HEAD])
|
||
continue
|
||
self.buf.append(b)
|
||
if b == TAIL or len(self.buf) >= 64:
|
||
if b == TAIL and len(self.buf) >= 4:
|
||
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 fmt_ver4(payload: bytes, *, kind: str = "semver") -> str:
|
||
"""解码版本字节为可读串。
|
||
|
||
kind:
|
||
semver — 前 3 字节 major.minor.patch（01 03/0C）
|
||
date — YY.M.D 日期（01 04 音源 / 01 05 UI），并展开为 20YY-MM-DD
|
||
"""
|
||
if len(payload) != 4:
|
||
return f"ver_raw={payload!r}"
|
||
a, b, c, _d = payload
|
||
if kind == "date":
|
||
return f"ver={a}.{b}.{c} date=20{a:02d}-{b:02d}-{c:02d}"
|
||
return f"ver={a}.{b}.{c}"
|
||
|
||
|
||
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, kind in (
|
||
(0x03, "固件主版本", "semver"),
|
||
(0x04, "音源版本", "date"),
|
||
(0x05, "UI版本", "date"),
|
||
):
|
||
r = sim.query(build(0x01, sub))
|
||
ok, d = expect(r, (0x60, 0x01, sub), 9)
|
||
if ok:
|
||
d += f" {fmt_ver4(bytes(r[4:-1]), kind=kind)}"
|
||
R.add(f"01 {sub:02X} {name}", "PASS" if ok else "FAIL", d)
|
||
|
||
# 01 0C：{branch}_{short6}[optional '*']，如 develop_0aedb4
|
||
r = sim.query(build(0x01, 0x0C))
|
||
ok, d = expect(r, (0x60, 0x01, 0x0C))
|
||
if ok:
|
||
raw = bytes(r[4:-1])
|
||
ok = (
|
||
3 <= len(raw) <= 25
|
||
and all(0x20 <= c < 0x7F for c in raw)
|
||
and b"_" in raw.rstrip(b"*")
|
||
)
|
||
build_s = raw.decode("ascii", "replace")
|
||
dirty = build_s.endswith("*")
|
||
core = build_s[:-1] if dirty else build_s
|
||
branch, _, short = core.rpartition("_")
|
||
extra = f" build={build_s!r}"
|
||
if len(short) == 6 and all(c in "0123456789abcdef" for c in short.lower()):
|
||
extra += f" branch={branch!r} commit={short.lower()}"
|
||
if dirty:
|
||
extra += " dirty=1"
|
||
d += extra
|
||
R.add("01 0C 用户固件版本", "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), 29) # 12B UID → 24 hex ASCII
|
||
if ok:
|
||
hx = bytes(r[4:-1])
|
||
ok = len(hx) == 24 and all(c in b"0123456789ABCDEF" for c in hx)
|
||
d += f" uid_hex={hx.decode('ascii', 'replace')}"
|
||
R.add("01 07 设备编码(UID)", "PASS" if ok else "FAIL", d)
|
||
|
||
# 01 0F：BLE-MIDI 安全子命令（原 01 FF 的 0xFF 非法出现在 SysEx 数据中）
|
||
r = sim.query(build(0x01, 0x0F))
|
||
ok, d = expect(r, (0x60, 0x01, 0x0F))
|
||
if ok:
|
||
code = bytes(r[4:-1])
|
||
ok = len(code) > 0 and all(0x20 <= c < 0x7F for c in code)
|
||
d += f" code={code!r}"
|
||
R.add("01 0F 固件编码", "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:
|
||
time.sleep(0.3)
|
||
# 禁止 query 自动重发：第二次会打在已断电的 BT 上
|
||
sim.drain()
|
||
sim.send(build(0x05, 0x00))
|
||
r = sim.read_frame(timeout=3.0)
|
||
ok, d = expect(r, (0x60, 0x05, 0x00), 6)
|
||
if not ok and r is None and hasattr(sim, "is_connected"):
|
||
# ACK 可能被软关机掐断 BT 抢走；链路断开也视为关机已执行
|
||
time.sleep(0.5)
|
||
if not sim.is_connected():
|
||
ok, d = True, "no ACK but BLE link dropped (soft-off likely)"
|
||
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()
|