roboto.domain.topics.topic_reader

Module Contents

type roboto.domain.topics.topic_reader.Timestamp = Union[int, float]

class roboto.domain.topics.topic_reader.TopicReader

Bases: abc.ABC

Private interface for retrieving topic data of a particular format.

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)

Abstractmethod:

Parameters:

message_paths_to_representations (collections.abc.Iterable[roboto.domain.topics.operations.MessagePathRepresentationMapping])

Return type:

bool

abstract 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[Timestamp, dict[str, Any]], None, None]

abstract 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]