K1Guitar/tools/test_protocol_ble.py

# -*- 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=3.0, retry=True):
        self.drain(quiet=0.15)
        self.send(frame)
        r = self.read_frame(timeout=timeout)
        if r is None and retry:
            # BLE 偶发丢 Notify：短间隔重发一次（关机指令不可重试）
            time.sleep(0.25)
            self.send(frame)
            r = self.read_frame(timeout=timeout)
        return r

    def is_connected(self) -> bool:
        c = self._client
        try:
            return bool(c is not None and c.is_connected)
        except Exception:
            return False

    async def _reconnect_clean(self, name, address, scan_timeout):
        """Drop session, unpair, rescan, reconnect — recovers Windows stale pairing."""
        await self._disconnect()
        self._notify_chars = []
        self._client = None
        await asyncio.sleep(0.8)
        self._unpair = True
        return await self._connect(name, address, scan_timeout)

    def probe_or_rebind(self, name=DEFAULT_NAME, address=None, scan_timeout=20.0):
        """Send 01 01; on timeout, unpair+reconnect once and retry."""
        r = self.query(build(0x01, 0x01), timeout=3.0)
        if r is not None and len(r) >= 5 and r[0] == HEAD and r[1] == 0x60:
            return True
        print("首包无应答 → 自动 unpair 并重连 …")
        try:
            self.info = self._run(
                self._reconnect_clean(name, address, scan_timeout), timeout=120.0
            )
        except Exception as e:
            print(f"重连失败: {e}")
            return False
        r = self.query(build(0x01, 0x01), timeout=3.0)
        ok = r is not None and len(r) >= 5 and r[0] == HEAD and r[1] == 0x60
        if not ok:
            print(
                "仍无应答。请 JLink/断电复位吉他（E49A 裸写或异常断连后模组常停广播），"
                "并确认琴已开机且蓝牙开关为开。"
            )
        return ok


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`=`Version[]`；`01 04/05`=日期；`01 0C`=`branch_short6`（如 develop_0aedb4）；`01 0F`=`BRS08L`。",
        "- `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=argparse.BooleanOptionalAction,
        default=(sys.platform == "win32"),
        help="连接前 unpair（Windows 默认开；--no-unpair 可关）",
    )
    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")

    print(f"选项: unpair={args.unpair} transport={args.transport}")
    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')} ==="
    )
    if not sim.probe_or_rebind(args.ble_name, args.ble_address):
        sim.close()
        path = args.report or os.path.normpath(
            os.path.join(
                os.path.dirname(os.path.abspath(__file__)),
                "..",
                "..",
                "..",
                "Doc",
                "reports",
                f"ble_sysex_{datetime.datetime.now():%Y%m%d_%H%M%S}.md",
            )
        )
        R = Results()
        R.add("probe 01 01", "FAIL", "timeout after unpair/reconnect")
        write_report(
            R,
            sim.info or {},
            path,
            notes=[
                "首包无应答：多半是 Windows 残留配对，或 ATS2853 被 E49A 裸写弄挂需复位。",
                "推荐: python tools/test_protocol_ble.py --unpair",
                "仍失败则 JLink reset / 给吉他断电再开机后再测。",
            ],
        )
        print(f"\n报告已写入: {path}")
        sys.exit(1)

    notes = [
        f"transport={sim.info.get('transport')}",
        f"unpair={args.unpair}",
        "Windows 默认 --unpair；全 timeout 时先复位吉他再测。",
        "勿用 --transport uart 裸写做冒烟（易停广播）。",
    ]
    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()