mcap-protobuf-support

import sys from mcap_protobuf.writer import Writer from mcap_protobuf.reader import read_protobuf_messages # Assuming you have a compiled protobuf message, e.g., 'my_message_pb2.py' # from my_message_pb2 import SimpleMessage, ComplexMessage # For demonstration, we'll create dummy classes that mimic protobuf messages class SimpleMessage: def __init__(self, data): self.data = data def SerializeToString(self): return f'SimpleMessage(data="{self.data}")'.encode('utf-8') # Dummy serialization @classmethod def FromString(cls, data): # Dummy deserialization - in a real scenario, this would parse protobuf bytes import re match = re.search(r'data="([^"]+)"', data.decode('utf-8')) return cls(match.group(1)) if match else cls('N/A') class ComplexMessage: def __init__(self, fieldA, fieldB): self.fieldA = fieldA self.fieldB = fieldB def SerializeToString(self): return f'ComplexMessage(fieldA="{self.fieldA}", fieldB="{self.fieldB}")'.encode('utf-8') # Dummy serialization @classmethod def FromString(cls, data): # Dummy deserialization import re match = re.search(r'fieldA="([^"]+)".*fieldB="([^"]+)"', data.decode('utf-8')) return cls(match.group(1), match.group(2)) if match else cls('N/A', 'N/A') # To make this runnable without actual .proto compilation, we need to mock # how the `Writer` would register these messages. In a real scenario, # SimpleMessage and ComplexMessage would be actual protobuf generated classes. # ---- Writing MCAP file with Protobuf messages ---- output_file = "example.mcap" with open(output_file, "wb") as f, Writer(f) as mcap_writer: # In a real application, SimpleMessage would be from `your_proto_file_pb2` mcap_writer.write_message( topic="/simple_messages", message=SimpleMessage(data="Hello MCAP protobuf world #1!"), log_time=1000, publish_time=1000, ) complex_message = ComplexMessage(fieldA="Field A 1", fieldB="Field B 1") mcap_writer.write_message( topic="/complex_messages", message=complex_message, log_time=2000, publish_time=2000, ) print(f"Wrote messages to {output_file}") # ---- Reading MCAP file with Protobuf messages ---- def register_dummy_message_classes(reader): # This function mocks how actual protobuf message classes would be registered # For real use, you'd import your compiled protobuf message classes reader.register_message("SimpleMessage", SimpleMessage) reader.register_message("ComplexMessage", ComplexMessage) print(f"\nReading messages from {output_file}:") for msg_info in read_protobuf_messages(output_file): # The msg_info.proto_msg will be an instance of your actual Protobuf class print(f"Topic: {msg_info.topic}, Message: {msg_info.proto_msg.data if hasattr(msg_info.proto_msg, 'data') else msg_info.proto_msg.fieldA}")