Fix Battery Guard BM6 empty realtime replies

bmslib/models/bm6.py

@@ -37,6 +37,7 @@
 
 CMD_REALTIME = bytes.fromhex('d1550700000000000000000000000000')
 CMD_VERSION = bytes.fromhex('d1550100000000000000000000000000')
+EMPTY_REALTIME = bytes.fromhex('d15507ff000000000000000000000000')
 
 _STATES = {0: 'ok', 1: 'low_voltage', 2: 'charging'}
 
@@ -166,8 +167,14 @@ def decode_bm2_realtime(plain: bytes) -> dict:
 
 def decode_realtime(plain: bytes) -> dict:
     """Decode a decrypted realtime reply (d1 55 07 ...)."""
-    if plain[:3] != b'\xd1\x55\x07':
+    if len(plain) != 16 or plain[:3] != b'\xd1\x55\x07':
         raise ValueError(f"not a BM6 realtime frame: {plain.hex()}")
+    # Some Battery Guard devices emit this exact empty sentinel while
+    # notifications are being initialized. Reject only the observed sentinel,
+    # not every unknown value in byte 3: this decoder serves other BM6 variants
+    # whose status extensions we cannot test here.
+    if plain == EMPTY_REALTIME:
+        raise ValueError(f"empty BM6 realtime frame: {plain.hex()}")
     temp = plain[4]
     if plain[3] == 1:
         temp = -temp
@@ -189,6 +196,7 @@ def __init__(self, address, **kwargs):
         kwargs.setdefault('_uses_pin', False)
         super().__init__(address, **kwargs)
         self._last_response = None
+        self._empty_realtime = asyncio.Event()
 
     def _notification_handler(self, sender, data):
         try:
@@ -201,8 +209,20 @@ def _notification_handler(self, sender, data):
         self._route(plain)
 
     def _route(self, plain: bytes):
-        if plain[:2] == b'\xd1\x55':
-            self._fetch_futures.set_result(plain[2], plain)
+        if plain[:2] != b'\xd1\x55':
+            return
+
+        cmd = plain[2]
+        if cmd == 0x07:
+            try:
+                decode_realtime(plain)
+            except ValueError as e:
+                self.logger.debug("%s ignoring invalid realtime notification: %s", self.name, e)
+                if plain == EMPTY_REALTIME:
+                    self._empty_realtime.set()
+                return
+
+        self._fetch_futures.set_result(cmd, plain)
 
     async def connect(self, **kwargs):
         try:
@@ -220,9 +240,41 @@ async def disconnect(self):
 
     async def _q(self, cmd: int):
         plain = {0x07: CMD_REALTIME, 0x01: CMD_VERSION}[cmd]
+
+        # Most BM6 variants expect notifications to be active before the request.
+        # The Battery Guard variant from #408 instead returns EMPTY_REALTIME for
+        # that ordering and only supplies measurements when the request is written
+        # before subscribing. Switch only devices that exhibit that exact reply.
+        if cmd == 0x07 and self._empty_realtime.is_set():
+            return await self._q_write_before_notify(cmd, plain)
+
+        with self._fetch_futures.acquire(cmd):
+            await self.client.write_gatt_char(self.UUID_TX, data=bm6_encrypt(plain, self.KEY),
+                                              response=True)
+            if cmd != 0x07:
+                return await self._fetch_futures.wait_for(cmd, self.TIMEOUT)
+
+            response_waiter = asyncio.create_task(self._fetch_futures.wait_for(cmd, self.TIMEOUT))
+            empty_waiter = asyncio.create_task(self._empty_realtime.wait())
+            done, pending = await asyncio.wait(
+                (response_waiter, empty_waiter), return_when=asyncio.FIRST_COMPLETED)
+            for waiter in pending:
+                waiter.cancel()
+            if pending:
+                await asyncio.gather(*pending, return_exceptions=True)
+
+            if response_waiter in done:
+                return response_waiter.result()
+
+        self.logger.debug("%s retrying realtime request before notify subscription", self.name)
+        return await self._q_write_before_notify(cmd, plain)
+
+    async def _q_write_before_notify(self, cmd: int, plain: bytes):
+        await self.stop_notify(self.UUID_RX)
         with self._fetch_futures.acquire(cmd):
             await self.client.write_gatt_char(self.UUID_TX, data=bm6_encrypt(plain, self.KEY),
                                               response=True)
