feat(ppclock): SUOTA 固件升级（ota.py + CLI ota 子命令 + 预检），APK互证报告入库，123 测试全绿

src/ppclock/cli.py

@@ -162,6 +162,10 @@ def build_parser() -> argparse.ArgumentParser:
     sp.add_argument("--channel", default="rxtx", choices=["rxtx", "epd"])
     sp.add_argument("--no-response", action="store_true")
 
+    sp = sub.add_parser("ota", help="SUOTA 固件升级（高危，需确认）")
+    sp.add_argument("file", help="固件 .img（pQ 头格式）")
+    sp.add_argument("--yes", action="store_true", help="确认执行（否则仅预检）")
+
     sp = sub.add_parser("batch", help="JSONL 批量执行（单次连接，agent 编排）")
     sp.add_argument("file", help="JSONL 文件，每行 {\"argv\": [...]}；'-' 读 stdin")
     sp.add_argument("--stop-on-error", action="store_true")
@@ -235,6 +239,24 @@ async def _run_command(args, t):
         else:
             await t.write_rxtx(data, response=not args.no_response)
         return {"sent": len(data)}
+    if c == "ota":
+        import zlib
+        from pathlib import Path
+        image = Path(args.file).read_bytes()
+        # 独立预检（pQ 头 + CRC32(body)），与 tools/fw.py 同规则
+        if len(image) < 64 or image[0:2] != b"pQ":
+            raise ValueError("固件格式错误（缺 pQ 头）")
+        body_size = int.from_bytes(image[4:8], "little")
+        crc_ok = (zlib.crc32(image[64:64 + body_size]) & 0xFFFFFFFF
+                  == int.from_bytes(image[8:12], "little"))
+        if not crc_ok:
+            raise ValueError("固件 CRC 校验失败，拒绝烧录")
+        if not args.yes:
+            return {"precheck": "ok", "body_size": body_size,
+                    "hint": "加 --yes 执行烧录"}
+        progress_log = []
+        result = await t.run_ota(image, on_progress=progress_log.append)
+        return {"ota": "done", **result, "blocks_ok": len(progress_log)}
     raise ValueError(f"未知命令 {c}")
 
 

src/ppclock/ota.py

