From ff46d2985bcbffcb485262e1d13b05683cd334fc Mon Sep 17 00:00:00 2001
From: agent <agent@local>
Date: Wed, 29 Jul 2026 04:20:24 +0000
Subject: [PATCH] =?UTF-8?q?feat(tools):=20gpio=5Fprobe.py=20GPIO=5FMAP=20?=
 =?UTF-8?q?=E5=80=99=E9=80=89=E5=AE=9E=E8=AF=81=E6=8E=A2=E9=92=88=EF=BC=88?=
 =?UTF-8?q?max=5Fblocks=3D2=20=E5=AE=89=E5=85=A8=E6=A8=A1=E5=BC=8F?=
 =?UTF-8?q?=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 tools/gpio_probe.py | 67 +++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 67 insertions(+)
 create mode 100644 tools/gpio_probe.py

diff --git a/tools/gpio_probe.py b/tools/gpio_probe.py
new file mode 100644
index 0000000..c8dd649
--- /dev/null
+++ b/tools/gpio_probe.py
@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+"""GPIO_MAP 候选实证探针（安全：max_blocks=2，不发 END，断连即弃）。
+用法：.venv/bin/python tools/gpio_probe.py [host] [port]
+"""
+import asyncio
+import json
+import sys
+import time
+
+HOST = sys.argv[1] if len(sys.argv) > 1 else "192.168.61.35"
+PORT = int(sys.argv[2]) if len(sys.argv) > 2 else 8971
+DEVICE = "18:BC:5A:5D:BF:28"
+IMAGE_HEX = open("analysis/firmware/fw_tst.img", "rb").read().hex()
+
+CANDIDATES = [
+    ("A 现行 {00,00,06,05}", "00000605"),
+    ("B LE(0x05060040) {40,00,06,05}", "40000605"),
+    ("C BE(0x05060040) {05,06,00,40}", "05060040"),
+]
+
+
+async def rpc(r, w, i, op, **kw):
+    w.write(json.dumps({"id": i, "op": op, **kw}).encode() + b"\n")
+    await w.drain()
+    events = []
+    while True:
+        msg = json.loads(await asyncio.wait_for(r.readline(), 120))
+        if "event" in msg:
+            events.append(msg)
+        else:
+            return msg, events
+
+
+async def probe(name, gpio_hex):
+    r, w = await asyncio.open_connection(HOST, PORT)
+    try:
+        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 "探针停止(无块确认)"
+        else:
+            verdict = f"❌ {resp.get('error')}"
+        print(f"{name}: {verdict}  状态码序列={codes}  ({time.time()-t0:.1f}s)")
+    finally:
+        w.close()
+
+
+async def main():
+    for name, gh in CANDIDATES:
+        await probe(name, gh)
+        await asyncio.sleep(3)
+
+
+asyncio.run(main())