asyncio_for_robotics 1.3.5

Asyncio interface for ROS 2, Zenoh, and other robotic middlewares.

Asyncio For Robotics

Requirements	Compatibility	Tests

The Asyncio For Robotics (afor) library makes asyncio usable with ROS 2, Zenoh and more, letting you write linear, testable, and non-blocking Python code.

• Better syntax.
• Only native python: Better docs and support.
• Simplifies testing.

Will this make my code slower? No.

Will this make my code faster? No. However, asyncio will help YOU write better, faster code.

[!TIP] asyncio_for_robotics interfaces do not replace their primary interfaces! We add capabilities, giving you more choices, not less.

Install

Barebone

Compatible with ROS 2 (jazzy,humble and newer) out of the box. This library is pure python (>=3.10), so it installs easily.

pip install asyncio_for_robotics

Along with Zenoh

pip install asyncio_for_robotics eclipse-zenoh

Read more

• Detailed ROS 2 tutorial
• Detailed examples
  • no talking 🦍 show me code 🦍
• Cross-Platform deployment even with ROS
• Usage for software testing

Available interfaces:

• Rate: Every tick of a clock. (native)
• TextIO: stdout lines of a Popen process (and other TextIO files). (native)
• ROS 2: Subscriber, Service Client, Service Server.
• Zenoh: Subscriber.
• Implement your own interface!

[!TIP] An interface is not required for every operation. ROS 2 native publishers and nodes work just fine. Furthermore, advanced behavior can be composed of generic afor object (see ROS2 Event Callback Example).

Additional Projects and Interfaces

• gogo_keyboard: Subscribe to keyboard key presses and release.
• asyncio_gazebo: Subscribe to Gazebo transport.

Code sample

Syntax is identical between ROS 2, Zenoh, TextIO, Rate...

Wait for messages one by one

Application:

• Get the latest sensor data
• Get clock value
• Wait for trigger
• Wait for next tick of the Rate
• Wait for system to be operational

sub = afor.Sub(...)

# Get the latest message
latest = await sub.wait_for_value()

# Get a new message
new = await sub.wait_for_new()

# Get the next message received
next = await sub.wait_for_next()

Continuously listen to a data stream

Application:

• Process a whole data stream
• React to changes in sensor data
• Execute on every tick of the Rate

# Continuously process the latest messages
async for msg in sub.listen():
    status = foo(msg)
    if status == DONE:
        break

# Continuously process all incoming messages
async for msg in sub.listen_reliable():
    status = foo(msg)
    if status == DONE:
        break

Improved Services / Queryable for ROS 2

[!NOTE] This is only for ROS 2.

Application:

• Client request reply from a server.
• Servers can delay their response without blocking (not possible in native ROS 2)

# Server is once again a afor subscriber, but generating responder objects
server = afor.Server(...)

# processes all requests.
# listen_reliable method is recommanded as it cannot skip requests
async for responder in server.listen_reliable():
    if responder.request == "PING!":
        reponder.response = "PONG!"
        await asyncio.sleep(...) # reply can be differed
        reponder.send()
    else:
        ... # reply is not necessary

# the client implements a async call method
client = afor.Client(...)

response = await client.call("PING!")

Process for the right amount of time

Application:

• Test if the system is responding as expected
• Run small tasks with small and local code

# Listen with a timeout
data = await afor.soft_wait_for(sub.wait_for_new(), timeout=1)
if isinstance(data, TimeoutError):
    pytest.fail(f"Failed to get new data in under 1 second")


# Process a codeblock with a timeout
async with afor.soft_timeout(1):
    sum = 0
    total = 0
    async for msg in sub.listen_reliable():
        number = process(msg)
        sum += number
        total += 1

last_second_average = sum/total
assert last_second_average == pytest.approx(expected_average)

Apply pre-processing to a data-stream

Application:

• Parse payload of different transport into a common type.

# ROS2 String type afor subscriber
inner_sub: Sub[String] = afor.ros2.Sub(String, "topic_name")
# converted into a subscriber generating python `str`
ros2str_func = lambda msg: msg.data
sub: Sub[str] = afor.ConverterSub(sub=inner_sub, convert_func=ros2str_func)

About Speed

The inevitable question: “But isn’t this slower than the ROS 2 executor? ROS 2 is the best!”

In short: rclpy's executor is the bottleneck.

• Comparing to the best ROS 2 Jazzy can do (SingleThreadedExecutor), afor increases latency from 110us to 150us.
• Comparing to other execution methods, afor is equivalent if not faster.
• If you find it slow, you should use C++ or Zenoh (or contribute to this repo?).

Benchmark code is available in ./tests/bench/, it consists in two pairs of pub/sub infinitely echoing a message (using one single node). The messaging rate, thus measures the request to response latency.

With afor	Transport	Executor		Frequency (kHz)	Latency (ms)
✔️	Zenoh	None		95	0.01
✔️	ROS 2	Experimental Asyncio		17	0.06
❌	ROS 2	Experimental Asyncio		13	0.08
❌	ROS 2	SingleThreaded		9	0.11
✔️	ROS 2	SingleThreaded		7	0.15
✔️	ROS 2	MultiThreaded		3	0.3
❌	ROS 2	MultiThreaded		3	0.3
✔️	ROS 2	ros_loop Method		3	0.3

Details:

• uvloop was used, replacing the asyncio executor (more or less doubles the performances for Zenoh)
• RMW was set to rmw_zenoh_cpp
• ROS2 benchmarks uses afor's ros2.ThreadedSession (the default in afor).
• Only the Benchmark of the ros_loop method uses afor's second type of session: ros2.SynchronousSession.
• ROS 2 executors can easily be changed in afor when creating a session.
• The experimental AsyncioExecutor PR on ros rolling by nadavelkabets is incredible https://github.com/ros2/rclpy/pull/1399. Maybe I will add proper support for it (but only a few will want to use an unmerged experimental PR of ROS 2 rolling).
• If there is interest in those benchmarks I will improve them, so others can run them all easily.

Analysis:

• Zenoh is extremely fast, proving that afor is not the bottleneck.
• This AsyncioExecutor having better perf when using afor is interesting, because afor does not bypass code.
  • I think this is due to AsyncioExecutor having some overhead that affects its own callback.
  • Without afor the ROS 2 callback executes some code and publishes.
  • With afor the ROS 2 callback returns immediately, and fully delegates execution to asyncio.
• The increase of latency on the SingleThreaded executors proves that getting data in and out of the rclpy executor and thread is the main bottleneck.
  • AsyncioExecutor does not have such thread, thus can directly communicate.
  • Zenoh has its own thread, however it is built exclusively for multi-thread operations, without any executor. Thus achieves far superior performances.
• MultiThreadedExecutor is just famously slow.
• Very surprisingly, the well known ros_loop method detailed here https://github.com/m2-farzan/ros2-asyncio is slow.