feat(bridge): alt(331f) 写通道——build 4 + CLI raw --channel alt，测试全绿

examples/wake_batch.jsonl

@@ -0,0 +1,10 @@
+{"argv": ["activate", "--show-id"]}
+{"argv": ["activate", "--auto"]}
+{"argv": ["time"]}
+{"argv": ["mode", "clock1"]}
+{"argv": ["image", "examples/checker_red.png", "--slot", "0"]}
+{"argv": ["mode", "image0"]}
+{"argv": ["countdown", "2026-12-31", "--prefix", "目标"]}
+{"argv": ["parking", "12345"]}
+{"argv": ["mode", "clock1"]}
+{"argv": ["clear"]}

src/ppclock/bridge.py

@@ -145,6 +145,10 @@ class BridgeTransport:
     async def write_rxtx(self, data: bytes, response: bool = True):
         await self._rpc("write", char="rxtx", data=data.hex(), response=response)
 
+    async def write_alt(self, data: bytes, response: bool = True):
+        """经 331f 通道写（实机证据：1f1f 仅 ATT-ACK，331f 为活命令通道）。"""
+        await self._rpc("write", char="alt", data=data.hex(), response=response)
+
     async def read_rxtx(self) -> bytes:
         return bytes.fromhex((await self._rpc("read", char="rxtx"))["data"])
 

src/ppclock/cli.py

@@ -172,7 +172,7 @@ def build_parser() -> argparse.ArgumentParser:
 
     sp = sub.add_parser("raw", help="高级命令：直发 hex")
     sp.add_argument("hex")
-    sp.add_argument("--channel", default="rxtx", choices=["rxtx", "epd"])
+    sp.add_argument("--channel", default="rxtx", choices=["rxtx", "epd", "alt"])
     sp.add_argument("--no-response", action="store_true")
 
     sp = sub.add_parser("ota", help="SUOTA 固件升级（高危，需确认）")
@@ -255,6 +255,8 @@ async def _run_command(args, t):
         data = bytes.fromhex(args.hex)
         if args.channel == "epd":
             await t.write_epd(data, response=not args.no_response)
+        elif args.channel == "alt":
+            await t.write_alt(data, response=not args.no_response)
         else:
             await t.write_rxtx(data, response=not args.no_response)
         return {"sent": len(data)}

tests/test_cli.py

@@ -28,6 +28,9 @@ class FakeTransport:
     async def write_rxtx(self, data, response=True):
         self.rxtx_writes.append(data)
 
+    async def write_alt(self, data, response=True):
+        self.rxtx_writes.append(data)
+
     async def request_device_id(self, timeout=8.0):
         return "DEVICE12345678"
 

tools/bridge_server.py

@@ -27,7 +27,7 @@ import platform
 import sys
 
 PROTOCOL_VERSION = 1
-SERVER_BUILD = 3  # 2026-07-28: notify 通道 1f1f/331f 双候选 + connect 返回 GATT 表
+SERVER_BUILD = 4  # 2026-07-28: write 增加 alt(331f) 通道 + notify 三订阅
 
 EPD_SERVICE_UUID = "13187b10-eba9-a3ba-044e-83d3217d9a38"
 EPD_CHAR_UUID = "4b646063-6264-f3a7-8941-e65356ea82fe"
@@ -52,7 +52,9 @@ BLOCK_SIZE = 240
 WRITE_CHUNK = 20
 STATUS_OK = 2
 
-CHARS = {"epd": EPD_CHAR_UUID, "rxtx": RXTX_CHAR_UUID}
+CHARS = {"epd": EPD_CHAR_UUID, "rxtx": RXTX_CHAR_UUID, "alt": ALT_NOTIFY_UUID}
+# 实机 GATT（2026-07-28 bridge 诊断）：1f1f(h=53)=仅write 无 notify；
+# 331f(h=57)=notify+write+read，为真交互通道；EPD(h=48)=notify+write
 
 
 def parse_device_id(data: bytes) -> str:
@@ -98,13 +100,12 @@ class BleSession:
             raise RuntimeError(f"设备解析失败（扫描未见 {address}）")
         self.client = BleakClient(dev, timeout=timeout)
         await self.client.connect()
-        # notify 通道：1f1f 优先，331f 兜底（固件两 handle 同入分发器，notify 位可能只在 331f）
+        # notify 全订阅：331f（主）+ 1f1f（部分固件）+ EPD（设备 ID 也可能经此应答）
-        self.notify_char = None
+        self.notify_char = []
-        for uuid in (RXTX_CHAR_UUID, ALT_NOTIFY_UUID):
+        for uuid in (ALT_NOTIFY_UUID, RXTX_CHAR_UUID, EPD_CHAR_UUID):
             try:
                 await self.client.start_notify(uuid, self._on_notify)
-                self.notify_char = uuid
+                self.notify_char.append(uuid)
-                break
             except Exception:
                 continue
         table = self._gatt_table()
@@ -134,7 +135,8 @@ class BleSession:
     async def request_device_id(self, timeout: float = 8.0):
         self._id_buf.clear()
         self._id_event.clear()
-        await self.client.write_gatt_char(RXTX_CHAR_UUID, bytes([0xEF, 0xEF]), response=True)
+        # EFEF 走 331f（实机证据：1f1f 仅 ATT-ACK 不执行命令，331f 为活通道）
+        await self.client.write_gatt_char(ALT_NOTIFY_UUID, bytes([0xEF, 0xEF]), response=True)
         await asyncio.wait_for(self._id_event.wait(), timeout)
         return {"device_id": parse_device_id(bytes(self._id_buf[:14]))}