296 lines
11 KiB
Python
296 lines
11 KiB
Python
|
# -*- 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()
|