68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
#!/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())
|