roboto.domain.datasets

Clear a single custom-field value on this dataset to None.

Clear multiple custom-field values on this dataset to None.

Create a new dataset in the Roboto platform.

Creates a new dataset with the specified properties and returns a Dataset instance for interacting with it. The dataset will be created in the caller’s organization unless a different organization is specified.

Usage

dataset = Dataset.create(
    name="Highway Test Session",
    description="Autonomous vehicle highway driving test data",
    tags=["highway", "autonomous", "test"],
    metadata={"vehicle_id": "vehicle_001", "test_type": "highway"},
)
print(dataset.dataset_id)
# ds_abc123

# Create minimal dataset
dataset = Dataset.create()
print(f"Created dataset: {dataset.dataset_id}")

Create a directory within the dataset.

Usage

Create a simple directory:

from roboto.domain import datasets
dataset = datasets.Dataset.from_id(...)
directory = dataset.create_directory("foo")
print(directory.relative_path)
# foo

Create a directory with intermediate directories:

directory = dataset.create_directory(
    name="final",
    parent_path=pathlib.Path("path/to/deep"),
    create_intermediate_dirs=True,
)
print(directory.relative_path)
# path/to/deep/final

Create a dataset if no existing dataset matches the specified query.

Searches for existing datasets using the provided RoboQL query. If a matching dataset is found, returns that dataset. If no match is found, creates a new dataset with the specified properties and returns it.

Concurrent calls with the same match_roboql_query in one organization create one dataset between them: the service runs them one at a time from the search through the create. Calls with different queries are not serialized, even when both queries would match the same dataset.

The dataset created must match match_roboql_query. If name, tags or metadata describe a dataset the query does not match, every later call creates another one. When several datasets match, which one is returned is not defined unless the query ends with a SORT BY clause.

Usage

Create a dataset only if no dataset with specific metadata exists:

dataset = Dataset.create_if_not_exists(
    match_roboql_query="dataset.metadata.vehicle_id = 'vehicle_001'",
    name="Vehicle 001 Test Session",
    description="Test data for vehicle 001",
    metadata={"vehicle_id": "vehicle_001", "test_type": "highway"},
    tags=["vehicle_001", "highway"],
)
print(dataset.dataset_id)
# ds_abc123

Create a dataset only if no dataset with specific tags exists:

dataset = Dataset.create_if_not_exists(
    match_roboql_query="dataset.tags CONTAINS 'unique_session_id_xyz'",
    name="Unique Test Session",
    tags=["unique_session_id_xyz", "test"],
)
# If a dataset with tag 'unique_session_id_xyz' already exists,
# that dataset is returned instead of creating a new one

Create a Session populated with files from this dataset.

By default, every file in the dataset is added to the new Session. include_patterns / exclude_patterns narrow that set using the same gitignore-style syntax as Dataset.list_files().

Usage

dataset = Dataset.from_id("ds_abc123")
session = dataset.create_session("flight-2026-04-23-001")

Notes

Convenience wrapper around Session.create() followed by Session.add_files(). A dataset may hold more files than one add request accepts: the files are sent in consecutive requests of at most MAX_FILES_AND_TOPICS_PER_REQUEST each. Every file goes in or none does: if any add request fails, or the platform refuses any one file, the partially populated Session is deleted before the exception propagates, so the call is safe to retry. If that cleanup fails too (e.g. a transient network error), the Session is left behind holding whatever files did go in; the cleanup failure is logged, and the original exception is what the caller sees.

Delete this dataset from the Roboto platform.

Permanently removes the dataset and all its associated files, metadata, and topics. This operation cannot be undone.

If a dataset’s files are hosted in Roboto managed S3 buckets or customer read/write bring-your-own-buckets, the files in this dataset will be deleted from S3 as well. For files hosted in customer read-only buckets, the files will not be deleted from S3, but the dataset record and all associated metadata will be deleted.

Usage

dataset = Dataset.from_id("ds_abc123")
dataset.delete()
# # Dataset and all its files are now permanently deleted

Delete files from this dataset based on pattern matching.

Deletes files that match the specified include patterns while excluding those that match exclude patterns. Uses gitignore-style pattern matching for flexible file selection.

Notes

Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.

Usage

