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

### Topic

```python
class roboto.domain.topics.topic.Topic(
    record: roboto.domain.topics.record.TopicRecord,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
    topic_data_service: Optional[roboto.domain.topics.topic_data_service.TopicDataService] = None,
)
```

`from roboto import Topic`

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

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

- **record** (`roboto.domain.topics.record.TopicRecord`)
- **roboto_client** (`Optional[roboto.http.RobotoClient]`)
- **topic_data_service** (`Optional[roboto.domain.topics.topic_data_service.TopicDataService]`)

**Properties**

- **Topic.association** (`roboto.association.Association`): Association linking this topic to its source entity (typically a file).

- **Topic.created** (`datetime.datetime`): Timestamp when this topic was created in the Roboto platform.

- **Topic.created_by** (`str`): Identifier of the user or system that created this topic.

- **Topic.dataset_id** (`str | None`): Unique identifier of the dataset containing this topic, if applicable.

  Return type: `Optional[str]`

- **Topic.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]`

- **Topic.end_time** (`int | None`): End time of the topic data in nanoseconds since UNIX epoch.

  Return type: `Optional[int]`

- **Topic.file_id** (`str | None`): Unique identifier of the file containing this topic, if applicable.

  Return type: `Optional[str]`

- **Topic.message_count** (`int | None`): Total number of messages in this topic.

  Return type: `Optional[int]`

- **Topic.message_paths** (`collections.abc.Sequence[roboto.domain.topics.record.MessagePathRecord]`): Sequence of message path records defining the topic's schema.

- **Topic.metadata** (`dict[str, Any]`): Metadata dictionary associated with this topic.

- **Topic.modified** (`datetime.datetime`): Timestamp when this topic was last modified.

- **Topic.modified_by** (`str`): Identifier of the user or system that last modified this topic.

- **Topic.name** (`str`): Name of the topic (e.g., '/camera/image', '/imu/data').

- **Topic.org_id** (`str`): Organization ID that owns this topic.

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

- **Topic.schema_checksum** (`str | None`): Checksum of the topic's message schema for validation.

  Return type: `Optional[str]`

- **Topic.schema_id** (`str | None`): ID of the schema for this topic.

  `None` if the topic has no schema, or if the schema has not yet been populated.

  Return type: `Optional[str]`

- **Topic.schema_name** (`str | None`): Name of the message schema (e.g., 'sensor_msgs/Image').

  Return type: `Optional[str]`

- **Topic.start_time** (`int | None`): Start time of the topic data in nanoseconds since UNIX epoch.

  Return type: `Optional[int]`

- **Topic.topic_id** (`str`): Unique identifier for this topic.

- **Topic.topic_name** (`str`): Name of the topic (e.g., '/camera/image', '/imu/data').

#### Topic.add_message_path()

```python
def add_message_path(
    message_path: str,
    data_type: str,
    canonical_data_type: roboto.domain.topics.record.CanonicalDataType,
    path_in_schema: Optional[list[str]] = None,
    metadata: Optional[dict[str, Any]] = None,
) -> roboto.domain.topics.record.MessagePathRecord
```

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

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.
- **path_in_schema** (`Optional[list[str]]`): List of path components representing the field's location in the source data schema. Unlike message_path, which assumes dots separate path parts implying nested data, this preserves the exact path from the source data for accurate attribute access.
- **metadata** (`Optional[dict[str, Any]]`): Additional metadata to associate with the message path.

**Returns**

- `roboto.domain.topics.record.MessagePathRecord`: MessagePathRecord representing the newly created message path.

**Raises**

- [`RobotoConflictException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoConflictException): Message path already exists for this topic.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to modify the topic.

**Usage**

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

#### Topic.add_message_path_representation()

```python
def add_message_path_representation(
    message_path_id: str,
    association: roboto.association.Association,
    storage_format: roboto.domain.topics.record.RepresentationStorageFormat,
    version: int,
    format: Optional[str] = None,
    transformations: Optional[list[str]] = None,
) -> roboto.domain.topics.record.RepresentationRecord
```

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

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

