Method read_lines

ci/util/ble_interface.py:98–116

Async iterator yielding lines received via BLE NOTIFY. Args: timeout: Maximum time to wait for lines in seconds. Yields: Lines from BLE notifications (stripped of whitespace).

(self, timeout: float)

        await self._client.write_gatt_char(BLE_CHAR_RX_UUID, payload, response=True)

    async def read_lines(self, timeout: float) -> AsyncIterator[str]:
        """Async iterator yielding lines received via BLE NOTIFY.

        Args:
            timeout: Maximum time to wait for lines in seconds.

        Yields:
            Lines from BLE notifications (stripped of whitespace).
        """
        deadline = asyncio.get_event_loop().time() + timeout
        while True:
            remaining = deadline - asyncio.get_event_loop().time()
            if remaining <= 0:
                break
            try:
                line = await asyncio.wait_for(self._rx_queue.get(), timeout=remaining)
                yield line
            except asyncio.TimeoutError:
                break