如何编写ROS2发布节点，从YAML文件定时发布消息运行Autoware节点？

我尝试运行Autoware生态中的单个ROS2节点，已通过ros2 bag record记录所有输入话题，得到若干包含消息内容的YAML文件（代码片段如下）。由于无法使用ros2 bag play进行仿真，我需要编写一个发布节点，从YAML文件中读取消息并每隔X秒发布一次，请问是否可行？

YAML消息片段：

header:
  stamp:
    sec: 1585897255
    nanosec: 366032919
  frame_id: map
child_frame_id: base_link
pose:
  pose:
    position:
      x: 89530.4904175517
      y: 42411.98934126079
      z: -3.608327379954426
    orientation:
      x: 0.003322876298139578
      y: -0.014172646821833243
      z: 0.8596727738629668
      w: 0.5106376567135674
  covariance:
  - 0.004823699025978851
  - 0.0014223050349292865
  - 0.0
  - 0.0
  - 0.0
  - -0.00033930002493972645
  - 0.0014223050349292876
  - 0.003066485845419473
  - 0.0
  - 0.0
  - 0.0
  - -0.00018709073063899757
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - -0.0003393000249397263
  - -0.00018709073063899755
  - 0.0
  - 0.0
  - 0.0
  - 6.080293039040077e-05
twist:
  twist:
    linear:
      x: 13.072203929530296
      y: 0.0
      z: 0.0
    angular:
      x: 0.0
      y: 0.0
      z: -0.008810670840086786
  covariance:
  - 0.03935738353508679
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - -5.688005653640222e-11
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - -5.688005653640594e-11
  - 0.0
  - 0.0
  - 0.0
  - 0.0
  - 0.0015608784670052972

完全可行，以下是实现思路和示例代码：

核心步骤 #

• 确定消息类型：从YAML结构判断，这是nav_msgs/msg/Odometry类型的消息，需确认目标节点订阅的话题类型与之匹配。
• 读取YAML文件：使用YAML解析库（如Python的pyyaml）加载文件内容。
• 构造ROS2消息：将YAML中的字段逐一映射到ROS2消息的对应属性上。
• 定时发布：借助ROS2的定时器，每隔X秒发布一次构造好的消息。

Python示例代码 #

import rclpy
from rclpy.node import Node
from nav_msgs.msg import Odometry
import yaml
from rclpy.time import Time

class YamlPublisher(Node):
    def __init__(self):
        super().__init__('yaml_odom_publisher')
        # 创建发布者，话题名和消息类型根据实际情况修改
        self.publisher_ = self.create_publisher(Odometry, '/odom', 10)
        # 加载YAML文件，替换为你的文件路径
        with open('/path/to/your/odom_msg.yaml', 'r') as f:
            self.yaml_data = yaml.safe_load(f)
        # 设置发布间隔，单位秒，替换为你需要的X值
        timer_period = 1.0
        self.timer = self.create_timer(timer_period, self.timer_callback)

    def timer_callback(self):
        msg = Odometry()
        # 填充header
        msg.header.stamp = Time(seconds=self.yaml_data['header']['stamp']['sec'], nanoseconds=self.yaml_data['header']['stamp']['nanosec']).to_msg()
        msg.header.frame_id = self.yaml_data['header']['frame_id']
        msg.child_frame_id = self.yaml_data['child_frame_id']
        # 填充pose
        msg.pose.pose.position.x = self.yaml_data['pose']['pose']['position']['x']
        msg.pose.pose.position.y = self.yaml_data['pose']['pose']['position']['y']
        msg.pose.pose.position.z = self.yaml_data['pose']['pose']['position']['z']
        msg.pose.pose.orientation.x = self.yaml_data['pose']['pose']['orientation']['x']
        msg.pose.pose.orientation.y = self.yaml_data['pose']['pose']['orientation']['y']
        msg.pose.pose.orientation.z = self.yaml_data['pose']['pose']['orientation']['z']
        msg.pose.pose.orientation.w = self.yaml_data['pose']['pose']['orientation']['w']
        msg.pose.covariance = self.yaml_data['pose']['covariance']
        # 填充twist
        msg.twist.twist.linear.x = self.yaml_data['twist']['twist']['linear']['x']
        msg.twist.twist.linear.y = self.yaml_data['twist']['twist']['linear']['y']
        msg.twist.twist.linear.z = self.yaml_data['twist']['twist']['linear']['z']
        msg.twist.twist.angular.x = self.yaml_data['twist']['twist']['angular']['x']
        msg.twist.twist.angular.y = self.yaml_data['twist']['twist']['angular']['y']
        msg.twist.twist.angular.z = self.yaml_data['twist']['twist']['angular']['z']
        msg.twist.covariance = self.yaml_data['twist']['covariance']
        # 发布消息
        self.publisher_.publish(msg)
        self.get_logger().info('发布Odometry消息')

def main(args=None):
    rclpy.init(args=args)
    yaml_publisher = YamlPublisher()
    rclpy.spin(yaml_publisher)
    yaml_publisher.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

注意事项 #

• 确保安装依赖：执行pip install pyyaml，同时正确配置ROS2环境。
• 若有多个YAML文件对应不同话题，需为每个话题单独编写发布逻辑，或在一个节点中创建多个发布者。
• 根据实际需求修改话题名、消息类型和YAML文件路径。

内容的提问来源于stack exchange，提问作者Yannic Neu