- **message_path_id** (`str`): Unique identifier of the message path.
- **association** (`roboto.association.Association`): Association pointing to the representation data.
- **storage_format** (`roboto.domain.topics.record.RepresentationStorageFormat`): Format of the representation data.
- **version** (`int`): Version number of the representation.
- **format** (`Optional[str]`): Content format descriptor (e.g. "jpeg", "sensor_msgs/Image").
- **transformations** (`Optional[list[str]]`): Transformation descriptors applied (e.g. ["downsample:0.5"]).

**Returns**

- `roboto.domain.topics.record.RepresentationRecord`: RepresentationRecord representing the newly created representation.

**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 modify the topic.

**Usage**

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

#### Topic.create()

```python
@classmethod
def create(
    file_id: str,
    topic_name: str,
    end_time: Optional[int] = None,
    message_count: Optional[int] = None,
    metadata: Optional[collections.abc.Mapping[str, Any]] = None,
    schema_checksum: Optional[str] = None,
    schema_name: Optional[str] = None,
    start_time: Optional[int] = None,
    message_paths: Optional[collections.abc.Sequence[roboto.domain.topics.operations.AddMessagePathRequest]] = None,
    caller_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Topic
```

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

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()`](/reference/python-sdk/roboto/domain/topics/topic#roboto.domain.topics.topic.Topic.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`: Topic instance representing the newly created topic.

**Raises**

