roboto.domain.topics.mcap_reader#

Module Contents#

class roboto.domain.topics.mcap_reader.McapReader(stream, message_paths, start_time=None, end_time=None)#

Reader for processing MCAP files with message path filtering.

Provides an iterator interface for reading decoded messages from MCAP files, with support for temporal filtering and message path selection. Handles multiple encoding formats including JSON, ROS1, and ROS2.

The reader automatically decodes messages using appropriate decoders and filters the output based on the specified message paths and time range.

Parameters:
property has_next: bool#

Check if there are more messages available to read.

Returns:

True if there are more messages to read, False otherwise.

Return type:

bool

property message_paths: list[str]#

Get the list of message paths being filtered for.

Returns the message path strings from the MessagePathRecord objects that were provided during initialization.

Returns:

List of message path strings in dot notation from the configured MessagePathRecord objects.

Return type:

list[str]

next()#

Read and return the next decoded message.

Advances the reader to the next message and returns it as a DecodedMessage object, or None if no more messages are available.

Returns:

DecodedMessage containing the next message data, or None if no more messages.

Return type:

roboto.domain.topics.decoded_message.DecodedMessage | None

Examples

>>> while reader.has_next:
...     message = reader.next()
...     if message:
...         data = message.to_dict()
...         print(f"Message at {data.get('log_time')}: {data}")
next_message_is_time_aligned(timestamp)#

Check if the next message has the specified timestamp.

Used for time-aligned reading when merging data from multiple readers.

Parameters:

timestamp (Union[int, float]) – Timestamp to check against in nanoseconds.

Returns:

True if the next message has the specified timestamp, False otherwise.

Return type:

bool

property next_timestamp: int | float#

Get the timestamp of the next message to be read.

Returns:

Timestamp of the next message in nanoseconds, or math.inf if no more messages.

Return type:

Union[int, float]

On this page