MCAP Protobuf Support for Python

Warnings

• breaking The high-level API for writing Protobuf messages to MCAP files has been significantly streamlined. Prior to `mcap-protobuf-support` v0.5.0, users often manually registered schemas using `mcap.mcap0.writer.Writer` and `mcap_protobuf.schema.register_schema`. The current recommended approach uses `mcap_protobuf.writer.Writer`, which abstracts away schema registration, simplifying usage.
  Fix: Migrate write operations to use `from mcap_protobuf.writer import Writer`. The `Writer` constructor takes a file-like object, and `write_message` directly accepts Protobuf message objects.
• gotcha Protobuf message definitions (`.proto` files) must be compiled into Python classes using the `protoc` compiler before they can be used with `mcap-protobuf-support`. The library does not dynamically parse `.proto` files at runtime.
  Fix: Run `protoc --python_out=. your_message.proto` to generate `your_message_pb2.py` files. Then, import your message classes from these generated files.
• gotcha While MCAP itself supports schema evolution, managing changes to Protobuf message schemas requires adherence to Protobuf's own forward and backward compatibility rules. Incompatible schema changes can lead to deserialization errors when reading older or newer MCAP files.
  Fix: Design Protobuf schemas with compatibility in mind (e.g., avoid changing field numbers, make new fields optional, avoid removing fields). Test schema evolution thoroughly across different versions of your data and code.

Install

• pip install mcap-protobuf-support Install latest version

Imports

• read_protobuf_messages

  from mcap_protobuf.reader import read_protobuf_messages

• Writer

  from mcap_protobuf.writer import Writer

  Older versions (e.g., v0.0.4) and low-level usage might involve manually registering schemas with `mcap.mcap0.writer`. The `mcap_protobuf.writer.Writer` provides a simplified, higher-level API.

Quickstart

This quickstart demonstrates how to write and read Protobuf messages to/from an MCAP file using `mcap-protobuf-support`. Before running, you would typically compile your `.proto` files into Python classes using `protoc` (e.g., `protoc --python_out=. your_message.proto`). The example uses dummy classes to simulate compiled Protobuf messages for executability. The `Writer` automatically handles schema registration, and `read_protobuf_messages` iterates through decoded messages.

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}")