ros2_rosbag2_py_create_bag_write_messages_sqlite3.py

1import rclpy
2from rclpy.serialization import serialize_message
3from std_msgs.msg import String
4
5import rosbag2_py
6
7def main():
8    writer = rosbag2_py.SequentialWriter()
9
10    storage_options = rosbag2_py.StorageOptions(
11        uri='my_bag',
12        storage_id='sqlite3')
13    converter_options = rosbag2_py.ConverterOptions(
14        input_serialization_format='cdr',
15        output_serialization_format='cdr')
16
17    writer.open(storage_options, converter_options)
18
19    topic_name = 'chatter'
20    topic_info = rosbag2_py.TopicMetadata(
21        id=0,
22        name=topic_name,
23        type='std_msgs/msg/String',
24        serialization_format='cdr')
25    writer.create_topic(topic_info)
26
27    time_stamp = 0
28    for i in range(10):
29        msg = String()
30        msg.data = f'Hello World {i}'
31
32        writer.write(
33            topic_name,
34            serialize_message(msg),
35            time_stamp)
36
37        time_stamp += 1000000000
38
39    # The writer goes out of scope and closes the bag automatically, 
40    # but it's good practice to delete it explicitly if in a long-running script.
41    del writer
42
43if __name__ == '__main__':
44    main()