417 lines
14 KiB
Python
417 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
test_protocol_ble.py — 通过笔记本蓝牙对吉他做全协议测试并出报告
|
||
|
||
协议来源: Doc/指令测试.docx
|
||
链路: 笔记本 BLE <-> "Smart Guitar MIDI" <-> 吉他 UART4
|
||
App->设备: F0 60 ... F7 ; 设备主动上报: F0 51 ... F7
|
||
|
||
GATT:
|
||
BLE-MIDI 03B80E5A-... / 7772E5DB-... (framed SysEx)
|
||
自定义串口 e49a25f8-... / e49a25e0(写) + e49a28e1(通知) (raw SysEx)
|
||
|
||
用法:
|
||
python test_protocol_ble.py
|
||
python test_protocol_ble.py --transport midi
|
||
python test_protocol_ble.py --transport uart
|
||
python test_protocol_ble.py --unpair
|
||
python test_protocol_ble.py --smoke
|
||
python test_protocol_ble.py --selftest
|
||
|
||
依赖: bleak (pip install bleak)
|
||
报告: Doc/reports/ble_sysex_<时间戳>.md
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
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, run_tests, hexs)
|
||
|
||
MIDI_SERVICE = "03B80E5A-EDE8-4B33-A751-6CE34EC4C700"
|
||
MIDI_CHAR = "7772E5DB-3868-4112-A1A9-F2669D106BF3"
|
||
UART_SERVICE = "e49a25f8-f69a-11e8-8eb2-f2801f1b9fd1"
|
||
UART_WRITE = "e49a25e0-f69a-11e8-8eb2-f2801f1b9fd1"
|
||
UART_NOTIFY = "e49a28e1-f69a-11e8-8eb2-f2801f1b9fd1"
|
||
DEFAULT_NAME = "Smart Guitar MIDI"
|
||
DEFAULT_MTU_PAYLOAD = 20
|
||
|
||
|
||
def ble_midi_encode_sysex(frame: bytes, max_payload: int):
|
||
ts_hi, ts_lo = 0x80, 0x80
|
||
pkts = []
|
||
cap = max_payload - 2
|
||
first = frame[:cap]
|
||
pkts.append(bytes([ts_hi, ts_lo]) + first)
|
||
rest = frame[len(first):]
|
||
while len(rest) > max_payload - 2:
|
||
pkts.append(bytes([ts_hi]) + rest[: max_payload - 1])
|
||
rest = rest[max_payload - 1 :]
|
||
if rest:
|
||
pkts.append(bytes([ts_hi, ts_lo]) + rest)
|
||
return pkts
|
||
|
||
|
||
def ble_midi_decode_packet(payload: bytes) -> bytes:
|
||
"""Strip BLE-MIDI header/timestamp bytes; keep App SysEx (F0…F7).
|
||
|
||
- Framed: [header ts][optional ts][F0 … ts … F7]
|
||
- Raw (some ATS2853 notifies): [F0 … F7] — must not treat F0 as header
|
||
(F0/F7 also have bit7=1).
|
||
"""
|
||
if not payload:
|
||
return b""
|
||
i = 0
|
||
if payload[0] not in (0xF0, 0xF7) and (payload[0] & 0x80):
|
||
i = 1 # BLE-MIDI header
|
||
out = bytearray()
|
||
while i < len(payload):
|
||
b = payload[i]
|
||
i += 1
|
||
if b in (0xF0, 0xF7):
|
||
out.append(b)
|
||
elif b & 0x80:
|
||
continue # timestamp inside/around SysEx
|
||
else:
|
||
out.append(b)
|
||
return bytes(out)
|
||
|
||
|
||
class BleMidiSim:
|
||
"""与 AppSim 同接口: send / read_frame / drain / query / close"""
|
||
|
||
def __init__(
|
||
self,
|
||
name=DEFAULT_NAME,
|
||
address=None,
|
||
scan_timeout=20.0,
|
||
transport="midi",
|
||
unpair=False,
|
||
):
|
||
from bleak import BleakClient, BleakScanner
|
||
|
||
self._BleakClient = BleakClient
|
||
self._BleakScanner = BleakScanner
|
||
self.parser = FrameParser()
|
||
self.pending = []
|
||
self._rx = queue.Queue()
|
||
self._client = None
|
||
self._write_char = None
|
||
self._notify_chars = []
|
||
self._transport = transport
|
||
self._unpair = unpair
|
||
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))
|
||
|
||
def _run(self, coro, timeout=90.0):
|
||
return asyncio.run_coroutine_threadsafe(coro, self._loop).result(timeout)
|
||
|
||
async def _find(self, name, address, scan_timeout):
|
||
if address:
|
||
return await self._BleakScanner.find_device_by_address(
|
||
address, timeout=scan_timeout
|
||
)
|
||
print(f"扫描 BLE 设备 ({scan_timeout:.0f}s) ...")
|
||
# name filter is more reliable than service UUID filter on Windows
|
||
t0 = time.monotonic()
|
||
while time.monotonic() - t0 < scan_timeout:
|
||
rem = max(1.0, scan_timeout - (time.monotonic() - t0))
|
||
d = await self._BleakScanner.find_device_by_filter(
|
||
lambda d, a: d.name and name.lower() in d.name.lower(),
|
||
timeout=min(8.0, rem),
|
||
)
|
||
if d:
|
||
print(f" 发现: {d.name!r} {d.address}")
|
||
return d
|
||
print(" ...")
|
||
return None
|
||
|
||
async def _connect(self, name, address, scan_timeout):
|
||
dev = await self._find(name, address, scan_timeout)
|
||
if dev is None:
|
||
raise RuntimeError(f"未找到设备 {name!r}")
|
||
|
||
if self._unpair:
|
||
try:
|
||
tmp = self._BleakClient(dev.address, timeout=20)
|
||
await tmp.connect()
|
||
try:
|
||
await tmp.unpair()
|
||
print("已 unpair Windows 残留配对")
|
||
except Exception as e:
|
||
print(f"unpair: {e}")
|
||
try:
|
||
await tmp.disconnect()
|
||
except Exception:
|
||
pass
|
||
await asyncio.sleep(1.5)
|
||
dev = await self._find(name, address, scan_timeout) or dev
|
||
except Exception as e:
|
||
print(f"unpair session: {e}")
|
||
|
||
client = self._BleakClient(dev, timeout=30)
|
||
await client.connect()
|
||
if not client.is_connected:
|
||
raise RuntimeError("BLE connect 后立即断开")
|
||
|
||
write_char = MIDI_CHAR if self._transport == "midi" else UART_WRITE
|
||
notify_list = (
|
||
[MIDI_CHAR]
|
||
if self._transport == "midi"
|
||
else [UART_NOTIFY, MIDI_CHAR]
|
||
)
|
||
|
||
max_payload = DEFAULT_MTU_PAYLOAD
|
||
for svc in client.services:
|
||
for c in svc.characteristics:
|
||
if c.uuid.lower() == write_char.lower():
|
||
try:
|
||
max_payload = c.max_write_without_response_size or max_payload
|
||
except Exception:
|
||
pass
|
||
|
||
def _on_notify(_sender, data):
|
||
self._rx.put(bytes(data))
|
||
|
||
for u in notify_list:
|
||
try:
|
||
await client.start_notify(u, _on_notify)
|
||
self._notify_chars.append(u)
|
||
except Exception as e:
|
||
print(f"notify fail {u[:8]}: {e}")
|
||
|
||
if not self._notify_chars:
|
||
raise RuntimeError("无法开启任何 notify（常见原因: Windows 残留配对）")
|
||
|
||
self._client = client
|
||
self._write_char = write_char
|
||
self._max_payload = max(DEFAULT_MTU_PAYLOAD, max_payload or DEFAULT_MTU_PAYLOAD)
|
||
return {
|
||
"name": getattr(dev, "name", None),
|
||
"address": getattr(dev, "address", address),
|
||
"max_payload": self._max_payload,
|
||
"transport": self._transport,
|
||
"notify": list(self._notify_chars),
|
||
}
|
||
|
||
async def _disconnect(self):
|
||
if self._client is None:
|
||
return
|
||
for u in self._notify_chars:
|
||
try:
|
||
await self._client.stop_notify(u)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
await self._client.disconnect()
|
||
except Exception:
|
||
pass
|
||
|
||
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
|
||
if self._transport == "midi":
|
||
midi = ble_midi_decode_packet(data)
|
||
else:
|
||
# uart notify may be raw SysEx, or occasionally BLE-MIDI wrapped
|
||
midi = (
|
||
ble_midi_decode_packet(data)
|
||
if data and (data[0] & 0x80)
|
||
else data
|
||
)
|
||
self.pending.extend(self.parser.feed(midi))
|
||
|
||
def send(self, frame: bytes):
|
||
if self._transport == "midi":
|
||
pkts = ble_midi_encode_sysex(frame, self._max_payload)
|
||
else:
|
||
pkts = [frame]
|
||
for pkt in pkts:
|
||
self._run(
|
||
self._client.write_gatt_char(self._write_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)
|
||
|
||
|
||
def write_report(R: Results, info: dict, path: str, notes: list[str] | None = None):
|
||
n = {
|
||
s: sum(1 for r in R.rows if r[1] == s)
|
||
for s in ("PASS", "FAIL", "SKIP", "SENT")
|
||
}
|
||
lines = [
|
||
"# BLE SysEx 协议测试报告",
|
||
"",
|
||
f"- 日期: {datetime.datetime.now():%Y-%m-%d %H:%M:%S}",
|
||
f"- 设备: {info.get('name')} ({info.get('address')})",
|
||
f"- 传输: {info.get('transport')} / bleak, payload={info.get('max_payload')}",
|
||
f"- notify: {info.get('notify')}",
|
||
"- 协议来源: Doc/指令测试.docx",
|
||
"- 测试脚本: tools/test_protocol_ble.py",
|
||
"",
|
||
"## 汇总",
|
||
"",
|
||
"| PASS | FAIL | SKIP | SENT |",
|
||
"|------|------|------|------|",
|
||
f"| {n['PASS']} | {n['FAIL']} | {n['SKIP']} | {n['SENT']} |",
|
||
"",
|
||
"## 明细",
|
||
"",
|
||
"| # | 用例 | 结果 | 详情 |",
|
||
"|---|------|------|------|",
|
||
]
|
||
for i, (name, status, detail) in enumerate(R.rows, 1):
|
||
lines.append(f"| {i} | {name} | {status} | {detail} |")
|
||
lines += ["", "## 备注", ""]
|
||
for note in notes or []:
|
||
lines.append(f"- {note}")
|
||
lines += [
|
||
"- `01 03` 为 MCU 实际版本；`01 0F` 固件编码暂 `BRS08L`（原 `01 FF`，BLE-MIDI 下 0xFF 非法）。",
|
||
"- `01 07` UID 为 24 字符 hex ASCII（96-bit），保证 SysEx 7-bit 安全。",
|
||
"- `04 xx`/`06 xx` 无应答，SENT 表示已发送。",
|
||
"- `FD 01` 升级指令不在本协议范围。",
|
||
]
|
||
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
|
||
|
||
|
||
def smoke(sim: BleMidiSim) -> Results:
|
||
R = Results()
|
||
r = sim.query(build(0x01, 0x01), timeout=3.0)
|
||
ok = r is not None and len(r) >= 5 and r[0] == HEAD and r[1] == 0x60
|
||
R.add("smoke 01 01 连接设备", "PASS" if ok else "FAIL", hexs(r))
|
||
if ok:
|
||
r2 = sim.query(build(0x01, 0x02), timeout=3.0)
|
||
ok2 = r2 is not None and r2[2:4] == bytes([0x01, 0x02])
|
||
R.add("smoke 01 02 设备名", "PASS" if ok2 else "FAIL", hexs(r2))
|
||
r3 = sim.query(build(0x02, 0x01), timeout=3.0)
|
||
ok3 = r3 is not None and len(r3) == 47
|
||
R.add("smoke 02 01 和弦表(长帧)", "PASS" if ok3 else "FAIL", hexs(r3))
|
||
return R
|
||
|
||
|
||
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]))
|
||
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)
|
||
R.add(
|
||
"encode/decode 47B",
|
||
"PASS" if back == frame and len(pkts) >= 3 else "FAIL",
|
||
f"{len(pkts)} pkts",
|
||
)
|
||
return R.summary()
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="BLE 吉他协议全量测试")
|
||
ap.add_argument("--ble-name", default=DEFAULT_NAME)
|
||
ap.add_argument("--ble-address")
|
||
ap.add_argument(
|
||
"--transport",
|
||
choices=("midi", "uart"),
|
||
default="midi",
|
||
help="midi=标准 BLE-MIDI 帧; uart=自定义 e49a 原始 SysEx",
|
||
)
|
||
ap.add_argument("--unpair", action="store_true", help="连接前先 unpair Windows 配对")
|
||
ap.add_argument("--allow-poweroff", action="store_true")
|
||
ap.add_argument("--smoke", action="store_true", help="仅冒烟: 01 01/02 + 02 01")
|
||
ap.add_argument("--selftest", action="store_true")
|
||
ap.add_argument("--report")
|
||
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,
|
||
transport=args.transport,
|
||
unpair=args.unpair,
|
||
)
|
||
print(
|
||
f"=== BLE 测试: {sim.info.get('name')} ({sim.info.get('address')}) "
|
||
f"transport={sim.info.get('transport')} payload={sim.info.get('max_payload')} ==="
|
||
)
|
||
notes = [
|
||
f"transport={sim.info.get('transport')}",
|
||
"若 FAIL 且 timeout: 检查手机 App 是否占用、Windows 是否残留配对(--unpair)。",
|
||
"BLE-MIDI Notify 中的时间戳字节已在解码时剥离。",
|
||
]
|
||
try:
|
||
R = smoke(sim) if args.smoke else 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.normpath(
|
||
os.path.join(
|
||
os.path.dirname(os.path.abspath(__file__)),
|
||
"..",
|
||
"..",
|
||
"..",
|
||
"Doc",
|
||
"reports",
|
||
f"ble_sysex_{ts}.md",
|
||
)
|
||
)
|
||
write_report(R, sim.info, path, notes=notes)
|
||
print(f"\n报告已写入: {path}")
|
||
sys.exit(code)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|