Python中使用mcap_ros2.writer添加元数据或追加至MCAP文件的方法咨询

一、使用mcap_ros2.writer.Writer添加元数据 #

mcap_ros2.writer.Writer内部封装了底层的mcap.writer.Writer实例，你可以直接访问该底层实例调用add_metadata()方法，同时还能实现自定义头部时间戳的需求，具体操作如下：

1. 直接访问底层实例添加元数据 #

from mcap_ros2.writer import Writer
from std_msgs.msg import String
import time

with open("output.mcap", "wb") as f:
    writer = Writer(f)
    # 调用底层Writer的add_metadata方法添加元数据
    writer._writer.add_metadata(key="project_name", value="my_ros2_demo")
    writer._writer.add_metadata(key="recorder_version", value="2.1.0")
    
    # 写入带自定义时间戳的消息
    msg = String(data="test message")
    custom_timestamp = int(time.time() * 1e9)  # 转成纳秒级时间戳
    writer.write_message(
        topic="/test/topic",
        msg=msg,
        timestamp=custom_timestamp
    )

2. 初始化时指定自定义头部时间戳 #

如果需要设置自定义的头部起始时间，可以先初始化底层mcap.Writer并传入start_time参数，再用它来创建mcap_ros2.Writer：

from mcap.writer import Writer as McapWriter
from mcap_ros2.writer import Writer
import time

custom_start_time = int(time.time() * 1e9) - 3600 * 1e9  # 自定义1小时前的起始时间
with open("output_with_custom_header.mcap", "wb") as f:
    mcap_writer = McapWriter(f, start_time=custom_start_time)
    ros2_writer = Writer(f, writer=mcap_writer)
    
    # 添加元数据
    mcap_writer.add_metadata(key="custom_header_start_time", value=str(custom_start_time))
    
    # 正常写入ROS2消息...

二、Python非CLI方式向现有MCAP文件追加元数据 #

MCAP格式支持内容追加，但直接修改原文件容易破坏结构，更稳妥的方式是读取原文件内容，复制到新文件的同时添加元数据，再替换原文件：

from mcap.reader import make_reader
from mcap.writer import Writer
import os

input_mcap = "existing.mcap"
temp_mcap = "temp.mcap"

with open(temp_mcap, "wb") as temp_f:
    with make_reader(input_mcap) as reader:
        # 沿用原文件的头部时间戳
        writer = Writer(temp_f, start_time=reader.header.start_time)
        # 添加新元数据
        writer.add_metadata(key="updated_at", value=str(int(time.time() * 1e9)))
        writer.add_metadata(key="note", value="metadata added via python script")
        
        # 复制原文件所有内容到新文件
        for schema, channel, message in reader.iter_messages():
            writer.write_message(
                topic=channel.topic,
                message=message.data,
                timestamp=message.timestamp,
                schema=schema,
                channel=channel
            )

# 替换原文件
os.replace(temp_mcap, input_mcap)

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