roboto-mcap-codec 0.1.16rc1

Shared Rust CDR/IDL decoder for ROS and DDS message data (omgidl, ros2idl, ros2msg, ros1msg).

roboto-mcap-codec

A fast, shared Rust decoder for ROS and DDS message data, packaged as a Python extension. It parses the schema languages used in MCAP and ROS bags — omgidl, ros2idl, ros2msg, ros1msg — and decodes message payloads across every common CDR framing (plain XCDR1/XCDR2, parameter-list PL_CDR, delimited D_CDR2) and the unaligned ROS1 wire, into plain Python dict / list / scalar values.

It powers message decoding in the Roboto Python SDK, where one Rust implementation serves both ingest-time field statistics and query-time message reads.

Install

pip install roboto-mcap-codec

Prebuilt abi3 wheels are published for CPython 3.10+ on Linux (manylinux2014 / glibc 2.17 and musllinux_1_2 / Alpine, x86-64 and aarch64), macOS (Intel and Apple Silicon), and Windows (x86-64 and ARM64). No Rust toolchain is needed to install.

Quick start

A RosCdrCodec is built once per channel schema, then reused to decode every message on that channel.

from mcap_codec import RosCdrCodec

# schema_encoding: "omgidl" | "ros2idl" | "ros2msg" | "ros1msg"
# schema_data:     the raw schema bytes (e.g. an MCAP Schema record's `data`)
# schema_name:     the fully-scoped root type name
codec = RosCdrCodec("ros2msg", schema_data, "sensor_msgs/msg/Imu")

# payload = the raw message bytes, including the 4-byte CDR encapsulation header
message = codec.decode(payload)        # -> nested dict / list / scalars

Decode only the fields you need (everything else is skipped, not materialized):

partial = codec.decode_fields(
    payload,
    ["angular_velocity.x", "header.stamp.sec"],
)
# -> {"angular_velocity": {"x": ...}, "header": {"stamp": {"sec": ...}}}

Output shape

• structs → dict; sequences and arrays → list (multi-dimensional arrays → nested lists); octet / uint8 arrays → bytes (matching mcap_ros2; signed int8 arrays stay list[int], since bytes can't hold negatives). The deprecated byte / char aliases follow their dialect: ROS2 byte→uint8 / char→int8, ROS1 the reverse
• scalars → native bool / int / float; strings → str
• enums → integers
• unions → {"$discriminator": <value>, "<active member>": <value>}
• Time / Duration → {"sec": int, "nsec": int} (ROS2 nanosec normalized to nsec)
• absent @optional members are omitted from the dict

Undecodable messages

Some fields have no portable CDR encoding and cannot be decoded reliably. When a message cannot be decoded for such a reason, decode() raises UnsupportedMessage (rather than returning a wrong value). Catch it to skip the message and keep reading the rest of the file:

from mcap_codec import RosCdrCodec, UnsupportedMessage

try:
    message = codec.decode(payload)
except UnsupportedMessage:
    message = None   # e.g. an implementation-defined `wstring`, or HASH autoid

This matches how the Roboto SDK and the Foxglove decoders treat the same cases.

Errors

• RosCdrCodec(...) raises ValueError if the schema cannot be parsed.
• decode(...) / decode_fields(...) raise ValueError on a genuine decode failure (malformed payload), and UnsupportedMessage for a message that is intentionally declined (see above).

License

Apache-2.0. Part of the Roboto platform.