roboto.domain.topics

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.

Usage

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

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.

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

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.

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

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.

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"])
# []

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.

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

Create a changeset that replaces all existing message paths.

Creates a changeset that will replace all existing message paths on a topic with the provided set of message paths. This is useful for completely redefining a topic’s schema.

Usage

from roboto.domain.topics import AddMessagePathRequest, CanonicalDataType
new_paths = [
    AddMessagePathRequest(
        message_path="velocity.x", data_type="float32", canonical_data_type=CanonicalDataType.Number
    )
]
changeset = MessagePathChangeset.from_replacement_message_paths(new_paths)

Check whether the changeset contains any actual changes.

Logical message path ancestors of this path.

Usage

Given a deeply nested field root.sub_obj_1.sub_obj_2.leaf_field:

field = "root.sub_obj_1.sub_obj_2.leaf_field"
record = MessagePathRecord(message_path=field)  # other fields omitted for brevity
print(record.parents())
# ['root.sub_obj_1.sub_obj_2', 'root.sub_obj_1', 'root']

Translate this record into the FieldSelection the format decoders accept.

Check whether a representation satisfies this selector’s criteria.

A representation matches when each non-None selector field is satisfied. For content_format, representations with no format set are treated as matching (legacy carve-out — see class docstring).

Select representations with no transformations applied (original data).

Select one representation per message path that matches this selector.

When the API returns multiple representations for the same message paths (e.g., both a raw MCAP and a processed JPEG MCAP for an image topic), this method picks a matching representation for each path and deduplicates so each message path appears in exactly one mapping.

Non-matching representations are excluded. When both an explicit format match and a legacy representation (no format set) cover the same message path, the explicit match wins. Message paths covered by no matching representation are dropped — callers must handle empty or partial results.

Add a new message path to this topic.

Creates a new message path within this topic, defining a specific field or signal that can be extracted from the topic’s data. Message paths use dot notation to specify nested attributes within the topic’s schema.

Usage

from roboto.domain.topics import CanonicalDataType
topic = Topic.from_id("topic_xyz789")
message_path = topic.add_message_path(
    message_path="pose.position.x",
    data_type="float64",
    canonical_data_type=CanonicalDataType.Number,
    metadata={"unit": "meters"},
)
print(message_path.message_path)
# pose.position.x

Add a representation for a specific message path.

Associates a message path with a data representation, enabling efficient access to specific fields within the topic data. Representations can be in different storage formats like MCAP or Parquet.

Usage

from roboto.association import Association
from roboto.domain.topics import RepresentationStorageFormat
topic = Topic.from_id("topic_xyz789")
representation = topic.add_message_path_representation(
    message_path_id="mp_123",
    association=Association.file("file_repr_456"),
    storage_format=RepresentationStorageFormat.MCAP,
    version=1,
)
print(representation.representation_id)
# repr_789

Create a new topic associated with a file.

Creates a new topic record in the Roboto platform, associating it with the specified file and defining its schema and temporal boundaries. This method is typically used during data ingestion to register topics found in robotics data files.

Usage

Create a basic topic for camera data:

topic = Topic.create(
    file_id="file_abc123",
    topic_name="/camera/image",
    schema_name="sensor_msgs/Image",
    start_time=1722870127699468923,
    end_time=1722870127799468923,
    message_count=100,
)
print(topic.topic_id)
# topic_xyz789

Create a topic with metadata and message paths:

from roboto.domain.topics import AddMessagePathRequest, CanonicalDataType
message_paths = [
    AddMessagePathRequest(
        message_path="header.stamp.sec",
        data_type="uint32",
        canonical_data_type=CanonicalDataType.Timestamp,
    )
]
topic = Topic.create(
    file_id="file_abc123",
    topic_name="/imu/data",
    schema_name="sensor_msgs/Imu",
    metadata={"sensor_type": "IMU", "frequency": 100},
    message_paths=message_paths,
)

Create a Topic from a pandas DataFrame and associate it with a file.

If a topic with the same name already exists for the specified file, it will be updated with the new data and schema.

Notes

• For most use cases, prefer File.add_topic() instead

Usage

Create a topic when you have file and dataset IDs:

