roboto.domain.topics.topic#

Module Contents#

class roboto.domain.topics.topic.Topic(record, roboto_client=None, topic_data_service=None)#

Represents a topic within the Roboto platform.

A topic is a sequence of structured time-series data linked to a source file, typically containing sensor readings, robot state information, or other timestamped data streams. Topics are fundamental building blocks for data analysis in robotics, providing organized access to time-synchronized data from various sources like ROS bags, MCAP files, or other structured data formats.

Each topic follows a defined schema where message paths represent the individual fields or signals within that schema. Topics enable efficient querying, filtering, and analysis of time-series data, supporting operations like temporal slicing, field selection, and data export to various formats including pandas DataFrames.

Topics are associated with files and inherit access permissions from their parent dataset. They provide the primary interface for accessing ingested robotics data in the Roboto platform, supporting both programmatic access through the SDK and visualization in the web interface.

The Topic class serves as the main interface for topic operations in the Roboto SDK, providing methods for data retrieval, message path management, metadata operations, and schema management.

Parameters:
add_message_path(message_path, data_type, canonical_data_type, metadata=None)#

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.

Parameters:

  • message_path (str) – Dot-delimited path to the attribute (e.g., “pose.position.x”).

  • data_type (str) – Native data type of the attribute as it appears in the original data source (e.g., “float32”, “uint8[]”, “geometry_msgs/Pose”). Used primarily for display purposes and should match the robot’s runtime language or schema definitions.

  • canonical_data_type (roboto.domain.topics.record.CanonicalDataType) – Normalized Roboto data type that enables specialized platform features for maps, images, timestamps, and other data with special interpretations.

  • metadata (Optional[dict[str, Any]]) – Additional metadata to associate with the message path.

Returns:

MessagePathRecord representing the newly created message path.

Raises:
Return type:

roboto.domain.topics.record.MessagePathRecord

Examples

>>> 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_message_path_representation(message_path_id, association, storage_format, version)#

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.

Parameters:
Returns:

RepresentationRecord representing the newly created representation.

Raises:
Return type:

roboto.domain.topics.record.RepresentationRecord

Examples

>>> 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
property association: roboto.association.Association#

Association linking this topic to its source entity (typically a file).

Return type:

roboto.association.Association

classmethod create(file_id, topic_name, end_time=None, message_count=None, metadata=None, schema_checksum=None, schema_name=None, start_time=None, message_paths=None, caller_org_id=None, roboto_client=None)#

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.

Note

On successful creation, the topic will be a metadata-only container and will not be visualizable or usable via data access methods like get_data_as_df() until one or more representations have been registered. This is typically handled automatically by ingestion actions, but power users may need to manage representations manually.

Parameters:

  • file_id (str) – Unique identifier of the file this topic is associated with.

  • topic_name (str) – Name of the topic (e.g., “/camera/image”, “/imu/data”).

  • end_time (Optional[int]) – End time of the topic data in nanoseconds since UNIX epoch.

  • message_count (Optional[int]) – Total number of messages in this topic.

  • metadata (Optional[collections.abc.Mapping[str, Any]]) – Additional metadata to associate with the topic.

  • schema_checksum (Optional[str]) – Checksum of the topic’s message schema for validation.

  • schema_name (Optional[str]) – Name of the message schema (e.g., “sensor_msgs/Image”).

  • start_time (Optional[int]) – Start time of the topic data in nanoseconds since UNIX epoch.

  • message_paths (Optional[collections.abc.Sequence[roboto.domain.topics.operations.AddMessagePathRequest]]) – Message paths to create along with the topic.

  • caller_org_id (Optional[str]) – Organization ID to create the topic in. Required for multi-org users.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Topic instance representing the newly created topic.

Raises:
Return type:

Topic

Examples

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
... )
property created: datetime.datetime#

Timestamp when this topic was created in the Roboto platform.

Return type:

datetime.datetime

property created_by: str#

Identifier of the user or system that created this topic.

Return type:

str

property dataset_id: str | None#

Unique identifier of the dataset containing this topic, if applicable.

Return type:

Optional[str]

property default_representation: roboto.domain.topics.record.RepresentationRecord | None#

Default representation used for accessing this topic’s data.

Return type:

Optional[roboto.domain.topics.record.RepresentationRecord]

delete()#

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.

Raises:
Return type:

None

Examples

>>> topic = Topic.from_id("topic_xyz789")
>>> topic.delete()
# Topic and all its data are now permanently deleted
property end_time: int | None#

End time of the topic data in nanoseconds since UNIX epoch.

Return type:

Optional[int]

property file_id: str | None#

Unique identifier of the file containing this topic, if applicable.

Return type:

Optional[str]

classmethod from_id(topic_id, roboto_client=None)#

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.

Parameters:

  • topic_id (str) – Unique identifier for the topic.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Topic instance representing the requested topic.

Raises:
Return type:

Topic

Examples

>>> topic = Topic.from_id("topic_xyz789")
>>> print(topic.name)
'/camera/image'
>>> print(topic.message_count)
100
classmethod from_name_and_file(topic_name, file_id, owner_org_id=None, roboto_client=None)#

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.

Parameters:

  • topic_name (str) – Name of the topic to retrieve.

  • file_id (str) – Unique identifier of the file containing the topic.

  • owner_org_id (Optional[str]) – Organization ID to scope the search. If None, uses caller’s org.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Topic instance representing the requested topic.