dataset = Dataset.from_id("ds_abc123")
# Delete all PNG files except those in back_camera directory
dataset.delete_files(include_patterns=["**/*.png"], exclude_patterns=["**/back_camera/**"])

# Delete all log files
dataset.delete_files(include_patterns=["**/*.log"])

Download files from this dataset to a local directory.

Downloads files that match the specified patterns to the given local directory. The directory structure from the dataset is preserved in the download location. If the output directory doesn’t exist, it will be created.

Notes

Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.

Usage

import pathlib
dataset = Dataset.from_id("ds_abc123")
downloaded = dataset.download_files(
    pathlib.Path("/tmp/dataset_download"),
    include_patterns=["**/*.bag"],
    exclude_patterns=["**/test/**"],
)
print(f"Downloaded {len(downloaded)} files")
# Downloaded 5 files

# Download all files
all_files = dataset.download_files(pathlib.Path("/tmp/all_files"))

Create a Dataset instance from a dataset ID.

Retrieves dataset information from the Roboto platform using the provided dataset ID and returns a Dataset instance for interacting with it.

Usage

dataset = Dataset.from_id("ds_abc123")
print(dataset.name)
# 'Highway Test Session'
print(len(list(dataset.list_files())))
# 42

Generate a new AI summary for this dataset.

Creates a new AI-generated summary that analyzes the dataset’s content, structure, and metadata. The summary generation is asynchronous and can be monitored through the returned StreamingAISummary object.

Usage

Generate a summary and wait for completion:

dataset = Dataset.from_id("ds_abc123")
summary = dataset.generate_summary()
complete_text = summary.complete_text
print(complete_text)
# 'This dataset contains 42 files with sensor data from highway driving tests...'

Generate a summary and stream the text as it’s generated:

dataset = Dataset.from_id("ds_abc123")
summary = dataset.generate_summary()
for text_chunk in summary.text_stream():
    print(text_chunk, end="", flush=True)

Check summary status without blocking:

dataset = Dataset.from_id("ds_abc123")
summary = dataset.generate_summary()
if summary.current and summary.current.status == AISummaryStatus.Complete:
    print("Summary is ready!")

Get a File instance for a file at the specified path in this dataset.

Retrieves a file by its relative path within the dataset. Optionally retrieves a specific version of the file.

Usage

dataset = Dataset.from_id("ds_abc123")
file = dataset.get_file_by_path("logs/session1.bag")
print(file.file_id)
# file_xyz789

# Get specific version
old_file = dataset.get_file_by_path("data/sensors.csv", version_id=1)
print(old_file.version)
# 1

Return custom metadata associated with this dataset.

Returns a copy of the dataset’s metadata dictionary containing arbitrary key-value pairs for storing custom information. Supports nested structures and dot notation for accessing nested fields.

Iterate over Sessions that include at least one file from this dataset.

A Session may draw files from one or more datasets; this method yields every Session whose current-version file list intersects this dataset.

Usage

dataset = Dataset.from_id("ds_abc123")
for session in dataset.get_sessions():
    print(session.session_id, session.name)

Retrieve this dataset’s existing AI summary.

Returns the dataset’s current AI summary if one exists. Reading never generates a summary as a side effect: a dataset that has never been summarized raises RobotoNotFoundException rather than implicitly kicking off — and paying for — generation. Call generate_summary() to create one explicitly.

Usage

Get the existing summary, generating one first if there is none:

from roboto.exceptions import RobotoNotFoundException
dataset = Dataset.from_id("ds_abc123")
try:
    summary = dataset.get_summary()
except RobotoNotFoundException:
    summary = dataset.generate_summary()
print(summary.complete_text)
# 'This dataset contains 42 files with sensor data from highway driving tests...'

Check whether a summary exists without generating one:

from roboto.exceptions import RobotoNotFoundException
dataset = Dataset.from_id("ds_abc123")
try:
    summary = dataset.get_summary()
    print(summary.complete_text)
except RobotoNotFoundException:
    print("No summary yet — call generate_summary() to create one.")

Get the earliest start and latest end across every topic in this dataset.

The same aggregate you would reach by folding start_time and end_time over get_topics(), computed server-side in one request instead of one per page of topics. Reach for it when you want the dataset’s time extent and not the topics themselves.

Usage