import pandas as pd
from roboto.domain.topics import Topic
df = pd.DataFrame(
    {
        "timestamp": [1763947309.4198897, 1763947316.7686195, 1763947335.0095527],
        "temperature": [20.5, 21.0, 20.8],
        "humidity": [45.2, 46.1, 45.8],
    }
)
topic = Topic.create_from_df(
    file_id="file_abc123",
    dataset_id="ds_xyz789",
    topic_name="sensor_data",
    df=df,
    timestamp_column="timestamp",
    timestamp_unit="s",
)
print(f"Created topic: {topic.name}")
# Created topic: sensor_data

Using File.add_topic() is typically more convenient:

from roboto import File
file = File.from_id("file_abc123")
topic = file.add_topic("sensor_data", df, timestamp_column="timestamp", timestamp_unit="s")

Delete this topic from the Roboto platform.

Permanently removes this topic and all its associated message paths and representations from the platform. This operation cannot be undone.

Usage

topic = Topic.from_id("topic_xyz789")
topic.delete()
# # Topic and all its data are now permanently deleted

Retrieve a topic by its unique identifier.

Fetches a topic record from the Roboto platform using its unique topic ID. This is the most direct way to access a specific topic when you know its identifier.

Usage

topic = Topic.from_id("topic_xyz789")
print(topic.name)
# '/camera/image'
print(topic.message_count)
# 100

Retrieve a topic by its name and associated file.

Fetches a topic record using its name and the file it’s associated with. This is useful when you know the topic name (e.g., “/camera/image”) and the file containing the topic data.

Usage

topic = Topic.from_name_and_file(topic_name="/camera/image", file_id="file_abc123")
print(topic.topic_id)
# topic_xyz789
print(len(topic.message_paths))
# 5

List all topics associated with files in a dataset.

Retrieves all topics from files within the specified dataset. If multiple files contain topics with the same name (e.g., chunked files with the same schema), they are returned as separate topic objects.

Usage

for topic in Topic.get_by_dataset("ds_abc123"):
    print(f"Topic: {topic.name} (File: {topic.file_id})")
# Topic: /camera/image (File: file_001)
# Topic: /imu/data (File: file_001)
# Topic: /camera/image (File: file_002)
# Topic: /imu/data (File: file_002)

# Count topics by name
from collections import Counter
topic_names = [topic.name for topic in Topic.get_by_dataset("ds_abc123")]
print(Counter(topic_names))
# Counter({'/camera/image': 2, '/imu/data': 2})

List all topics associated with a specific file.

Retrieves all topics contained within the specified file. This is useful for exploring the structure of robotics data files and understanding what data streams are available.

Usage

for topic in Topic.get_by_file("file_abc123"):
    print(f"Topic: {topic.name} ({topic.message_count} messages)")
# Topic: /camera/image (150 messages)
# Topic: /imu/data (1500 messages)
# Topic: /gps/fix (50 messages)

# Get topics with specific schema
camera_topics = [topic for topic in Topic.get_by_file("file_abc123") if "camera" in topic.name]

Return this topic’s underlying data.

Retrieves and yields data records from this topic, with optional filtering by message paths and time range. Each yielded datum is a dictionary that matches this topic’s schema.

Notes

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

{

“angular_velocity”: {

“x”: <uint32>, “y”: <uint32>, “z”: <uint32>

}, “orientation”: { “x”: <uint32>, “y”: <uint32>, “z”: <uint32>, “w”: <uint32> }

}

Usage

Print all data to stdout:

topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data():
    print(timestamp, record)

Only include the “angular_velocity” sub-object, but filter out its “y” property:

topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data(
    message_paths_include=["angular_velocity"],
    message_paths_exclude=["angular_velocity.y"],
):
    ...

Only include data between two timestamps:

topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data(
    start_time=1722870127699468923,
    end_time=1722870127699468924,
):
    ...

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

topic = Topic.from_name_and_file(...)
df = topic.get_data_as_df()

Get the JPEG-encoded version of image data:

topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data(
    representation_selector=RepresentationSelector(content_format="jpeg"),
):
    ...

Return this topic’s underlying data as a pandas DataFrame.

Retrieves topic data and converts it to a pandas DataFrame for analysis and visualization. The DataFrame is indexed by log time and contains columns for each message path in the topic data.

Notes

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

An array-typed message path (for example a float32[3] acceleration) becomes a single column whose values are lists. Individual array elements are not addressable on their own — neither as separate DataFrame columns nor as message paths in message_paths_include / message_paths_exclude. Filter by the array’s path and unpack the column with numpy, as in the example below. np.stack requires every row to have the same length; a variable-length array path (for example a point cloud) must instead be processed row-wise, e.g. with df[col].map(...).

