# -*- coding: utf-8 -*- """ test_protocol_ble.py — 通过笔记本蓝牙(BLE-MIDI)对吉他做全协议测试并出报告 协议来源: Doc/指令测试.docx 链路: 笔记本 BLE <-> "Smart Guitar MIDI" (BLE-MIDI) <-> 吉他 UART4 App->设备: F0 60 ... F7 ; 设备主动上报: F0 51 ... F7 BLE-MIDI: service 03B80E5A-EDE8-4B33-A751-6CE34EC4C700 characteristic 7772E5DB-3868-4112-A1A9-F2669D106BF3 (notify + write-no-resp) 用法: python test_protocol_ble.py # 扫描并按名连接 python test_protocol_ble.py --ble-address AA:BB:CC:... # 按地址连接 python test_protocol_ble.py --allow-poweroff # 含 05 00 关机用例 python test_protocol_ble.py --selftest # 无硬件自检(编解码) 依赖: bleak (pip install bleak) 报告: Doc/reports/ble_sysex_<时间戳>.md 退出码: 0 = 全部通过, 1 = 有 FAIL """ import argparse import asyncio import datetime import os import queue import sys import threading import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from test_protocol_app_sim import ( # noqa: E402 HEAD, DEV_ID, build, FrameParser, Results, expect, run_tests, hexs) MIDI_SERVICE = "03B80E5A-EDE8-4B33-A751-6CE34EC4C700" MIDI_CHAR = "7772E5DB-3868-4112-A1A9-F2669D106BF3" DEFAULT_NAME = "Smart Guitar MIDI" DEFAULT_MTU_PAYLOAD = 20 # ATT 默认 23 -> 应用载荷 20 # ---------------------------------------------------------- BLE-MIDI codec def ble_midi_encode_sysex(frame: bytes, max_payload: int): """完整 SysEx(F0...F7) -> BLE-MIDI 包列表 (ts=0)""" ts_hi, ts_lo = 0x80, 0x80 pkts = [] cap = max_payload - 2 # header + timestamp first = frame[:cap] pkts.append(bytes([ts_hi, ts_lo]) + first) rest = frame[len(first):] while len(rest) > max_payload - 2: # 中间包: header + data pkts.append(bytes([ts_hi]) + rest[:max_payload - 1]) rest = rest[max_payload - 1:] if rest: # 结束包: header + ts + data..F7 pkts.append(bytes([ts_hi, ts_lo]) + rest) return pkts def ble_midi_decode_packet(payload: bytes) -> bytes: """单个 BLE-MIDI 通知包 -> MIDI 字节流 (仅 SysEx 场景: 数据均为 7bit)""" if len(payload) < 2: return b"" i = 1 # 跳过 header if payload[i] & 0x80: # 时间戳(起始/结束包); 续包无 i += 1 return bytes(payload[i:]) # ---------------------------------------------------------- BLE transport class BleMidiSim: """与 AppSim 同接口: send / read_frame / drain / query / close""" def __init__(self, name=DEFAULT_NAME, address=None, scan_timeout=15.0): from bleak import BleakClient, BleakScanner self._BleakClient = BleakClient self._BleakScanner = BleakScanner self.parser = FrameParser() self.pending = [] self._rx = queue.Queue() self._client = None self._char = None self._loop = asyncio.new_event_loop() self._thread = threading.Thread(target=self._loop.run_forever, daemon=True) self._thread.start() self.info = self._run(self._connect(name, address, scan_timeout)) # ---- async helpers ---- def _run(self, coro, timeout=60.0): return asyncio.run_coroutine_threadsafe(coro, self._loop).result(timeout) async def _connect(self, name, address, scan_timeout): dev = None if address: dev = await self._BleakScanner.find_device_by_address( address, timeout=scan_timeout) else: print(f"扫描 BLE 设备 ({scan_timeout:.0f}s) ...") devs = await self._BleakScanner.discover( timeout=scan_timeout, service_uuids=[MIDI_SERVICE]) for d in devs: print(f" 发现: {d.name!r} {d.address}") if d.name and name.lower() in d.name.lower(): dev = d if dev is None and devs: dev = devs[0] print(f" (未匹配名称 {name!r}, 使用第一个 MIDI 设备)") if dev is None: raise RuntimeError(f"未找到 BLE-MIDI 设备 {name!r}") client = self._BleakClient(dev) await client.connect() char = None for svc in client.services: if svc.uuid.lower() == MIDI_SERVICE.lower(): for c in svc.characteristics: if c.uuid.lower() == MIDI_CHAR.lower(): char = c if char is None: raise RuntimeError("设备无 BLE-MIDI 特征") try: max_payload = char.max_write_without_response_size except Exception: max_payload = DEFAULT_MTU_PAYLOAD if not max_payload or max_payload < DEFAULT_MTU_PAYLOAD: max_payload = DEFAULT_MTU_PAYLOAD def _on_notify(_sender, data): self._rx.put(bytes(data)) await client.start_notify(char, _on_notify) self._client, self._char = client, char self._max_payload = max_payload return {"name": dev.name, "address": dev.address, "max_payload": max_payload} async def _disconnect(self): if self._client is not None: try: await self._client.stop_notify(self._char) except Exception: pass await self._client.disconnect() # ---- AppSim-compatible sync API ---- def close(self): try: self._run(self._disconnect(), timeout=10) except Exception: pass self._loop.call_soon_threadsafe(self._loop.stop) self._thread.join(timeout=5) def _pump(self): while True: try: data = self._rx.get_nowait() except queue.Empty: break self.pending.extend(self.parser.feed(ble_midi_decode_packet(data))) def send(self, frame: bytes): for pkt in ble_midi_encode_sysex(frame, self._max_payload): self._run(self._client.write_gatt_char( self._char, pkt, response=False), timeout=10) def read_frame(self, timeout=1.0, want_dev=False): 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) # ---------------------------------------------------------- report def write_report(R: Results, info: dict, path: str): n = {s: sum(1 for r in R.rows if r[1] == s) for s in ("PASS", "FAIL", "SKIP", "SENT")} lines = [ "# BLE MIDI SysEx 协议测试报告", "", f"- 日期: {datetime.datetime.now():%Y-%m-%d %H:%M:%S}", f"- 设备: {info.get('name')} ({info.get('address')})", f"- 传输: BLE-MIDI / bleak, 单包载荷 {info.get('max_payload')} 字节", "- 协议来源: Doc/指令测试.docx", "- 测试脚本: tools/test_protocol_ble.py (复用 test_protocol_app_sim.py 用例)", "", "## 汇总", "", "| PASS | FAIL | SKIP | SENT |", "|------|------|------|------|", f"| {n['PASS']} | {n['FAIL']} | {n['SKIP']} | {n['SENT']} |", "", "## 明细", "", "| # | 用例 | 结果 | 详情(RX/说明) |", "|---|------|------|----------------|", ] for i, (name, status, detail) in enumerate(R.rows, 1): lines.append(f"| {i} | {name} | {status} | {detail} |") lines += [ "", "## 备注", "", "- `01 03` 固件主版本回复为 MCU 实际版本 (0.2.5 -> 00 02 05 01),", " 与文档样例 `21 05 7F 01` 编码不同 (文档未给编码规则)。", "- `01 FF` 固件编码暂回 ASCII `BRS08L`, 文档标注待定。", "- `04 xx`/`06 xx` 为无应答下发命令, SENT 表示已发送。", "- `FD 01` 升级指令不属于本 UART4/BLE 应用协议范围, 未测。", "- 蓝牙广播名由 Dream 模块固件决定 (Smart Guitar MIDI), 与 `01 02` 协议设备名无关。", ] os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as f: f.write("\n".join(lines) + "\n") return path # ---------------------------------------------------------- selftest def selftest(): R = Results() # 短帧: 单包 pkts = ble_midi_encode_sysex(bytes.fromhex("F0600101F7"), 20) ok = pkts == [bytes.fromhex("8080F0600101F7")] R.add("encode short", "PASS" if ok else "FAIL", str([hexs(p) for p in pkts])) # 长帧: 47B -> 多包, 且可解码还原 frame = bytes([0xF0, 0x60, 0x02, 0x01]) + bytes(range(1, 43)) + bytes([0xF7]) pkts = ble_midi_encode_sysex(frame, 20) back = b"".join(ble_midi_decode_packet(p) for p in pkts) ok = back == frame and len(pkts) >= 3 R.add("encode/decode 47B roundtrip", "PASS" if ok else "FAIL", f"{len(pkts)} pkts") # 解码: 单包带时间戳 ok = ble_midi_decode_packet(bytes.fromhex("8080F0600101F7")) == bytes.fromhex("F0600101F7") R.add("decode single pkt", "PASS" if ok else "FAIL", "") # 解码: 续包(无时间戳) ok = ble_midi_decode_packet(bytes.fromhex("80010203")) == bytes.fromhex("010203") R.add("decode continuation", "PASS" if ok else "FAIL", "") return R.summary() def main(): ap = argparse.ArgumentParser(description="BLE-MIDI 吉他协议全量测试") ap.add_argument("--ble-name", default=DEFAULT_NAME, help=f"广播名匹配 (默认 {DEFAULT_NAME!r})") ap.add_argument("--ble-address", help="直接按 MAC/地址连接") ap.add_argument("--allow-poweroff", action="store_true", help="允许执行 05 00 关机用例") ap.add_argument("--selftest", action="store_true", help="无硬件自检") ap.add_argument("--report", help="报告输出路径 (默认 Doc/reports/ble_sysex_<时间戳>.md)") args = ap.parse_args() if args.selftest: sys.exit(selftest()) try: import bleak # noqa: F401 except ImportError: sys.exit("缺少 bleak: pip install bleak") sim = BleMidiSim(name=args.ble_name, address=args.ble_address) print(f"=== BLE 协议测试开始: {sim.info['name']} ({sim.info['address']}) " f"payload={sim.info['max_payload']} ===") try: R = run_tests(sim, allow_poweroff=args.allow_poweroff) finally: sim.close() code = R.summary() ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") path = args.report or os.path.join( os.path.dirname(os.path.abspath(__file__)), "..", "..", "..", "Doc", "reports", f"ble_sysex_{ts}.md") path = os.path.normpath(path) write_report(R, sim.info, path) print(f"\n报告已写入: {path}") sys.exit(code) if __name__ == "__main__": main()