dora.dora

class Ros2Context:

ROS2 Context holding all messages definition for receiving and sending messages to ROS2.

By default, Ros2Context will use env AMENT_PREFIX_PATH to search for message definition.

AMENT_PREFIX_PATH folder structure should be the following:

  • For messages: /msg/.msg
  • For services: /srv/.srv

You can also use ros_paths if you don't want to use env variable.

warning:: dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.

context = Ros2Context()
def new_node(self, /, name, namespace, options):

Create a new ROS2 node

ros2_node = ros2_context.new_node(
    "turtle_teleop",
    "/ros2_demo",
    Ros2NodeOptions(rosout=True),
)

warning:: dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.

class Ros2Node:

ROS2 Node

warnings::

  • dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.
  • There's a known issue about ROS2 nodes not being discoverable by ROS2 See: https://github.com/jhelovuo/ros2-client/issues/4
def create_topic(self, /, name, message_type, qos):

Create a ROS2 topic to connect to.

turtle_twist_topic = ros2_node.create_topic(
    "/turtle1/cmd_vel", "geometry_msgs/Twist", topic_qos
)
def create_publisher(self, /, topic, qos=None):

Create a ROS2 publisher

pose_publisher = ros2_node.create_publisher(turtle_pose_topic)

warnings:

  • dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.
def create_subscription(self, /, topic, qos=None):

Create a ROS2 subscription

pose_reader = ros2_node.create_subscription(turtle_pose_topic)

warnings:

  • dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.
class Ros2NodeOptions:

ROS2 Node Options

class Ros2Topic:

ROS2 Topic

warnings:

  • dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.
class Ros2Publisher:

ROS2 Publisher

warnings:

  • dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.
def publish(self, /, data):

Publish a message into ROS2 topic.

Remember that the data format should respect the structure of the ROS2 message using an arrow Structure.

ex:

gripper_command.publish(
    pa.array(
        [
            {
                "name": "gripper",
                "cmd": np.float32(5),
            }
        ]
    ),
)
class Ros2Subscription:

ROS2 Subscription

warnings:

  • dora Ros2 bridge functionality is considered unstable. It may be changed at any point without it being considered a breaking change.
def next(self, /):
class Ros2QosPolicies:

ROS2 QoS Policy

class Ros2Durability:

DDS 2.2.3.4 DURABILITY

class Ros2Liveliness:

DDS 2.2.3.11 LIVELINESS

ManualByParticipant = Ros2Liveliness.ManualByParticipant
def start_runtime():

Start a runtime for Operators

def run(dataflow_path, uv=None):

Run a Dataflow, exactly the same way as dora run command line tool.

def build( dataflow_path, uv=None, coordinator_addr=None, coordinator_port=None, force_local=False):

Build a Dataflow, exactly the same way as dora build command line tool.

class Node:

The custom node API lets you integrate dora into your application. It allows you to retrieve input and send output in any fashion you want.

Use with:

from dora import Node

node = Node()
def next(self, /, timeout=None):

.next() gives you the next input that the node has received. It blocks until the next event becomes available. You can use timeout in seconds to return if no input is available. It will return None when all senders has been dropped.

event = node.next()

You can also iterate over the event stream with a loop

for event in node:
   match event["type"]:
       case "INPUT":
           match event["id"]:
                case "image":
def recv_async(self, /, timeout=None):

.recv_async() gives you the next input that the node has received asynchronously. It does not blocks until the next event becomes available. You can use timeout in seconds to return if no input is available. It will return an Error if the timeout is reached. It will return None when all senders has been dropped.

warning:: This feature is experimental as pyo3 async (rust-python FFI) is still in development.

event = await node.recv_async()

You can also iterate over the event stream with a loop

def send_output(self, /, output_id, data, metadata=None):

send_output send data from the node.

Args:
   output_id: str,
   data: pyarrow.Array,
   metadata: Option[Dict],

ex:

node.send_output("string", b"string", {"open_telemetry_context": "7632e76"})
def dataflow_descriptor(self, /):

Returns the full dataflow descriptor that this node is part of.

This method returns the parsed dataflow YAML file.

def node_config(self, /):

Returns the node configuration.

def dataflow_id(self, /):

Returns the dataflow id.

def merge_external_events(self, /, subscription):

Merge an external event stream with dora main loop. This currently only work with ROS2.