roboto-mcap-codec

Decodes one channel of an MCAP file into Apache Arrow, and single messages into Python objects. open_mcap_file reads the file through a byte-range reader you supply: its footer and summary, then only the chunks that hold the channel's messages within the requested row range and, for a window on log time, overlap that window. It decodes ROS 1 messages, ROS 2 and OMG IDL messages over CDR, and JSON or MessagePack messages described by a JSON Schema. The rows arrive as pyarrow.RecordBatch values, one per batch, each owning its data, so it stays valid after the file and the cursor that read it are gone.

Each open_mcap_file call opens a cursor over one channel of one file, and cursors on separate threads run in parallel, since each decodes without holding the interpreter lock. With the default limits, one cursor reads messages of up to 64 MiB, such as camera images and lidar point clouds.

The package ships prebuilt wheels for Linux (manylinux2014 and musllinux_1_2, x86-64 and aarch64), macOS (Intel and Apple Silicon) and Windows (x86-64 and ARM64). Each wheel targets CPython's stable ABI, so one wheel per platform serves CPython 3.10 and later. The package is published as these wheels only, so pip cannot install it on any other platform.

Install

pip install "roboto-mcap-codec[pyarrow]"

open_mcap_file needs PyArrow 20 or later, which the pyarrow extra installs. Without it, pip install roboto-mcap-codec installs only the per-message RosCdrCodec, and open_mcap_file raises ImportError naming the extra. PyArrow publishes no Windows ARM64 wheel, so there the extra installs nothing and open_mcap_file needs a PyArrow built separately.

Quick start

import os

import mcap_codec

path = "recording.mcap"
with open(path, "rb") as file:

    # Called for each byte range the cursor reads; return exactly `length` bytes.
    def read_bytes(offset: int, length: int) -> bytes:
        file.seek(offset)
        return file.read(length)

    with mcap_codec.open_mcap_file(
        read_bytes,
        os.path.getsize(path),
        channel=mcap_codec.TopicName("/odometry"),
        projection=mcap_codec.Projection(include=[["speed"], ["pose", "x"]]),
        timestamp=mcap_codec.TimestampSource("message_log_time"),
        time_window=(1_700_000_000_000_000_000, 1_700_000_060_000_000_000),
    ) as cursor:
        for batch in cursor:
            ...  # use the rows of this pyarrow.RecordBatch

• Column 0 of every batch is the row number: the message's zero-based position among the channel's messages, in the order the file stores them. Column 1 is the timestamp in nanoseconds, from the source the timestamp argument names. The projected fields follow: a Projection's include paths, or every field without include, minus its exclude paths. An open_mcap_file call without projection gets every top-level field, and include=[] gets none.
• Read each cursor from one thread at a time: a second call on a cursor while the first still runs raises RuntimeError.
• cursor.to_record_batch_reader() returns a pyarrow.RecordBatchReader over the remaining batches, which PyArrow, DuckDB and Polars accept.
• Close a cursor you stop reading, so it makes no further read_bytes calls; the with statement closes it when its block ends.
• A failure raises CodecError, a ValueError whose code names it, such as unknown_channel, corrupt_input or resource_limit. The exceptions: an exception your read_bytes raises propagates unchanged, and a value of the wrong type, including a read_bytes result that is not bytes, raises TypeError.

The docstrings, such as help(mcap_codec.open_mcap_file), document every argument, class and error code.

Decoding single messages

A RosCdrCodec decodes one message at a time, from its bytes, without reading a file. Constructing one parses a schema, so build one per schema and reuse it for every message that schema describes:

from mcap_codec import RosCdrCodec, UnsupportedMessage

# schema_data: the schema's bytes, such as an MCAP Schema record's `data`
codec = RosCdrCodec("ros2msg", "cdr", schema_data, "sensor_msgs/msg/Imu")

# payload: a Message record's `data`; a CDR payload starts with its 4-byte
# encapsulation header
try:
    message = codec.decode(payload)
    partial = codec.decode_fields(
        payload, ["angular_velocity.x", "header.stamp.sec"]
    )
except UnsupportedMessage:
    message = partial = None  # a message the decoder declines; skip it

decode returns the message as dicts, lists and scalars, with each array of a fixed-width numeric type as an array.array. decode_fields returns the fields at the dotted paths it is given, nested as in the whole message, and builds no Python objects for the others. A message the decoder declines, such as one holding a wstring in a ROS 2 message or outside an IDL @mutable struct, raises UnsupportedMessage, a CodecError you can catch to skip it.

Changes from 0.1

• McapBatchDecoder is removed; use open_mcap_file.
• UnsupportedMessage derives from CodecError, and so from ValueError, where 0.1 derived it from Exception. Place an except UnsupportedMessage clause before any except ValueError clause, which would otherwise catch it.
• RosCdrCodec raises CodecError, still a ValueError, with message text that differs from 0.1's; match on code, not on the text.

License

The package is licensed under MPL-2.0; see LICENSE. Versions 0.1.19 and earlier were published under Apache-2.0 and remain under it. NOTICE and THIRD_PARTY_LICENSES/ cover the third-party components compiled into the wheels.