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

### TopicDataService

```python
class roboto.domain.topics.topic_data_service.TopicDataService(
    roboto_client: roboto.http.RobotoClient,
    cache_dir: Union[str, pathlib.Path, None] = None,
)
```

`from roboto.domain.topics import TopicDataService`

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

Internal service for retrieving topic data.

This service handles the low-level operations for accessing topic data that has been ingested by the Roboto platform. It manages downloads, filtering, and processing various data formats to provide efficient access to time-series robotics data.

> **Note**
>
> This is not intended as a public API. To access topic data, prefer the `get_data` or `get_data_as_df` methods on [`Topic`](/reference/python-sdk/roboto/domain/topics/topic#roboto.domain.topics.topic.Topic), [`MessagePath`](/reference/python-sdk/roboto/domain/topics/message_path#roboto.domain.topics.message_path.MessagePath), or [`Event`](/reference/python-sdk/roboto/domain/events/event#roboto.domain.events.event.Event).

**Parameters**

- **roboto_client** (`roboto.http.RobotoClient`)
- **cache_dir** (`Union[str, pathlib.Path, None]`)

**Attributes**

- **TopicDataService.DEFAULT_CACHE_DIR** (`ClassVar[pathlib.Path]`)

#### TopicDataService.get_data()

```python
def get_data(
    topic_id: str,
    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_override: 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_data_service.py#L76-L131)

Retrieve data for a specific topic with optional filtering.

**Parameters**

- **topic_id** (`str`): Unique identifier of the topic to retrieve data for.
- **message_paths_include** (`Optional[collections.abc.Iterable[str]]`): Dot notation paths to include in the results. If None, all paths are included.
- **message_paths_exclude** (`Optional[collections.abc.Iterable[str]]`): Dot notation paths to exclude from the results. If None, no paths are excluded.
- **start_time** (`Optional[roboto.time.Time]`): Start time (inclusive) for temporal filtering.
- **end_time** (`Optional[roboto.time.Time]`): End time (exclusive) for temporal filtering.
- **cache_dir_override** (`Union[str, pathlib.Path, None]`): Override the default cache directory for downloads.
- **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).

**Yields**

- Tuple of (timestamp, record) where timestamp is in nanoseconds since Unix epoch.

**Returns**

- `collections.abc.Generator[tuple[roboto.domain.topics.topic_reader.Timestamp, dict[str, Any]], None, None]`

#### TopicDataService.get_data_as_df()

```python
def get_data_as_df(
    topic_id: str,
    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_override: 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_data_service.py#L133-L203)

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

**Parameters**

- **topic_id** (`str`): Unique identifier of the topic to retrieve data for.
- **message_paths_include** (`Optional[collections.abc.Iterable[str]]`): Dot notation paths to include in the results. If None, all paths are included.
- **message_paths_exclude** (`Optional[collections.abc.Iterable[str]]`): Dot notation paths to exclude from the results. If None, no paths are excluded.
- **start_time** (`Optional[roboto.time.Time]`): Start time (inclusive) for temporal filtering.
- **end_time** (`Optional[roboto.time.Time]`): End time (exclusive) for temporal filtering.
- **cache_dir_override** (`Union[str, pathlib.Path, None]`): Override the default cache directory for downloads.
- **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).

**Returns**

- `pandas.DataFrame`: pandas.DataFrame The index of the DataFrame (`df.index`) is a pandas.DateTimeIndex, labeling the timestamp of each row.

### iterable_to_set()

```python
def roboto.domain.topics.topic_data_service.iterable_to_set(
    arg: collections.abc.Iterable[str],
) -> set[str]
```

`from roboto.domain.topics.topic_data_service import iterable_to_set`

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

**Parameters**

- **arg** (`collections.abc.Iterable[str]`)

**Returns**

- `set[str]`

### logger

```python
roboto.domain.topics.topic_data_service.logger
```

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

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