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

roboto.domain.topics.message_path

Module Contents

MessagePath

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

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]

Attributes

MessagePath.DELIMITER

DELIMITER ClassVar = '.' #

Properties

MessagePath.canonical_data_type

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

MessagePath.count

count PreComputedStat #

Number of data points available for this message path.

Return type: PreComputedStat

MessagePath.created

created datetime.datetime #

Timestamp when this message path was created.

Return type: datetime.datetime

MessagePath.created_by

created_by str #

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

Return type: str

MessagePath.data_type

data_type str #

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

Return type: str

MessagePath.from_id()

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

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

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.

Usage

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

MessagePath.get_data()

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

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[tuple[roboto.domain.topics.topic_reader.Timestamp, 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>
    }
}

Usage

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())

MessagePath.get_data_as_df()

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

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

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

Raises

ImportError

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

Notes

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

Usage

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

Properties

MessagePath.max

max PreComputedStat #

Maximum value observed for this message path.

Return type: PreComputedStat

MessagePath.mean

mean PreComputedStat #

Mean (average) value for this message path.

Return type: PreComputedStat

MessagePath.median

median PreComputedStat #

Median value for this message path.

Return type: PreComputedStat

MessagePath.message_path_id

message_path_id str #

Unique identifier for this message path.

Return type: str

MessagePath.metadata

metadata dict[str, Any] #

Metadata dictionary associated with this message path.

Return type: dict[str, Any]

MessagePath.min

min PreComputedStat #

Minimum value observed for this message path.

Return type: PreComputedStat

MessagePath.modified

modified datetime.datetime #

Timestamp when this message path was last modified.

Return type: datetime.datetime

MessagePath.modified_by

modified_by str #

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

Return type: str

MessagePath.org_id

org_id str #

Organization ID that owns this message path.

Return type: str

MessagePath.p25

p25 PreComputedStat #

25th percentile of the values observed for this message path.

Return type: PreComputedStat

MessagePath.p75

p75 PreComputedStat #

75th percentile of the values observed for this message path.

Return type: PreComputedStat

MessagePath.p95

p95 PreComputedStat #

95th percentile of the values observed for this message path.

Return type: PreComputedStat

MessagePath.p99

p99 PreComputedStat #

99th percentile of the values observed for this message path.

Return type: PreComputedStat

MessagePath.parents()

static parents(path_in_schema)#View Source

Get parent paths for a message path.

Given a path_in_schema (list of path components), returns a list of its parent paths ordered from most specific to least specific.

Parameters

path_in_schema list[str]

List of path components (e.g., [“pose”, “pose”, “position”, “x”]).

Returns

list[str]

List of parent paths in dot notation, ordered from most to least specific.

Raises

TypeError

If a string is passed instead of a list. This method previously accepted a dot-delimited string; passing a string now would silently iterate over its characters and produce wrong results.

Usage

path_in_schema = ["pose", "pose", "position", "x"]
MessagePath.parents(path_in_schema)
# ['pose.pose.position', 'pose.pose', 'pose']
# Single level path has no parents
MessagePath.parents(["velocity"])
# []

Properties

MessagePath.path

path str #

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

Return type: str

MessagePath.record

record roboto.domain.topics.record.MessagePathRecord #

Underlying MessagePathRecord for this message path.

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

MessagePath.stddev

stddev PreComputedStat #

Standard deviation of the values observed for this message path.

Return type: PreComputedStat

MessagePath.to_association()

to_association()#View Source

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

roboto.association.Association

Association object representing this message path.

Usage

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

Properties

MessagePath.topic_id

topic_id str #

Unique identifier of the topic containing this message path.

Return type: str

PreComputedStat

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

Was this page helpful?