- [`RobotoInvalidRequestException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoInvalidRequestException): Invalid topic parameters.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to create topics.

**Usage**

Create a basic topic for camera data:

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

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

#### Topic.create_from_df()

```python
@classmethod
def create_from_df(
    file_id: str,
    dataset_id: str,
    topic_name: str,
    df: pandas.DataFrame,
    timestamp_column: Optional[str] = None,
    timestamp_unit: Optional[Union[str, roboto.time.TimeUnit]] = None,
    caller_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Topic
```

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

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.

**Parameters**

- **file_id** (`str`): ID of the file to associate this topic with.
- **dataset_id** (`str`): ID of the dataset containing the file.
- **topic_name** (`str`): Name for the topic. Must be unique within the file.
- **df** (`pandas.DataFrame`): pandas DataFrame containing the data to ingest. Must include a timestamp column (either explicitly specified or automatically detectable).
- **timestamp_column** (`Optional[str]`): Name of the column to use as the timestamp. If not provided, the method will attempt to automatically detect a timestamp column by looking for the first column that is a timezone-aware timestamp type.
- **timestamp_unit** (`Optional[Union[str, roboto.time.TimeUnit]]`): Unit of the timestamp column values. Required when timestamp_column contains numeric values (int, float, decimal). Valid values include "s", "ms", "us", "ns". Not needed for datetime columns or when timestamp_column is not specified.
- **caller_org_id** (`Optional[str]`): Organization ID of the caller. If not provided, uses the default from the client context.
- **roboto_client** (`Optional[roboto.http.RobotoClient]`): Roboto client instance. If not provided, uses the default client.

**Returns**

- `Topic`: The created or updated Topic instance.

**Raises**

- [`IngestionException`](/reference/python-sdk/roboto/exceptions/ingestion#roboto.exceptions.ingestion.IngestionException): If the timestamp column cannot be determined, is not present in the DataFrame, has an invalid type, or if the timestamp unit is required but not provided.
- `ImportError`: If pandas or pyarrow are not installed. Install with `pip install roboto[ingestion]` to use this feature.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): If the caller lacks permission to create topics or upload files to the specified dataset.

**Notes**

- For most use cases, prefer [`File.add_topic()`](/reference/python-sdk/roboto/domain/files/file#roboto.domain.files.file.File.add_topic) instead

**Usage**

Create a topic when you have file and dataset IDs:

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

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

#### Topic.delete()

```python
def delete() -> None
```

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

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

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): Topic does not exist or has already been deleted.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to delete the topic.

**Returns**

- `None`

**Usage**

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

#### Topic.from_id()

```python
@classmethod
def from_id(
    topic_id: str,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Topic
```

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

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`: Topic instance representing the requested topic.

**Raises**

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): Topic with the given ID does not exist.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to access the topic.

**Usage**

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

#### Topic.from_name_and_file()

```python
@classmethod
def from_name_and_file(
    topic_name: str,
    file_id: str,
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> Topic
```

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

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`: Topic instance representing the requested topic.

**Raises**

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): Topic with the given name does not exist in the specified file.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to access the topic.

**Usage**

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

#### Topic.get_by_dataset()

```python
@classmethod
def get_by_dataset(
    dataset_id: str,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> collections.abc.Generator[Topic, None, None]
```

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

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.

> **Note**
>
> This method returns topics WITHOUT message_paths populated (for performance). The message_paths list will be empty. If you need message_paths, use Topic.get_by_file() for each file instead.

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

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): Dataset with the given ID does not exist.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to access the dataset.

**Returns**

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

**Usage**

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

```python
# 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})
```

#### Topic.get_by_file()

```python
@classmethod
def get_by_file(
    file_id: str,
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> collections.abc.Generator[Topic, None, None]
```

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

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

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): File with the given ID does not exist.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to access the file.

**Returns**

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

**Usage**

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

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

#### Topic.get_data()

```python
def get_data(
    message_paths_include: Optional[collections.abc.Iterable[str]] = None,
    message_paths_exclude: Optional[collections.abc.Iterable[str]] = None,
    start_time: Optional[roboto.time.Time] = None,
    end_time: Optional[roboto.time.Time] = None,
    cache_dir: Union[str, pathlib.Path, None] = None,
    representation_selector: roboto.domain.topics.record.RepresentationSelector = RepresentationSelector.raw(),
) -> 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/topic.py#L888-L996)

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.Iterable[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.Iterable[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()`](/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).
- **representation_selector** (`roboto.domain.topics.record.RepresentationSelector`): Criteria for selecting among multiple representations. Defaults to [`RepresentationSelector.raw()`](/reference/python-sdk/roboto/domain/topics/record#roboto.domain.topics.record.RepresentationSelector.raw) — original, untransformed data. Pass a `RepresentationSelector` to request a specific content format or transformation pipeline.

**Yields**

- Timestamp and dictionary records that match this topic's schema, filtered according to the parameters.

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

```
{
```

> **"angular_velocity": {**
>
> "x": \<uint32>, "y": \<uint32>, "z": \<uint32>
>
> }, "orientation": { "x": \<uint32>, "y": \<uint32>, "z": \<uint32>, "w": \<uint32> }

}

**Usage**

Print all data to stdout:

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

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

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

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

Get the JPEG-encoded version of image data:

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

> **Tip**
>
> For many topics, 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 t: list(t.get_data()), topics))
> merged = list(chain.from_iterable(results))
> ```

#### Topic.get_data_as_df()

```python
def get_data_as_df(
    message_paths_include: Optional[collections.abc.Iterable[str]] = None,
    message_paths_exclude: Optional[collections.abc.Iterable[str]] = None,
    start_time: Optional[roboto.time.Time] = None,
    end_time: Optional[roboto.time.Time] = None,
    cache_dir: Union[str, pathlib.Path, None] = None,
    representation_selector: roboto.domain.topics.record.RepresentationSelector = RepresentationSelector.raw(),
) -> pandas.DataFrame
```

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

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.Iterable[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.Iterable[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()`](/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).
- **representation_selector** (`roboto.domain.topics.record.RepresentationSelector`): Criteria for selecting among multiple representations. Defaults to [`RepresentationSelector.raw()`](/reference/python-sdk/roboto/domain/topics/record#roboto.domain.topics.record.RepresentationSelector.raw) — original, untransformed data. Pass a `RepresentationSelector` to request a specific content format or transformation pipeline.

**Returns**

- `pandas.DataFrame`: pandas DataFrame containing the topic 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.

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

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

```python
# 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']
```

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

> **Tip**
>
> For many topics, 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 t: t.get_data_as_df(), topics))
> combined = pd.concat(dfs).sort_index()
> ```

#### Topic.get_message_path()

```python
def get_message_path(message_path: str) -> roboto.domain.topics.message_path.MessagePath
```

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

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

- `roboto.domain.topics.message_path.MessagePath`: MessagePath instance for the specified path.

**Raises**

- `ValueError`: No message path with the given name exists in this topic.

**Usage**

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

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

#### Topic.get_schema()

```python
def get_schema() -> Optional[roboto.domain.topics.topic_schema.TopicSchema]
```

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

Retrieve the schema for this topic, if one exists.

**Returns**

- `Optional[roboto.domain.topics.topic_schema.TopicSchema]`: A [`TopicSchema`](/reference/python-sdk/roboto/domain/topics/topic_schema#roboto.domain.topics.topic_schema.TopicSchema) describing the message structure of this topic, or `None` if this topic has no schema.

**Raises**

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): `schema_id` references a schema that no longer exists.

**Usage**

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

#### Topic.get_time_bounds_by_association()

```python
@classmethod
def get_time_bounds_by_association(
    association: roboto.association.Association,
    owner_org_id: Optional[str] = None,
    roboto_client: Optional[roboto.http.RobotoClient] = None,
) -> roboto.domain.topics.record.TopicTimeBounds
```

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

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.

**Parameters**

- **association** (`roboto.association.Association`): The file or dataset whose topics are aggregated, e.g. `Association.file("file_abc123")`.
- **owner_org_id** (`Optional[str]`): Organization ID to scope the lookup. If None, uses caller's org.
- **roboto_client** (`Optional[roboto.http.RobotoClient]`): HTTP client for API communication. If None, uses the default client.

**Returns**

- `roboto.domain.topics.record.TopicTimeBounds`: Bounds in nanoseconds since the Unix epoch. Both fields are None when the association holds no topics, and either is None when no topic of the association carries that timestamp.

**Raises**

- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to access the file or dataset.

**Usage**

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

#### Topic.refresh()

```python
def refresh() -> None
```

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

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

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

**Returns**

- `None`

#### Topic.set_default_representation()

```python
def set_default_representation(
    association: roboto.association.Association,
    storage_format: roboto.domain.topics.record.RepresentationStorageFormat,
    version: int,
    format: Optional[str] = None,
    transformations: Optional[list[str]] = None,
) -> roboto.domain.topics.record.RepresentationRecord
```

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

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

- **association** (`roboto.association.Association`): Association pointing to the representation data.
- **storage_format** (`roboto.domain.topics.record.RepresentationStorageFormat`): Format of the representation data.
- **version** (`int`): Version number of the representation.
- **format** (`Optional[str]`): Content format descriptor (e.g. "jpeg", "sensor_msgs/Image").
- **transformations** (`Optional[list[str]]`): Transformation descriptors applied (e.g. ["downsample:0.5"]).

**Returns**

- `roboto.domain.topics.record.RepresentationRecord`: RepresentationRecord representing the newly set default representation.

**Raises**

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): Specified representation does not exist.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to modify the topic.

**Usage**

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

#### Topic.to_association()

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

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

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

- `roboto.association.Association`: Association object representing this topic.

**Usage**

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

#### Topic.update()

```python
def update(
    end_time: Union[Optional[int], roboto.sentinels.NotSetType] = NotSet,
    message_count: Union[int, roboto.sentinels.NotSetType] = NotSet,
    schema_checksum: Union[Optional[str], roboto.sentinels.NotSetType] = NotSet,
    schema_name: Union[Optional[str], roboto.sentinels.NotSetType] = NotSet,
    start_time: Union[Optional[int], roboto.sentinels.NotSetType] = NotSet,
    metadata_changeset: Union[roboto.updates.MetadataChangeset, roboto.sentinels.NotSetType] = NotSet,
    message_path_changeset: Union[roboto.domain.topics.operations.MessagePathChangeset, roboto.sentinels.NotSetType] = NotSet,
) -> Topic
```

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

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

**Parameters**

- **schema_name** (`Union[Optional[str], roboto.sentinels.NotSetType]`): topic schema name. Setting to `None` clears the attribute.
- **schema_checksum** (`Union[Optional[str], roboto.sentinels.NotSetType]`): topic schema checksum. Setting to `None` clears the attribute.
- **start_time** (`Union[Optional[int], roboto.sentinels.NotSetType]`): topic data start time, in epoch nanoseconds. Must be non-negative. Setting to `None` clears the attribute.
- **end_time** (`Union[Optional[int], roboto.sentinels.NotSetType]`): topic data end time, in epoch nanoseconds. Must be non-negative, and greater than start_time. Setting to None clears the attribute.
- **message_count** (`Union[int, roboto.sentinels.NotSetType]`): number of messages recorded for this topic. Must be non-negative.
- **metadata_changeset** (`Union[roboto.updates.MetadataChangeset, roboto.sentinels.NotSetType]`): a set of changes to apply to the topic's metadata
- **message_path_changeset** (`Union[roboto.domain.topics.operations.MessagePathChangeset, roboto.sentinels.NotSetType]`): a set of additions, deletions or updates to this topic's message paths. Updating or deleting non-existent message paths has no effect. Attempting to (re-)add existing message paths raises `RobotoConflictException`, unless the changeset's `replace_all` flag is set to `True`

**Returns**

- `Topic`: this `Topic` object with any updates applied

**Raises**

- [`RobotoInvalidRequestException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoInvalidRequestException): if any method argument has an invalid value, e.g. a negative `message_count`
- [`RobotoConflictException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.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`

#### Topic.update_message_path()

```python
def update_message_path(
    message_path: str,
    metadata_changeset: Union[roboto.updates.TaglessMetadataChangeset, roboto.sentinels.NotSetType] = NotSet,
    data_type: Union[str, roboto.sentinels.NotSetType] = NotSet,
    canonical_data_type: Union[roboto.domain.topics.record.CanonicalDataType, roboto.sentinels.NotSetType] = NotSet,
    path_in_schema: Union[list[str], roboto.sentinels.NotSetType] = NotSet,
) -> roboto.domain.topics.message_path.MessagePath
```

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

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

- **message_path** (`str`): Name of the message path to update (e.g., "pose.position.x").
- **metadata_changeset** (`Union[roboto.updates.TaglessMetadataChangeset, roboto.sentinels.NotSetType]`): Metadata changeset to apply to any existing metadata.
- **data_type** (`Union[str, roboto.sentinels.NotSetType]`): Native (application-specific) message path data type.
- **canonical_data_type** (`Union[roboto.domain.topics.record.CanonicalDataType, roboto.sentinels.NotSetType]`): Canonical Roboto data type corresponding to the native data type.
- **path_in_schema** (`Union[list[str], roboto.sentinels.NotSetType]`)

**Returns**

- `roboto.domain.topics.message_path.MessagePath`: MessagePath instance representing the updated message path.

**Raises**

- [`RobotoNotFoundException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoNotFoundException): No message path with the given name exists for this topic.
- [`RobotoUnauthorizedException`](/reference/python-sdk/roboto/exceptions/domain#roboto.exceptions.domain.RobotoUnauthorizedException): Caller lacks permission to modify the topic.

**Usage**

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

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

### logger

```python
roboto.domain.topics.topic.logger
```

`from roboto.domain.topics.topic import logger`

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