dataset = Dataset.from_id("ds_abc123")
bounds = dataset.get_topic_time_bounds()
print(bounds.start_time, bounds.end_time)
# 1722870127699468923 1722870187004821001

Get all topics associated with files in this dataset, with optional filtering.

Retrieves all topics that were extracted from files in this dataset during ingestion. If multiple files have topics with the same name (e.g., chunked files with the same schema), they are returned as separate topic objects.

Topics can be filtered by name using include/exclude patterns. Topics specified on both the inclusion and exclusion lists will be excluded.

Usage

dataset = Dataset.from_id("ds_abc123")
for topic in dataset.get_topics():
    print(f"Topic: {topic.name}")
# Topic: /camera/image
# Topic: /imu/data
# Topic: /gps/fix

# Only get camera topics
camera_topics = list(dataset.get_topics(include=["/camera/image", "/camera/info"]))
print(f"Found {len(camera_topics)} camera topics")

# Exclude diagnostic topics
data_topics = list(dataset.get_topics(exclude=["/diagnostics"]))

Get all topics associated with a specific file in this dataset.

Retrieves all topics that were extracted from the specified file during ingestion. This is a convenience method that combines file lookup and topic retrieval.

Usage

dataset = Dataset.from_id("ds_abc123")
for topic in dataset.get_topics_by_file("logs/session1.bag"):
    print(f"Topic: {topic.name}")
# Topic: /camera/image
# Topic: /imu/data
# Topic: /gps/fix

Yield every directory in this dataset, at any depth.

Usage

dataset = Dataset.from_id("ds_abc123")
for directory in dataset.list_directories():
    print(directory.relative_path)
# logs
# logs/session1

List files in this dataset with optional pattern-based filtering.

Returns all files in the dataset that match the specified include patterns while excluding those that match exclude patterns. Uses gitignore-style pattern matching for flexible file selection.

Notes

Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.

Usage

dataset = Dataset.from_id("ds_abc123")
for file in dataset.list_files():
    print(file.relative_path)
# logs/session1.bag
# data/sensors.csv
# images/camera_001.jpg

# List only image files, excluding back camera
for file in dataset.list_files(
    include_patterns=["**/*.png", "**/*.jpg"], exclude_patterns=["**/back_camera/**"]
):
    print(file.relative_path)
# images/front_camera_001.jpg
# images/side_camera_001.jpg

Add or update metadata fields for this dataset.

Sets each key-value pair in the provided dictionary as dataset metadata. If a key doesn’t exist, it will be created. If it exists, the value will be overwritten. Keys must be strings and dot notation is supported for nested keys.

Usage

dataset = Dataset.from_id("ds_abc123")
dataset.put_metadata(
    {
        "vehicle_id": "vehicle_001",
        "test_type": "highway_driving",
        "weather.condition": "sunny",
        "weather.temperature": 25,
    }
)
print(dataset.metadata["vehicle_id"])
# 'vehicle_001'
print(dataset.metadata["weather"]["condition"])
# 'sunny'

Add or update tags for this dataset.

Adds each tag in the provided sequence to the dataset. If a tag already exists, it will not be duplicated. This operation replaces the current tag list with the provided tags.

Usage

dataset = Dataset.from_id("ds_abc123")
dataset.put_tags(["highway", "autonomous", "test", "sunny"])
print(dataset.tags)
# ['highway', 'autonomous', 'test', 'sunny']

Query datasets using a specification with filters and pagination.

Searches for datasets matching the provided query specification. Results are returned as a generator that automatically handles pagination, yielding Dataset instances as they are retrieved from the API.

Usage

from roboto.query import Comparator, Condition, QuerySpecification
spec = QuerySpecification(
    condition=Condition(field="name", comparator=Comparator.Contains, value="Roboto")
)
for dataset in Dataset.query(spec):
    print(f"Found dataset: {dataset.name}")
# Found dataset: Roboto Test
# Found dataset: Other Roboto Test

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

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

Usage

dataset = Dataset.from_id("ds_abc123")
# Dataset may have been updated by another process
refreshed_dataset = dataset.refresh()
print(f"Current file count: {len(list(refreshed_dataset.list_files()))}")

Remove each key in this sequence from dataset metadata if it exists. Keys must be strings. Dot notation is supported for nested keys.

Usage

