ha-inlite

commit c89368ead58a300078e6c8dc09db456aec5a188b
parent 4f432941252424bc5eedc2fc8126a96bbb7fe335
Author: Stéphan Kochen <git@stephank.nl>
Date: Sat, 05 Sep 2026 22:11:33 +0200

Handle stream retransmit

diff --git a/custom_components/inlite/lib/inlite_ble/hub.py b/custom_components/inlite/lib/inlite_ble/hub.py
index e2544b7a461b5246e91b8572b4ad2350b857e4f5..04c41587dd89922742ecd74ff89cd3abeac2c75d 100644
--- a/custom_components/inlite/lib/inlite_ble/hub.py
+++ b/custom_components/inlite/lib/inlite_ble/hub.py
@@ -48,6 +48,7 @@ ACK_TIMEOUT = 2.0   # seconds to wait for hub ACK
 STREAM_TIMEOUT = 3.0  # seconds to wait for STREAM response
 BLE_PACKET_PART_SIZE = 78  # Matches the official app's GATT packet splitter.
 STREAM_DATA_SIZE = 62  # Maximum official-app payload per PKT_BLOCK_DATA.
+STREAM_RETRY_LIMIT = 3  # Bound retransmits or out-of-order response packets.
 SMART_HUB_75_PRODUCT_ID = 0x1D
 
 
@@ -492,6 +493,7 @@
     async def _receive_response_stream(self) -> bytes | None:
         """Drain the hub's FLUSH → STREAM* → FLUSH response exchange."""
         dest = self._device_id
+        retries = 0
 
         # The next queued control packet is the hub's response FLUSH(0).
         await self._wait_ack(timeout=STREAM_TIMEOUT)
@@ -502,9 +504,28 @@         try:
             while True:
                 response_type, payload = await self._wait_response_part()
                 if response_type == PKT_BLOCK_STREAM:
+                    offset = payload[0] | (payload[1] << 8) if len(payload) >= 2 else None
+                    if offset != len(self._stream_buffer):
+                        # The hub retransmits a segment when its ACK was lost.
+                        # Match the official app: retain the contiguous data and
+                        # re-ACK the offset we have already received.
+                        retries += 1
+                        if retries > STREAM_RETRY_LIMIT:
+                            _LOGGER.warning("Discarding malformed STREAM response")
+                            return None
+                        _LOGGER.debug(
+                            "Out-of-order STREAM segment at offset %s (expected %d); re-ACKing",
+                            offset,
+                            len(self._stream_buffer),
+                        )
+                        await self._write_mesh(
+                            dest, PKT_BLOCK_ACK, build_ack_payload(len(self._stream_buffer))
+                        )
+                        continue
                     if not self._accept_stream_segment(payload):
                         _LOGGER.warning("Discarding malformed STREAM response")
                         return None
+                    retries = 0
                     # Android acknowledges every segment by its cumulative stream
                     # offset; wait for the next segment or final FLUSH afterwards.
                     await self._write_mesh(
@@ -517,6 +538,20 @@                     payload[0] | (payload[1] << 8)
                     if len(payload) >= 2
                     else len(self._stream_buffer)
                 )
+                if flush_count != len(self._stream_buffer):
+                    retries += 1
+                    if retries > STREAM_RETRY_LIMIT:
+                        _LOGGER.warning("Discarding malformed STREAM response")
+                        return None
+                    _LOGGER.debug(
+                        "STREAM end FLUSH at offset %d (expected %d); re-ACKing",
+                        flush_count,
+                        len(self._stream_buffer),
+                    )
+                    await self._write_mesh(
+                        dest, PKT_BLOCK_ACK, build_ack_payload(len(self._stream_buffer))
+                    )
+                    continue
                 await asyncio.sleep(WRITE_DELAY)
                 # The 0xef terminator is a third byte, not an arithmetic adjustment
                 # of the 16-bit byte count.
diff --git a/tests/test_hub.py b/tests/test_hub.py
index e226dfc2541c0f8ea882f3919173a3f07173f9b6..8beb70785f8bffa2e34dc04e9597dfc5006664bd 100644
--- a/tests/test_hub.py
+++ b/tests/test_hub.py
@@ -191,6 +191,38 @@             (0x72, b"\x05\x00"),
             (0x72, b"\x05\x00\xef"),
         ]
 
+    def test_response_stream_reacks_a_retransmitted_segment(self) -> None:
+        async def run() -> list[tuple[int, bytes]]:
+            hub = InliteHub(device_id=1, passphrase="test")
+            writes: list[tuple[int, bytes]] = []
+
+            async def write_mesh(_dest: int, packet_type: int, data: bytes) -> None:
+                writes.append((packet_type, data))
+
+            hub._write_mesh = write_mesh  # type: ignore[method-assign]
+            hub._ack_queue.put_nowait(b"\x00\x00")
+            parts = iter(
+                [
+                    (PKT_BLOCK_STREAM, b"\x00\x00abc"),
+                    (PKT_BLOCK_STREAM, b"\x00\x00abc"),
+                    (PKT_BLOCK_FLUSH, b"\x03\x00"),
+                ]
+            )
+
+            async def wait_response_part() -> tuple[int, bytes]:
+                return next(parts)
+
+            hub._wait_response_part = wait_response_part  # type: ignore[method-assign]
+            assert await hub._receive_response_stream() == b"abc"
+            return writes
+
+        assert asyncio.run(run()) == [
+            (0x72, b"\x00\x00"),
+            (0x72, b"\x03\x00"),
+            (0x72, b"\x03\x00"),
+            (0x72, b"\x03\x00\xef"),
+        ]
+
     def test_complete_notification_reassembles_continuation(self) -> None:
         received: list[dict] = []
         hub = InliteHub(device_id=1, passphrase="test")