feat(bridge): build 11 ota gpio_map 参数+max_blocks 探针模式

tools/bridge_server.py

@@ -27,7 +27,7 @@ import platform
 import sys
 
 PROTOCOL_VERSION = 1
-SERVER_BUILD = 10  # 2026-07-29: 状态码 8/16 按会话信息码处理（固件反汇编实证）+ 阶段状态取证 + SPOTA 写特征入 CHARS
+SERVER_BUILD = 11  # 2026-07-29: ota 支持 gpio_map 参数与 max_blocks 探针模式（GPIO 引脚实证用）
 
 EPD_SERVICE_UUID = "13187b10-eba9-a3ba-044e-83d3217d9a38"
 EPD_CHAR_UUID = "4b646063-6264-f3a7-8941-e65356ea82fe"
@@ -151,7 +151,9 @@ class BleSession:
         await asyncio.wait_for(self._id_event.wait(), timeout)
         return {"device_id": parse_device_id(bytes(self._id_buf[:14]))}
 
-    async def ota(self, image: bytes, status_timeout: float = 60.0):
+    async def ota(self, image: bytes, status_timeout: float = 60.0,
+                  gpio_map: bytes = GPIO_MAP_DEFAULT, max_blocks: int = 0):
+        """max_blocks>0 时只烧前 N 块即停（GPIO 探针用，不写 END）。"""
         if SPOTA_SERVICE_UUID not in [str(s.uuid) for s in self.client.services]:
             raise RuntimeError("设备不存在 fef5 SUOTA 服务")
         status_q: asyncio.Queue[int] = asyncio.Queue()
@@ -185,11 +187,14 @@ class BleSession:
             yield {"event": "ota_stage", "stage": "mem_dev_ok"}
             async for ev in drain("mem_dev", 1.0):
                 yield ev
-            await w(SPOTA_GPIO_MAP_UUID, GPIO_MAP_DEFAULT)
+            await w(SPOTA_GPIO_MAP_UUID, gpio_map)
             yield {"event": "ota_stage", "stage": "gpio_map_ok"}
             async for ev in drain("gpio_map", 1.0):
                 yield ev
             for idx, block in enumerate(blocks, 1):
+                if max_blocks and idx > max_blocks:
+                    yield {"event": "ota_stage", "stage": f"probe_stop_at_{max_blocks}"}
+                    return
                 await w(SPOTA_PATCH_LEN_UUID, len(block).to_bytes(2, "little"))
                 for i in range(0, len(block), WRITE_CHUNK):
                     await self.client.write_gatt_char(SPOTA_PATCH_DATA_UUID,
@@ -261,7 +266,10 @@ class BridgeServer:
                 try:
                     async with self._ble_lock:
                         if op == "ota":
-                            gen = self.ble.ota(bytes.fromhex(req["image"]))
+                            gen = self.ble.ota(
+                                bytes.fromhex(req["image"]),
+                                gpio_map=bytes.fromhex(req["gpio_map"]) if req.get("gpio_map") else GPIO_MAP_DEFAULT,
+                                max_blocks=int(req.get("max_blocks", 0)))
                             async for ev in gen:
                                 await send(ev)
                             data = {"done": True}