from roboto.domain import datasets
dataset = datasets.Dataset(...)
dataset.remove_metadata(["foo", "baz.qux"])

Remove each tag in this sequence if it exists

Rename or move a directory within this dataset.

Both old_path and new_path are relative to the dataset root. Pass a new_path with fewer path components to move the directory up the tree, or a different leaf name at the same depth to rename in place.

Usage

dataset = Dataset.from_id("ds_abc123")
dataset.rename_directory("logs/session1", "session1")

Rename or move a file within this dataset.

new_path is relative to the dataset root. Pass a path with fewer components to move the file up the tree, a different name at the same depth to rename in place, or a path under a different directory to move sideways.

The file’s storage URI is unchanged; only the logical location in the dataset hierarchy moves.

Usage

dataset = Dataset.from_id("ds_abc123")
record = dataset.rename_file("file_xyz789", "file.bag")
record.relative_path
# 'file.bag'

Set a single custom-field value on this dataset.

name must be the name of a Ready custom field for this dataset’s org and the Dataset entity type; value must satisfy the field’s declared type.

Set or overwrite multiple custom-field values on this dataset.

Each key must name a Ready custom field for this dataset’s org and the Dataset entity type; each value must satisfy the field’s declared type.

Set the device ID for this dataset.

Explicitly set the AI summary text for this dataset.

This method is intended to be used in cases where an action or other active component is able to generate a more specialized summary than Dataset::generate_summary would, and you want to make that summary canonical from the perspective of the UI and Dataset::get_summary.

Convert this dataset to a dictionary representation.

Returns the dataset’s data as a JSON-serializable dictionary containing all dataset attributes and metadata.

Usage

dataset = Dataset.from_id("ds_abc123")
dataset_dict = dataset.to_dict()
print(dataset_dict["name"])
# 'Highway Test Session'
print(dataset_dict["metadata"])
# {'vehicle_id': 'vehicle_001', 'test_type': 'highway'}

Update this dataset’s properties.

Updates various properties of the dataset including name, description, and metadata. Only specified parameters are updated; others remain unchanged.

Usage

dataset = Dataset.from_id("ds_abc123")
updated_dataset = dataset.update(
    name="Updated Highway Test Session", description="Updated description with more details"
)
print(updated_dataset.name)
# 'Updated Highway Test Session'

# Update with metadata changes
from roboto.updates import MetadataChangeset
changeset = MetadataChangeset(put_fields={"processed": True})
updated_dataset = dataset.update(metadata_changeset=changeset)

# Clear the device association
updated_dataset = dataset.update(device_id=None)

# Clear the description
updated_dataset = dataset.update(description=None)

Uploads all files and directories recursively from the specified directory path. You can use include_patterns and exclude_patterns to control what files and directories are uploaded, and can use delete_after_upload to clean up your local filesystem after the uploads succeed.

Usage

from roboto import Dataset
dataset = Dataset(...)
dataset.upload_directory(
    pathlib.Path("/path/to/directory"),
    exclude_patterns=[
        "__pycache__/",
        "*.pyc",
        "node_modules/",
        "**/*.log",
    ],
)

Notes

• Both include_patterns and exclude_patterns follow the ‘gitignore’ pattern format described in https://git-scm.com/docs/gitignore#_pattern_format.
• If both include_patterns and exclude_patterns are provided, files matching exclude_patterns will be excluded even if they match include_patterns.

Upload a single file to the dataset. If file_destination_path is not provided, the file will be uploaded to the top-level of the dataset.

Usage

from roboto.domain import datasets
dataset = datasets.Dataset(...)
dataset.upload_file(
    pathlib.Path("/path/to/file.txt"),
    file_destination_path="foo/bar.txt",
)

Upload multiple files to the dataset.

If file_destination_paths is not provided, files will be uploaded to the top-level of the dataset.

Usage

import pathlib
from roboto.domain import datasets
dataset = datasets.Dataset.from_id("ds_abc123")
file_ids = dataset.upload_files(
    [pathlib.Path("/path/to/file.txt")],
    file_destination_paths={
        pathlib.Path("/path/to/file.txt"): "foo/bar.txt",
    },
)
file_ids[pathlib.Path("/path/to/file.txt")]
# 'fl_0123456789abcdef'