如何使用ros2bag读取自定义消息类型？代码反序列化报错

问题核心在于deserialize_cdr函数需要传入实际的Python消息类，而非字符串形式的消息类型名称（比如your_custom_package/MyMsg）。自定义消息不会被rosbags自动识别，必须先生成对应的Python接口并导入。

步骤1：生成自定义消息的Python代码 #

确保你的自定义ROS2包已经编译，生成对应的Python消息模块：

colcon build --packages-select your_custom_package

编译完成后，Python接口会生成在install/your_custom_package/lib/python3.x/site-packages目录下（x是你的Python版本号）。

步骤2：在代码中导入自定义消息类 #

先将安装目录添加到Python路径，再导入目标消息类：

import sys
# 替换为你的install路径和Python版本
sys.path.insert(0, '/path/to/your/workspace/install/lib/python3.10/site-packages')
from your_custom_package.msg import MyCustomMsg  # 替换为你的消息类名

步骤3：映射消息类型字符串到Python类 #

创建一个字典，把rosbag中读取到的字符串类型名映射到对应的Python类：

msg_type_map = {
    'your_custom_package/MyCustomMsg': MyCustomMsg,
    # 如果有多个自定义消息，继续添加键值对
}

步骤4：修改反序列化逻辑 #

替换原代码中deserialize_cdr的调用部分：

if connection.topic == topic:
    print(connection.topic)
    print(connection.msgtype)
    
    # 获取对应的消息类
    msg_class = msg_type_map.get(connection.msgtype)
    if not msg_class:
        print(f"未支持的消息类型：{connection.msgtype}")
        continue
    
    # 反序列化消息
    data = deserialize_cdr(rawdata, msg_class)
    print(data)

完整修改后的代码 #

from rosbags.rosbag2 import Reader as ROS2Reader
import sqlite3
from rosbags.serde import deserialize_cdr
import matplotlib.pyplot as plt
import os
import collections
import argparse
import sys

# 添加自定义消息所在的Python路径
sys.path.insert(0, '/path/to/your/workspace/install/lib/python3.10/site-packages')
from your_custom_package.msg import MyCustomMsg  # 替换为你的消息类

parser = argparse.ArgumentParser(description="Extract images from rosbag.")
parser.add_argument('--input','-i',type=str, help="rosbag input location")
args = parser.parse_args()

rosbag_dir = args.input
topic = "/topic/name"
frame_counter = 0

# 消息类型映射字典
msg_type_map = {
    'your_custom_package/MyCustomMsg': MyCustomMsg,
}

with ROS2Reader(rosbag_dir) as ros2_reader:
    ros2_conns = [x for x in ros2_reader.connections]
    print([x.topic for x in ros2_conns])

    ros2_messages = ros2_reader.messages(connections=ros2_conns)
    
    for m, msg in enumerate(ros2_messages):
        (connection, timestamp, rawdata) = msg
        
        if connection.topic == topic:
            print(connection.topic)
            print(connection.msgtype)
            
            msg_class = msg_type_map.get(connection.msgtype)
            if not msg_class:
                print(f"未支持的消息类型：{connection.msgtype}")
                continue
            
            data = deserialize_cdr(rawdata, msg_class)
            print(data)

额外提示 #

• 可以通过设置环境变量PYTHONPATH避免在代码中手动添加路径：

  export PYTHONPATH=$PYTHONPATH:/path/to/your/workspace/install/lib/python3.x/site-packages
  

• 确保自定义包的package.xml中已经添加了rosidl_default_generators和rosidl_default_runtime依赖，并且CMakeLists.txt中正确配置了消息生成规则。

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