mcap-ros2-support

import os from pathlib import Path from mcap.mcap_writer import McapWriter from mcap_ros2.writer import Ros2RawWriter from mcap_ros2.reader import Ros2Reader # Assuming a ROS2 message type is available, e.g., from an installed ROS2 package # For demonstration, we'll mock a simple message structure for 'std_msgs/msg/String' # In a real scenario, you would import the actual message type and populate it. class MockStringMsg: _TYPE = 'std_msgs/msg/String' _HAS_HEADER = False _LAYOUT = [ ('data', 'string'), ] _REGISTERED_TYPES = {} _CDR_TYPES = {} def __init__(self, data=''): self.data = data def __repr__(self): return f"MockStringMsg(data='{self.data}')" @staticmethod def get_fields_and_field_types(): return [('data', 'string')] def get_type_description_interfaces(self): # Simplified mock for the purpose of the quickstart return {'type_description': 'string data'} # Write a simple MCAP file with a ROS2 message output_file = Path('ros2_messages.mcap') with open(output_file, 'wb') as f, McapWriter(f) as mcap_writer, Ros2RawWriter(mcap_writer) as ros2_writer: # In a real ROS2 environment, you'd create actual ROS2 messages # from std_msgs.msg import String # msg = String(data='Hello ROS2 from MCAP!') mock_msg = MockStringMsg(data='Hello ROS2 from MCAP!') topic = '/chatter' log_time_ns = 123456789 # The Ros2RawWriter can infer the schema from a ROS2 message object. # For this mock, we ensure the mock_msg has the necessary _TYPE and field info. ros2_writer.write_message( topic=topic, message=mock_msg, log_time_ns=log_time_ns, publish_time_ns=log_time_ns ) print(f"Wrote ROS2 message to {output_file}") # Read the MCAP file and verify with open(output_file, 'rb') as f: reader = Ros2Reader(f) for msg, schema, channel in reader.iter_decoded_messages(): print(f"Read message from topic '{channel.topic}': {msg}") assert msg.data == 'Hello ROS2 from MCAP!' break # Only read one message for this example # Clean up the created file os.remove(output_file) print(f"Cleaned up {output_file}")