diff --git a/examples/checker_red.png b/examples/checker_red.png new file mode 100644 index 0000000..49b64fa Binary files /dev/null and b/examples/checker_red.png differ diff --git a/examples/wake_batch.jsonl b/examples/wake_batch.jsonl new file mode 100644 index 0000000..134df30 --- /dev/null +++ b/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"]} diff --git a/src/ppclock/bridge.py b/src/ppclock/bridge.py index b7e542a..bd80cdc 100644 --- a/src/ppclock/bridge.py +++ b/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"]) diff --git a/src/ppclock/cli.py b/src/ppclock/cli.py index 1f8450e..fd8dd02 100644 --- a/src/ppclock/cli.py +++ b/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)} diff --git a/tests/test_cli.py b/tests/test_cli.py index 58badde..ef8012e 100644 --- a/tests/test_cli.py +++ b/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" diff --git a/tools/bridge_server.py b/tools/bridge_server.py index b023ddf..af88a8f 100644 --- a/tools/bridge_server.py +++ b/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) - self.notify_char = None - for uuid in (RXTX_CHAR_UUID, ALT_NOTIFY_UUID): + # notify 全订阅:331f(主)+ 1f1f(部分固件)+ EPD(设备 ID 也可能经此应答) + self.notify_char = [] + 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 - break + self.notify_char.append(uuid) 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]))}