Raises:
Return type:

Topic

Examples

>>> 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
classmethod get_by_dataset(dataset_id, roboto_client=None)#

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.

Parameters:

  • dataset_id (str) – Unique identifier of the dataset to search.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Topic instances associated with files in the dataset.

Raises:
Return type:

collections.abc.Generator[Topic, None, None]

Examples

>>> 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})
classmethod get_by_file(file_id, owner_org_id=None, roboto_client=None)#

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.

Parameters:

  • file_id (str) – Unique identifier of the file to search.

  • owner_org_id (Optional[str]) – Organization ID to scope the search. If None, uses caller’s org.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Topic instances associated with the specified file.

Raises:
Return type:

collections.abc.Generator[Topic, None, None]

Examples

>>> 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
... ]
get_data(message_paths_include=None, message_paths_exclude=None, start_time=None, end_time=None, cache_dir=None)#

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.

Parameters:

  • message_paths_include (Optional[collections.abc.Sequence[str]]) – Dot notation paths that match attributes of individual data records to include. If None, all paths are included.

  • message_paths_exclude (Optional[collections.abc.Sequence[str]]) – Dot notation paths that match attributes of individual data records to exclude. If None, no paths are excluded.

  • 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 that match this topic’s schema, filtered according to the parameters.

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

{
“angular_velocity”: {

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

}, “orientation”: {

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

}

}

Examples

Print all data to stdout:

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

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

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

Only include data between two timestamps:

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

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_data_as_df(message_paths_include=None, message_paths_exclude=None, start_time=None, end_time=None, cache_dir=None)#

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.

Parameters:

  • message_paths_include (Optional[collections.abc.Sequence[str]]) – Dot notation paths that match attributes of individual data records to include. If None, all paths are included.

  • message_paths_exclude (Optional[collections.abc.Sequence[str]]) – Dot notation paths that match attributes of individual data records to exclude. If None, no paths are excluded.

  • 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 topic 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")
>>> 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']
get_message_path(message_path)#

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.

Parameters:

message_path (str) – Dot-delimited path to the desired attribute (e.g., “pose.position.x”).

Returns:

MessagePath instance for the specified path.

Raises:

ValueError – No message path with the given name exists in this topic.

Return type:

roboto.domain.topics.message_path.MessagePath

Examples

>>> 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
property message_count: int | None#

Total number of messages in this topic.

Return type:

Optional[int]

property message_paths: collections.abc.Sequence[roboto.domain.topics.record.MessagePathRecord]#

Sequence of message path records defining the topic’s schema.

Return type:

collections.abc.Sequence[roboto.domain.topics.record.MessagePathRecord]

property metadata: dict[str, Any]#

Metadata dictionary associated with this topic.

Return type:

dict[str, Any]

property modified: datetime.datetime#

Timestamp when this topic was last modified.

Return type:

datetime.datetime

property modified_by: str#

Identifier of the user or system that last modified this topic.

Return type:

str

property name: str#

Name of the topic (e.g., ‘/camera/image’, ‘/imu/data’).

Return type:

str

property org_id: str#

Organization ID that owns this topic.

Return type:

str

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

Topic representation in the Roboto database.

This property is on the path to deprecation. All TopicRecord attributes are accessible directly using a Topic instance.

Return type:

roboto.domain.topics.record.TopicRecord

refresh()#

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.

Examples

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

None

property schema_checksum: str | None#

Checksum of the topic’s message schema for validation.

Return type:

Optional[str]

property schema_name: str | None#

Name of the message schema (e.g., ‘sensor_msgs/Image’).

Return type:

Optional[str]

set_default_representation(association, storage_format, version)#

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.

Parameters:
Returns:

RepresentationRecord representing the newly set default representation.

Raises:
Return type:

roboto.domain.topics.record.RepresentationRecord

Examples

>>> 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
property start_time: int | None#

Start time of the topic data in nanoseconds since UNIX epoch.

Return type:

Optional[int]

to_association()#

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.

Returns:

Association object representing this topic.

Return type:

roboto.association.Association

Examples

>>> topic = Topic.from_id("topic_xyz789")
>>> association = topic.to_association()
>>> print(association.association_type)
AssociationType.Topic
>>> print(association.association_id)
topic_xyz789
property topic_id: str#

Unique identifier for this topic.

Return type:

str

property topic_name: str#

Name of the topic (e.g., ‘/camera/image’, ‘/imu/data’).

Return type:

str

update(end_time=NotSet, message_count=NotSet, schema_checksum=NotSet, schema_name=NotSet, start_time=NotSet, metadata_changeset=NotSet, message_path_changeset=NotSet)#

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

Parameters:
Returns:

this Topic object with any updates applied

Raises:

  • RobotoInvalidRequestException – if any method argument has an invalid value, e.g. a negative message_count

  • RobotoConflictException – if, as part of the update, an attempt is made to add an already extant message path, and to this topic, and replace_all is not toggled on the message_path_changeset

Return type:

Topic

update_message_path(message_path, metadata_changeset=NotSet, data_type=NotSet, canonical_data_type=NotSet)#

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.

Parameters:
Returns:

MessagePath instance representing the updated message path.

Raises:
Return type:

roboto.domain.topics.message_path.MessagePath

Examples

>>> 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
... )
property url_quoted_name: str#

URL-encoded version of the topic name for use in API calls.

Return type:

str

roboto.domain.topics.topic.logger#
On this page