MCAP Python Library
The `mcap` library provides classes for reading and writing the MCAP file format. MCAP is a modular container format primarily intended for robotics applications, supporting pub/sub messages with arbitrary message serialization, including JSON, Protobuf, ROS 1, and ROS 2. It emphasizes high-performance writing, efficient seeking, and data recoverability. The project maintains an active development pace with frequent releases across its core library and various serialization support packages.
Warnings
- gotcha Modules, variables, classes, and attributes prefixed with a single underscore (`_`) are considered internal API and may change between minor or patch releases without prior notice. Relying on them can lead to unexpected breakages.
- gotcha Reading topics from MCAP files that were converted from ROS 1 `.bag` files might exhibit slower performance compared to reading directly from the original `.bag` files using native ROS 1 tools.
- gotcha The `mcap` library provides raw message data (byte streams). Users are responsible for decoding complex or compressed payloads (e.g., images, point clouds) into usable Python objects using appropriate image processing or serialization libraries. The library does not automatically handle this decoding beyond schema registration.
- gotcha The core `mcap` Python library is primarily designed for creating new files or reading existing ones. Direct programmatic appending to an *already closed* MCAP file is not a officially supported operation and can lead to unexpected behavior or data corruption.
Install
-
pip install mcap -
pip install mcap-protobuf-support -
pip install mcap-ros1-support -
pip install mcap-ros2-support
Imports
- make_reader
from mcap.reader import make_reader
- Writer
from mcap.writer import Writer
- read_protobuf_messages
from mcap_protobuf.reader import read_protobuf_messages
- DecoderFactory
from mcap_ros1.decoder import DecoderFactory
- read_ros2_messages
from mcap_ros2.reader import read_ros2_messages
- Writer (ROS2)
from mcap_ros2.writer import Writer
Quickstart
import json
import sys
from time import time_ns
from mcap.writer import Writer
from mcap.reader import make_reader
file_path = "example.mcap"
# --- Writing an MCAP file ---
with open(file_path, "wb") as stream:
writer = Writer(stream)
writer.start(profile="x-my-profile", library="my-writer-v1")
schema_id = writer.register_schema(
name="sample",
encoding="jsonschema",
data=json.dumps({
"type": "object",
"properties": {
"sample": {
"type": "string"
}
}
}).encode(),
)
channel_id = writer.register_channel(
schema_id=schema_id,
topic="/sample_topic",
message_encoding="json",
)
writer.add_message(
channel_id=channel_id,
log_time=time_ns(),
data=json.dumps({"sample": "hello world"}).encode("utf-8"),
publish_time=time_ns(),
)
writer.finish()
print(f"Wrote {file_path}")
# --- Reading an MCAP file ---
with open(file_path, "rb") as f:
reader = make_reader(f)
for schema, channel, message in reader.iter_messages(topics=["/sample_topic"]):
print(f"Read: Topic='{channel.topic}' (Schema='{schema.name}'): {message.data.decode('utf-8')}")