Usage

topic = Topic.from_name_and_file("/imu/data", "file_abc123")
df = topic.get_data_as_df()
print(df.head())
# angular_velocity.x  angular_velocity.y  ...
# log_time
# 1722870127699468923                  0.1                 0.2  ...
# 1722870127699468924                  0.15                0.25 ...

# Filter specific message paths
df_filtered = topic.get_data_as_df(
    message_paths_include=["angular_velocity"], message_paths_exclude=["angular_velocity.z"]
)
print(df_filtered.columns.tolist())
# ['angular_velocity.x', 'angular_velocity.y']

# Unpack a fixed-length array-typed path (a list-valued column) into a numpy array
import numpy as np
df_accel = topic.get_data_as_df(message_paths_include=["acceleration"])
xyz = np.stack(df_accel["acceleration"].to_numpy())  # shape (N, 3)

Get a specific message path from this topic.

Retrieves a MessagePath object for the specified path, enabling access to individual fields or signals within the topic’s data schema.

Usage

topic = Topic.from_name_and_file("/imu/data", "file_abc123")
angular_vel_x = topic.get_message_path("angular_velocity.x")
print(angular_vel_x.canonical_data_type)
# CanonicalDataType.Number

# Access message path statistics
print(angular_vel_x.mean)
# 0.125
print(angular_vel_x.std_dev)
# 0.05

Retrieve the schema for this topic, if one exists.

Usage

topic = Topic.from_id("topic_xyz789")
schema = topic.get_schema()
if schema is not None:
    for field in schema.fields:
        print(field.name, field.data_type)

Get the earliest start and latest end across every topic of a file or dataset.

The same aggregate you would reach by folding start_time and end_time over the topics of that file or dataset, computed server-side in one request instead of one per page of topics.

Usage

from roboto.association import Association
bounds = Topic.get_time_bounds_by_association(Association.file("file_abc123"))
print(bounds.start_time, bounds.end_time)
# 1722870127699468923 1722870187004821001

Refresh this topic instance with the latest data from the platform.

Fetches the current state of the topic from the Roboto platform and updates this instance’s data. Useful when the topic may have been modified by other processes or users.

Usage

topic = Topic.from_id("topic_xyz789")
# Topic may have been updated by another process
topic.refresh()
print(f"Current message count: {topic.message_count}")

Set the default representation for this topic.

Designates a specific representation as the default for this topic, which will be used when accessing topic data without specifying a particular representation.

Usage

from roboto.association import Association
from roboto.domain.topics import RepresentationStorageFormat
topic = Topic.from_id("topic_xyz789")
default_repr = topic.set_default_representation(
    association=Association.file("file_repr_456"),
    storage_format=RepresentationStorageFormat.MCAP,
    version=2,
)
print(topic.default_representation.representation_id)
# repr_789

Convert this topic to an Association object.

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

Usage

topic = Topic.from_id("topic_xyz789")
association = topic.to_association()
print(association.association_type)
# AssociationType.Topic
print(association.association_id)
# topic_xyz789

Updates a topic’s attributes and (optionally) its message paths.

Update the metadata and attributes of a message path.

Modifies an existing message path within this topic, allowing updates to its metadata, data type, and canonical data type. This is useful for correcting or enhancing message path definitions after initial creation.

Usage

from roboto.updates import TaglessMetadataChangeset
from roboto.domain.topics import CanonicalDataType
topic = Topic.from_id("topic_xyz789")

# Update metadata for a message path
changeset = TaglessMetadataChangeset(put_fields={"unit": "meters"})
updated_path = topic.update_message_path(message_path="pose.position.x", metadata_changeset=changeset)
print(updated_path.metadata["unit"])
# meters

# Update data type and canonical type
updated_path = topic.update_message_path(
    message_path="velocity", data_type="float64", canonical_data_type=CanonicalDataType.Number
)

Retrieve data for a specific topic with optional filtering.

Retrieve data for a specific topic as a pandas DataFrame with optional filtering.

Retrieve a schema by its ID.

Usage

from roboto.domain.topics import TopicSchema
schema = TopicSchema.from_id("ts_abc123")
for field in schema.fields:
    print(field.path_in_schema, field.data_type)

Parse a "<kind>:<param>" transformation descriptor into its kind and raw parameter.