roboto.domain.topics.omgidl.decoder_factory

MCAP decoder factory for omgidl and ros2idl schemas.

Exposes an mcap.decoder.DecoderFactory that decodes CDR messages whose schema is OMG IDL (omgidl / ros2idl), for use by McapReader. Decoding is delegated to the shared Rust decoder (the mcap_codec extension built from roboto-mcap-codec – the same decoder the ingestion action uses), which parses the schema and decodes every CDR framing (plain XCDR1/XCDR2 and the parameter-list PL_CDR/PL_CDR2 families) into nested dict / sequence / scalar values. Numeric arrays and sequences come back as the compact array.array (one C buffer rather than a list of boxed Python objects) and octet sequences as bytes; nested/heterogeneous sequences (arrays of structs, arrays of sequences, the outer dimension of a multidimensional array) are plain list. This mirrors the ROS2 decoder, which likewise returns bytes for uint8/octet arrays. Callers that need JSON-serializable output (array.array and bytes are not JSON-native) should convert at the edge, e.g. via array.tolist().

Module Contents

exception roboto.domain.topics.omgidl.decoder_factory.OmgidlDecodeError

Bases: mcap.exceptions.McapError

Raised when an omgidl / ros2idl message cannot be decoded.

class roboto.domain.topics.omgidl.decoder_factory.OmgidlDecoderFactory

Bases: mcap.decoder.DecoderFactory

Decode CDR messages described by omgidl or ros2idl schemas.

Supply an instance to mcap.reader.make_reader(). The schema is parsed once per schema id (by the shared Rust mcap_codec decoder); the returned callable then decodes each message into nested dict / sequence / scalar values – numeric arrays/sequences as array.array, octet sequences as bytes, and nested/heterogeneous sequences as list (see the module docstring).

decoder_for(message_encoding, schema)

If the message encoding and schema arguments can be decoded by this decoder factory, returns a callable to decode message bytes.

Parameters:

• message_encoding (str)

• schema (Optional[mcap.records.Schema])

Return type:

Optional[Callable[[bytes], Any]]

roboto.domain.topics.omgidl.decoder_factory.UNDECODABLE_MESSAGE

roboto.domain.topics.omgidl.decoder_factory.make_omgidl_decoder_factory()

Build a decoder factory for omgidl / ros2idl CDR messages, backed by the shared Rust mcap_codec decoder.

Return type:

OmgidlDecoderFactory