roboto.domain.topics.topic

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
)