roboto.domain.topics.message_path#

Module Contents#

class roboto.domain.topics.message_path.MessagePath(record, roboto_client=None, topic_data_service=None)#

Represents a message path within a topic in the Roboto platform.

A message path defines a specific field or signal within a topic’s data schema, using dot notation to specify nested attributes. Message paths enable fine-grained access to individual data elements within time-series robotics data, supporting operations like statistical analysis, data filtering, and visualization.

Each message path has an associated data type (both native and canonical), metadata, and statistical information computed from the underlying data. Message paths are the fundamental building blocks for data analysis in Roboto, allowing users to work with specific signals or measurements from complex robotics data structures.

Message paths support temporal filtering, data export to various formats including pandas DataFrames, and integration with the broader Roboto analytics ecosystem. They provide efficient access to time-series data while maintaining the semantic structure of the original robotics messages.

The MessagePath class serves as the primary interface for accessing individual data signals within topics, providing methods for data retrieval, statistical analysis, and metadata management.

Parameters:

record (roboto.domain.topics.record.MessagePathRecord)
roboto_client (Optional[roboto.http.RobotoClient])
topic_data_service (Optional[roboto.domain.topics.topic_data_service.TopicDataService])

DELIMITER: ClassVar = '.'#

property canonical_data_type: roboto.domain.topics.record.CanonicalDataType#

Canonical Roboto data type corresponding to the native data type.

Return type:: roboto.domain.topics.record.CanonicalDataType

property count: PreComputedStat#

Number of data points available for this message path.

Return type:: PreComputedStat

property created: datetime.datetime#

Timestamp when this message path was created.

Return type:: datetime.datetime

property created_by: str#

Identifier of the user or system that created this message path.

Return type:: str

property data_type: str#

Native data type for this message path, e.g. ‘float32’

Return type:: str

classmethod from_id(message_path_id, roboto_client=None, topic_data_service=None)#

Retrieve a message path by its unique identifier.

Fetches a message path record from the Roboto platform using its unique ID. This is useful when you have a message path identifier from another operation.

Parameters:

message_path_id (str) – Unique identifier for the message path.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.
topic_data_service (Optional[roboto.domain.topics.topic_data_service.TopicDataService]) – Service for accessing topic data. If None, creates a default instance.

Returns:

MessagePath instance representing the requested message path.

Raises:

RobotoNotFoundException – Message path with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the message path.

Return type:

MessagePath

Examples

>>> message_path = MessagePath.from_id("mp_abc123")
>>> print(message_path.path)
'angular_velocity.x'
>>> print(message_path.canonical_data_type)
CanonicalDataType.Number

get_data(start_time=None, end_time=None, cache_dir=None)#

Return data for this specific message path.

Retrieves and yields data records containing only the values for this message path, with optional temporal filtering. This provides a focused view of a single signal or field within the broader topic data.

Parameters:

