rosbag2_py_sequential_writer_mcap_string_messages.py

1import rosbagpy
2from std_msgs.msg import String
3import rclpy
4from rclpy.serialization import serialize_message
5
6import rosbag2_py
7
8def write_bag():
9    writer = rosbag2_py.SequentialWriter()
10
11    storage_options = rosbag2_py.StorageOptions(
12        uri='my_bag',
13        storage_id='mcap')
14    converter_options = rosbag2_py.ConverterOptions(
15        input_serialization_format='cdr',
16        output_serialization_format='cdr')
17
18    writer.open(storage_options, converter_options)
19
20    topic_info = rosbag2_py.TopicMetadata(
21        name='test_topic',
22        type='std_msgs/msg/String',
23        serialization_format='cdr')
24    writer.create_topic(topic_info)
25
26    time_stamp = 0
27    for i in range(10):
28        msg = String()
29        msg.data = f'Hello World {i}'
30        
31        writer.write(
32            'test_topic',
33            serialize_message(msg),
34            time_stamp)
35        time_stamp += 1000000000  # nanoseconds
36
37    # The writer will close automatically when the object goes out of scope
38    # or you can delete it explicitly
39    del writer
40
41if __name__ == '__main__':
42    write_bag()