---
sidebar:
  hidden: true
title: roboto.domain.topics.message_path
---
## Module Contents

### MessagePath

```python
class roboto.domain.topics.message_path.MessagePath(
    record: roboto.domain.topics.record.MessagePathRecord,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
    topic_data_service: Optional[roboto.domain.topics.topic_data_service.TopicDataService] = None,
)
```

`from roboto import MessagePath`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/topics/message_path.py#L31-L421)

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** (`ClassVar`) = `'.'`

**Properties**

- **MessagePath.canonical_data_type** (`roboto.domain.topics.record.CanonicalDataType`): Canonical Roboto data type corresponding to the native data type.
- **MessagePath.count** (`PreComputedStat`): Number of data points available for this message path.
- **MessagePath.created** (`datetime.datetime`): Timestamp when this message path was created.
- **MessagePath.created_by** (`str`): Identifier of the user or system that created this message path.
- **MessagePath.data_type** (`str`): Native data type for this message path, e.g. 'float32'
- **MessagePath.max** (`PreComputedStat`): Maximum value observed for this message path.
- **MessagePath.mean** (`PreComputedStat`): Mean (average) value for this message path.
- **MessagePath.median** (`PreComputedStat`): Median value for this message path.
- **MessagePath.message_path_id** (`str`): Unique identifier for this message path.
- **MessagePath.metadata** (`dict[str, Any]`): Metadata dictionary associated with this message path.
- **MessagePath.min** (`PreComputedStat`): Minimum value observed for this message path.
- **MessagePath.modified** (`datetime.datetime`): Timestamp when this message path was last modified.
- **MessagePath.modified_by** (`str`): Identifier of the user or system that last modified this message path.
- **MessagePath.org_id** (`str`): Organization ID that owns this message path.
- **MessagePath.p25** (`PreComputedStat`): 25th percentile of the values observed for this message path.
- **MessagePath.p75** (`PreComputedStat`): 75th percentile of the values observed for this message path.
- **MessagePath.p95** (`PreComputedStat`): 95th percentile of the values observed for this message path.
- **MessagePath.p99** (`PreComputedStat`): 99th percentile of the values observed for this message path.
- **MessagePath.path** (`str`): Dot-delimited path to the attribute (e.g., 'pose.position.x').
- **MessagePath.record** (`roboto.domain.topics.record.MessagePathRecord`): Underlying MessagePathRecord for this message path.
- **MessagePath.stddev** (`PreComputedStat`): Standard deviation of the values observed for this message path.
- **MessagePath.topic_id** (`str`): Unique identifier of the topic containing this message path.

#### MessagePath.from_id()

```python
@classmethod
def from_id(
    message_path_id: str,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
    topic_data_service: Optional[roboto.domain.topics.topic_data_service.TopicDataService] = None,
) -> MessagePath
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/topics/message_path.py#L101-L133)

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`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): Message path with the given ID does not exist.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to access the message path.

**Usage**

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

```python
def get_data(
    start_time: Optional[roboto.time.Time] = None,
    end_time: Optional[roboto.time.Time] = None,
    cache_dir: Union[str, pathlib.Path, None] = None,
) -> collections.abc.Generator[tuple[roboto.domain.topics.topic_reader.Timestamp, dict[str, Any]], None, None]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/topics/message_path.py#L263-L341)

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()`](/reference/python-sdk/roboto/time#roboto.time.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()`](/reference/python-sdk/roboto/time#roboto.time.to_epoch_nanoseconds).
- **cache_dir** (`Union[str, pathlib.Path, None]`): Directory where topic data will be downloaded if necessary. Defaults to [`DEFAULT_CACHE_DIR`](/reference/python-sdk/roboto/domain/topics/topic_data_service#roboto.domain.topics.topic_data_service.TopicDataService.DEFAULT_CACHE_DIR).

**Yields**

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

**Returns**

- `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:

```python
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:

```python
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):

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

> **Tip**
>
> For many message paths, parallelize with a thread pool:
>
> ```python
> from concurrent.futures import ThreadPoolExecutor
> from itertools import chain
> with ThreadPoolExecutor(max_workers=16) as ex:  # tune for your workload
>     results = list(ex.map(lambda mp: list(mp.get_data()), message_paths))
> merged = list(chain.from_iterable(results))
> ```

#### MessagePath.get_data_as_df()

```python
def get_data_as_df(
    start_time: Optional[roboto.time.Time] = None,
    end_time: Optional[roboto.time.Time] = None,
    cache_dir: Union[str, pathlib.Path, None] = None,
) -> pandas.DataFrame
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/topics/message_path.py#L343-L399)

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()`](/reference/python-sdk/roboto/time#roboto.time.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()`](/reference/python-sdk/roboto/time#roboto.time.to_epoch_nanoseconds).
- **cache_dir** (`Union[str, pathlib.Path, None]`): Directory where topic data will be downloaded if necessary. Defaults to [`DEFAULT_CACHE_DIR`](/reference/python-sdk/roboto/domain/topics/topic_data_service#roboto.domain.topics.topic_data_service.TopicDataService.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**

```python
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
```

> **Tip**
>
> For many message paths, parallelize with a thread pool:
>
> ```python
> import pandas as pd
> from concurrent.futures import ThreadPoolExecutor
> with ThreadPoolExecutor(max_workers=16) as ex:  # tune for your workload
>     dfs = list(ex.map(lambda mp: mp.get_data_as_df(), message_paths))
> combined = pd.concat(dfs).sort_index()
> ```

#### MessagePath.parents()

```python
@staticmethod
def parents(path_in_schema: list[str]) -> list[str]
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/topics/message_path.py#L61-L98)

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**

```python
path_in_schema = ["pose", "pose", "position", "x"]
MessagePath.parents(path_in_schema)
# ['pose.pose.position', 'pose.pose', 'pose']
```

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

#### MessagePath.to_association()

```python
def to_association() -> roboto.association.Association
```

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/topics/message_path.py#L401-L418)

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**

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

### PreComputedStat

```python
type roboto.domain.topics.message_path.PreComputedStat = typing.Optional[typing.Union[int, float]]
```

`from roboto.domain.topics.message_path import PreComputedStat`

[Source](https://github.com/roboto-ai/roboto-python-sdk/blob/main/src/roboto/domain/topics/message_path.py#L28-L28)