refactor(tools): gpio_probe v2 单会话+复位+守候

tools/gpio_probe.py

@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
-"""GPIO_MAP 候选实证探针（安全：max_blocks=2，不发 END，断连即弃）。
+"""GPIO_MAP 候选实证探针 v2（单 TCP/单 BLE 会话；候选间 SPOTA 复位）。
+安全：max_blocks=2 不写 END；MEM_DEV FF 复位会话；断连即弃。
 用法：.venv/bin/python tools/gpio_probe.py [host] [port]
 """
 import asyncio
@@ -17,51 +18,73 @@ CANDIDATES = [
     ("B LE(0x05060040) {40,00,06,05}", "40000605"),
     ("C BE(0x05060040) {05,06,00,40}", "05060040"),
 ]
+RESET_MEM_DEV = "000000ff"
 
 
-async def rpc(r, w, i, op, **kw):
+class Bridge:
-    w.write(json.dumps({"id": i, "op": op, **kw}).encode() + b"\n")
+    def __init__(self):
-    await w.drain()
+        self.next_id = 0
+
+    async def open(self):
+        self.r, self.w = await asyncio.open_connection(HOST, PORT)
+
+    async def rpc(self, op, timeout=120, **kw):
+        self.next_id += 1
+        rid = self.next_id
+        self.w.write(json.dumps({"id": rid, "op": op, **kw}).encode() + b"\n")
+        await self.w.drain()
         events = []
         while True:
-        msg = json.loads(await asyncio.wait_for(r.readline(), 120))
+            line = await asyncio.wait_for(self.r.readline(), timeout)
+            if not line:
+                raise ConnectionError("bridge EOF")
+            msg = json.loads(line)
             if "event" in msg:
                 events.append(msg)
             else:
                 return msg, events
 
-
+    async def connect_until_awake(self, tries=12):
-async def probe(name, gpio_hex):
+        for i in range(tries):
-    r, w = await asyncio.open_connection(HOST, PORT)
+            resp, _ = await self.rpc("connect", 150, address=DEVICE,
-    try:
+                                     timeout=30, subscribe=False)
-        c, _ = await rpc(r, w, 1, "connect", address=DEVICE, timeout=40,
-                         subscribe=False)
-        if not c.get("ok"):
-            print(f"{name}: 连接失败 {c.get('error')}")
-            return
-        t0 = time.time()
-        resp, events = await rpc(r, w, 2, "ota", image=IMAGE_HEX,
-                                 gpio_map=gpio_hex, max_blocks=2)
-        verdict = "?"
-        codes = []
-        for ev in events:
-            if ev.get("stage", "").endswith("_status") or "info_status" in ev.get("stage", ""):
-                codes += ev.get("codes", [])
-            if ev.get("event") == "ota_progress":
-                verdict = f"✅ 块{ev['block']} 确认(2)"
             if resp.get("ok"):
-            verdict = verdict if "✅" in verdict else "探针停止(无块确认)"
+                print(f"连接成功（第 {i + 1} 次）")
-        else:
+                return True
-            verdict = f"❌ {resp.get('error')}"
+            print(f"  等待设备唤醒... ({i + 1}/{tries})")
-        print(f"{name}: {verdict}  状态码序列={codes}  ({time.time()-t0:.1f}s)")
+            await asyncio.sleep(15)
-    finally:
+        return False
-        w.close()
 
 
 async def main():
+    b = Bridge()
+    await b.open()
+    if not await b.connect_until_awake():
+        print("设备持续未醒，退出")
+        return
     for name, gh in CANDIDATES:
-        await probe(name, gh)
+        t0 = time.time()
-        await asyncio.sleep(3)
+        resp, events = await b.rpc("ota", 180, image=IMAGE_HEX,
+                                   gpio_map=gh, max_blocks=2)
+        codes, confirmed = [], 0
+        for ev in events:
+            st = ev.get("stage", "")
+            if st.endswith("_status") or "info_status" in st:
+                codes += ev.get("codes", [])
+            if ev.get("event") == "ota_progress":
+                confirmed = ev["block"]
+        if confirmed:
+            verdict = f"✅ 块{confirmed} 确认（引脚正确！）"
+        elif resp.get("ok"):
+            verdict = "探针停（无块确认）"
+        else:
+            verdict = f"❌ {resp.get('error')}"
+        print(f"{name}: {verdict}  状态码={codes}  ({time.time() - t0:.1f}s)")
+        # SPOTA 会话复位，避免缓冲跨候选累积
+        await b.rpc("write", 30, char="spota_mem_dev",
+                    data=RESET_MEM_DEV, response=True)
+        await asyncio.sleep(2)
+    print("探针完成")
 
 
 asyncio.run(main())