How to make bleak compatibile with cmd, cmd2, or other terminal apps?

[dpkristensen]

I've had some trouble getting bleak to work inside cmd2, but the problem would also be present in other frameworks such as cmd or GUI/TUI applications like tkinter or Textual. All of this is happening on Windows, but I would imagine the same restrictions may appear on other OS's like Linux or MacOS/BSD.

Bleak is built from the ground up around asyncio, meaning it expects the main thread to be running async from the top down on the main process thread. There's several cases where this is not feasible:

• cmd/cmd2 will block on the main thread to read the command line
• TUI programs receive console events on stdin, (resizing the window, events from Windows, mouse clicks, keyboard input, etc...)
• GUI programs must use an event loop to check for events from the OS

I wanted background scanning functionality but bleak doesn't support that, so I set my expectations to a time-bounded scanning operation.

I tried creating a separate thread (threading module) and run the thread as async with a threading.Queue to process control commands, but I ran into asyncio exceptions like "operation cancelled by user", and the BLE connection would not stay alive when I connected to the peripheral. I'm guessing this is because blocking the second thread for events somehow prevented the host from sending during the connection windows.

So I removed the thread and created an async loop through which to funnel all of my async interactions with bleak:

asyncio_main_loop = asyncio.get_event_loop()

def async_run(func: typing.Coroutine) -> typing.Any:
    return asyncio_main_loop.run_until_complete(func)

This worked fine until I connected to a GATT server and tried to set up notify callbacks. The notify callback was not received until after I quit the program.

I think the answer must rely on getting threading and asyncio to play nicely, so what's the right way to make this possible using an event-driven control thread without a continuous asyncio loop?

[dlech]

There have been several other discussions about using Bleak in a non-asyncio app, e.g #826.

TL;DR; there should only be one asynio.run() in the app or thread that never stops until the app exists rather than running a loop intermittently.

[dpkristensen]

It's frustrating that there's no built-in functionality to the asyncio library that does this, as it seems this is a too-common problem with a not-so-common solution.

I had searched for cmd and cmd2 in the discussion here with no results so maybe this will at least improve the visibility of this solution. Not sure if there are any libraries out there that provide this functionality, but maybe someone will make one. It might have been nice if bleak did not rely on asyncio in the public interface, but it's too late to change it. Maybe for bleak2! 😄

I got this working by creating a singleton object that encapsulates a thread and the asyncio loop execution. Anyone may feel free to use this or put it into a PyPi library:

class AsyncIoThread:
    _instance = None

    def __init__(self):
        self._loop: asyncio.AbstractEventLoop
        self._thread: threading.Thread
        raise RuntimeError("Use instance() instead")

    @classmethod
    def instance(cls) -> "AsyncIoThread":
        # pylint: disable=protected-access
        if cls._instance is None:
            cls._instance = cls.__new__(cls)
            cls._instance._loop = asyncio.get_event_loop()
            cls._instance._thread = threading.Thread(
                target=cls._instance._thread_func,
                name="AsyncIo"
            )
            cls._instance._thread.start()
        return cls._instance

    @typing.overload
    def call(self, func, default_ret: typing.Any=None, timeout: float=None) -> typing.Any: ...

    def call(self, func, **kwargs) -> typing.Any:
        """
        Call func on the thread and wait for the result to return

        timeout and default_ret must be specified as keyword arguments.

        If default_ret is specified, asyncio.CancelledError and
        asyncio.TimeoutError will cause the given value to be returned
        instead of raising an exception (regardless of whether timeout
        was specified).

        asyncio.CancelledError and asyncio.TimeoutError can also be
        caught with the python standard TimeoutError exception
        """
        if asyncio.coroutines.iscoroutine(func):
            future = asyncio.run_coroutine_threadsafe(func, self._loop)
            timeout = kwargs.get("timeout", None)
            try:
                return future.result(timeout=timeout)
            except (asyncio.CancelledError, asyncio.TimeoutError) as e:
                if "default_ret" in kwargs:
                    return kwargs["default_ret"]
                else:
                    # Re-raise to allow catching as a core exception
                    raise TimeoutError() from e
        else:
            raise RuntimeError(f"Unwaitable object {func}")

    def schedule(self, func):
        """Schedule the execution of func without waiting for the result"""
        if asyncio.coroutines.iscoroutine(func):
            asyncio.run_coroutine_threadsafe(func, self._loop)
        elif callable(func):
            self._loop.call_soon_threadsafe(func)
        else:
            raise RuntimeError(f"Uncallable object {func}")

    def stop(self):
        """
        Stop the asyncio thread and wait for shutdown

        This should be done just prior to exiting the program so that all scheduled
        operations can be stopped cleanly to allow for cleanup of resources.
        """
        # Schedule the stop so it can be done after all other queued items
        self.schedule(self._thread_stop)
        self._thread.join()

    def _thread_stop(self):
        self._loop.stop()

    def _thread_func(self):
        self._loop.run_forever()

def get_asyncio_thread() -> AsyncIoThread:
    return AsyncIoThread.instance()

I discovered that background scanning was indeed functional, just not working due to blocking the asyncio loop. I don't know what the CPU usage of this loop is while idle, but I suppose it depends on the asyncio implementation; my application is not performance critical anyway.

Example: A scheduled operation that does not wait for return:

class MyBleAdapter:
    def __init__(self):
        self._scanner: bleak.BleakScanner = None

    def scan_start(self):
        get_asyncio_thread().schedule(self._scan_start_async())

    async def _scan_start_async(self):
        self._scanner = bleak.BleakScanner(
            detection_callback=_on_scan_detection,
            service_uuids=[MY_SERVICE_UUID]
        )
        await _scanner.start()
        log.info("Scanning started")

Or for a synchronous operation that handles the exception implicitly:

from bleak import BleakClient
from bleak.backends.characteristic import BleakGATTCharacteristic

def read_gatt_chrc(client: BleakClient, chrc: BleakGATTCharacteristic) -> bytes:
    return get_asyncio_thread().call(
        _read_gatt_chrc_async(client, chrc),
        timeout=10.0,
        default_ret=bytes()
    )

async def _read_gatt_chrc_async(client: BleakClient, chrc: BleakGATTCharacteristic) -> bytes:
    if client.is_connected:
        return bytes(await client.read_gatt_char(chrc))
    else:
        return bytes()