使用Dora-rs项目实现ROS2与Dora的Marker消息桥接

概述

Dora-rs是一个新兴的数据流编程框架，它提供了与ROS2系统的桥接能力。本文将详细介绍如何利用Dora-rs项目实现ROS2系统中的Marker消息类型在Dora框架中的发布，这对于需要在Dora环境中可视化数据的开发者来说非常有用。

Marker消息简介

Marker消息是ROS2中visualization_msgs包定义的一种消息类型，广泛用于在RViz等可视化工具中显示各种形状和标记。它支持多种基本形状如立方体、球体、箭头等，并可以设置位置、朝向、颜色、尺寸等属性。

Dora-rs中的ROS2桥接实现

Dora-rs项目通过其ros2_bridge模块提供了与ROS2系统的交互能力。要实现Marker消息的发布，需要以下几个关键步骤：

1. 初始化ROS2上下文：创建ROS2运行环境
2. 创建ROS2节点：在ROS2网络中注册一个节点
3. 定义QoS策略：设置消息发布的质量服务参数
4. 创建主题：指定要发布的Marker消息主题
5. 创建发布者：用于实际发布消息的对象

Python实现详解

以下是使用Python在Dora中发布Marker消息的完整实现：

from typing import Callable, Optional
import pyarrow as pa
from dora import DoraStatus
import dora
import numpy as np

# 定义Marker消息数据结构
marker = pa.array(
    [
        {
            "header": {"frame_id": "world"},
            "ns": "my_namespace",
            "id": np.int32(1),
            "type": np.int32(0),  # ARROW类型
            "action": np.int32(0),  # ADD动作
            "lifetime": {"sec": np.int32(1), "nanosec": np.uint32(2)},
            "pose": {
                "position": {
                    "x": np.float64(1.0),
                    "y": np.float64(2.0),
                    "z": np.float64(3.0)
                },
                "orientation": {
                    "x": np.float64(0.0),
                    "y": np.float64(0.0),
                    "z": np.float64(0.0),
                    "w": np.float64(1.0)
                }
            },
            "scale": {
                "x": np.float64(1.0),
                "y": np.float64(1.0),
                "z": np.float64(1.0)
            },
            "color": {
                "r": np.float32(1.0),
                "g": np.float32(0.0),
                "b": np.float32(0.0),
                "a": np.float32(1.0)
            },
            "frame_locked": False,
            "points": [{"x": np.float64(1.0), "y": np.float64(1.0), "z": np.float64(1.0)}],
            "colors": [{"r": np.float32(1.0), "g": np.float32(1.0), "b": np.float32(1.0), "a": np.float32(1.0)}],
            "texture_resource": "",
            "uv_coordinates": [{}],
            "text": "",
            "mesh_resource": "",
            "mesh_use_embedded_materials": False
        }
    ]
)

class Operator:
    def __init__(self) -> None:
        # 初始化ROS2上下文
        self.ros2_context = dora.experimental.ros2_bridge.Ros2Context()
        
        # 创建ROS2节点
        self.ros2_node = self.ros2_context.new_node(
            "marker2ros",
            "/ros2_bridge",
            dora.experimental.ros2_bridge.Ros2NodeOptions(rosout=True),
        )
        
        # 设置QoS策略
        self.topic_qos = dora.experimental.ros2_bridge.Ros2QosPolicies(
            reliable=True, max_blocking_time=0.1
        )
        
        # 创建Marker主题
        self.marker_data_topic = self.ros2_node.create_topic(
            "/ros2_bridge/visualization_marker",
            "visualization_msgs/Marker",
            self.topic_qos,
        )
        
        # 创建发布者
        self.marker_data_publisher = self.ros2_node.create_publisher(
            self.marker_data_topic
        )

    def on_event(self, dora_event, send_output) -> DoraStatus:
        if dora_event["type"] == "INPUT":
            # 发布Marker消息
            self.marker_data_publisher.publish(marker)
        return DoraStatus.CONTINUE

关键点解析

1. 数据结构定义：使用PyArrow库定义Marker消息的数据结构，注意各种数据类型的正确使用：

   • 字符串类型直接使用Python字符串
   • 数值类型使用对应的numpy类型
   • 嵌套结构使用字典表示

2. ROS2节点配置：

   • 节点名称和命名空间需要符合ROS2规范
   • QoS策略决定了消息传输的可靠性和实时性

3. 消息发布时机：在Operator的on_event方法中响应输入事件并发布消息

实际应用建议

1. 动态Marker生成：可以根据实际需求动态修改Marker的属性，如位置、颜色等
2. 性能优化：对于高频更新的Marker，可以考虑调整QoS策略
3. 错误处理：增加对发布失败情况的处理逻辑
4. 多Marker支持：通过修改id和ns字段实现多个Marker的同时显示

总结

通过Dora-rs项目的ROS2桥接功能，开发者可以方便地在Dora数据流中生成并发布ROS2的Marker消息，实现与RViz等可视化工具的无缝集成。这种集成方式既保留了Dora数据流编程的优势，又能利用ROS2丰富的可视化生态系统，为机器人开发和数据分析提供了强大的工具支持。