Harden BLE protocol test for Windows pairing and flaky notifies.

Default --unpair on Windows, probe/reconnect on first timeout, resync FrameParser on duplicate F0, and retry query once.

Co-authored-by: Cursor <cursoragent@cursor.com>

tools/_rtt_ble_diagnose.py

@@ -138,7 +138,7 @@ async def main():
     if not lines:
         print("WARNING: no RTT yet — touch screen or wait; continuing BLE test", flush=True)
 
-    # 1) BLE-MIDI framed (stable path historically)
+    # 1) BLE-MIDI framed (stable path)
     await ble_write(
         "midi-framed",
         MIDI,
@@ -147,15 +147,15 @@ async def main():
     )
     await asyncio.sleep(2.0)
 
-    # 2) custom uart raw (may drop link)
+    # E49A raw write often drops link / stops advertising — skip by default
-    await ble_write(
+    if "--with-uartw" in sys.argv:
-        "uartw-raw",
+        await ble_write(
-        UARTW,
+            "uartw-raw",
-        bytes.fromhex("F0600101F7"),
+            UARTW,
-        [UARTN, MIDI, EFF2],
+            bytes.fromhex("F0600101F7"),
-    )
+            [UARTN, MIDI, EFF2],
-
+        )
-    await asyncio.sleep(2.0)
+        await asyncio.sleep(2.0)
     stop = True
     time.sleep(0.4)
     j.close()

tools/test_protocol_app_sim.py

@@ -46,9 +46,13 @@ class FrameParser:
                 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:
+                if b == TAIL and len(self.buf) >= 4:
                     out.append(bytes(self.buf))
                 self.buf.clear()
         return out

tools/test_protocol_ble.py

@@ -270,10 +270,47 @@ class BleMidiSim:
         while self.read_frame(timeout=quiet) is not None:
             pass
 
-    def query(self, frame: bytes, timeout=1.0):
+    def query(self, frame: bytes, timeout=3.0):
-        self.drain()
+        self.drain(quiet=0.15)
         self.send(frame)
-        return self.read_frame(timeout=timeout)
+        r = self.read_frame(timeout=timeout)
+        if r is None:
+            # BLE 偶发丢 Notify：短间隔重发一次
+            time.sleep(0.25)
+            self.send(frame)
+            r = self.read_frame(timeout=timeout)
+        return r
+
+    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):
@@ -358,9 +395,14 @@ def main():
         "--transport",
         choices=("midi", "uart"),
         default="midi",
-        help="midi=标准 BLE-MIDI 帧; uart=自定义 e49a 原始 SysEx",
+        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("--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")
@@ -374,6 +416,7 @@ def main():
     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,
@@ -384,10 +427,39 @@ def main():
         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')}",
-        "若 FAIL 且 timeout: 检查手机 App 是否占用、Windows 是否残留配对(--unpair)。",
+        f"unpair={args.unpair}",
-        "BLE-MIDI Notify 中的时间戳字节已在解码时剥离。",
+        "Windows 默认 --unpair；全 timeout 时先复位吉他再测。",
+        "勿用 --transport uart 裸写做冒烟（易停广播）。",
     ]
     try:
         R = smoke(sim) if args.smoke else run_tests(sim, allow_poweroff=args.allow_poweroff)