roboto.domain.topics.message_path_accessor#

Module Contents#

roboto.domain.topics.message_path_accessor.Accessor#

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

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

class roboto.domain.topics.message_path_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 consuming topics that happen to share a path_in_schema tuple but resolve through different schemas (e.g., ROS1 vs. ROS2 timestamps) cannot pollute each other.

get_or_compile(message_paths, sample, getter)#

Return accessors for these paths, 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:
Return type:

list[Accessor]

roboto.domain.topics.message_path_accessor.Accumulator#

Output dict the accessors write into. Nested keys materialize as nested dicts.

roboto.domain.topics.message_path_accessor.PathInSchema#

One field’s path through a message schema, e.g. ("header", "stamp", "sec").

roboto.domain.topics.message_path_accessor.compile_accessors(message_paths, sample, getter)#

Compile one accessor per message path. 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:
Return type:

tuple[list[Accessor], bool]

On this page