ha-inlite

commit 971f56d4a6de6747af36fd8869a31937804e7531
parent 358ca1f20db5de7d29e3da06fc78c09c34bc755d
Author: Stéphan Kochen <git@stephank.nl>
Date: Thu, 03 Sep 2026 20:29:14 +0200

Segment outgoing CSRmesh request streams

diff --git a/custom_components/inlite/lib/inlite_ble/hub.py b/custom_components/inlite/lib/inlite_ble/hub.py
index dc667745b7b34e05014aa4d2415c19dd49d99dfe..9fa4e2af23a657c694f65c5be7fdd3f528147b4a 100644
--- a/custom_components/inlite/lib/inlite_ble/hub.py
+++ b/custom_components/inlite/lib/inlite_ble/hub.py
@@ -46,6 +46,7 @@ WRITE_DELAY = 0.06  # 60ms between BLE writes (matches app timing)
 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.
 
 
 class ZoneState:
@@ -420,20 +421,41 @@         self._stream_buffer.clear()
         self._stream_invalid = False
         while not self._ack_queue.empty():
             self._ack_queue.get_nowait()
+
+        # ``build_block_data_payload`` retains its offset prefix for callers
+        # and tests. The stream transport owns that prefix, however, so strip
+        # its initial zero offset before transmitting numbered segments.
+        # Association is already an offset-free vendor payload.
+        stream_payload = block_data[2:] if block_data.startswith(b"\x00\x00") else block_data
         await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(0))
         await self._wait_ack()
 
-        await self._write_mesh(dest, PKT_BLOCK_DATA, block_data)
-        ack = await self._wait_ack()
-
-        # Parse acked byte count
-        if len(ack) >= 2:
-            acked_bytes = ack[0] | (ack[1] << 8)
-        else:
-            acked_bytes = len(block_data)
+        offset = 0
+        while offset < len(stream_payload):
+            segment = stream_payload[offset : offset + STREAM_DATA_SIZE]
+            await self._write_mesh(
+                dest,
+                PKT_BLOCK_DATA,
+                offset.to_bytes(2, "little") + segment,
+            )
+            ack = await self._wait_ack()
+            if len(ack) >= 2:
+                acked_bytes = ack[0] | (ack[1] << 8)
+                # Acknowledgements carry the total stream offset. Avoid a
+                # malformed/stale ACK causing a non-progressing loop.
+                if offset < acked_bytes <= offset + len(segment):
+                    offset = acked_bytes
+                    continue
+                _LOGGER.warning(
+                    "Unexpected DATA ACK offset %d (expected %d..%d)",
+                    acked_bytes,
+                    offset + 1,
+                    offset + len(segment),
+                )
+            offset += len(segment)
 
         # Step 3: Flush (end)
-        await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(acked_bytes))
+        await self._write_mesh(dest, PKT_BLOCK_FLUSH, build_flush_payload(offset))
         ack = await self._wait_ack()
 
         # Check for completion marker (0xef suffix)
diff --git a/tests/test_hub.py b/tests/test_hub.py
index 5e6c3841ca68d9eaaed6c83d3f8d7e4266b26055..a625d0b24d3d547605e59e3b9388d25271de7dbe 100644
--- a/tests/test_hub.py
+++ b/tests/test_hub.py
@@ -2,7 +2,7 @@ """Tests for inlite_ble hub module — ZoneState and notification safety."""
 
 import asyncio
 
-from inlite_ble.hub import BLE_PACKET_PART_SIZE, InliteHub, ZoneState
+from inlite_ble.hub import BLE_PACKET_PART_SIZE, STREAM_DATA_SIZE, InliteHub, ZoneState
 from inlite_ble.protocol import CHAR_CONTINUATION_UUID, CHAR_WRITE_UUID
 
 
@@ -131,6 +131,32 @@         client = asyncio.run(run())
         assert client.writes == [
             (CHAR_CONTINUATION_UUID, b"x" * BLE_PACKET_PART_SIZE, True),
             (CHAR_WRITE_UUID, b"x", True),
+        ]
+
+    def test_request_stream_is_segmented_and_offset_acknowledged(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))
+
+            acks = iter((b"\x00\x00", b"\x3e\x00", b"\x3f\x00", b"\x3f\x00\xef"))
+
+            async def wait_ack(_timeout: float = 0) -> bytes:
+                return next(acks)
+
+            hub._write_mesh = write_mesh  # type: ignore[method-assign]
+            hub._wait_ack = wait_ack  # type: ignore[method-assign]
+            assert await hub._send_raw_stream(b"\x00\x00" + b"x" * 63)
+            return writes
+
+        writes = asyncio.run(run())
+        assert writes == [
+            (0x70, b"\x00\x00"),
+            (0x71, b"\x00\x00" + b"x" * STREAM_DATA_SIZE),
+            (0x71, b"\x3e\x00x"),
+            (0x70, b"\x3f\x00"),
         ]
 
     def test_teach_in_tlv_is_forwarded(self) -> None: