roboto #

roboto.domain.files.DirectoryRecord

Examples

>>> 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_directory(name, error_if_exists=False, create_intermediate_dirs=False, parent_path=None, origination=None)#

Create a directory within the dataset.

Parameters:

name (str) – Name of the directory to create.
error_if_exists (bool) – If True, raises an exception if the directory already exists.
parent_path (Optional[pathlib.Path]) – Path of the parent directory. If None, creates the directory in the root of the dataset.
origination (Optional[str]) – Optional string describing the source or context of the directory creation.
create_intermediate_dirs (bool) – If True, creates intermediate directories in the path if they don’t exist. If False, requires all parent directories to already exist.

Raises:

RobotoConflictException – If the directory already exists and error_if_exists is True.
RobotoUnauthorizedException – If the caller lacks permission to create the directory.
RobotoInvalidRequestException – If the directory name is invalid or the parent path does not exist (when create_intermediate_dirs is False).

Returns:

DirectoryRecord of the created directory.

Return type:

Examples

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="path/to/deep",
...     create_intermediate_dirs=True
... )
>>> print(directory.relative_path)
path/to/deep/final

classmethod create_if_not_exists(match_roboql_query, description=None, metadata=None, name=None, tags=None, caller_org_id=None, roboto_client=None)#

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.

Parameters:

match_roboql_query (str) – RoboQL query string to search for existing datasets. If this query matches any dataset, that dataset will be returned instead of creating a new one.
description (Optional[str]) – Optional human-readable description of the dataset.
metadata (Optional[dict[str, Any]]) – Optional key-value metadata pairs to associate with the dataset.
name (Optional[str]) – Optional short name for the dataset (max 120 characters).
tags (Optional[list[str]]) – Optional list of tags for dataset discovery and organization.
caller_org_id (Optional[str]) – Organization ID to create the dataset in. Required for multi-org users.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Dataset instance representing either the existing matched dataset or the newly created dataset.

Raises:

RobotoInvalidRequestException – Invalid dataset parameters or malformed RoboQL query.
RobotoUnauthorizedException – Caller lacks permission to create datasets or search existing ones.

Return type:

Examples

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

property created: datetime.datetime#

Timestamp when this dataset was created.

Returns the UTC datetime when this dataset was first created in the Roboto platform. This property is immutable.

Return type:: datetime.datetime

property created_by: str#

Identifier of the user who created this dataset.

Returns the identifier of the person or service which originally created this dataset in the Roboto platform.

Return type:: str

property dataset_id: str#

Unique identifier for this dataset.

Returns the globally unique identifier assigned to this dataset when it was created. This ID is immutable and used to reference the dataset across the Roboto platform. It is always prefixed with ‘ds_’ to distinguish it from other Roboto resource IDs.

Return type:: str

delete()#

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.

Raises:

RobotoNotFoundException – Dataset does not exist or has already been deleted.
RobotoUnauthorizedException – Caller lacks permission to delete the dataset.

Return type:

None

Examples

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

delete_files(include_patterns=None, exclude_patterns=None)#

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.

Parameters:

include_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to include. If None, all files are considered for deletion.
exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude from deletion. Takes precedence over include patterns.

Raises:

RobotoUnauthorizedException – Caller lacks permission to delete files.

Return type:

None

Notes

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

Examples

>>> 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"])

property description: str | None#

Human-readable description of this dataset.

Returns the optional description text that provides details about the dataset’s contents, purpose, or context. Can be None if no description was provided.

Return type:: Optional[str]

download_files(out_path, include_patterns=None, exclude_patterns=None)#

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.

Parameters:

out_path (pathlib.Path) – Local directory path where files should be downloaded.
include_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to include. If None, all files are downloaded.
exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude from download. Takes precedence over include patterns.

Returns:

List of tuples containing (FileRecord, local_path) for each downloaded file.

Raises:

RobotoUnauthorizedException – Caller lacks permission to download files.

Return type:

list[tuple[roboto.domain.files.FileRecord, pathlib.Path]]

Notes

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

Examples

>>> 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"))

classmethod from_id(dataset_id, roboto_client=None)#

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.

Parameters:

dataset_id (str) – Unique identifier for the dataset.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Dataset instance representing the requested dataset.

Raises:

RobotoNotFoundException – Dataset with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the dataset.

Return type:

Examples

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

generate_summary()#

Generate a new AI-powered summary of this dataset.

Creates a new AI-generated summary that analyzes the dataset’s contents, including files, metadata, and topics. If a summary already exists, it will be overwritten. The results are persisted and can be retrieved later with get_summary().

Returns:

AISummary object containing the generated summary text and creation timestamp.

Raises:

RobotoUnauthorizedException – Caller lacks permission to generate summaries.
RobotoInvalidRequestException – Dataset is not suitable for summarization.

Return type:

Examples

>>> dataset = Dataset.from_id("ds_abc123")
>>> summary = dataset.generate_summary()
>>> print(summary.text)
This dataset contains autonomous vehicle sensor data from a highway
driving session, including camera images, LiDAR point clouds, and
GPS coordinates collected over a 30-minute period.
>>> print(summary.created)
2024-01-15 10:30:00+00:00

get_file_by_path(relative_path, version_id=None)#

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.

Parameters:

relative_path (Union[str, pathlib.Path]) – Path of the file relative to the dataset root.
version_id (Optional[int]) – Specific version of the file to retrieve. If None, gets the latest version.

Returns:

File instance representing the file at the specified path.

Raises:

RobotoNotFoundException – File at the given path does not exist in the dataset.
RobotoUnauthorizedException – Caller lacks permission to access the file.

Return type:

roboto.domain.files.File

Examples

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

get_summary()#

Get the latest AI-generated summary of this dataset.

Retrieves the most recent AI-generated summary for this dataset. If no summary exists, one will be automatically generated (equivalent to calling generate_summary()).

Once a summary is generated, it persists and is returned by this method until generate_summary() is explicitly called again. The summary does not automatically update when the dataset or its files change.

Returns:: AISummary object containing the summary text and creation timestamp.
Raises:: RobotoUnauthorizedException – Caller lacks permission to access summaries.
Return type:: roboto.ai.summary.AISummary

Examples

>>> dataset = Dataset.from_id("ds_abc123")
>>> summary = dataset.get_summary()
>>> print(summary.text)
This dataset contains autonomous vehicle sensor data from a highway
driving session, including camera images, LiDAR point clouds, and
GPS coordinates collected over a 30-minute period.

>>> # Summary is cached - subsequent calls return the same summary
>>> cached_summary = dataset.get_summary()
>>> assert summary.created == cached_summary.created

get_summary_sync(timeout=60, poll_interval=2)#

Poll the summary endpoint until a summary’s status is COMPLETED, or raise an exception if the status is FAILED or the configurable timeout is reached.

This method will call get_summary() repeatedly until the summary reaches a terminal status. If no summary exists when this method is called, one will be generated automatically.

Parameters:

timeout (float) – The maximum amount of time, in seconds, to wait for the summary to complete. Defaults to 1 minute.
poll_interval (roboto.waiters.Interval) – The amount of time, in seconds, to wait between polling iterations. Defaults to 2 seconds.

Return type:

Returns: An AI Summary object containing a full LLM summary of the dataset.

Raises:

RobotoFailedToGenerateException – If the summary status becomes FAILED.
TimeoutError – If the timeout is reached before the summary completes.

Parameters:

timeout (float)
poll_interval (roboto.waiters.Interval)

Return type:

Example

>>> from roboto import Dataset
>>> dataset = Dataset.from_id("ds_abc123")
>>> summary = dataset.get_summary_sync(timeout=60)
>>> print(summary.text)
This dataset contains ...

get_topics(include=None, exclude=None)#

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.

Parameters:

include (Optional[collections.abc.Sequence[str]]) – If provided, only topics with names in this sequence are yielded.
exclude (Optional[collections.abc.Sequence[str]]) – If provided, topics with names in this sequence are skipped. Takes precedence over include list.

Yields:

Topic instances associated with files in this dataset, filtered according to the parameters.

Return type:

collections.abc.Generator[roboto.domain.topics.Topic, None, None]

Examples

>>> 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_topics_by_file(relative_path)#

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.

Parameters:

relative_path (Union[str, pathlib.Path]) – Path of the file relative to the dataset root.

Yields:

Topic instances associated with the specified file.

Raises:

RobotoNotFoundException – File at the given path does not exist in the dataset.
RobotoUnauthorizedException – Caller lacks permission to access the file or its topics.

Return type:

collections.abc.Generator[roboto.domain.topics.Topic, None, None]

Examples

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

list_directories()#

Return type:: collections.abc.Generator[roboto.domain.files.DirectoryRecord, None, None]

list_files(include_patterns=None, exclude_patterns=None)#

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.

Parameters:

include_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to include. If None, all files are considered.
exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude. Takes precedence over include patterns.

Yields:

File instances that match the specified patterns.

Raises:

RobotoUnauthorizedException – Caller lacks permission to list files.

Return type:

collections.abc.Generator[roboto.domain.files.File, None, None]

Notes

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

Examples

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

property metadata: dict[str, Any]#

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.

Return type:: dict[str, Any]

property modified: datetime.datetime#

Timestamp when this dataset was last modified.

Returns the UTC datetime when this dataset was most recently updated. This includes changes to metadata, tags, description, or other properties.

Return type:: datetime.datetime

property modified_by: str#

Identifier of the user or service which last modified this dataset.

Returns the identifier of the person or service which most recently updated this dataset’s metadata, tags, description, or other properties.

Return type:: str

property name: str | None#

Human-readable name of this dataset.

Returns the optional display name for this dataset. Can be None if no name was provided during creation. For users whose organizations have their own idiomatic internal dataset IDs, it’s recommended to set the name to the organization’s internal dataset ID, since the Roboto dataset_id property is randomly generated.

Return type:: Optional[str]

property org_id: str#

Organization identifier that owns this dataset.

Returns the unique identifier of the organization that owns and has primary access control over this dataset.

Return type:: str

put_metadata(metadata)#

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.

Parameters:: metadata (dict[str, Any]) – Dictionary of metadata key-value pairs to add or update.
Raises:: RobotoUnauthorizedException – Caller lacks permission to update the dataset.
Return type:: None

Examples

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

put_tags(tags)#

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.

Parameters:: tags (roboto.updates.StrSequence) – Sequence of tag strings to set on the dataset.
Raises:: RobotoUnauthorizedException – Caller lacks permission to update the dataset.
Return type:: None

Examples

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

classmethod query(spec=None, roboto_client=None, owner_org_id=None)#

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.

Parameters:

spec (Optional[roboto.query.QuerySpecification]) – Query specification with filters, sorting, and pagination options. If None, returns all accessible datasets.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.
owner_org_id (Optional[str]) – Organization ID to scope the query. If None, uses caller’s org.

Yields:

Dataset instances matching the query specification.

Raises:

ValueError – Query specification references unknown dataset attributes.
RobotoUnauthorizedException – Caller lacks permission to query datasets.

Return type:

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

Examples

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

property record: roboto.domain.datasets.record.DatasetRecord#

Underlying data record for this dataset.

Returns the raw DatasetRecord that contains all the dataset’s data fields. This provides access to the complete dataset state as stored in the platform.

Return type:: roboto.domain.datasets.record.DatasetRecord

refresh()#

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.

Returns:

This Dataset instance with refreshed data.

Raises:

RobotoNotFoundException – Dataset no longer exists.
RobotoUnauthorizedException – Caller lacks permission to access the dataset.

Return type:

roboto.domain.files.DirectoryRecord

Examples

>>> 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_metadata(metadata)#

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

Example

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

Parameters:: metadata (roboto.updates.StrSequence)
Return type:: None

remove_tags(tags)#

Remove each tag in this sequence if it exists

Parameters:: tags (roboto.updates.StrSequence)
Return type:: None

rename_directory(old_path, new_path)#

Parameters:

old_path (str)
new_path (str)

Return type:

property tags: list[str]#

List of tags associated with this dataset.

Returns a copy of the list of string tags that have been applied to this dataset for categorization and filtering purposes.

Return type:: list[str]

to_association()#

Return type:: roboto.association.Association

to_dict()#

Convert this dataset to a dictionary representation.

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

Returns:: Dictionary representation of the dataset data.
Return type:: dict[str, Any]

Examples

>>> 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(conditions=None, description=None, metadata_changeset=None, name=None)#

Update this dataset’s properties.

Updates various properties of the dataset including name, description, and metadata. Only specified parameters are updated; others remain unchanged. Optionally supports conditional updates based on current field values.

Parameters:

conditions (Optional[list[roboto.updates.UpdateCondition]]) – Optional list of conditions that must be met for the update to proceed.
description (Optional[str]) – New description for the dataset.
metadata_changeset (Optional[roboto.updates.MetadataChangeset]) – Metadata changes to apply (add, update, or remove fields/tags).
name (Optional[str]) – New name for the dataset.

Returns:

Updated Dataset instance with the new properties.

Raises:

RobotoUnauthorizedException – Caller lacks permission to update the dataset.
RobotoConditionalUpdateFailedException – Update conditions were not met.

Return type:

Examples

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

upload_directory(directory_path, include_patterns=None, exclude_patterns=None, delete_after_upload=False, max_batch_size=MAX_FILES_PER_MANIFEST, print_progress=True)#

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.

Example

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

Parameters:

directory_path (pathlib.Path)
include_patterns (Optional[list[str]])
exclude_patterns (Optional[list[str]])
delete_after_upload (bool)
max_batch_size (int)
print_progress (bool)

Return type:

None

upload_file(file_path, file_destination_path=None, print_progress=True)#

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.

Example

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

Parameters:

file_path (pathlib.Path)
file_destination_path (Optional[str])
print_progress (bool)

Return type:

None

upload_files(files, file_destination_paths={}, max_batch_size=MAX_FILES_PER_MANIFEST, print_progress=True)#

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

Example

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

Parameters:

files (collections.abc.Iterable[pathlib.Path])
file_destination_paths (collections.abc.Mapping[pathlib.Path, str])
max_batch_size (int)
print_progress (bool)

class roboto.DatasetCredentials(/, **data)#

Bases: pydantic.BaseModel

Handle credentials for dataset file access

Parameters:: data (Any)

access_key_id: str#

bucket: str#

expiration: datetime.datetime#

is_expired()#

Return type:: bool

region: str#

required_prefix: str#

secret_access_key: str#

session_token: str#

to_dict()#

Return type:: dict[str, Any]

to_s3_credentials()#

Return type:: S3Credentials

class roboto.DatasetRecord(/, **data)#

Bases: pydantic.BaseModel

Wire-transmissible representation of a dataset in the Roboto platform.

DatasetRecord contains all the metadata and properties associated with a dataset, including its identification, timestamps, metadata, tags, and organizational information. This is the data structure used for API communication and persistence.

DatasetRecord instances are typically created by the platform during dataset creation operations and are updated as datasets are modified. The Dataset domain class wraps DatasetRecord to provide a more convenient interface for dataset operations.

The record includes audit information (created/modified timestamps and users), organizational context, and user-defined metadata and tags for discovery and organization purposes.

Parameters:: data (Any)

administrator: str = 'Roboto'#: Deprecated field maintained for backwards compatibility. Always defaults to ‘Roboto’.

created: datetime.datetime#: Timestamp when this dataset was created in the Roboto platform.

created_by: str#: User ID or service account that created this dataset.

dataset_id: str#: Unique identifier for this dataset within the Roboto platform.

description: str | None = None#: Human-readable description of the dataset’s contents and purpose.

device_id: str | None = None#: Optional identifier of the device that generated this dataset’s data.

metadata: dict[str, Any] = None#: User-defined key-value pairs for storing additional dataset information.

modified: datetime.datetime#: Timestamp when this dataset was last modified.

modified_by: str#: User ID or service account that last modified this dataset.

name: str | None = None#: A short name for this dataset. This may be an org-specific unique ID that’s more meaningful than the dataset_id, or a short summary of the dataset’s contents. If provided, must be 120 characters or less.

org_id: str#: Organization ID that owns this dataset.

roboto_record_version: int = 0#: Internal version number for this record, automatically incremented on updates.

storage_ctx: dict[str, Any] = None#: Deprecated storage context field maintained for backwards compatibility with SDK versions prior to 0.10.0.

storage_location: str = 'S3'#: Deprecated storage location field maintained for backwards compatibility. Always defaults to ‘S3’.

tags: list[str] = None#: List of tags for categorizing and discovering this dataset.

class roboto.DeleteFileRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for deleting a file from the platform.

This request is used internally by the platform to delete files and their associated data. The file is identified by its storage URI.

Parameters:: data (Any)

uri: str#

//bucket/path/to/file.bag’).

Type:: Storage URI of the file to delete (e.g., ‘s3

class roboto.DeleteMessagePathRequest(/, **data)#

Bases: pydantic.BaseModel

Request to delete a message path from a topic.

Removes a message path from a topic’s schema. This operation cannot be undone and will remove all associated data and metadata for the specified path.

Parameters:: data (Any)

message_path: str#: Message path name.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.Device(record, roboto_client=None)#

A device is a non-human entity which can interact with Roboto on behalf of a specific organization. Each device is identified by a device_id, which is unique among all devices in the organization to which it belongs.

The most typical device is a robot which uploads its log data to Roboto, either directly from its on-board software stack, or indirectly through an automatic upload station or human-in-the-loop upload process. Its device ID may be a string that represents its serial number in a scheme native to its organization.

A dedicated uploader station which connects to many different robots and uploads data on their behalf could also be modeled as a device.

API access tokens can be allocated for devices, and these tokens can be used to authenticate Roboto requests made on behalf of a device.

Parameters:

record (roboto.domain.devices.record.DeviceRecord)
roboto_client (Optional[roboto.http.RobotoClient])

classmethod create(device_id, caller_org_id=None, roboto_client=None)#

Registers a device with Roboto.

Parameters:

device_id (str)
caller_org_id (Optional[str])
roboto_client (Optional[roboto.http.RobotoClient])

Return type:

Device

create_token(expiry_days=366, name=None, description=None)#

Parameters:

expiry_days (int)
name (Optional[str])
description (Optional[str])

Return type:

tuple[roboto.domain.tokens.Token, str]

property created: datetime.datetime#

Return type:: datetime.datetime

property created_by: str#

Return type:: str

delete()#

Deletes this device.

Return type:: None

property device_id: str#

This device’s ID. Device ID is a user-provided identifier for a device, which is unique within the device’s org.

Return type:: str

classmethod for_org(org_id, roboto_client=None)#

List all devices registered for a given org.

Parameters:

org_id (str) – The org to list devices for
roboto_client (Optional[roboto.http.RobotoClient]) – Common parameters required to construct any Device object

Returns:

A generator of Device objects. For orgs with a large number of devices, this may involve multiple service calls, and the generator will yield results as they become available.

Return type:

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

classmethod from_id(device_id, roboto_client=None, org_id=None)#

Parameters:

device_id (str) – The device ID to look up. See Device.device_id() for more details.
roboto_client (Optional[roboto.http.RobotoClient]) – Common parameters required to construct any Device object
org_id (Optional[str]) – The org to which the device belongs. If not specified by a caller who only belongs to one org, will default to the org_id of that org. If not specified by a caller who belongs to multiple orgs, will raise an exception.

Returns:

A Device object representing the specified device

Raises:

RobotoNotFoundException – If the specified device is not registered with Roboto

Return type:

Device

property modified: datetime.datetime#

Return type:: datetime.datetime

property modified_by: str#

Return type:: str

property org_id: str#

The ID of the org to which this device belongs.

Return type:: str

property record: roboto.domain.devices.record.DeviceRecord#

The underlying DeviceRecord object which represents this device. This is often used as the wire representation of a device during API requests, and is subject to evolve over time. You should not program against this if avoidable.

Return type:: roboto.domain.devices.record.DeviceRecord

tokens()#

Return type:: collections.abc.Sequence[roboto.domain.tokens.Token]

class roboto.DeviceRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of a device.

Parameters:: data (Any)

created: datetime.datetime = None#

created_by: str = None#

device_id: str = None#

modified: datetime.datetime = None#

modified_by: str = None#

org_id: str = None#

class roboto.EvaluateTriggersRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to manually evaluate specific triggers.

Used to force evaluation of triggers outside of their normal automatic evaluation cycle. This is typically used for testing or debugging trigger behavior.

Note

This is primarily used internally by the Roboto platform and is not commonly needed by SDK users.

Parameters:: data (Any)

trigger_evaluation_ids: collections.abc.Iterable[int]#: Collection of trigger evaluation IDs to process.

class roboto.Event(record, roboto_client=None)#

Represents an event within the Roboto platform.

An event is a time-anchored annotation that relates Roboto entities (datasets, files, topics, and message paths) to specific time periods. Events enable temporal analysis, data correlation, and annotation of activities across different data sources.

Events serve as temporal markers that can:

Annotate specific time periods in your data
Associate multiple entities (datasets, files, topics, message paths) with time ranges
Enable time-based data retrieval and analysis
Support metadata and tagging for organization and search
Provide visual markers in timeline views and analysis tools

Events can represent instantaneous moments (point in time) or time ranges. They are particularly useful for marking significant occurrences like sensor anomalies, system events, behavioral patterns, or any other time-based phenomena in your data.

Events cannot be instantiated directly through the constructor. Use the class methods Event.create() to create new events or Event.from_id() to load existing events.

Parameters:

record (roboto.domain.events.record.EventRecord)
roboto_client (Optional[roboto.http.RobotoClient])

property color: str | None#

Display color for the event, if set.

Return type:: Optional[str]

classmethod create(name, start_time, end_time=None, associations=None, dataset_ids=None, file_ids=None, topic_ids=None, message_path_ids=None, description=None, metadata=None, tags=None, display_options=None, caller_org_id=None, roboto_client=None)#

Create a new event associated with at least one dataset, file, topic, or message path.

Creates a time-anchored event that can be associated with various Roboto entities. For instantaneous events (a point in time), only start_time is required. Otherwise, both start_time and end_time should be provided. These fields accept nanoseconds since the UNIX epoch, or any other compatible representations supported by to_epoch_nanoseconds().

Events must be associated with at least one entity. While associations, file_ids, topic_ids, dataset_ids and message_path_ids are all optional, at least one of them must contain a valid association for the event.

Parameters:

name (str) – Human-readable name for the event. Required.
start_time (roboto.time.Time) – Start timestamp of the event as nanoseconds since UNIX epoch, or any value convertible by to_epoch_nanoseconds().
end_time (Optional[roboto.time.Time]) – End timestamp of the event. If not provided, defaults to start_time for instantaneous events.
associations (Optional[collections.abc.Collection[roboto.association.Association]]) – Collection of Association objects linking the event to specific entities.
dataset_ids (Optional[collections.abc.Collection[str]]) – Dataset IDs to associate the event with.
file_ids (Optional[collections.abc.Collection[str]]) – File IDs to associate the event with.
topic_ids (Optional[collections.abc.Collection[str]]) – Topic IDs to associate the event with.
message_path_ids (Optional[collections.abc.Collection[str]]) – Message path IDs to associate the event with.
description (Optional[str]) – Optional human-readable description of the event.
metadata (Optional[dict[str, Any]]) – Key-value metadata for discovery and search.
tags (Optional[list[str]]) – Tags for categorizing and searching the event.
display_options (Optional[roboto.domain.events.operations.EventDisplayOptions]) – Visual display options such as color.
caller_org_id (Optional[str]) – Organization ID of the SDK caller. If not provided, uses the caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Event instance with the provided attributes and associations.

Raises:

RobotoInvalidRequestException – Invalid parameters (e.g., start_time > end_time), or associations point to non-existent resources.
RobotoUnauthorizedException – Caller lacks permission to access associated entities.

Return type:

Event

Examples

Create an event for a sensor anomaly on a specific topic:

>>> from roboto.domain.events import Event
>>> event = Event.create(
...     name="Temperature Spike",
...     start_time=1722870127699468923,
...     end_time=1722870127799468923,
...     description="Unusual temperature readings detected",
...     topic_ids=["tp_abc123"],
...     tags=["anomaly", "temperature"],
...     metadata={"severity": "high", "sensor_id": "temp_01"}
... )

Create an instantaneous event on a file:

>>> event = Event.create(
...     name="System Boot",
...     start_time="1722870127.699468923",  # String format also supported
...     file_ids=["fl_xyz789"],
...     tags=["system", "boot"]
... )

Create an event with display options:

>>> from roboto.domain.events import EventDisplayOptions
>>> event = Event.create(
...     name="Critical Alert",
...     start_time=1722870127699468923,
...     end_time=1722870127799468923,
...     dataset_ids=["ds_abc123"],
...     display_options=EventDisplayOptions(color="red"),
...     metadata={"alert_type": "critical", "component": "engine"}
... )

property created: datetime.datetime#

Date and time when this event was created.

Return type:: datetime.datetime

property created_by: str#

User who created this event.

Return type:: str

dataset_ids(strict_associations=False)#

Get dataset IDs associated with this event.

Parameters:: strict_associations (bool) – If True, only return datasets with direct associations. If False (default), also return datasets inferred from file and topic associations.
Returns:: List of unique dataset IDs associated with this event.
Return type:: list[str]

Examples

Get all associated dataset IDs:

>>> event = Event.from_id("ev_abc123")
>>> dataset_ids = event.dataset_ids()
>>> print(f"Associated with {len(dataset_ids)} datasets")

Get only directly associated datasets:

>>> strict_dataset_ids = event.dataset_ids(strict_associations=True)
>>> print(f"Directly associated with {len(strict_dataset_ids)} datasets")

delete()#

Delete this event permanently.

This operation cannot be undone. The event and all its associations will be permanently removed from the platform.

Raises:

RobotoUnauthorizedException – Caller lacks permission to delete this event.
RobotoNotFoundException – Event has already been deleted or does not exist.

Return type:

None

Examples

Delete an event:

>>> event = Event.from_id("ev_abc123")
>>> event.delete()
>>> # Event is now permanently deleted

Conditional deletion:

>>> event = Event.from_id("ev_abc123")
>>> if "temporary" in event.tags:
...     event.delete()
...     print("Temporary event deleted")

property description: str | None#

Optional human-readable description of the event.

Return type:: Optional[str]

property display_options: roboto.domain.events.operations.EventDisplayOptions | None#

Display options for the event, such as color.

Return type:: Optional[roboto.domain.events.operations.EventDisplayOptions]

property end_time: int#

End time of the event in nanoseconds since UNIX epoch.

Return type:: int

property event_id: str#

Unique identifier for this event.

Return type:: str

file_ids(strict_associations=False)#

Get file IDs associated with this event.

Parameters:: strict_associations (bool) – If True, only return files with direct associations. If False (default), also return files inferred from topic and message path associations.
Returns:: List of unique file IDs associated with this event.
Return type:: list[str]

Examples

Get all associated file IDs:

>>> event = Event.from_id("ev_abc123")
>>> file_ids = event.file_ids()
>>> print(f"Associated with {len(file_ids)} files")

Get only directly associated files:

>>> strict_file_ids = event.file_ids(strict_associations=True)
>>> for file_id in strict_file_ids:
...     print(f"Directly associated file: {file_id}")

classmethod from_id(event_id, roboto_client=None)#

Load an existing event by its ID.

Parameters:

event_id (str) – Unique identifier of the event to retrieve.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

Event instance for the specified ID.

Raises:

RobotoNotFoundException – Event with the specified ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the event.

Return type:

Event

Examples

Load an event by ID:

>>> event = Event.from_id("ev_abc123")
>>> print(f"Event: {event.name}")
>>> print(f"Created: {event.created}")

Load and update an event:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.set_description("Updated description")
>>> print(f"New description: {updated_event.description}")

classmethod get_by_associations(associations, roboto_client=None)#

Retrieve all events associated with the provided associations.

Returns events that match any of the provided associations. Events that you don’t have access to will be filtered out of the response rather than raising an exception.

Parameters:

associations (collections.abc.Collection[roboto.association.Association]) – Collection of Association objects to query events for.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Event instances associated with any of the specified associations.

Return type:

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

Examples

Query events for multiple associations:

>>> from roboto import Association
>>> associations = [
...     Association.topic("tp_abc123"),
...     Association.file("fl_xyz789")
... ]
>>> events = list(Event.get_by_associations(associations))
>>> for event in events:
...     print(f"Event: {event.name}")

Query events for a specific dataset and file combination:

>>> associations = [
...     Association.dataset("ds_abc123"),
...     Association.file("fl_xyz789")
... ]
>>> events = list(Event.get_by_associations(associations))

classmethod get_by_dataset(dataset_id, roboto_client=None, strict_associations=False)#

Retrieve all events associated with a specific dataset.

Returns events that are associated with the given dataset. By default, this includes events associated with the dataset itself, as well as events associated with any files or topics within that dataset. Use strict_associations=True to only return events with direct dataset associations.

Parameters:

dataset_id (str) – ID of the dataset to query events for.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.
strict_associations (bool) – If True, only return events with direct dataset associations. If False (default), also return events associated with files or topics within the dataset.

Yields:

Event instances associated with the specified dataset.

Return type:

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

Examples

Get all events for a dataset (including file and topic events):

>>> events = list(Event.get_by_dataset("ds_abc123"))
>>> for event in events:
...     print(f"Event: {event.name} at {event.start_time}")

Get only events directly associated with the dataset:

>>> strict_events = list(Event.get_by_dataset("ds_abc123", strict_associations=True))
>>> print(f"Found {len(strict_events)} dataset-level events")

Process events in batches:

>>> for event in Event.get_by_dataset("ds_abc123"):
...     if "anomaly" in event.tags:
...         print(f"Anomaly event: {event.name}")

classmethod get_by_file(file_id, roboto_client=None)#

Retrieve all events with a direct association to a specific file.

Parameters:

file_id (str) – ID of the file to query events for.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Event instances directly associated with the specified file.

Return type:

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

Examples

Get all events for a specific file:

>>> events = list(Event.get_by_file("fl_xyz789"))
>>> for event in events:
...     print(f"File event: {event.name}")

Check if a file has any events:

>>> file_events = list(Event.get_by_file("fl_xyz789"))
>>> if file_events:
...     print(f"File has {len(file_events)} events")
... else:
...     print("No events found for this file")

classmethod get_by_message_path(message_path_id, roboto_client=None)#

Retrieve all events with a direct association to a specific message path.

Parameters:

message_path_id (str) – ID of the message path to query events for.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Event instances directly associated with the specified message path.

Return type:

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

Examples

Get all events for a specific message path:

>>> events = list(Event.get_by_message_path("mp_abc123"))
>>> for event in events:
...     print(f"Message path event: {event.name}")

Find events within a time range for a message path:

>>> events = Event.get_by_message_path("mp_abc123")
>>> filtered_events = [
...     event for event in events
...     if event.start_time >= 1722870127699468923
... ]

classmethod get_by_topic(topic_id, roboto_client=None)#

Retrieve all events with a direct association to a specific topic.

Parameters:

topic_id (str) – ID of the topic to query events for.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Yields:

Event instances directly associated with the specified topic.

Return type:

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

Examples

Get all events for a specific topic:

>>> events = list(Event.get_by_topic("tp_abc123"))
>>> for event in events:
...     print(f"Topic event: {event.name}")

Analyze event patterns for a topic:

>>> events = list(Event.get_by_topic("tp_abc123"))
>>> anomaly_events = [e for e in events if "anomaly" in e.tags]
>>> print(f"Found {len(anomaly_events)} anomaly events")

get_data(message_paths_include=None, message_paths_exclude=None, topic_name=None, topic_data_service=None, cache_dir=None, strict_associations=False)#

Iteratively yield records of the underlying topic data this event annotates.

An event can be associated with data at multiple resolutions:

as an event on its containing dataset, file, and/or topic,
but also directly with the message path (“signal data”)

A single event can also span signals that share a timeline, so it may annotate multiple topics in a file, or even multiple files in a dataset.

For now, getting the underlying signal data associated with an event only works for events that can be sourced to a single topic (extracted from one file, uploaded to one dataset). This means that the event must have been made on either a single file, a single topic, or one or many message paths within that topic.

If the event was made on a file, topic_name must be provided, and either or both of message_paths_include or message_paths_exclude may be provided, but are optional.

If the event was made on a topic, either or both of message_paths_include or message_paths_exclude may be provided, but are optional. topic_name, if provided in this instance, is ignored.

If the event was made on one or many message paths, each of those message paths must be found in the same topic. topic_name, message_paths_include, and message_paths_exclude, if provided in this instance, are ignored.

If the event is associated with data at multiple resolutions (e.g., two message paths, one topic, one file), this method will consider the lowest resolution associations first (message path), then topic, then file.

If message_paths_include or message_paths_exclude are defined, they should be dot notation paths that match attributes of individual data records. If a partial path is provided, it is treated as a wildcard, matching all subpaths.

For example, given topic data with the following interface:

{
    "velocity": {
        "x": <uint32>,
        "y": <uint32>,
        "z": <uint32>
    }
}

Calling get_data on an Event associated with that topic like:

>>> event.get_data(message_paths_include=["velocity.x", "velocity.y"])

is expected to give the same output as:

>>> event.get_data(message_paths_include=["velocity"], message_paths_exclude=["velocity.z"])

Parameters:

message_paths_include (Optional[collections.abc.Sequence[str]])
message_paths_exclude (Optional[collections.abc.Sequence[str]])
topic_name (Optional[str])
topic_data_service (Optional[roboto.domain.topics.TopicDataService])
cache_dir (Union[str, pathlib.Path, None])
strict_associations (bool)

Return type:

collections.abc.Generator[dict[str, Any], None, None]

get_data_as_df(message_paths_include=None, message_paths_exclude=None, topic_name=None, topic_data_service=None, cache_dir=None, strict_associations=False)#

Return the underlying topic data this event annotates as a pandas DataFrame.

Collects all data from get_data() and returns it as a pandas DataFrame with the log time as the index. Requires installing this package using the roboto[analytics] extra.

Parameters:

message_paths_include (Optional[collections.abc.Sequence[str]]) – Dot notation paths to include in the data.
message_paths_exclude (Optional[collections.abc.Sequence[str]]) – Dot notation paths to exclude from the data.
topic_name (Optional[str]) – Required when event is associated with a file.
topic_data_service (Optional[roboto.domain.topics.TopicDataService]) – Service for accessing topic data.
cache_dir (Union[str, pathlib.Path, None]) – Directory for caching downloaded data.
strict_associations (bool)

Returns:

DataFrame containing the event’s underlying topic data, indexed by log time.

Raises:

ImportError – If pandas is not installed (install with roboto[analytics]).
RobotoInvalidRequestException – Invalid parameters or event associations.

Return type:

pandas.DataFrame

Examples

Get event data as a DataFrame:

>>> event = Event.from_id("ev_abc123")
>>> df = event.get_data_as_df()
>>> print(f"Data shape: {df.shape}")
>>> print(df.head())

Get specific message paths as DataFrame:

>>> df = event.get_data_as_df(
...     message_paths_include=["velocity.x", "velocity.y"]
... )
>>> print(df.columns.tolist())

Analyze event data:

>>> df = event.get_data_as_df()
>>> print(f"Event duration: {df.index.max() - df.index.min()} ns")
>>> print(f"Data points: {len(df)}")

message_path_ids()#

Get message path IDs directly associated with this event.

Returns:: List of unique message path IDs directly associated with this event.
Return type:: list[str]

Examples

Get message path IDs:

>>> event = Event.from_id("ev_abc123")
>>> msgpath_ids = event.message_path_ids()
>>> print(f"Associated with {len(msgpath_ids)} message paths")

property metadata: dict[str, Any]#

Key-value metadata associated with this event.

Return type:: dict[str, Any]

property modified: datetime.datetime#

Date and time when this event was last modified.

Return type:: datetime.datetime

property modified_by: str#

User who last modified this event.

Return type:: str

property name: str#

Human-readable name of the event.

Return type:: str

put_metadata(metadata)#

Add or update metadata fields for this event.

Parameters:: metadata (dict[str, Any]) – Dictionary of key-value pairs to add or update.
Returns:: Updated Event instance.
Return type:: Event

Examples

Add metadata to an event:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.put_metadata({
...     "severity": "high",
...     "component": "engine",
...     "alert_id": "alert_001"
... })
>>> print(updated_event.metadata["severity"])
'high'

put_tags(tags)#

Replace all tags for this event.

Parameters:: tags (list[str]) – List of tags to set for this event.
Returns:: Updated Event instance.
Return type:: Event

Examples

Set tags for an event:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.put_tags(["anomaly", "critical", "engine"])
>>> print(updated_event.tags)
['anomaly', 'critical', 'engine']

property record: roboto.domain.events.record.EventRecord#

Underlying event record data.

Return type:: roboto.domain.events.record.EventRecord

refresh()#

Refresh this event’s data from the server.

Fetches the latest version of this event from the server, updating all properties to reflect any changes made by other processes.

Returns:: This Event instance with refreshed data.
Return type:: Event

Examples

Refresh an event to get latest changes:

>>> event = Event.from_id("ev_abc123")
>>> # Event may have been updated by another process
>>> refreshed_event = event.refresh()
>>> print(f"Current description: {refreshed_event.description}")

remove_metadata(metadata)#

Remove metadata fields from this event.

Parameters:: metadata (roboto.updates.StrSequence) – Sequence of metadata field names to remove. Supports dot notation for nested fields.
Returns:: Updated Event instance.
Return type:: Event

Examples

Remove specific metadata fields:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.remove_metadata(["severity", "temp_data.max"])
>>> # Fields 'severity' and nested 'temp_data.max' are now removed

remove_tags(tags)#

Remove specific tags from this event.

Parameters:: tags (roboto.updates.StrSequence) – Sequence of tag names to remove from this event.
Returns:: Updated Event instance.
Return type:: Event

Examples

Remove specific tags:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.remove_tags(["temporary", "draft"])
>>> # Tags 'temporary' and 'draft' are now removed

set_color(color)#

Set the display color for this event.

Parameters:: color (Optional[str]) – CSS-compatible color value (e.g., “red”, “#ff0000”, “rgb(255,0,0)”). Use None to clear the color and use automatic coloring.
Returns:: Updated Event instance.
Return type:: Event

Examples

Set event color to red:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.set_color("red")
>>> print(updated_event.color)
'red'

Clear event color:

>>> updated_event = event.set_color(None)
>>> print(updated_event.color)
None

set_description(description)#

Set the description for this event.

Parameters:: description (Optional[str]) – New description for the event. Use None to clear the description.
Returns:: Updated Event instance.
Return type:: Event

Examples

Set event description:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.set_description("Updated event description")
>>> print(updated_event.description)
'Updated event description'

Clear event description:

>>> updated_event = event.set_description(None)
>>> print(updated_event.description)
None

set_name(name)#

Set the name for this event.

Parameters:: name (str) – New name for the event.
Returns:: Updated Event instance.
Return type:: Event

Examples

Update event name:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.set_name("Critical System Alert")
>>> print(updated_event.name)
'Critical System Alert'

property start_time: int#

Start time of the event in nanoseconds since UNIX epoch.

Return type:: int

property tags: list[str]#

Tags associated with this event for categorization and search.

Return type:: list[str]

to_dict()#

Convert this event to a dictionary representation.

Returns:: Dictionary containing all event data in JSON-serializable format.
Return type:: dict[str, Any]

Examples

Convert event to dictionary:

>>> event = Event.from_id("ev_abc123")
>>> event_dict = event.to_dict()
>>> print(event_dict["name"])
>>> print(event_dict["start_time"])

topic_ids(strict_associations=False)#

Get topic IDs associated with this event.

Parameters:: strict_associations (bool) – If True, only return topics with direct associations. If False (default), also return topics inferred from message path associations.
Returns:: List of unique topic IDs associated with this event.
Return type:: list[str]

Examples

Get all associated topic IDs:

>>> event = Event.from_id("ev_abc123")
>>> topic_ids = event.topic_ids()
>>> print(f"Associated with {len(topic_ids)} topics")

Get only directly associated topics:

>>> strict_topic_ids = event.topic_ids(strict_associations=True)
>>> for topic_id in strict_topic_ids:
...     print(f"Directly associated topic: {topic_id}")

update(name=NotSet, start_time=NotSet, end_time=NotSet, description=NotSet, metadata_changeset=NotSet, display_options_changeset=NotSet)#

Update this event’s attributes.

Updates various properties of the event including name, time range, description, metadata, and display options. Only specified parameters are updated; others remain unchanged.

When provided, start_time and end_time should be integers representing nanoseconds since the UNIX epoch, or convertible to such integers by to_epoch_nanoseconds().

Parameters:

name (Union[str, roboto.sentinels.NotSetType]) – New human-readable name for the event.
start_time (Union[roboto.time.Time, roboto.sentinels.NotSetType]) – New start timestamp for the event.
end_time (Union[roboto.time.Time, roboto.sentinels.NotSetType]) – New end timestamp for the event.
description (Union[str, None, roboto.sentinels.NotSetType]) – New description for the event. Set to None to clear existing description.
metadata_changeset (Union[roboto.updates.MetadataChangeset, roboto.sentinels.NotSetType]) – Changes to apply to the event’s metadata and tags.
display_options_changeset (Union[roboto.domain.events.operations.EventDisplayOptionsChangeset, roboto.sentinels.NotSetType]) – Changes to apply to the event’s display options.

Returns:

This Event instance with attributes updated accordingly.

Raises:

ValueError – If start_time or end_time are negative.
RobotoIllegalArgumentException – If start_time > end_time.
RobotoUnauthorizedException – Caller lacks permission to edit this event.

Return type:

Event

Examples

Update event name and description:

>>> event = Event.from_id("ev_abc123")
>>> updated_event = event.update(
...     name="Critical System Alert",
...     description="Updated description with more details"
... )

Update event time range:

>>> updated_event = event.update(
...     start_time=1722870127699468923,
...     end_time=1722870127799468923
... )

Update metadata and display options:

>>> from roboto.updates import MetadataChangeset
>>> from roboto.domain.events import EventDisplayOptionsChangeset
>>> updated_event = event.update(
...     metadata_changeset=MetadataChangeset(
...         put_fields={"severity": "high"},
...         put_tags=["critical", "urgent"]
...     ),
...     display_options_changeset=EventDisplayOptionsChangeset(color="red")
... )

class roboto.EventDisplayOptions(/, **data)#

Bases: pydantic.BaseModel

Display options for an event.

Parameters:: data (Any)

color: str | None = None#

Display color for the event.

Used to visually distinguish events on a timeline, and optionally to signal semantic information about the event (e.g. “red” for events representing critical issues).

Any value that is permissible in CSS to define a valid color can be used here, encoded as a string. For instance, the following are all valid: “red”, “#ff0000”, “rgb(255 0 0)”.

has_options()#

Checks whether any display options have been specified.

Return type:: bool

class roboto.EventDisplayOptionsChangeset(/, **data)#

Bases: pydantic.BaseModel

A set of changes to the display options of an event.

Parameters:: data (Any)

apply_to(display_options)#

Applies this changeset to some existing display options.

Parameters:: display_options (EventDisplayOptions)
Return type:: EventDisplayOptions

color: str | None | roboto.sentinels.NotSetType#

An update to an event’s color.

Use None to clear any previously set color value. On the Roboto website, the event will be displayed using an automatically selected color.

has_changes()#

Checks whether this changeset contains any changes.

Return type:: bool

class roboto.EventRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of an event.

Parameters:: data (Any)

associations: list[roboto.association.Association] = None#: Datasets, files, topics and message paths which this event pertains to.

created: datetime.datetime#: Date/time when this event was created.

created_by: str#: The user who created this event.

description: str | None = None#: An optional human-readable description of the event.

display_options: roboto.domain.events.operations.EventDisplayOptions | None = None#: Display options for the event, such as color.

end_time: int#: The end time of the event, in nanoseconds since epoch (assumed Unix epoch). This can be equal to start_time if the event is discrete, but can never be less than start_time.

event_id: str#: A globally unique ID used to reference an event.

metadata: dict[str, Any] = None#: Key-value pairs to associate with this event for discovery and search.

modified: datetime.datetime#: Date/time when this event was last modified.

modified_by: str#: The user who last modified this event.

name: str#: A brief human-readable name for the event. Many events can have the same name. “Takeoff”, “Crash”, “CPU Spike”, “Bad Image Quality”, and “Unexpected Left” are a few potential examples.

org_id: str#: The organization to which this event belongs.

start_time: int#: The start time of the event, in nanoseconds since epoch (assumed Unix epoch).

tags: list[str] = None#: Tags to associate with this event for discovery and search.

class roboto.ExecutableProvenance(/, **data)#

Bases: pydantic.BaseModel

Provenance information for an action executable

Parameters:: data (Any)

container_image_digest: str | None = None#

container_image_uri: str | None = None#

class roboto.ExecutorContainer#

Bases: enum.Enum

Type of container running as part of an action invocation

Action = 'action'#

LogRouter = 'firelens_log_router'#

Monitor = 'monitor'#

OutputHandler = 'output_handler'#

Setup = 'setup'#

class roboto.File(record, roboto_client=None)#

Represents a file within the Roboto platform.

Files are the fundamental data storage unit in Roboto. They can be uploaded to datasets, imported from external sources, or created as outputs from actions. Once in the platform, files can be tagged with metadata, post-processed by actions, added to collections, visualized in the web interface, and searched using the query system.

Files contain structured data that can be ingested into topics for analysis and visualization. Common file formats include ROS bags, MCAP files, ULOG files, CSV files, and many others. Each file has an associated ingestion status that tracks whether its data has been processed and made available for querying.

Files are versioned entities - each modification creates a new version while preserving the history. Files are associated with datasets and inherit access permissions from their parent dataset.

The File class provides methods for downloading, updating metadata, managing tags, accessing topics, and performing other file operations. It serves as the primary interface for file manipulation in the Roboto SDK.

Parameters:

record (roboto.domain.files.record.FileRecord)
roboto_client (Optional[roboto.http.RobotoClient])

static construct_s3_obj_arn(bucket, key, partition='aws')#

Construct an S3 object ARN from bucket and key components.

Parameters:

bucket (str) – S3 bucket name.
key (str) – S3 object key (path within the bucket).
partition (str) – AWS partition name, defaults to “aws”.

Returns:

Complete S3 object ARN string.

Return type:

str

Examples

>>> arn = File.construct_s3_obj_arn("my-bucket", "path/to/file.bag")
>>> print(arn)
'arn:aws:s3:::my-bucket/path/to/file.bag'

static construct_s3_obj_uri(bucket, key, version=None)#

Construct an S3 object URI from bucket, key, and optional version.

Parameters:

bucket (str) – S3 bucket name.
key (str) – S3 object key (path within the bucket).
version (Optional[str]) – Optional S3 object version ID.

Returns:

Complete S3 object URI string.

Return type:

str

Examples

>>> uri = File.construct_s3_obj_uri("my-bucket", "path/to/file.bag")
>>> print(uri)
's3://my-bucket/path/to/file.bag'

>>> versioned_uri = File.construct_s3_obj_uri("my-bucket", "path/to/file.bag", "abc123")
>>> print(versioned_uri)
's3://my-bucket/path/to/file.bag?versionId=abc123'

property created: datetime.datetime#

Timestamp when this file was created.

Returns the UTC datetime when this file was first uploaded or created in the Roboto platform. This timestamp is immutable.

Return type:: datetime.datetime

property created_by: str#

Identifier of the user who created this file.

Returns the user ID or identifier of the person or service that originally uploaded or created this file in the Roboto platform.

Return type:: str

property dataset_id: str#

Identifier of the dataset that contains this file.

Returns the unique identifier of the dataset that this file belongs to. Files are always associated with exactly one dataset.

Return type:: str

delete()#

Delete this file from the Roboto platform.

Permanently removes the file and all its associated data, including topics and metadata. This operation cannot be undone.

For files that were imported from customer S3 buckets (read-only BYOB integrations), this method does not delete the file content from S3. It only removes the metadata and references within the Roboto platform.

Raises:

RobotoNotFoundException – File does not exist or has already been deleted.
RobotoUnauthorizedException – Caller lacks permission to delete the file.

Return type:

None

Examples

>>> file = File.from_id("file_abc123")
>>> file.delete()
# File is now permanently deleted

property description: str | None#

Human-readable description of this file.

Returns the optional description text that provides details about the file’s contents, purpose, or context. Can be None if no description was provided.

Return type:: Optional[str]

download(local_path, credential_provider=None, progress_monitor_factory=NoopProgressMonitorFactory())#

Download this file to a local path.

Downloads the file content from cloud storage to the specified local path. The parent directories are created automatically if they don’t exist.

Parameters:

local_path (pathlib.Path) – Local filesystem path where the file should be saved.
credential_provider (Optional[roboto.domain.files.file_creds.CredentialProvider]) – Custom credentials for accessing the file storage. If None, uses default credentials for the file’s dataset.
progress_monitor_factory (roboto.domain.files.progress.ProgressMonitorFactory) – Factory for creating progress monitors to track download progress. Defaults to no progress monitoring.

Raises:

RobotoUnauthorizedException – Caller lacks permission to download the file.
FileNotFoundError – File content is not available in storage.

Examples

>>> import pathlib
>>> file = File.from_id("file_abc123")
>>> local_path = pathlib.Path("/tmp/downloaded_file.bag")
>>> file.download(local_path)
>>> print(f"Downloaded to {local_path}")

>>> # Download with progress monitoring
>>> from roboto.domain.files.progress import TqdmProgressMonitorFactory
>>> progress_factory = TqdmProgressMonitorFactory()
>>> file.download(local_path, progress_monitor_factory=progress_factory)

property file_id: str#

Unique identifier for this file.

Returns the globally unique identifier assigned to this file when it was created. This ID is immutable and used to reference the file across the Roboto platform.

Return type:: str

classmethod from_id(file_id, version_id=None, roboto_client=None)#

Create a File instance from a file ID.

Retrieves file information from the Roboto platform using the provided file ID and optionally a specific version.

Parameters:

file_id (str) – Unique identifier for the file.
version_id (Optional[int]) – Specific version of the file to retrieve. If None, gets the latest version.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

File instance representing the requested file.

Raises:

RobotoNotFoundException – File with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the file.

Return type:

Examples

>>> file = File.from_id("file_abc123")
>>> print(file.relative_path)
'data/sensor_logs.bag'

>>> old_version = File.from_id("file_abc123", version_id=1)
>>> print(old_version.version)
1

classmethod from_path_and_dataset_id(file_path, dataset_id, version_id=None, roboto_client=None)#

Create a File instance from a file path and dataset ID.

Retrieves file information using the file’s relative path within a specific dataset. This is useful when you know the file’s location within a dataset but not its file ID.

Parameters:

file_path (Union[str, pathlib.Path]) – Relative path of the file within the dataset.
dataset_id (str) – ID of the dataset containing the file.
version_id (Optional[int]) – Specific version of the file to retrieve. If None, gets the latest version.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

File instance representing the requested file.

Raises:

RobotoNotFoundException – File at the given path does not exist in the dataset.
RobotoUnauthorizedException – Caller lacks permission to access the file or dataset.

Return type:

Examples

>>> file = File.from_path_and_dataset_id("logs/session1.bag", "ds_abc123")
>>> print(file.file_id)
'file_xyz789'

>>> file = File.from_path_and_dataset_id(pathlib.Path("data/sensors.csv"), "ds_abc123")
>>> print(file.relative_path)
'data/sensors.csv'

static generate_s3_client(credential_provider, tcp_keepalive=True)#

Generate a configured S3 client using Roboto credentials.

Creates an S3 client with refreshable credentials obtained from the provided credential provider. The client is configured with the appropriate region and connection settings.

Parameters:

credential_provider (roboto.domain.files.file_creds.CredentialProvider) – Function that returns AWS credentials for S3 access.
tcp_keepalive (bool) – Whether to enable TCP keepalive for the S3 connection.

Returns:

Configured boto3 S3 client instance.

Examples

>>> from roboto.domain.files.file_creds import FileCredentialsHelper
>>> helper = FileCredentialsHelper(roboto_client)
>>> cred_provider = helper.get_dataset_download_creds_provider("ds_123", "bucket")
>>> s3_client = File.generate_s3_client(cred_provider)

generate_summary()#

Generate a new AI generated summary of this file. If a summary already exists, it will be overwritten. The results of this call are persisted and can be retrieved with get_summary().

Returns: An AISummary object containing the summary text and the creation timestamp.

Example

>>> from roboto import File
>>> fl = File.from_id("fl_abc123")
>>> summary = fl.generate_summary()
>>> print(summary.text)
This file contains ...

Return type:: roboto.ai.summary.AISummary

get_signed_url(override_content_type=None, override_content_disposition=None)#

Generate a signed URL for direct access to this file.

Creates a time-limited URL that allows direct access to the file content without requiring Roboto authentication. Useful for sharing files or integrating with external systems.

Parameters:

override_content_type (Optional[str]) – Custom MIME type to set in the response headers.
override_content_disposition (Optional[str]) – Custom content disposition header value (e.g., “attachment; filename=myfile.bag”).

Returns:

Signed URL string that provides temporary access to the file.

Raises:

RobotoUnauthorizedException – Caller lacks permission to access the file.

Return type:

str

Examples

>>> file = File.from_id("file_abc123")
>>> url = file.get_signed_url()
>>> print(f"Direct access URL: {url}")

>>> # Force download with custom filename
>>> download_url = file.get_signed_url(
...     override_content_disposition="attachment; filename=data.bag"
... )

get_summary()#

Get the latest AI generated summary of this file. If no summary exists, one will be generated, equivalent to a call to generate_summary().

After the first summary for a file is generated, it will be persisted and returned by this method until generate_summary() is explicitly called again. This applies even if the file or its topics/metadata change.

Returns: An AISummary object containing the summary text and the creation timestamp.

Example

>>> from roboto import File
>>> fl = File.from_id("fl_abc123")
>>> summary = fl.get_summary()
>>> print(summary.text)
This file contains ...

Return type:: roboto.ai.summary.AISummary

get_summary_sync(timeout=60, poll_interval=2)#

Poll the summary endpoint until a summary’s status is COMPLETED, or raise an exception if the status is FAILED or the configurable timeout is reached.

This method will call get_summary() repeatedly until the summary reaches a terminal status. If no summary exists when this method is called, one will be generated automatically.

Parameters:

timeout (float) – The maximum amount of time, in seconds, to wait for the summary to complete. Defaults to 1 minute.
poll_interval (roboto.waiters.Interval) – The amount of time, in seconds, to wait between polling iterations. Defaults to 2 seconds.

Return type:

Returns: An AI Summary object containing a full LLM summary of the file.

Raises:

RobotoFailedToGenerateException – If the summary status becomes FAILED.
TimeoutError – If the timeout is reached before the summary completes.

Parameters:

timeout (float)
poll_interval (roboto.waiters.Interval)

Return type:

roboto.domain.topics.Topic

Example

>>> from roboto import File
>>> fl = File.from_id("fl_abc123")
>>> summary = fl.get_summary_sync(timeout=60)
>>> print(summary.text)
This file contains ...

get_topic(topic_name)#

Get a specific topic from this file by name.

Retrieves a topic with the specified name that is associated with this file. Topics contain the structured data extracted from the file during ingestion.

Parameters:

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

Returns:

Topic instance for the specified topic name.

Raises:

RobotoNotFoundException – Topic with the given name does not exist in this file.
RobotoUnauthorizedException – Caller lacks permission to access the topic.

Return type:

Examples

>>> file = File.from_id("file_abc123")
>>> camera_topic = file.get_topic("/camera/image")
>>> print(f"Topic schema: {camera_topic.schema}")

>>> # Access topic data
>>> for record in camera_topic.get_data():
...     print(f"Timestamp: {record['timestamp']}")

get_topics(include=None, exclude=None)#

Get all topics associated with this file, with optional filtering.

Retrieves all topics that were extracted from this file during ingestion. Topics can be filtered by name using include/exclude patterns.

Parameters:

include (Optional[collections.abc.Sequence[str]]) – If provided, only topics with names in this sequence are yielded.
exclude (Optional[collections.abc.Sequence[str]]) – If provided, topics with names in this sequence are skipped.

Yields:

Topic instances associated with this file, filtered according to the parameters.

Return type:

collections.abc.Generator[roboto.domain.topics.Topic, None, None]

Examples

>>> file = File.from_id("file_abc123")
>>> for topic in file.get_topics():
...     print(f"Topic: {topic.name}")
Topic: /camera/image
Topic: /imu/data
Topic: /gps/fix

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

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

classmethod import_batch(requests, roboto_client=None, caller_org_id=None)#

Import files from customer S3 bring-your-own buckets into Roboto datasets.

This is the ingress point for importing data stored in customer-owned S3 buckets that have been registered as read-only bring-your-own bucket (BYOB) integrations with Roboto. Files remain in their original S3 locations while metadata is registered with Roboto for discovery, processing, and analysis.

This method only works with S3 URIs from buckets that have been properly registered as BYOB integrations for your organization. It performs batch operations to efficiently import multiple files in a single API call, reducing overhead and improving performance.

Parameters:

requests (collections.abc.Sequence[roboto.domain.files.operations.ImportFileRequest]) – Sequence of import requests, each specifying file details and metadata.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.
caller_org_id (Optional[str]) – Organization ID of the caller. Required for multi-org users.

Returns:

Sequence of File objects representing the imported files.

Raises:

RobotoInvalidRequestException – If any URI is not a valid S3 URI, if the batch exceeds 500 items, or if bucket integrations are not properly configured.
RobotoUnauthorizedException – If the caller lacks upload permissions for target datasets or if buckets don’t belong to the caller’s organization.

Return type:

collections.abc.Sequence[File]

Notes

Only works with S3 URIs from registered read-only BYOB integrations
Files are not copied; only metadata is imported into Roboto
Batch size is limited to 500 items per request
All S3 buckets must be registered to the caller’s organization

Examples

>>> from roboto.domain.files import ImportFileRequest
>>> requests = [
...     ImportFileRequest(
...         dataset_id="ds_abc123",
...         relative_path="logs/session1.bag",
...         uri="s3://my-bucket/data/session1.bag",
...         size=1024000
...     ),
...     ImportFileRequest(
...         dataset_id="ds_abc123",
...         relative_path="logs/session2.bag",
...         uri="s3://my-bucket/data/session2.bag",
...         size=2048000
...     )
... ]
>>> files = File.import_batch(requests)
>>> print(f"Imported {len(files)} files")
Imported 2 files

classmethod import_one(dataset_id, relative_path, uri, description=None, tags=None, metadata=None, roboto_client=None)#

Import a single file from an external bucket into a Roboto dataset. This currently only supports AWS S3.

This is a convenience method for importing a single file from customer-owned buckets that have been registered as bring-your-own bucket (BYOB) integrations with Roboto. Unlike import_batch(), this method automatically determines the file size by querying the object store and verifies that the object actually exists before importing, providing additional validation and convenience for single-file operations.

The file remains in its original location while metadata is registered with Roboto for discovery, processing, and analysis. This method currently only works with S3 URIs from buckets that have been properly registered as BYOB integrations for your organization.

Parameters:

dataset_id (str) – ID of the dataset to import the file into.
relative_path (str) – Path of the file relative to the dataset root (e.g., logs/session1.bag).
uri (str) – URI where the file is located (e.g., s3://my-bucket/path/to/file.bag). Must be from a registered BYOB integration.
description (Optional[str]) – Optional human-readable description of the file.
tags (Optional[list[str]]) – Optional list of tags for file discovery and organization.
metadata (Optional[dict[str, Any]]) – Optional key-value metadata pairs to associate with the file.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

File object representing the imported file.

Raises:

RobotoInvalidRequestException – If the URI is not a valid URI or if the bucket integration is not properly configured.
RobotoNotFoundException – If the specified object does not exist.
RobotoUnauthorizedException – If the caller lacks upload permissions for the target dataset or if the bucket doesn’t belong to the caller’s organization.

Return type:

Notes

Only works with S3 URIs from registered BYOB integrations
File size is automatically determined from the object metadata
The file is not copied; only metadata is imported into Roboto
For importing multiple files efficiently, use import_batch() instead

Examples

Import a single ROS bag file:

>>> from roboto.domain.files import File
>>> file = File.import_one(
...     dataset_id="ds_abc123",
...     relative_path="logs/session1.bag",
...     uri="s3://my-bucket/data/session1.bag"
... )
>>> print(f"Imported file: {file.relative_path}")
Imported file: logs/session1.bag

Import a file with metadata and tags:

>>> file = File.import_one(
...     dataset_id="ds_abc123",
...     relative_path="sensors/lidar_data.pcd",
...     uri="s3://my-bucket/sensors/lidar_data.pcd",
...     description="LiDAR point cloud from highway test",
...     tags=["lidar", "highway", "test"],
...     metadata={"sensor_type": "Velodyne", "resolution": "high"}
... )
>>> print(f"File size: {file.size} bytes")

property ingestion_status: roboto.domain.files.record.IngestionStatus#

Current ingestion status of this file.

Returns the status indicating whether this file has been processed and its data extracted into topics. Used to track ingestion pipeline progress.

Return type:: roboto.domain.files.record.IngestionStatus

mark_ingested()#

Mark this file as fully ingested and ready for post-processing.

Updates the file’s ingestion status to indicate that all data has been successfully processed and extracted into topics. This enables triggers and other automated workflows that depend on complete ingestion.

Returns:: Updated File instance with ingestion status set to Ingested.
Raises:: RobotoUnauthorizedException – Caller lacks permission to update the file.
Return type:: File

Notes

This method is typically called by ingestion actions after they have successfully processed all data in the file. Once marked as ingested, the file becomes eligible for additional post-processing actions.

Examples

>>> file = File.from_id("file_abc123")
>>> print(file.ingestion_status)
IngestionStatus.NotIngested
>>> updated_file = file.mark_ingested()
>>> print(updated_file.ingestion_status)
IngestionStatus.Ingested

property metadata: dict[str, Any]#

Custom metadata associated with this file.

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

Return type:: dict[str, Any]

property modified: datetime.datetime#

Timestamp when this file was last modified.

Returns the UTC datetime when this file’s metadata, tags, or other properties were most recently updated. The file content itself is immutable, but metadata can be modified.

Return type:: datetime.datetime

property modified_by: str#

Identifier of the user who last modified this file.

Returns the user ID or identifier of the person who most recently updated this file’s metadata, tags, or other mutable properties.

Return type:: str

property org_id: str#

Organization identifier that owns this file.

Returns the unique identifier of the organization that owns and has primary access control over this file.

Return type:: str

put_metadata(metadata)#

Add or update metadata fields for this file.

Adds new metadata fields or updates existing ones. Existing fields not specified in the metadata dict are preserved.

Parameters:: metadata (dict[str, Any]) – Dictionary of metadata key-value pairs to add or update.
Returns:: Updated File instance with the new metadata.
Raises:: RobotoUnauthorizedException – Caller lacks permission to update the file.
Return type:: File

Examples

>>> file = File.from_id("file_abc123")
>>> updated_file = file.put_metadata({
...     "vehicle_id": "vehicle_001",
...     "session_type": "highway_driving",
...     "weather": "sunny"
... })
>>> print(updated_file.metadata["vehicle_id"])
'vehicle_001'

put_tags(tags)#

Add or update tags for this file.

Replaces the file’s current tags with the provided list. To add tags while preserving existing ones, retrieve current tags first and combine them.

Parameters:: tags (list[str]) – List of tag strings to set on the file.
Returns:: Updated File instance with the new tags.
Raises:: RobotoUnauthorizedException – Caller lacks permission to update the file.
Return type:: File

Examples

>>> file = File.from_id("file_abc123")
>>> updated_file = file.put_tags(["sensor-data", "highway", "sunny"])
>>> print(updated_file.tags)
['sensor-data', 'highway', 'sunny']

classmethod query(spec=None, roboto_client=None, owner_org_id=None)#

Query files using a specification with filters and pagination.

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

Parameters:

spec (Optional[roboto.query.QuerySpecification]) – Query specification with filters, sorting, and pagination options. If None, returns all accessible files.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.
owner_org_id (Optional[str]) – Organization ID to scope the query. If None, uses caller’s org.

Yields:

File instances matching the query specification.

Raises:

ValueError – Query specification references unknown file attributes.
RobotoUnauthorizedException – Caller lacks permission to query files.

Return type:

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

Examples

>>> from roboto.query import Comparator, Condition, QuerySpecification
>>> spec = QuerySpecification(
...     condition=Condition(
...         field="tags",
...         comparator=Comparator.Contains,
...         value="sensor-data"
...     ))
>>> for file in File.query(spec):
...     print(f"Found file: {file.relative_path}")
Found file: logs/sensors_2024_01_01.bag
Found file: logs/sensors_2024_01_02.bag

>>> # Query with metadata filter
>>> spec = QuerySpecification(
...     condition=Condition(
...         field="metadata.vehicle_id",
...         comparator=Comparator.Equals,
...         value="vehicle_001"
...     ))
>>> files = list(File.query(spec))
>>> print(f"Found {len(files)} files for vehicle_001")

property record: roboto.domain.files.record.FileRecord#

Underlying data record for this file.

Returns the raw FileRecord that contains all the file’s data fields. This provides access to the complete file state as stored in the platform.

Return type:: roboto.domain.files.record.FileRecord

refresh()#

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

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

Returns:

This File instance with refreshed data.

Raises:

RobotoNotFoundException – File no longer exists.
RobotoUnauthorizedException – Caller lacks permission to access the file.

Return type:

roboto.domain.files.record.FileRecord

Examples

>>> file = File.from_id("file_abc123")
>>> # File may have been updated by another process
>>> refreshed_file = file.refresh()
>>> print(f"Current version: {refreshed_file.version}")

property relative_path: str#

Path of this file relative to its dataset root.

Returns the file path within the dataset, using forward slashes as separators regardless of the operating system. This path uniquely identifies the file within its dataset.

Return type:: str

rename_file(file_id, new_path)#

Rename this file to a new path within its dataset.

Changes the relative path of the file within its dataset. This updates the file’s location identifier but does not move the actual file content.

Parameters:

file_id (str) – File ID (currently unused, kept for API compatibility).
new_path (str) – New relative path for the file within the dataset.

Returns:

Updated FileRecord with the new path.

Raises:

RobotoUnauthorizedException – Caller lacks permission to rename the file.
RobotoInvalidRequestException – New path is invalid or conflicts with existing file.

Return type:

Examples

>>> file = File.from_id("file_abc123")
>>> print(file.relative_path)
'old_logs/session1.bag'
>>> updated_record = file.rename_file("file_abc123", "logs/session1.bag")
>>> print(updated_record.relative_path)
'logs/session1.bag'

property tags: list[str]#

List of tags associated with this file.

Returns the list of string tags that have been applied to this file for categorization and filtering purposes.

Return type:: list[str]

to_association()#

Convert this file to an Association reference.

Creates an Association object that can be used to reference this file in other contexts, such as when creating collections or specifying action inputs.

Returns:: Association object referencing this file and its current version.
Return type:: roboto.association.Association

Examples

>>> file = File.from_id("file_abc123")
>>> association = file.to_association()
>>> print(f"Association: {association.association_type}:{association.association_id}")
Association: file:file_abc123

to_dict()#

Convert this file to a dictionary representation.

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

Returns:: Dictionary representation of the file data.
Return type:: dict[str, Any]

Examples

>>> file = File.from_id("file_abc123")
>>> file_dict = file.to_dict()
>>> print(file_dict["relative_path"])
'logs/session1.bag'
>>> print(file_dict["metadata"])
{'vehicle_id': 'vehicle_001', 'session_type': 'highway'}

update(description=NotSet, metadata_changeset=NotSet, ingestion_complete=NotSet)#

Update this file’s properties.

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

Parameters:

description (Optional[Union[str, roboto.sentinels.NotSetType]]) – New description for the file. Use NotSet to leave unchanged.
metadata_changeset (Union[roboto.updates.MetadataChangeset, roboto.sentinels.NotSetType]) – Metadata changes to apply (add, update, or remove fields/tags). Use NotSet to leave metadata unchanged.
ingestion_complete (Union[Literal[True], roboto.sentinels.NotSetType]) – Set to True to mark the file as fully ingested. Use NotSet to leave ingestion status unchanged.

Returns:

Updated File instance with the new properties.

Raises:

RobotoUnauthorizedException – Caller lacks permission to update the file.

Return type:

Examples

>>> file = File.from_id("file_abc123")
>>> updated_file = file.update(description="Updated sensor data from highway test")
>>> print(updated_file.description)
'Updated sensor data from highway test'

>>> # Update metadata and mark as ingested
>>> from roboto.updates import MetadataChangeset
>>> changeset = MetadataChangeset(put_fields={"processed": True})
>>> updated_file = file.update(
...     metadata_changeset=changeset,
...     ingestion_complete=True
... )

property uri: str#

Storage URI for this file’s content.

Returns the storage location URI where the file’s actual content is stored. This is typically an S3 URI or similar cloud storage reference.

Return type:: str

property version: int#

Version number of this file.

Returns the version number that increments each time the file’s metadata or properties are updated. The file content itself is immutable, but metadata changes create new versions.

Return type:: int

class roboto.FileRecord(/, **data)#

Bases: pydantic.BaseModel

Wire-transmissible representation of a file in the Roboto platform.

FileRecord contains all the metadata and properties associated with a file, including its location, status, ingestion state, and user-defined metadata. This is the data structure used for API communication and persistence.

FileRecord instances are typically created by the platform during file import or upload operations, and are updated as files are processed and modified. The File domain class wraps FileRecord to provide a more convenient interface for file operations.

Parameters:: data (Any)

association_id: str#

property bucket: str#

Return type:: str

created: datetime.datetime#

created_by: str = ''#

description: str | None = None#

device_id: str | None = None#

file_id: str#

fs_type: FSType#

ingestion_status: IngestionStatus#

property key: str#

Return type:: str

metadata: dict[str, Any] = None#

modified: datetime.datetime#

modified_by: str#

name: str#

org_id: str#

origination: str = ''#

parent_id: str | None = None#

relative_path: str#

size: int#

status: FileStatus#

storage_type: FileStorageType#

tags: list[str] = None#

upload_id: str = 'NO_ID'#

uri: str#

version: int#

class roboto.FileRecordRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for upserting a file record.

Used to create or update file metadata records in the platform. This is typically used during file import or metadata update operations.

Parameters:: data (Any)

file_id: str#: Unique identifier for the file.

metadata: dict[str, Any] = None#: Key-value metadata pairs to associate with the file.

tags: list[str] = None#: List of tags to associate with the file for discovery and organization.

class roboto.FileStatus#

Bases: str, enum.Enum

Enumeration of possible file status values in the Roboto platform.

File status tracks the lifecycle state of a file from initial upload through to availability for use. This status is managed automatically by the platform and affects file visibility and accessibility.

The typical file lifecycle is: Reserved → Available → (optionally) Deleted.

Available = 'available'#

File upload is complete and the file is ready for use.

Files with this status are visible in dataset listings, searchable through the query system, and available for download and processing by actions.

Deleted = 'deleted'#

File is marked for deletion and is no longer accessible.

Files with this status are not visible in listings and cannot be accessed. This status may be temporary during the deletion process.

Reserved = 'reserved'#

File upload has been initiated but not yet completed.

Files with this status are not yet available for use and are not visible in dataset listings. This is the initial status when an upload begins.

class roboto.FileTag#

Bases: enum.Enum

Enumeration of system-defined file tag types.

These tags are used internally by the platform for indexing and organizing files. They are automatically applied during file operations and should not be manually modified by users.

CommonPrefix = 'common_prefix'#: Tag containing the common path prefix for files in a batch operation.

DatasetId = 'dataset_id'#: Tag containing the ID of the dataset that contains this file.

OrgId = 'org_id'#: Tag containing the organization ID that owns this file.

TransactionId = 'transaction_id'#: Tag containing the transaction ID for files uploaded in a batch.

class roboto.FilesChangesetFileManager#

This class is used to pre-write tags/metadata updates to files which haven’t been uploaded yet, but will be at the conclusion of an action, by virtue of being in that action’s output directory.

It uses a “file changeset” file to accumulate these pending updates during an action’s runtime, and then applies them automatically at the end of an action, after the action’s output directory has been uploaded.

The most common way to get access to this would be via roboto.action_runtime.ActionRuntime.

put_fields(relative_path, metadata)#

Adds metadata key/value pairs to a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 2, "trucks": 3})
>>>
>>> # Actually there was a 3rd car I missed in the first pass, and a plane, let me fix that...
>>> file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 3, "planes": 1})