@@ -0,0 +1,116 @@
+"""OTA/SUOTA（Dialog SPOTA 自定义 UUID 映射）。
+证据：analysis/apk/apk-analysis.md §e（APK z7.b0/a8.g 反编译）。
+流程：fef5 服务探测 → MTU/高优先级 → SERV_STATUS notify → MEM_DEV←SPI
+→ GPIO_MAP → 240B 块（PATCH_LEN + 20B 写块 PATCH_DATA + 等状态2）
+→ END_SIGNAL → 等状态2 → REBOOT_SIGNAL。
+"""
+from __future__ import annotations
+
+# GATT（自定义映射，与 Dialog 标准不同 —— 证据 z7_b0.java:665-679）
+SPOTA_SERVICE_UUID = "0000fef5-0000-1000-8000-00805f9b34fb"
+SPOTA_SERV_STATUS_UUID = "64b4e8b5-0de5-401b-a21d-acc8db3b913a"
+SPOTA_MEM_DEV_UUID = "8082caa8-41a6-4021-91c6-56f9b954cc34"
+SPOTA_GPIO_MAP_UUID = "724249f0-5ec3-4b5f-8804-42345af08651"
+SPOTA_PATCH_DATA_UUID = "457871e8-d516-4ca1-9116-57d0b17b9cb2"
+SPOTA_PATCH_LEN_UUID = "9d84b9a3-000c-49d8-9183-855b673fda31"
+
+BLOCK_SIZE = 240       # z7.h0 默认块长
+WRITE_CHUNK = 20       # MTU 23 → 20B 写块
+
+MEM_DEV_SPI_FLASH = bytes([0x00, 0x00, 0x00, 0x13])   # mem type 0x13=SPI Flash
+GPIO_MAP_DEFAULT = bytes([0x00, 0x00, 0x06, 0x05])    # miso5/mosi6/cs3/sck0 编码
+END_SIGNAL = bytes([0x00, 0x00, 0x00, 0xFE])
+REBOOT_SIGNAL = bytes([0x00, 0x00, 0x00, 0xFD])
+
+STATUS_OK = 2
+STATUS_ERRORS = {
+    3: "强制退出 SPOTA",
+    4: "Patch Data CRC 失败",
+    5: "长度与 PATCH_LEN 不符",
+    6: "外部存储写失败",
+    7: "内部存储不足",
+    8: "无效存储器类型",
+    9: "应用层错误",
+    17: "无效镜像 Bank",
+    18: "无效镜像头部",
+    19: "无效镜像大小",
+    20: "无效产品头部",
+    21: "相同镜像错误",
+    22: "外部存储读失败",
+    0xFFFF: "通信错误",
+    0xFFFE: "设备不支持 SUOTA",
+}
+
+
+class OtaError(Exception):
+    pass
+
+
+def xor_checksum(data: bytes) -> int:
+    """全镜像 XOR 折叠为 1 字节（z7.h0 追加的尾校验）。"""
+    acc = 0
+    for b in data:
+        acc ^= b
+    return acc
+
+
+def package_image(image: bytes, block_size: int = BLOCK_SIZE) -> list[bytes]:
+    """镜像 + 尾 XOR 校验字节 → 240B 块列表。"""
+    payload = image + bytes([xor_checksum(image)])
+    return [payload[i:i + block_size] for i in range(0, len(payload), block_size)]
+
+
+def patch_len_frame(block_len: int) -> bytes:
+    return block_len.to_bytes(2, "little")
+
+
+def write_chunks(block: bytes, chunk: int = WRITE_CHUNK):
+    for i in range(0, len(block), chunk):
+        yield block[i:i + chunk]
+
+
+async def run_ota(client, image: bytes, on_progress=None, status_timeout=60.0):
+    """在已连接的 BleakClient 上执行完整 SUOTA。on_progress(dict) 回报进度。
+    client 为 bleak.BleakClient（需已连接且存在 fef5 服务）。"""
+    import asyncio
+    from bleak.exc import BleakError
+
+    if SPOTA_SERVICE_UUID not in [str(s.uuid) for s in client.services]:
+        raise OtaError("设备不存在 fef5 SUOTA 服务")
+
+    status_q: asyncio.Queue[int] = asyncio.Queue()
+
+    def _on_status(_sender, data: bytearray):
+        status_q.put_nowait(int.from_bytes(bytes(data), "little"))
+
+    async def wait_ok(what: str):
+        try:
+            status = await asyncio.wait_for(status_q.get(), status_timeout)
+        except asyncio.TimeoutError as e:
+            raise OtaError(f"等待设备状态超时（{what}）") from e
+        if status != STATUS_OK:
+            raise OtaError(f"{what} 失败: {STATUS_ERRORS.get(status, f'未知状态 {status}')}")
+
+    await client.start_notify(SPOTA_SERV_STATUS_UUID, _on_status)
+    try:
+        await client.write_gatt_char(SPOTA_MEM_DEV_UUID, MEM_DEV_SPI_FLASH, response=True)
+        await client.write_gatt_char(SPOTA_GPIO_MAP_UUID, GPIO_MAP_DEFAULT, response=True)
+        blocks = package_image(image)
+        for idx, block in enumerate(blocks, 1):
+            await client.write_gatt_char(SPOTA_PATCH_LEN_UUID,
+                                         patch_len_frame(len(block)), response=True)
+            for chunk in write_chunks(block):
+                await client.write_gatt_char(SPOTA_PATCH_DATA_UUID, chunk, response=False)
+            await wait_ok(f"块 {idx}/{len(blocks)}")
+            if on_progress:
+                on_progress({"block": idx, "total": len(blocks),
+                             "bytes_sent": sum(len(b) for b in blocks[:idx])})
+        await client.write_gatt_char(SPOTA_MEM_DEV_UUID, END_SIGNAL, response=True)
+        await wait_ok("镜像校验")
+        await client.write_gatt_char(SPOTA_MEM_DEV_UUID, REBOOT_SIGNAL, response=True)
+    finally:
+        try:
+            await client.stop_notify(SPOTA_SERV_STATUS_UUID)
+        except BleakError:
+            pass
+    return {"blocks": len(blocks), "bytes": len(image) + 1}

src/ppclock/transport.py

@@ -89,3 +89,8 @@ class BLETransport:
 
     async def delay(self, seconds: float):
         await asyncio.sleep(seconds)
+
+    async def run_ota(self, image: bytes, on_progress=None):
+        """SUOTA 固件升级（证据 analysis/apk/apk-analysis.md §e）。"""
+        from . import ota
+        return await ota.run_ota(self._client, image, on_progress)