start_time (Optional[roboto.time.Time]) – Start time (inclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
end_time (Optional[roboto.time.Time]) – End time (exclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
cache_dir (Union[str, pathlib.Path, None]) – Directory where topic data will be downloaded if necessary. Defaults to DEFAULT_CACHE_DIR.

Yields:

Dictionary records containing the log_time and the value for this message path.

Return type:

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

Notes

For each example below, assume the following is a sample datum record that can be found in this message path’s associated topic:

{
    "angular_velocity": {
        "x": <uint32>,
        "y": <uint32>,
        "z": <uint32>
    },
    "orientation": {
        "x": <uint32>,
        "y": <uint32>,
        "z": <uint32>,
        "w": <uint32>
    }
}

Examples

Print all data for a specific message path:

>>> topic = Topic.from_name_and_file("/imu/data", "file_abc123")
>>> angular_velocity_x = topic.get_message_path("angular_velocity.x")
>>> for record in angular_velocity_x.get_data():
...     print(f"Time: {record['log_time']}, Value: {record['angular_velocity']['x']}")

Get data within a time range:

>>> for record in angular_velocity_x.get_data(
...     start_time=1722870127699468923,
...     end_time=1722870127799468923
... ):
...     print(record)

Collect data into a dataframe (requires installing the roboto[analytics] extra):

>>> df = angular_velocity_x.get_data_as_df()
>>> import math
>>> assert math.isclose(angular_velocity_x.mean, df[angular_velocity_x.path].mean())

get_data_as_df(start_time=None, end_time=None, cache_dir=None)#

Return this message path’s data as a pandas DataFrame.

Retrieves message path data and converts it to a pandas DataFrame for analysis and visualization. The DataFrame is indexed by log time and contains a column for this message path’s values.

Parameters:

start_time (Optional[roboto.time.Time]) – Start time (inclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
end_time (Optional[roboto.time.Time]) – End time (exclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
cache_dir (Union[str, pathlib.Path, None]) – Directory where topic data will be downloaded if necessary. Defaults to DEFAULT_CACHE_DIR.

Returns:

pandas DataFrame containing the message path data, indexed by log time.

Raises:

ImportError – pandas is not installed. Install with roboto[analytics] extra.

Return type:

pandas.DataFrame

Notes

Requires installing this package using the roboto[analytics] extra.

Examples

>>> topic = Topic.from_name_and_file("/imu/data", "file_abc123")
>>> angular_velocity_x = topic.get_message_path("angular_velocity.x")
>>> df = angular_velocity_x.get_data_as_df()
>>> print(df.head())
                        angular_velocity.x
log_time
1722870127699468923                  0.1
1722870127699468924                  0.15
>>> print(f"Mean: {df[angular_velocity_x.path].mean()}")
Mean: 0.125

property max: PreComputedStat#

Maximum value observed for this message path.

Return type:: PreComputedStat

property mean: PreComputedStat#

Mean (average) value for this message path.

Return type:: PreComputedStat

property median: PreComputedStat#

Median value for this message path.

Return type:: PreComputedStat

property message_path_id: str#

Unique identifier for this message path.

Return type:: str

property metadata: dict[str, Any]#

Metadata dictionary associated with this message path.

Return type:: dict[str, Any]

property min: PreComputedStat#

Minimum value observed for this message path.

Return type:: PreComputedStat

property modified: datetime.datetime#

Timestamp when this message path was last modified.

Return type:: datetime.datetime

property modified_by: str#

Identifier of the user or system that last modified this message path.

Return type:: str

property org_id: str#

Organization ID that owns this message path.

Return type:: str

static parents(path)#

Get parent paths for a message path in dot notation.

Given a message path in dot notation, returns a list of its parent paths ordered from most specific to least specific.

Parameters:: path (str) – Message path in dot notation (e.g., “pose.pose.position.x”).
Returns:: List of parent paths in dot notation, ordered from most to least specific.
Return type:: list[str]

Examples

>>> path = "pose.pose.position.x"
>>> MessagePath.parents(path)
['pose.pose.position', 'pose.pose', 'pose']

>>> # Single level path has no parents
>>> MessagePath.parents("velocity")
[]

static parts(path)#

Split message path in dot notation into its constituent parts.

Splits a message path string into individual components, useful for programmatic manipulation of message path hierarchies.

Parameters:: path (str) – Message path in dot notation (e.g., “pose.pose.position.x”).
Returns:: List of path components in order from root to leaf.
Return type:: list[str]

Examples

>>> path = "pose.pose.position.x"
>>> MessagePath.parts(path)
['pose', 'pose', 'position', 'x']

>>> # Single component path
>>> MessagePath.parts("velocity")
['velocity']

property path: str#

Dot-delimited path to the attribute (e.g., ‘pose.position.x’).

Return type:: str

property record: roboto.domain.topics.record.MessagePathRecord#

Underlying MessagePathRecord for this message path.

Return type:: roboto.domain.topics.record.MessagePathRecord

to_association()#

Convert this message path to an Association object.

Creates an Association object that can be used to reference this message path in other parts of the Roboto platform.

Returns:: Association object representing this message path.
Return type:: roboto.association.Association

Examples

>>> message_path = MessagePath.from_id("mp_abc123")
>>> association = message_path.to_association()
>>> print(association.association_type)
AssociationType.MessagePath
>>> print(association.association_id)
mp_abc123

property topic_id: str#

Unique identifier of the topic containing this message path.

Return type:: str

type roboto.domain.topics.message_path.PreComputedStat = Optional[Union[int, float]]#