100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
"""GPIO_MAP 候选实证探针 v2（单 TCP/单 BLE 会话；候选间 SPOTA 复位）。
|
||
安全：max_blocks=2 不写 END；MEM_DEV FF 复位会话；断连即弃。
|
||
用法：.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"),
|
||
]
|
||
RESET_MEM_DEV = "000000ff"
|
||
|
||
|
||
class Bridge:
|
||
def __init__(self):
|
||
self.next_id = 0
|
||
|
||
async def open(self):
|
||
try:
|
||
self.w.close()
|
||
except Exception:
|
||
pass
|
||
self.r, self.w = await asyncio.open_connection(HOST, PORT)
|
||
|
||
async def rpc(self, op, wait=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:
|
||
line = await asyncio.wait_for(self.r.readline(), wait)
|
||
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):
|
||
for i in range(tries):
|
||
try:
|
||
resp, _ = await self.rpc("connect", 150, address=DEVICE,
|
||
timeout=30, subscribe=False)
|
||
except (ConnectionError, asyncio.TimeoutError):
|
||
print(" 连接通道中断，重开 TCP...")
|
||
await self.open()
|
||
continue
|
||
if resp.get("ok"):
|
||
print(f"连接成功（第 {i + 1} 次）")
|
||
return True
|
||
print(f" 等待设备唤醒... ({i + 1}/{tries})")
|
||
await asyncio.sleep(15)
|
||
return False
|
||
|
||
|
||
async def main():
|
||
b = Bridge()
|
||
await b.open()
|
||
if not await b.connect_until_awake():
|
||
print("设备持续未醒，退出")
|
||
return
|
||
for name, gh in CANDIDATES:
|
||
t0 = time.time()
|
||
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())
|