Parameters:

relative_path (str)
metadata (dict[str, Any])

put_tags(relative_path, tags)#

Adds tags to a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_tags("images/front0_raw_000734.jpg", ["cloudy", "rainy"]})

Parameters:

relative_path (str)
tags (list[str])

remove_fields(relative_path, keys)#

Removes metadata key/value pairs from a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime. You’ll generally only need this to remove values which were added by a previous call to put_fields().

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 2, "trucks": 3})
>>>
>>> # Whoops, actually I don't want to count those trucks...
>>> file_changeset_manager.remove_fields("images/front0_raw_000734.jpg", ["trucks"])

Parameters:

relative_path (str)
keys (list[str])

remove_tags(relative_path, tags)#

Removes tags from a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime. You’ll generally only need this to remove tags which were added by a previous call to put_tags().

This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.put_tags("images/front0_raw_000734.jpg", ["cloudy", "rainy"]})
>>>
>>> # Actually this is just Seattle's aggressive mist, that's not really rainy...
>>> file_changeset_manager.remove_tags("images/front0_raw_000734.jpg", ["rainy"]})

Parameters:

relative_path (str)
tags (list[str])

set_description(relative_path, description)#

Sets the human-readable description of a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.

Examples

>>> from roboto import ActionRuntime
>>> action_runtime = ActionRuntime.from_env()
>>> file_changeset_manager = action_runtime.file_changeset_manager
>>>
>>> # This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
>>> file_changeset_manager.set_description("images/front0_raw_000734.jpg", "This image was over-exposed")