+            await self.start_notify(self.UUID_RX, self._notification_handler)
             return await self._fetch_futures.wait_for(cmd, self.TIMEOUT)
 
     async def fetch(self) -> BmsSample:

bmslib/test/test_bm6_decode.py

@@ -11,6 +11,7 @@
 CMD_PLAIN = bytes.fromhex('d1550700000000000000000000000000')
 NOTIFY_CIPHER = bytes.fromhex('5a7a41c3a57ca1fa9247f76557c5d618')
 NOTIFY_PLAIN = bytes.fromhex('d155070017010004ab00000000020000')
+INVALID_INITIAL_PLAIN = bytes.fromhex('d15507ff000000000000000000000000')
 
 
 def test_aes_fips197_vector():
@@ -34,6 +35,18 @@ def test_bm6_decode_realtime():
     assert decode_realtime(bytes(neg))['temperature'] == -5
 
 
+def test_bm6_rejects_invalid_initial_realtime_frame():
+    with pytest.raises(ValueError, match="empty BM6 realtime frame"):
+        decode_realtime(INVALID_INITIAL_PLAIN)
+
+
+def test_bm6_keeps_unknown_nonempty_status_compatible():
+    variant = bytearray(NOTIFY_PLAIN)
+    variant[3] = 0x02
+
+    assert decode_realtime(bytes(variant))["temperature"] == 23
+
+
 def test_bm6_fetch_sample():
     bms = Bm6Bt("00:11:22:33:44:55", name="bm6")
     sample = run_fetch_with_response(bms, NOTIFY_PLAIN)
@@ -55,6 +68,65 @@ async def run():
     assert asyncio.run(run()) == NOTIFY_PLAIN
 
 
+def test_bm6_ignores_initial_frame_and_waits_for_measurement():
+    async def run():
+        bms = Bm6Bt("00:11:22:33:44:55", name="bm6")
+        with bms._fetch_futures.acquire(0x07):
+            bms._notification_handler(None, bm6_encrypt(INVALID_INITIAL_PLAIN))
+            bms._notification_handler(None, NOTIFY_CIPHER)
+            return await bms._fetch_futures.wait_for(0x07, 1)
+
+    assert asyncio.run(run()) == NOTIFY_PLAIN
+
+
+def test_bm6_empty_reply_switches_to_write_before_notify():
+    async def run():
+        bms = Bm6Bt("00:11:22:33:44:55", name="bm6")
+        calls = []
+
+        class Client:
+            async def write_gatt_char(self, *_args, **_kwargs):
+                calls.append("write")
+                if calls.count("write") == 1:
+                    bms._notification_handler(None, bm6_encrypt(INVALID_INITIAL_PLAIN))
+
+        async def stop_notify(_uuid):
+            calls.append("stop_notify")
+
+        async def start_notify(_uuid, callback):
+            calls.append("start_notify")
+            callback(None, NOTIFY_CIPHER)
+
+        bms.client = Client()
+        bms.stop_notify = stop_notify
+        bms.start_notify = start_notify
+        result = await bms._q(0x07)
+        return result, calls
+
+    result, calls = asyncio.run(run())
+    assert result == NOTIFY_PLAIN
+    assert calls == ["write", "stop_notify", "write", "start_notify"]
+
+
+def test_bm6_valid_reply_keeps_master_notify_order():
+    async def run():
+        bms = Bm6Bt("00:11:22:33:44:55", name="bm6")
+
+        class Client:
+            async def write_gatt_char(self, *_args, **_kwargs):
+                bms._notification_handler(None, NOTIFY_CIPHER)
+
+        async def unexpected_notify_change(_uuid, *_args):
+            pytest.fail("valid BM6 replies must not change the notification subscription")
+
+        bms.client = Client()
+        bms.stop_notify = unexpected_notify_change
+        bms.start_notify = unexpected_notify_change
+        return await bms._q(0x07)
+
+    assert asyncio.run(run()) == NOTIFY_PLAIN
+
+
 def test_bm6_rejects_wrong_length():
     with pytest.raises(ValueError):
         bm6_decrypt(b'\x00' * 15)