diff --git a/tools/bridge_server.py b/tools/bridge_server.py index 61f5c00..3c192eb 100644 --- a/tools/bridge_server.py +++ b/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}