synchros2

At its core, synchros2 is nothing but a collection of utilities and wrappers built on top of rclpy. When used in concert, these utilities and wrappers simplify ROS 2 usage by enabling standard, idiomatic, synchronous Python programming. To that end, synchros2 relies on heavy yet implicit concurrency and thus there is overhead in its simplicity.

Table of contents

• Features
  • Process-wide APIs
  • Actionable and serviced APIs
  • Message feed APIs
• Guidelines
  • Integration testing

Features

Process-wide APIs

Process-wide APIs are built around the notion of a ROS 2 aware scope. ROS 2 aware scopes manage the lifetime of a thread local graph of ROS 2 nodes, along with an executor that dispatches work for them. ROS 2 nodes may be loaded and unloaded (i.e. instantiated and put to spin, and explicitly destroyed, respectively) or have their entire lifecycle be managed (i.e. bound to a context manager). A ROS 2 aware scope may also define a main ROS 2 node (for ease of use, for log forwarding, etc.). ROS 2 aware scopes may be nested, enforcing locality of ROS 2 usage in a codebase, though the innermost scope is always accessible through synchros2.scope module-level APIs.

A ROS 2 aware scope may also be process local (i.e. global within a process), which allows for the notion of a ROS 2 aware process. Only one ROS 2 aware process may be active at any point in time as a top-level scope i.e. a ROS 2 aware process may not start within an existing ROS 2 aware scope. The current ROS 2 aware process and associated scope are always accessible process-wide through synchros2.process module-level APIs.

These notions afford process-wide (and thread-wide) access to locally managed ROS 2 entities and thus ownership and lifetime is well defined. Moreover, callbacks are dispatched in the background by default, enabling both synchronous and asynchronous programming out-of-the-box.

These APIs are also quite handy to reconcile past rospy experience with ROS 2.

Common use cases

Setting up single node processes

To make use of ROS 2 without getting into the details, just decorate your executable entrypoint (or main function) with synchros2.process.main:

import logging
import time

import synchros2.process as ros_process

import std_msgs.msg

@ros_process.main()
def entrypoint() -> None:
    # no need to initialize, it is automatic
    node = ros_process.node()  # or entrypoint.node
    assert node is not None
    pub = node.create_publisher(std_msgs.msg.String, "test", 1)

    def callback() -> None:
        time.sleep(10)  # you can block in a callback
        pub.publish(std_msgs.msg.String(data="testing"))
    executor = ros_process.executor()  # or entrypoint.executor
    assert executor is not None
    executor.create_task(callback)  # dispatch callback for execution

    time.sleep(10)  # you can block in the main thread

    node.get_logger().info("testing")
    logging.info("testing")  # you can use Python logging

    try: 
        ros_process.wait_for_shutdown()  # you can wait for Ctrl + C
    except KeyboardInterrupt:
        pass  # to avoid traceback printing

    return  # no need to cleanup or shutdown, it is automatic

if __name__ == "__main__":
    entrypoint()

Note a ROS 2 node and an executor are accessible process-wide through synchros2.process module APIs. This is ideal for quick prototyping and simple scripts, as the UX is largely intuitive e.g. you can make blocking calls from virtually anywhere.

Setting up multi-node processes

You can spin as many ROS 2 nodes as you need, and skip the default process-wide node if unnecessary. Here’s an example that loads three (3) ROS 2 nodes and spins them indefinitely:

```python import logging import time

from typing import Any, List

import synchros2.process as ros_process from synchros2.node import Node

File truncated at 100 lines see the full file