How to bridge rclpy with Python's asyncio?

Question

In my opinion, the correct way of running ROS2 nodes inside a larger asynchronous Python application (e.g., an application built around asyncio) is not particularly obvious. I think a good solution should (i) not busy-wait the middleware and (ii) should handle cancellation properly.

I have had go at this and found a solution (see my answer) and would like to know if anyone else has a better solution or can see any flaws in my solution.

Answer 1

The following code creates a bridge between rclpy.spin and an application built around Python's asyncio. The guard condition is used to interrupt rmw_wait and hence return early from spin_once (to handle the case where nothing else is happening). I am using Python 3.6, but I think the same principles apply to later versions of Python with the addition of some nicer APIs being available.

import threading
import asyncio
import rclpy
from rclpy.node import Node

async def spin(node: Node):
    cancel = node.create_guard_condition(lambda: None)
    def _spin(node: Node,
              future: asyncio.Future,
              event_loop: asyncio.AbstractEventLoop):
        while not future.cancelled():
            rclpy.spin_once(node)
        if not future.cancelled():
            event_loop.call_soon_threadsafe(future.set_result, None)
    event_loop = asyncio.get_event_loop()
    spin_task = event_loop.create_future()
    spin_thread = threading.Thread(target=_spin, args=(node, spin_task, event_loop))
    spin_thread.start()
    try:
        await spin_task
    except asyncio.CancelledError:
        cancel.trigger()
    spin_thread.join()
    node.destroy_guard_condition(cancel)

This code can then be used, for example, like this:

async def main():
    # create a node without any work to do
    node = Node('unused')

    # create tasks for spinning and sleeping
    spin_task = asyncio.get_event_loop().create_task(spin(node))
    sleep_task = asyncio.get_event_loop().create_task(asyncio.sleep(5.0))

    # concurrently execute both tasks
    await asyncio.wait([spin_task, sleep_task], return_when=asyncio.FIRST_COMPLETED)

    # cancel tasks
    if spin_task.cancel():
        await spin_task
    if sleep_task.cancel():
        await sleep_task

# run program
rclpy.init()
asyncio.get_event_loop().run_until_complete(main())
asyncio.get_event_loop().close()
rclpy.shutdown()