roboto.domain.topics.mcap_topic_reader#

Module Contents#

class roboto.domain.topics.mcap_topic_reader.McapTopicReader(roboto_client)#

Private interface for retrieving topic data stored in MCAP files.

Uses HTTP Range requests to efficiently fetch only required chunks from remote storage, avoiding full file downloads.

Note

This is not intended as a public API. To access topic data, prefer the get_data or get_data_as_df methods on Topic, MessagePath, or Event.

static accepts(message_paths_to_representations)#

Parameters:: message_paths_to_representations (collections.abc.Iterable[roboto.domain.topics.operations.MessagePathRepresentationMapping])
Return type:: bool

get_data(message_paths_to_representations, start_time=None, end_time=None, timestamp_message_path_representation_mapping=None)#

Parameters:

message_paths_to_representations (collections.abc.Iterable[roboto.domain.topics.operations.MessagePathRepresentationMapping])
start_time (Optional[int])
end_time (Optional[int])
timestamp_message_path_representation_mapping (Optional[roboto.domain.topics.operations.MessagePathRepresentationMapping])

Return type:

collections.abc.Generator[tuple[roboto.domain.topics.topic_reader.Timestamp, dict[str, Any]], None, None]

get_data_as_df(message_paths_to_representations, start_time=None, end_time=None, timestamp_message_path_representation_mapping=None)#

Parameters:

message_paths_to_representations (collections.abc.Iterable[roboto.domain.topics.operations.MessagePathRepresentationMapping])
start_time (Optional[int])
end_time (Optional[int])
timestamp_message_path_representation_mapping (Optional[roboto.domain.topics.operations.MessagePathRepresentationMapping])

Return type:

tuple[pandas.Series, pandas.DataFrame]