diff --git a/src/ppclock/client.py b/src/ppclock/client.py index 4a2fdbe..923681d 100644 --- a/src/ppclock/client.py +++ b/src/ppclock/client.py @@ -31,6 +31,18 @@ class PPClient: def __init__(self, transport): self._t = transport + @property + def transport_info(self) -> dict: + """返回传输层公开诊断信息；bridge/fake 等未提供字段时为空。""" + info = {} + firmware_revision = getattr(self._t, "firmware_revision", None) + rxtx_uuid = getattr(self._t, "rxtx_uuid", None) + if firmware_revision: + info["firmware_revision"] = firmware_revision + if rxtx_uuid: + info["rxtx_uuid"] = rxtx_uuid + return info + # ---------- 生命周期 ---------- async def __aenter__(self): diff --git a/src/ppclock/device_manager.py b/src/ppclock/device_manager.py index de75603..cd3f417 100644 --- a/src/ppclock/device_manager.py +++ b/src/ppclock/device_manager.py @@ -67,6 +67,7 @@ class DeviceManager: "connect_timeout": self._connect_timeout, "idle_timeout": self._idle_timeout, } + info.update(self._client.transport_info) try: info["device_id"] = await self._client.get_device_id() except Exception: # noqa: BLE001 - 状态查询尽力而为 diff --git a/src/ppclock/protocol.py b/src/ppclock/protocol.py index 7aea6eb..d15f2d5 100644 --- a/src/ppclock/protocol.py +++ b/src/ppclock/protocol.py @@ -10,11 +10,25 @@ EPD_CHAR_UUID = "4b646063-6264-f3a7-8941-e65356ea82fe" RXTX_SERVICE_UUID = "00001f10-0000-1000-8000-00805f9b34fb" RXTX_CHAR_UUID = "0000331f-0000-1000-8000-00805f9b34fb" # 活命令通道(h=57, notify+write+read) RXTX_CHAR_UUID_LEGACY = "00001f1f-0000-1000-8000-00805f9b34fb" # h=53 仅write，本设备不执行命令 +FIRMWARE_REVISION_UUID = "00002a26-0000-1000-8000-00805f9b34fb" DEVICE_NAME_PREFIX = "NRF-" BLOCK_PAYLOAD = 240 # my_step=480 hex 字符 = 240 数据字节 COMMIT = 0xE2 + +def select_rxtx_uuid(firmware_revision: str | None) -> str: + """按固件家族选择控制特征。 + + PP 4.2 实机的 1F1F 虽可写但不执行控制命令，活通道是 331F； + HM42_AIO 实机则与 Web-BLE / 厂商 App 一致，仅执行 1F1F。 + 未知或读不到 DIS 版本时保留既有 331F 行为，避免回归已验证的 PP 设备。 + """ + revision = (firmware_revision or "").strip().upper() + if revision.startswith("HM42_AIO"): + return RXTX_CHAR_UUID_LEGACY + return RXTX_CHAR_UUID + MODES = { **{f"image{i}": bytes([0xE1, 0x00, 0x00 + i]) for i in range(4)}, "calendar1": bytes.fromhex("e10103"), diff --git a/src/ppclock/transports/local.py b/src/ppclock/transports/local.py index 1692279..094288d 100644 --- a/src/ppclock/transports/local.py +++ b/src/ppclock/transports/local.py @@ -35,6 +35,9 @@ class BLETransport: self._client: BleakClient | None = None self._id_buf = bytearray() self._id_event = asyncio.Event() + self.firmware_revision: str | None = None + self.rxtx_uuid = P.RXTX_CHAR_UUID + self._notify_started = False async def __aenter__(self): await self.connect() @@ -54,13 +57,37 @@ class BLETransport: await self._client.connect() except BleakError as e: raise TransportError(f"连接失败: {e}") from e - await self._client.start_notify(P.RXTX_CHAR_UUID, self._on_notify) + try: + raw_revision = await self._client.read_gatt_char(P.FIRMWARE_REVISION_UUID) + self.firmware_revision = bytes(raw_revision).decode( + "utf-8", errors="replace").rstrip("\x00") + except Exception: # DIS 版本是兼容性提示，不应阻断基础连接 + self.firmware_revision = None + + preferred = P.select_rxtx_uuid(self.firmware_revision) + alternate = ( + P.RXTX_CHAR_UUID_LEGACY + if preferred == P.RXTX_CHAR_UUID else P.RXTX_CHAR_UUID + ) + selected = self._client.services.get_characteristic(preferred) + if selected is None: + selected = self._client.services.get_characteristic(alternate) + if selected is None: + raise TransportError("未找到 RXTX 控制特征 1F1F/331F") + self.rxtx_uuid = selected.uuid + + properties = set(selected.properties) + if properties.intersection({"notify", "indicate"}): + await self._client.start_notify(selected, self._on_notify) + self._notify_started = True async def close(self): if self._client and self._client.is_connected: try: - await self._client.stop_notify(P.RXTX_CHAR_UUID) + if self._notify_started: + await self._client.stop_notify(self.rxtx_uuid) finally: + self._notify_started = False await self._client.disconnect() def _on_notify(self, _sender, data: bytearray): @@ -72,12 +99,17 @@ class BLETransport: await self._client.write_gatt_char(P.EPD_CHAR_UUID, data, response=response) async def write_rxtx(self, data: bytes, response: bool = True): - await self._client.write_gatt_char(P.RXTX_CHAR_UUID, data, response=response) + await self._client.write_gatt_char(self.rxtx_uuid, data, response=response) async def read_rxtx(self) -> bytes: - return bytes(await self._client.read_gatt_char(P.RXTX_CHAR_UUID)) + return bytes(await self._client.read_gatt_char(self.rxtx_uuid)) async def request_device_id(self, timeout: float = 8.0) -> str: + if not self._notify_started: + raise TransportError( + f"固件 {self.firmware_revision or 'unknown'} 的控制特征不支持通知，" + "无法读取设备 ID" + ) self._id_buf.clear() self._id_event.clear() await self.write_rxtx(P.request_id_frame()) diff --git a/tests/test_device_manager.py b/tests/test_device_manager.py index 2391552..dd42b8c 100644 --- a/tests/test_device_manager.py +++ b/tests/test_device_manager.py @@ -19,6 +19,8 @@ class FakeTransport: # fail_next_write 触发时抛出的异常实例（默认 SDK 包装后的 TransportError； # 真实硬件路径常是 bleak 裸抛的 BleakError，测试可替换） self.fail_exc = TransportError("connection lost") + self.firmware_revision = "HM42_AIO_V1.0.9" + self.rxtx_uuid = "00001f1f-0000-1000-8000-00805f9b34fb" async def __aenter__(self): self.enters += 1 @@ -204,6 +206,8 @@ class TestStatus: assert s["connect_timeout"] == 90.0 assert s["idle_timeout"] == 300.0 assert s["idle_seconds"] is None # 无先前活动；status 自身不得把读数归零 + assert s["firmware_revision"] == "HM42_AIO_V1.0.9" + assert s["rxtx_uuid"].startswith("00001f1f") await m.close() @pytest.mark.asyncio diff --git a/tests/test_protocol.py b/tests/test_protocol.py index 8b8bdf7..811dfca 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -61,6 +61,12 @@ class TestSetTime: class TestCommandFrames: + def test_rxtx_channel_by_firmware_family(self): + assert P.select_rxtx_uuid("HM42_AIO_V1.0.9") == P.RXTX_CHAR_UUID_LEGACY + assert P.select_rxtx_uuid("hm42_aio_v2") == P.RXTX_CHAR_UUID_LEGACY + assert P.select_rxtx_uuid("PP_da14585_4.2") == P.RXTX_CHAR_UUID + assert P.select_rxtx_uuid(None) == P.RXTX_CHAR_UUID + def test_parking(self): """停车牌 "12" → EF1122（向量#4）""" assert P.parking_frame("12") == bytes([0xEF, 0x11, 0x22]) diff --git a/tests/test_transport_local.py b/tests/test_transport_local.py new file mode 100644 index 0000000..932c36a --- /dev/null +++ b/tests/test_transport_local.py @@ -0,0 +1,109 @@ +"""本机 BLE 传输层的固件通道选择测试。""" + +import pytest + +from ppclock import protocol as P +from ppclock.transports import local + + +class FakeCharacteristic: + def __init__(self, uuid, properties): + self.uuid = uuid + self.properties = properties + + +class FakeServices: + def __init__(self): + self.characters = { + P.RXTX_CHAR_UUID_LEGACY: FakeCharacteristic( + P.RXTX_CHAR_UUID_LEGACY, ["write"]), + P.RXTX_CHAR_UUID: FakeCharacteristic( + P.RXTX_CHAR_UUID, ["read", "write", "notify"]), + } + + def get_characteristic(self, uuid): + return self.characters.get(str(uuid).lower()) + + +class FakeBleakClient: + firmware_revision = "PP_da14585_4.2" + fail_firmware_read = False + + def __init__(self, address, timeout): + self.address = address + self.timeout = timeout + self.services = FakeServices() + self.is_connected = False + self.notifications = [] + self.writes = [] + + async def connect(self): + self.is_connected = True + + async def disconnect(self): + self.is_connected = False + + async def read_gatt_char(self, uuid): + if str(uuid).lower() == P.FIRMWARE_REVISION_UUID: + if self.fail_firmware_read: + raise RuntimeError("DIS unavailable") + return self.firmware_revision.encode() + return b"" + + async def write_gatt_char(self, uuid, data, response=True): + target = uuid.uuid if hasattr(uuid, "uuid") else uuid + self.writes.append((target, bytes(data), response)) + + async def start_notify(self, uuid, callback): + target = uuid.uuid if hasattr(uuid, "uuid") else uuid + self.notifications.append(target) + + async def stop_notify(self, uuid): + self.notifications.remove(uuid) + + +@pytest.mark.asyncio +async def test_hm42_selects_write_only_1f1f(monkeypatch): + class HM42Client(FakeBleakClient): + firmware_revision = "HM42_AIO_V1.0.9" + + monkeypatch.setattr(local, "BleakClient", HM42Client) + transport = local.BLETransport("18:BC:5A:7E:BB:1E") + await transport.connect() + + assert transport.firmware_revision == "HM42_AIO_V1.0.9" + assert transport.rxtx_uuid == P.RXTX_CHAR_UUID_LEGACY + assert not transport._notify_started + await transport.write_rxtx(b"\xe1\x02\x02") + assert transport._client.writes[-1][0] == P.RXTX_CHAR_UUID_LEGACY + with pytest.raises(local.TransportError, match="不支持通知"): + await transport.request_device_id() + await transport.close() + + +@pytest.mark.asyncio +async def test_pp42_keeps_331f_with_notifications(monkeypatch): + monkeypatch.setattr(local, "BleakClient", FakeBleakClient) + transport = local.BLETransport("18:BC:5A:5D:BF:28") + await transport.connect() + + assert transport.rxtx_uuid == P.RXTX_CHAR_UUID + assert transport._notify_started + assert transport._client.notifications == [P.RXTX_CHAR_UUID] + await transport.write_rxtx(b"\xe2") + assert transport._client.writes[-1][0] == P.RXTX_CHAR_UUID + await transport.close() + + +@pytest.mark.asyncio +async def test_unknown_firmware_preserves_331f_default(monkeypatch): + class NoDISClient(FakeBleakClient): + fail_firmware_read = True + + monkeypatch.setattr(local, "BleakClient", NoDISClient) + transport = local.BLETransport("AA:BB:CC:DD:EE:FF") + await transport.connect() + + assert transport.firmware_revision is None + assert transport.rxtx_uuid == P.RXTX_CHAR_UUID + await transport.close()