Parameters:

relative_path (str)
description (Optional[str])

class roboto.ImportFileRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for importing an existing file into a dataset.

Used to register files that already exist in storage (such as customer S3 buckets) with the Roboto platform. The file content remains in its original location while metadata is stored in Roboto for discovery and processing.

Parameters:: data (Any)

dataset_id: str#: ID of the dataset to import the file into.

description: str | None = None#: Optional human-readable description of the file.

metadata: dict[str, Any] | None = None#: Optional key-value metadata pairs to associate with the file.

relative_path: str#: Path of the file relative to the dataset root (e.g., logs/session1.bag).

size: int | None = None#: Size of the file in bytes. When importing a single file, you can omit the size, as Roboto will look up the size from the object store. When calling import_batch, you must provide the size explicitly.

tags: list[str] | None = None#: Optional list of tags for file discovery and organization.

uri: str#

//bucket/path/to/file.bag`).

Type:: Storage URI where the file is located (e.g., `s3

class roboto.IngestionStatus#

Bases: str, enum.Enum

Enumeration of file ingestion status values in the Roboto platform.

Ingestion status tracks whether a file’s data has been processed and extracted into topics for analysis and visualization. This status determines what platform features are available for the file and whether it can trigger automated workflows.

File ingestion happens as a post-upload processing step. Roboto supports many common robotics log formats (ROS bags, MCAP files, ULOG files, etc.) out-of-the-box. Custom ingestion actions can be written for other formats.

When writing custom ingestion actions, be sure to update the file’s ingestion status to mark it as fully ingested. This enables triggers and other automated workflows that depend on complete ingestion.

Ingested files have first-class visualization support and can be queried through the topic data system.

Ingested = 'ingested'#

All topics from this file have been fully processed and recorded.

Files with this status have complete topic data available for visualization, analysis, and querying. They are eligible for post-ingestion triggers and automated workflows that depend on complete data extraction.

NotIngested = 'not_ingested'#

No topics from this file have been processed or recorded.

Files with this status have not undergone data extraction. They cannot be visualized through the topic system and are not eligible for topic-based triggers or analysis workflows.

PartlyIngested = 'partly_ingested'#

Some but not all topics from this file have been processed.

Files with this status have at least one topic record but ingestion is incomplete. Some visualization and analysis features may be available, but the file is not yet eligible for post-ingestion triggers.

class roboto.Invocation(record, roboto_client=None)#

An instance of an execution of an action, initiated manually by a user or automatically by a trigger.

An Invocation represents a single execution of an Action with specific inputs, parameters, and configuration. It tracks the execution lifecycle from creation through completion, including status updates, logs, and results.

Invocations are created by calling Action.invoke() or through the UI. They cannot be created directly through the constructor. Each invocation has a unique ID and maintains a complete audit trail of its execution.

Key features:

Status tracking (Queued, Running, Completed, Failed, etc.)
Input data specification and parameter values
Compute requirement and container parameter overrides
Log collection and output file management
Progress monitoring and result retrieval

Parameters:

record (roboto.domain.actions.invocation_record.InvocationRecord)
roboto_client (Optional[roboto.http.RobotoClient])

property action: roboto.domain.actions.invocation_record.ActionProvenance#

Provenance information about the action that was invoked.

Return type:: roboto.domain.actions.invocation_record.ActionProvenance

cancel()#

Cancel this invocation if it is not already in a terminal status.

Attempts to cancel the invocation. If the invocation has already completed, failed, or reached another terminal status, this method has no effect.

Raises:

RobotoNotFoundException – If the invocation is not found.
RobotoUnauthorizedException – If the caller lacks permission to cancel the invocation.

Return type:

None

Examples

Cancel a running invocation:

>>> invocation = Invocation.from_id("iv_12345")
>>> if not invocation.reached_terminal_status:
...     invocation.cancel()

property compute_requirements: roboto.domain.actions.action_record.ComputeRequirements#

The compute requirements (CPU, memory) used for this invocation.

Return type:: roboto.domain.actions.action_record.ComputeRequirements

property container_parameters: roboto.domain.actions.action_record.ContainerParameters#

The container parameters used for this invocation.

Return type:: roboto.domain.actions.action_record.ContainerParameters

property created: datetime.datetime#

The timestamp when this invocation was created.

Return type:: datetime.datetime

property current_status: roboto.domain.actions.invocation_record.InvocationStatus#

The current status of this invocation (e.g., Queued, Running, Completed).

Return type:: roboto.domain.actions.invocation_record.InvocationStatus

property data_source: roboto.domain.actions.invocation_record.InvocationDataSource#

The data source that provided input data for this invocation.

Return type:: roboto.domain.actions.invocation_record.InvocationDataSource

property executable: roboto.domain.actions.invocation_record.ExecutableProvenance#

Provenance information about the executable (container) that was run.

Return type:: roboto.domain.actions.invocation_record.ExecutableProvenance

classmethod from_id(invocation_id, roboto_client=None)#

Load an existing invocation by its ID.

Retrieves an invocation from the Roboto platform using its unique identifier.

Parameters:

invocation_id (str) – The unique ID of the invocation to retrieve.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses default if not provided.

Returns:

The Invocation instance.

Raises:

RobotoNotFoundException – If the invocation is not found.
RobotoUnauthorizedException – If the caller lacks permission to access the invocation.

Return type:

Invocation

Examples

Load an invocation and check its status:

>>> invocation = Invocation.from_id("iv_12345")
>>> print(f"Status: {invocation.current_status}")
>>> print(f"Created: {invocation.created}")

get_logs(page_token=None)#

Retrieve runtime STDOUT/STDERR logs generated during this invocation’s execution.

Fetches log records from the invocation’s container execution, with support for pagination to handle large log volumes.

Parameters:

page_token (Optional[str]) – Optional token for pagination. If provided, starts retrieving logs from that point.

Yields:

LogRecord instances containing log messages and metadata.

Raises:

RobotoNotFoundException – If the invocation is not found.
RobotoUnauthorizedException – If the caller lacks permission to access logs.

Return type:

collections.abc.Generator[roboto.domain.actions.invocation_record.LogRecord, None, None]

property id: str#

The unique identifier for this invocation.

Return type:: str

property input_data: roboto.domain.actions.invocation_record.InvocationInput | None#

The input data specification for this invocation, if any.

Return type:: Optional[roboto.domain.actions.invocation_record.InvocationInput]

is_queued_for_scheduling()#

An invocation is queued for scheduling if:: 1. its most recent status is “Queued” 3. and is not “Deadly”

Return type:: bool

property org_id: str#

The organization ID that owns this invocation.

Return type:: str

property parameter_values: dict[str, Any]#

The parameter values that were provided when this invocation was created.

Return type:: dict[str, Any]

classmethod query(spec=None, owner_org_id=None, roboto_client=None)#

Query invocations with optional filtering and pagination.

Searches for invocations based on the provided query specification. Can filter by status, action name, creation time, and other attributes.

Parameters:

spec (Optional[roboto.query.QuerySpecification]) – Query specification with filters, sorting, and pagination. If not provided, returns all accessible invocations.
owner_org_id (Optional[str]) – Organization ID to search within. If not provided, searches in the caller’s organization.
roboto_client (Optional[roboto.http.RobotoClient]) – Roboto client instance. Uses default if not provided.

Yields:

Invocation instances matching the query criteria.

Raises:

ValueError – If the query specification contains unknown fields.
RobotoUnauthorizedException – If the caller lacks permission to query invocations.

Return type:

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

Examples

Query all invocations:

>>> for invocation in Invocation.query():
...     print(f"Invocation: {invocation.id}")

Query completed invocations:

>>> from roboto.query import QuerySpecification
>>> from roboto.domain.actions import InvocationStatus
>>> spec = QuerySpecification().where("current_status").equals(InvocationStatus.Completed)
>>> completed = list(Invocation.query(spec))

Query recent invocations:

>>> spec = QuerySpecification().order_by("created", ascending=False).limit(10)
>>> recent = list(Invocation.query(spec))

property reached_terminal_status: bool#

True if this invocation has reached a terminal status (Completed, Failed, etc.).

Return type:: bool

property record: roboto.domain.actions.invocation_record.InvocationRecord#

The underlying invocation record containing all invocation data.

Return type:: roboto.domain.actions.invocation_record.InvocationRecord

refresh()#

Return type:: Invocation

set_container_image_digest(digest)#

This is an admin-only operation to memorialize the digest of the container image that was pulled in the course of invoking the action.

Parameters:: digest (str)
Return type:: Invocation

set_logs_location(logs)#

This is an admin-only operation to memorialize the base location where invocation logs are saved.

Use the “get_logs” or “stream_logs” methods to access invocation logs.

Parameters:: logs (roboto.domain.actions.invocation_record.LogsLocation)
Return type:: Invocation

property source: roboto.domain.actions.invocation_record.SourceProvenance#

Provenance information about the source that initiated this invocation.

Return type:: roboto.domain.actions.invocation_record.SourceProvenance

property status_log: list[roboto.domain.actions.invocation_record.InvocationStatusRecord]#

The complete history of status changes for this invocation.

Return type:: list[roboto.domain.actions.invocation_record.InvocationStatusRecord]

stream_logs(last_read=None)#

Parameters:: last_read (Optional[str])
Return type:: collections.abc.Generator[roboto.domain.actions.invocation_record.LogRecord, None, Optional[str]]

property timeout: int#

The timeout in minutes for this invocation.

Return type:: int

to_dict()#

Return type:: dict[str, Any]

update_status(next_status, detail=None)#

Parameters:

next_status (roboto.domain.actions.invocation_record.InvocationStatus)
detail (Optional[str])

Return type:

Invocation

property upload_destination: roboto.domain.actions.invocation_record.InvocationUploadDestination | None#

The destination where output files from this invocation will be uploaded.

Return type:: Optional[roboto.domain.actions.invocation_record.InvocationUploadDestination]

wait_for_terminal_status(timeout=60 * 5, poll_interval=5)#

Wait for the invocation to reach a terminal status.

Throws a TimeoutError if the timeout is reached.

Parameters:

timeout (float) – The maximum amount of time, in seconds, to wait for the invocation to reach a terminal status.
poll_interval (roboto.waiters.Interval) – The amount of time, in seconds, to wait between polling iterations.

Return type:

None

class roboto.InvocationDataSource(/, **data)#

Bases: pydantic.BaseModel

Abstracted data source that can be provided to an invocation.

Represents a source of input data for action invocations. The data source type determines how the ID should be interpreted (e.g., as a dataset ID).

Parameters:: data (Any)

data_source_id: str#: The ID of the data source. For Dataset type, this is a dataset ID.

data_source_type: InvocationDataSourceType#: The type of data source (currently only Dataset).

is_unspecified()#

Check if this data source is unspecified.

Returns:: True if this is an unspecified data source, False otherwise.
Return type:: bool

static unspecified()#

Returns a special value indicating that no invocation source is specified.

Returns:: An InvocationDataSource instance representing an unspecified data source.
Return type:: InvocationDataSource

class roboto.InvocationDataSourceType#

Bases: enum.Enum

Source of data for an action’s input binding.

Defines the type of data source that provides input data to an action invocation. Currently supports datasets, with potential for future expansion to other data source types.

Dataset = 'Dataset'#

class roboto.InvocationProvenance(/, **data)#

Bases: pydantic.BaseModel

Provenance information for an invocation

Parameters:: data (Any)

action: ActionProvenance#: The Action that was invoked.

executable: ExecutableProvenance#: The underlying executable (e.g., Docker image) that was run.

source: SourceProvenance#: The source of the invocation.

class roboto.InvocationRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of an invocation.

Parameters:: data (Any)

compute_requirements: roboto.domain.actions.action_record.ComputeRequirements#

container_parameters: roboto.domain.actions.action_record.ContainerParameters#

created: datetime.datetime#

data_source: InvocationDataSource#

duration: datetime.timedelta = None#

idempotency_id: str | None = None#

input_data: list[str]#

invocation_id: str#

last_heartbeat: datetime.datetime | None = None#

last_status: InvocationStatus#

org_id: str#

parameter_values: dict[str, Any] = None#

provenance: InvocationProvenance#

rich_input_data: InvocationInput | None = None#

status: list[InvocationStatusRecord] = None#

timeout: int#

upload_destination: InvocationUploadDestination | None = None#

class roboto.InvocationSource#

Bases: enum.Enum

Method by which an invocation was run

Manual = 'Manual'#

Trigger = 'Trigger'#

class roboto.InvocationStatus#

Bases: int, enum.Enum

Invocation status enum

Cancelled = 997#

Completed = 5#

Deadly = 999#

Downloading = 2#

Failed = 998#

Processing = 3#

Queued = 0#

Scheduled = 1#

Uploading = 4#

can_transition_to(other)#

Parameters:: other (InvocationStatus)
Return type:: bool

static from_value(v)#

Parameters:: v (Union[int, str])
Return type:: InvocationStatus

is_running()#

Return type:: bool

is_terminal()#

Return type:: bool

next()#

Return type:: Optional[InvocationStatus]

class roboto.InvocationStatusRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of an invocation status.

Parameters:: data (Any)

detail: str | None = None#

status: InvocationStatus#

timestamp: datetime.datetime#

to_presentable_dict()#

Return type:: dict[str, Optional[str]]

class roboto.InvocationUploadDestination(/, **data)#

Bases: pydantic.BaseModel

Default destination to which invocation outputs - if any - should be uploaded.

Specifies where files generated during action execution should be stored. Actions can write files to their output directory, and this destination determines where those files are uploaded after execution completes.

Parameters:: data (Any)

classmethod dataset(dataset_id)#

Create a dataset upload destination with the given ID.

Parameters:: dataset_id (str) – The ID of the dataset where outputs should be uploaded.
Returns:: An InvocationUploadDestination configured for the specified dataset.
Return type:: InvocationUploadDestination

destination_id: str | None = None#: Optional identifier for the upload destination. In the case of a dataset, it would be the dataset ID.

destination_type: UploadDestinationType#: Type of upload destination. By default, outputs are uploaded to a dataset.

property is_dataset: bool#

True if this is a dataset destination with a dataset ID, False otherwise.

Returns:: True if this destination is configured for a dataset with a valid ID.
Return type:: bool

property is_unknown: bool#

True if the upload destination is not of a supported type, False otherwise.

Return type:: bool

classmethod pre_validate_destination_type(value)#

Parameters:: value (Any)
Return type:: Any

class roboto.LogRecord(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of a log record.

Parameters:: data (Any)

log: str#

partial_id: str | None = None#

process: roboto.domain.actions.action_record.ExecutorContainer#

timestamp: datetime.datetime#

class roboto.LogsLocation(/, **data)#

Bases: pydantic.BaseModel

Invocation log storage location

Parameters:: data (Any)

bucket: str#

prefix: str#

class roboto.MessagePath(record, roboto_client=None, topic_data_service=None)#

Represents a message path within a topic in the Roboto platform.

A message path defines a specific field or signal within a topic’s data schema, using dot notation to specify nested attributes. Message paths enable fine-grained access to individual data elements within time-series robotics data, supporting operations like statistical analysis, data filtering, and visualization.

Each message path has an associated data type (both native and canonical), metadata, and statistical information computed from the underlying data. Message paths are the fundamental building blocks for data analysis in Roboto, allowing users to work with specific signals or measurements from complex robotics data structures.

Message paths support temporal filtering, data export to various formats including pandas DataFrames, and integration with the broader Roboto analytics ecosystem. They provide efficient access to time-series data while maintaining the semantic structure of the original robotics messages.

The MessagePath class serves as the primary interface for accessing individual data signals within topics, providing methods for data retrieval, statistical analysis, and metadata management.

Parameters:

record (roboto.domain.topics.record.MessagePathRecord)
roboto_client (Optional[roboto.http.RobotoClient])
topic_data_service (Optional[roboto.domain.topics.topic_data_service.TopicDataService])

DELIMITER: ClassVar = '.'#

property canonical_data_type: roboto.domain.topics.record.CanonicalDataType#

Canonical Roboto data type corresponding to the native data type.

Return type:: roboto.domain.topics.record.CanonicalDataType

property count: PreComputedStat#

Number of data points available for this message path.

Return type:: PreComputedStat

property created: datetime.datetime#

Timestamp when this message path was created.

Return type:: datetime.datetime

property created_by: str#

Identifier of the user or system that created this message path.

Return type:: str

property data_type: str#

Native data type for this message path, e.g. ‘float32’

Return type:: str

classmethod from_id(message_path_id, roboto_client=None, topic_data_service=None)#

Retrieve a message path by its unique identifier.

Fetches a message path record from the Roboto platform using its unique ID. This is useful when you have a message path identifier from another operation.

Parameters:

message_path_id (str) – Unique identifier for the message path.
roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.
topic_data_service (Optional[roboto.domain.topics.topic_data_service.TopicDataService]) – Service for accessing topic data. If None, creates a default instance.

Returns:

MessagePath instance representing the requested message path.

Raises:

RobotoNotFoundException – Message path with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the message path.

Return type:

MessagePath

Examples

>>> message_path = MessagePath.from_id("mp_abc123")
>>> print(message_path.path)
'angular_velocity.x'
>>> print(message_path.canonical_data_type)
CanonicalDataType.Number

get_data(start_time=None, end_time=None, cache_dir=None)#

Return data for this specific message path.

Retrieves and yields data records containing only the values for this message path, with optional temporal filtering. This provides a focused view of a single signal or field within the broader topic data.

Parameters:

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 containing the log_time and the value for this message path.

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 message path’s associated topic:

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

Examples

Print all data for a specific message path:

>>> topic = Topic.from_name_and_file("/imu/data", "file_abc123")
>>> angular_velocity_x = topic.get_message_path("angular_velocity.x")
>>> for record in angular_velocity_x.get_data():
...     print(f"Time: {record['log_time']}, Value: {record['angular_velocity']['x']}")

Get data within a time range:

>>> for record in angular_velocity_x.get_data(
...     start_time=1722870127699468923,
...     end_time=1722870127799468923
... ):
...     print(record)

Collect data into a dataframe (requires installing the roboto[analytics] extra):

>>> df = angular_velocity_x.get_data_as_df()
>>> import math
>>> assert math.isclose(angular_velocity_x.mean, df[angular_velocity_x.path].mean())

get_data_as_df(start_time=None, end_time=None, cache_dir=None)#

Return this message path’s data as a pandas DataFrame.

Retrieves message path data and converts it to a pandas DataFrame for analysis and visualization. The DataFrame is indexed by log time and contains a column for this message path’s values.

Parameters:

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 message path 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")
>>> angular_velocity_x = topic.get_message_path("angular_velocity.x")
>>> df = angular_velocity_x.get_data_as_df()
>>> print(df.head())
                        angular_velocity.x
log_time
1722870127699468923                  0.1
1722870127699468924                  0.15
>>> print(f"Mean: {df[angular_velocity_x.path].mean()}")
Mean: 0.125

property max: PreComputedStat#

Maximum value observed for this message path.

Return type:: PreComputedStat

property mean: PreComputedStat#

Mean (average) value for this message path.

Return type:: PreComputedStat

property median: PreComputedStat#

Median value for this message path.

Return type:: PreComputedStat

property message_path_id: str#

Unique identifier for this message path.

Return type:: str

property metadata: dict[str, Any]#

Metadata dictionary associated with this message path.

Return type:: dict[str, Any]

property min: PreComputedStat#

Minimum value observed for this message path.

Return type:: PreComputedStat

property modified: datetime.datetime#

Timestamp when this message path was last modified.

Return type:: datetime.datetime

property modified_by: str#

Identifier of the user or system that last modified this message path.

Return type:: str

property org_id: str#

Organization ID that owns this message path.

Return type:: str

static parents(path)#

Get parent paths for a message path in dot notation.

Given a message path in dot notation, returns a list of its parent paths ordered from most specific to least specific.

Parameters:: path (str) – Message path in dot notation (e.g., “pose.pose.position.x”).
Returns:: List of parent paths in dot notation, ordered from most to least specific.
Return type:: list[str]

Examples

>>> path = "pose.pose.position.x"
>>> MessagePath.parents(path)
['pose.pose.position', 'pose.pose', 'pose']

>>> # Single level path has no parents
>>> MessagePath.parents("velocity")
[]

static parts(path)#

Split message path in dot notation into its constituent parts.

Splits a message path string into individual components, useful for programmatic manipulation of message path hierarchies.

Parameters:: path (str) – Message path in dot notation (e.g., “pose.pose.position.x”).
Returns:: List of path components in order from root to leaf.
Return type:: list[str]

Examples

>>> path = "pose.pose.position.x"
>>> MessagePath.parts(path)
['pose', 'pose', 'position', 'x']

>>> # Single component path
>>> MessagePath.parts("velocity")
['velocity']

property path: str#

Dot-delimited path to the attribute (e.g., ‘pose.position.x’).

Return type:: str

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

Underlying MessagePathRecord for this message path.

Return type:: roboto.domain.topics.record.MessagePathRecord

to_association()#

Convert this message path to an Association object.

Creates an Association object that can be used to reference this message path in other parts of the Roboto platform.

Returns:: Association object representing this message path.
Return type:: roboto.association.Association

Examples

>>> message_path = MessagePath.from_id("mp_abc123")
>>> association = message_path.to_association()
>>> print(association.association_type)
AssociationType.MessagePath
>>> print(association.association_id)
mp_abc123

property topic_id: str#

Unique identifier of the topic containing this message path.

Return type:: str

class roboto.MessagePathChangeset(/, **data)#

Bases: pydantic.BaseModel

Changeset for batch operations on topic message paths.

Defines a collection of add, delete, and update operations to be applied to a topic’s message paths in a single atomic operation. Useful for making multiple schema changes efficiently.

Parameters:: data (Any)

check_replace_all_correctness()#

Return type:: MessagePathChangeset

classmethod from_replacement_message_paths(message_paths)#

Create a changeset that replaces all existing message paths.

Creates a changeset that will replace all existing message paths on a topic with the provided set of message paths. This is useful for completely redefining a topic’s schema.

Parameters:: message_paths (collections.abc.Sequence[AddMessagePathRequest]) – Sequence of message path requests to replace existing paths.
Returns:: MessagePathChangeset configured to replace all existing message paths.
Return type:: MessagePathChangeset

Examples

>>> from roboto.domain.topics import AddMessagePathRequest, CanonicalDataType
>>> new_paths = [
...     AddMessagePathRequest(
...         message_path="velocity.x",
...         data_type="float32",
...         canonical_data_type=CanonicalDataType.Number
...     )
... ]
>>> changeset = MessagePathChangeset.from_replacement_message_paths(new_paths)

has_changes()#

Check whether the changeset contains any actual changes.

Returns:: True if the changeset contains operations that would modify the topic’s message paths.
Return type:: bool

message_paths_to_add: collections.abc.Sequence[AddMessagePathRequest] | None = None#: Message paths to add to a topic.

message_paths_to_delete: collections.abc.Sequence[DeleteMessagePathRequest] | None = None#: Message paths to delete from a topic.

message_paths_to_update: collections.abc.Sequence[UpdateMessagePathRequest] | None = None#: Message paths to update on a topic.

replace_all: bool = False#

Flag indicating whether this changeset should replace all message paths on a topic.

It assumes that the replacement message paths will be provided via message_paths_to_add. Rather than setting this flag directly, use appropriate class methods such as from_replacement_message_paths.

class roboto.MessagePathRecord(/, **data)#

Bases: pydantic.BaseModel

Record representing a message path within a topic.

Defines a specific field or signal within a topic’s data schema, including its data type, metadata, and statistical information. Message paths use dot notation to specify nested attributes within complex message structures.

Message paths are the fundamental units for accessing individual data elements within time-series robotics data, enabling fine-grained analysis and visualization of specific signals or measurements.

Parameters:: data (Any)

canonical_data_type: CanonicalDataType#: Normalized data type, used primarily internally by the Roboto Platform.

created: datetime.datetime#

created_by: str#

data_type: str#: ‘Native’/framework-specific data type of the attribute at this path. E.g. “float32”, “uint8[]”, “geometry_msgs/Pose”, “string”.

message_path: str#: Dot-delimited path to the attribute within the datum record.

message_path_id: str#

metadata: collections.abc.Mapping[str, Any] = None#: Key-value pairs to associate with this metadata for discovery and search, e.g. { ‘min’: ‘0.71’, ‘max’: ‘1.77 }

modified: datetime.datetime#

modified_by: str#

org_id: str#: This message path’s organization ID, which is the organization ID of the containing topic.

representations: collections.abc.MutableSequence[RepresentationRecord] = None#: Zero to many Representations of this MessagePath.

topic_id: str#

class roboto.MessagePathStatistic#

Bases: enum.Enum

Statistics computed by Roboto in our standard ingestion actions.

Count = 'count'#

Max = 'max'#

Mean = 'mean'#

Median = 'median'#

Min = 'min'#

class roboto.QueryDatasetFilesRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for querying files within a dataset.

Used to retrieve files from a dataset with optional pattern-based filtering and pagination support. Supports gitignore-style patterns for flexible file selection.

Parameters:: data (Any)

exclude_patterns: list[str] | None = None#: List of gitignore-style patterns for files to exclude from results.

include_patterns: list[str] | None = None#: List of gitignore-style patterns for files to include in results.

page_token: str | None = None#: Token for retrieving the next page of results in paginated queries.

class roboto.QueryDatasetsRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for querying datasets with filters.

Used to search for datasets based on various criteria such as metadata, tags, and other dataset properties. The filters are applied server-side to efficiently return matching datasets.

Parameters:: data (Any)

filters: dict[str, Any] = None#: Dictionary of filter criteria to apply when searching for datasets.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.QueryFilesRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for querying files with filters.

Used to search for files based on various criteria such as metadata, tags, ingestion status, and other file properties. The filters are applied server-side to efficiently return matching files.

Parameters:: data (Any)

filters: dict[str, Any] = None#: Dictionary of filter criteria to apply when searching for files.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.QueryTriggersRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to query triggers with filters.

Used to search for triggers based on various criteria such as name, status, or other attributes.

Parameters:: data (Any)

filters: dict[str, Any] = None#: Dictionary of filter criteria to apply to the trigger search.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.ReportTransactionProgressRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for reporting file upload transaction progress.

Used to notify the platform about the completion status of individual files within a batch upload transaction. This enables progress tracking and partial completion handling for large file uploads.

Parameters:: data (Any)

manifest_items: list[str]#: List of manifest item identifiers that have completed upload.

class roboto.RepresentationRecord(/, **data)#

Bases: pydantic.BaseModel

Record representing a data representation for topic content.

A representation is a pointer to processed topic data stored in a specific format and location. Representations enable efficient access to topic data by providing multiple storage formats optimized for different use cases.

Most message paths within a topic point to the same representation (e.g., an MCAP or Parquet file containing all topic data). However, some message paths may have multiple representations for analytics or preview formats.

Representations are versioned and associated with specific files or storage locations through the association field.

Parameters:: data (Any)

association: roboto.association.Association#: Identifier and entity type with which this Representation is associated. E.g., a file, a database.

created: datetime.datetime#

modified: datetime.datetime#

representation_id: str#

storage_format: RepresentationStorageFormat#

topic_id: str#

version: int#

class roboto.RepresentationStorageFormat#

Bases: enum.Enum

Supported storage formats for topic data representations.

Defines the available formats for storing and accessing topic data within the Roboto platform. Each format has different characteristics and use cases.

MCAP = 'mcap'#: MCAP format - optimized for robotics time-series data with efficient random access.

PARQUET = 'parquet'#: Parquet format - columnar storage optimized for analytics and large-scale data processing.

class roboto.RobotoClient(endpoint, auth_decorator, http_client_kwargs=None)#

A client for making HTTP requests against Roboto service

Parameters:

endpoint (str)
auth_decorator (Optional[roboto.http.request.HttpRequestDecorator])
http_client_kwargs (Optional[dict[str, Any]])

classmethod defaulted(client=None)#

Parameters:: client (Optional[RobotoClient])
Return type:: RobotoClient

delete(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])

Return type:

property endpoint: str#

Return type:: str

classmethod from_config(config)#

Parameters:: config (roboto.config.RobotoConfig)
Return type:: RobotoClient

classmethod from_env()#

Return type:: RobotoClient

property frontend_endpoint: str#

Return type:: str

get(path, caller_org_id=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])

Return type:

property http_client: roboto.http.http_client.HttpClient#

Return type:: roboto.http.http_client.HttpClient

patch(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])

Return type:

post(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])

Return type:

put(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])

Return type:

roboto.domain.topics.record.MessagePathRecord

class roboto.RobotoConfig(/, **data)#

Bases: pydantic.BaseModel

RobotoConfig captures an api_key and endpoint required to programmatically interact with Roboto. Multiple profiles can be configured if desired.

Parameters:: data (Any)

api_key: str#

endpoint: str = 'https://api.roboto.ai'#

classmethod from_env(profile_override=None)#

Parameters:: profile_override (Optional[str])
Return type:: RobotoConfig

class roboto.RobotoEnv(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_file=ENV_FILE_SENTINEL, _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, **values)#

Bases: pydantic_settings.BaseSettings

Enivronment within an action invocation runtime

Parameters:

_case_sensitive (bool | None)
_nested_model_default_partial_update (bool | None)
_env_prefix (str | None)
_env_file (pydantic_settings.sources.DotenvType | None)
_env_file_encoding (str | None)
_env_ignore_empty (bool | None)
_env_nested_delimiter (str | None)
_env_nested_max_split (int | None)
_env_parse_none_str (str | None)
_env_parse_enums (bool | None)
_cli_prog_name (str | None)
_cli_parse_args (bool | list[str] | tuple[str, Ellipsis] | None)
_cli_settings_source (pydantic_settings.sources.CliSettingsSource[Any] | None)
_cli_parse_none_str (str | None)
_cli_hide_none_type (bool | None)
_cli_avoid_json (bool | None)
_cli_enforce_required (bool | None)
_cli_use_class_docs_for_groups (bool | None)
_cli_exit_on_error (bool | None)
_cli_prefix (str | None)
_cli_flag_prefix_char (str | None)
_cli_implicit_flags (bool | None)
_cli_ignore_unknown_args (bool | None)
_cli_kebab_case (bool | None)
_cli_shortcuts (collections.abc.Mapping[str, str | list[str]] | None)
_secrets_dir (pydantic_settings.sources.PathType | None)
values (Any)

action_inputs_manifest_file: pathlib.Path | None = None#

action_parameters_file: str | None = None#

action_timeout: str | None = None#

api_key: str | None = None#

config_file: str | None = None#: An override for the location of a roboto config file. If not provided, the default ~/.roboto/config.json will be used (subject to RobotoConfig’s implementation)

dataset_id: str | None = None#

dataset_metadata_changeset_file: str | None = None#

classmethod default()#

Return type:: RobotoEnv

file_metadata_changeset_file: str | None = None#

get_env_var(var_name, default_value=None)#

Parameters:

var_name (str)
default_value (Optional[str])

Return type:

Optional[str]

input_dir: str | None = None#

invocation_id: str | None = None#

org_id: str | None = None#

output_dir: str | None = None#

profile: str | None = None#: The profile name to use if getting RobotoConfig from a config file.

roboto_env: str | None = None#

roboto_service_endpoint: str | None = None#

//api.roboto.ai

Type:: A Roboto Service API endpoint to send requests to, typically https

roboto_service_url: str | None = None#: Deprecated, use roboto_service_endpoint instead. Left here until 0.3.3 is released so we can migrate existing actions to use the new env var.

class roboto.RobotoRegion#

Bases: str, enum.Enum

The geographic region of a Roboto resource. Used when configuring org-level default behavior for data storage, in order to ensure that data is close to your users.

EU_CENTRAL = 'eu-central'#

US_EAST = 'us-east'#

US_GOV_WEST = 'us-gov-west'#

US_WEST = 'us-west'#

class roboto.RobotoSearch(query_client=None)#

A high-level interface for querying the Roboto data platform.

In most cases, using this class should be as simple as:

>>> from roboto import RobotoSearch
>>> robosearch = RobotoSearch()
>>> for dataset in robosearch.find_datasets(...):
...     ...

Parameters:: query_client (Optional[roboto.query.QueryClient])

find_collections(query=None, timeout_seconds=math.inf)#

Parameters:

query (Optional[roboto.query.Query])
timeout_seconds (float)

Return type:

collections.abc.Generator[roboto.domain.collections.Collection, None, None]

find_datasets(query=None, timeout_seconds=math.inf)#

Parameters:

query (Optional[roboto.query.Query])
timeout_seconds (float)

Return type:

collections.abc.Generator[roboto.domain.datasets.Dataset, None, None]

find_events(query=None, timeout_seconds=math.inf)#

Parameters:

query (Optional[roboto.query.Query])
timeout_seconds (float)

Return type:

collections.abc.Generator[roboto.domain.events.Event]

find_files(query=None, timeout_seconds=math.inf)#

Parameters:

query (Optional[roboto.query.Query])
timeout_seconds (float)

Return type:

collections.abc.Generator[roboto.domain.files.File, None, None]

find_message_paths(query=None, timeout_seconds=math.inf)#

Parameters:

query (Optional[roboto.query.Query])
timeout_seconds (float)

Return type:

collections.abc.Generator[roboto.domain.topics.MessagePath, None, None]

find_topics(query=None, timeout_seconds=math.inf)#

Examples

>>> import matplotlib.pyplot as plt
>>> import pandas as pd
>>> from roboto import RobotoSearch
>>> robosearch = RobotoSearch()
>>> for topic in robosearch.find_topics("msgpaths[cpuload.load].max > 0.9"):
...     topic_data = list(topic.get_data())
...     df = pd.json_normalize(topic_data)
... plt.plot(df["log_time"], df["load"], label=f"{topic.topic_id}")
...
>>> plt.legend()
>>> plt.show()

Parameters:

query (Optional[roboto.query.Query])
timeout_seconds (float)

Return type:

collections.abc.Generator[roboto.domain.topics.Topic, None, None]

class roboto.S3Credentials#

Bases: TypedDict

This interface is driven by botocore.credentials.RefreshableCredentials

access_key: str#

expiry_time: str | None#

region: str#

secret_key: str#

token: str#

class roboto.SetActionAccessibilityRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to set action accessibility.

Used to change whether an action is private to the organization or published publicly in the Action Hub.

Parameters:: data (Any)

accessibility: roboto.domain.actions.action_record.Accessibility#: The new accessibility level (Organization or ActionHub).

digest: str | None = None#: Specific version of Action. If not specified, the latest version’s accessibility will be updated.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.SetContainerInfoRequest(/, **data)#

Bases: pydantic.BaseModel

Request to set container information for an invocation.

Used internally by the Roboto platform to record container details after the action image has been pulled and inspected.

Note

This is typically called from the monitor process and is not intended for direct use by SDK users.

Parameters:: data (Any)

image_digest: str#: The digest of the container image that was pulled.

class roboto.SetDefaultRepresentationRequest(/, **data)#

Bases: BaseAddRepresentationRequest

Request to set the default representation for a topic.

Designates a specific representation as the default for accessing topic data. The default representation is used when no specific representation is requested for data access operations.

Parameters:: data (Any)

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.SetLogsLocationRequest(/, **data)#

Bases: pydantic.BaseModel

Request to set the location where invocation logs are stored.

Used internally by the Roboto platform to record where log files are saved for later retrieval.

Note

This is typically called from the invocation-scheduling-service and is not intended for direct use by SDK users.

Parameters:: data (Any)

bucket: str#: S3 bucket name where logs are stored.

prefix: str#: S3 key prefix for the log files.

class roboto.SourceProvenance(/, **data)#

Bases: pydantic.BaseModel

Provenance information for an invocation source

Parameters:: data (Any)

source_id: str#

source_type: InvocationSource#

class roboto.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:

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

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:

RobotoConflictException – Message path already exists for this topic.
RobotoUnauthorizedException – Caller lacks permission to modify the topic.

Return type:

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:

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.

Returns:

RepresentationRecord representing the newly created representation.

Raises:

RobotoNotFoundException – Message path with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to modify the topic.

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:

RobotoInvalidRequestException – Invalid topic parameters.
RobotoUnauthorizedException – Caller lacks permission to create topics.

Return type:

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:

RobotoNotFoundException – Topic does not exist or has already been deleted.
RobotoUnauthorizedException – Caller lacks permission to delete the topic.

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:

RobotoNotFoundException – Topic with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the topic.

Return type:

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:

RobotoNotFoundException – Topic with the given name does not exist in the specified file.
RobotoUnauthorizedException – Caller lacks permission to access the topic.

Return type:

roboto.domain.topics.record.RepresentationRecord

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:

RobotoNotFoundException – Dataset with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the dataset.

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:

RobotoNotFoundException – File with the given ID does not exist.
RobotoUnauthorizedException – Caller lacks permission to access the file.

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:

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.

Returns:

RepresentationRecord representing the newly set default representation.

Raises:

RobotoNotFoundException – Specified representation does not exist.
RobotoUnauthorizedException – Caller lacks permission to modify the topic.

Return type:

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:

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:

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: