roboto.formats.mcap.accessor#

Module Contents#

roboto.formats.mcap.accessor.Accessor#

Reads one path’s value out of a decoded message and writes it into an Accumulator.

The accessor is compiled once per fields set and reused for every subsequent message in the same read pass. Compilation resolves time-field name remapping and sequence boundaries against a sample message, so per-call work is just attribute access plus accumulator writes.

class roboto.formats.mcap.accessor.AccessorCache#

Holds compiled Accessor callables for the lifetime of a single read pass.

The cache lives on the reader rather than at module scope so that two readers projecting the same paths against differently-shaped messages (e.g. a time field present in one topic and absent in another) cannot pollute each other.

get_or_compile(fields, sample, getter)#

Return accessors for these fields, compiling against sample on first call.

Compilations that hit an empty sequence in the sample are speculative — the inner shape past the empty point can’t be observed — and are returned without caching so a later message with a non-empty sequence triggers a fresh, complete compile.

Parameters:

fields (collections.abc.Sequence[FieldSelection])
sample (Any)
getter (AttrGetter)

Return type:

list[Accessor]

roboto.formats.mcap.accessor.Accumulator#: Output dict the accessors write into. Nested keys materialize as nested dicts.

class roboto.formats.mcap.accessor.AttrGetter#

Bases: abc.ABC

Abstract base class for reading attributes from a decoded message.

Decoded MCAP messages are dictionaries – JSON and the shared mcap_codec decoder both materialize to dict – so DictAttrGetter is the only implementation. The abstraction is kept so the accessor compiler takes a getter rather than hard-coding dict access.

static get_attribute(value, attribute)#

Abstractmethod:
Return type:: Any

Get the value of a specific attribute from the given value.

Parameters:

value – The decoded message value to access.
attribute – Name of the attribute to retrieve.

Returns:

The value of the specified attribute.

Return type:

Any

static get_attribute_names(value)#

Abstractmethod:
Return type:: collections.abc.Sequence[str]

Get the names of all attributes available in the given value.

Parameters:: value – The decoded message value to inspect.
Returns:: Sequence of attribute names available in the value.
Return type:: collections.abc.Sequence[str]

static has_attribute(value, attribute)#

Abstractmethod:
Parameters:: attribute (str)
Return type:: bool

Check if the given value has a specific attribute.

Parameters:

value – The decoded message value to inspect.
attribute (str) – Name of the attribute to check for.

Returns:

True if the value has the specified attribute, False otherwise.

Return type:

bool

static has_sub_attributes(value)#

Abstractmethod:
Return type:: bool

Check if the given value has nested attributes that can be accessed.

Parameters:: value – The decoded message value to inspect.
Returns:: True if the value has nested attributes, False otherwise.
Return type:: bool

class roboto.formats.mcap.accessor.DictAttrGetter#

Bases: AttrGetter

Attribute getter for decoded messages represented as dictionaries.

Both JSON and the shared mcap_codec decoder materialize messages and their nested structs as plain dicts. The isinstance guards let a non-dict value – a JSON null, scalar, or list message, or a scalar leaf reached mid-walk – resolve to “no such attribute” instead of raising, so the accessor compiler simply stops descending.

static get_attribute(value, attribute)#

Get the value of a specific attribute from the given value.

Parameters:

value – The decoded message value to access.
attribute – Name of the attribute to retrieve.

Returns:

The value of the specified attribute.

static get_attribute_names(value)#

Get the names of all attributes available in the given value.

Parameters:: value – The decoded message value to inspect.
Returns:: Sequence of attribute names available in the value.

static has_attribute(value, attribute)#

Check if the given value has a specific attribute.

Parameters:

value – The decoded message value to inspect.
attribute (str) – Name of the attribute to check for.

Returns:

True if the value has the specified attribute, False otherwise.

Return type:

bool

static has_sub_attributes(value)#

Check if the given value has nested attributes that can be accessed.

Parameters:: value – The decoded message value to inspect.
Returns:: True if the value has nested attributes, False otherwise.

roboto.formats.mcap.accessor.PathInSchema#: One field’s path through a message schema, e.g. ("header", "stamp", "sec").

roboto.formats.mcap.accessor.Resolution#

a no-op, a simple attribute chain, or a per-element sequence crossing.

Build one with none_resolution(), simple_resolution(), or sequence_resolution(), then compile it with build_accessor().

Type:: A resolved accessor path

roboto.formats.mcap.accessor.build_accessor(resolution)#

Compile a resolution into an Accessor that reads its path into an accumulator.

The resolution carries the (possibly time-remapped) structure; this only selects the matching runtime walk. It does not sample, so a caller that built the resolution from a schema can compile without a message in hand.

Parameters:: resolution (Resolution)
Return type:: Accessor

roboto.formats.mcap.accessor.compile_accessors(fields, sample, getter)#

Compile one accessor per field. Does not cache; callers manage caching.

Returns a tuple of (accessors, fully_resolved). fully_resolved is False if any path traversed an empty sequence in sample and the inner shape past it had to be guessed. Callers maintaining a cross-message cache should not cache speculative compilations, since the next message may need a different shape.

Parameters:

fields (collections.abc.Sequence[FieldSelection])
sample (Any)
getter (AttrGetter)

Return type:

tuple[list[Accessor], bool]

roboto.formats.mcap.accessor.getter_for(message)#

The shared attribute getter for a decoded message.

Every decoder materializes messages as dicts (and the dict getter resolves a non-dict JSON payload to no attributes), so one getter serves every message.

Parameters:: message (Any)
Return type:: AttrGetter

roboto.formats.mcap.accessor.is_codec_time_value(val)#

Whether val is a ROS2 Time / Duration decoded by the shared mcap_codec decoder.

The codec surfaces ROS2 Time / Duration with the schema’s own member names – a plain {"sec", "nanosec"} dict, no marker type – so the match is structural. Recognizing it lets the accessor remap a legacy nsec path component (how topics ingested before the move to wire-true field names recorded the sub-second leaf) onto the codec’s nanosec key; without the remap the nanosecond component is silently dropped. ROS1’s Time is natively {"sec", "nsec"} and JSON time values likewise use nsec, so neither matches here nor needs a remap.

Parameters:: val (Any)
Return type:: bool

roboto.formats.mcap.accessor.none_resolution()#

A resolution whose accessor is a no-op (the path is absent on a message).

Return type:: Resolution

roboto.formats.mcap.accessor.remap_time_fields(resolution, sample, getter)#

Substitute the decoder’s runtime time-field name into resolution, observed against sample.

A legacy topic’s message paths address a ROS2 time struct’s sub-second leaf as nsec (how it was recorded before the move to wire-true field names), but the shared mcap_codec decoder now materializes that value as a {"sec", "nanosec"} dict. This walks sample along the resolution and rewrites a trailing nsec past any such time value to nanosec, so the built accessor reads the right key.

Returns (remapped, time_resolved). time_resolved is False only when a time-bearing leaf sits past a sequence that is empty in sample — its element cannot be observed, so the runtime names stay a guess and the caller should re-resolve against a later, non-empty message. A resolution with no time component is returned unchanged with True. Paths that already name the leaf nanosec (ROS2 wire-true), ROS1 nsec, and JSON nsec values are all no-ops.

Parameters:

resolution (Resolution)
sample (Any)
getter (AttrGetter)

Return type:

tuple[Resolution, bool]

roboto.formats.mcap.accessor.sequence_resolution(pre_path, sub)#

A resolution that crosses the sequence at pre_path, applying sub per element.

Parameters:

pre_path (collections.abc.Sequence[str])
sub (Resolution)

Return type:

Resolution

roboto.formats.mcap.accessor.simple_resolution(path)#

A resolution for a straight attribute chain (no sequence crossing).

Parameters:: path (collections.abc.Sequence[str])
Return type:: Resolution