Skip to content
Roboto
Esc
↑↓navigate↵open⌘Jpreview
On this page

roboto.domain.topics.topic_reader

Module Contents

Timestamp

type roboto.domain.topics.topic_reader.Timestamp = typing.Union[int, float]#View Source

TopicReader

class roboto.domain.topics.topic_reader.TopicReader#View Source

Bases: abc.ABC

Private interface for retrieving topic data of a particular format.

TopicReader.accepts()

abstract static accepts(message_paths_to_representations)#View Source

Parameters

message_paths_to_representations collections.abc.Iterable[roboto.domain.topics.record.MessagePathRepresentationMapping]

Return type

bool

TopicReader.get_data()

abstract get_data(message_paths_to_representations, start_time=None, end_time=None, timestamp_message_path_representation_mapping=None)#View Source

Parameters

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

Return type

collections.abc.Generator[tuple[Timestamp, dict[str, Any]], None, None]

TopicReader.get_data_as_df()

abstract get_data_as_df(message_paths_to_representations, start_time=None, end_time=None, timestamp_message_path_representation_mapping=None)#View Source

Parameters

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

Return type

tuple[pandas.Series, pandas.DataFrame]

Was this page helpful?