roboto.domain.files#

Submodules#

Package Contents#

class roboto.domain.files.CreateDirectoryRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to create a directory among the files of one association.

Parameters:: data (Any)

create_intermediate_dirs: bool = False#: If True, creates intermediate directories in the path if they don’t exist. If False, requires all parent directories to already exist.

error_if_exists: bool = False#

name: str#

origination: str | None = None#

parent_path: str | None = None#

class roboto.domain.files.CreateLinkRequest(/, **data)#

Bases: pydantic.BaseModel

Request body for PUT /v1/files/association/id/<association_id>/link.

Parameters:: data (Any)

relative_path: str#: Where the link sits among the association’s files. Missing parent directories are created.

target_file_id: str#: ID of the file the link points at. It must be a file, not a link or a directory, in the same org.

target_version: int | None = None#: Version of the target to pin. Defaults to the target’s current version.

class roboto.domain.files.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.domain.files.DirectoryContentsPage(/, **data)#

Bases: pydantic.BaseModel

Response containing the contents of a dataset directory page.

Represents a paginated view of files and subdirectories within a dataset directory. Used when browsing dataset contents hierarchically.

Parameters:: data (Any)

directories: collections.abc.Sequence[roboto.domain.files.record.DirectoryRecord]#: Subdirectories contained in this directory page.

files: collections.abc.Sequence[roboto.domain.files.record.FileRecord]#: Files contained in this directory page.

next_token: str | None = None#: Token for retrieving the next page of results, if any.

class roboto.domain.files.DirectoryRecord(/, **data)#

Bases: pydantic.BaseModel

Wire-transmissible representation of a directory within a dataset.

DirectoryRecord represents a logical directory structure within a dataset, containing metadata about the directory’s location and contents. Directories are used to organize files hierarchically within datasets.

Directory records are typically returned when browsing dataset contents or when performing directory-based operations like bulk deletion.

Parameters:: data (Any)

association_id: str#

created: datetime.datetime#

created_by: str#

description: str | None = None#

directory_id: str#

fs_type: FSType#

metadata: dict[str, Any] = None#

modified: datetime.datetime#

modified_by: str#

name: str#: Name of the directory (the final component of the path).

org_id: str#

origination: str#

parent_id: str | None = None#

relative_path: str#

status: FileStatus#

storage_type: FileStorageType#

tags: list[str] = None#

upload_id: str#

class roboto.domain.files.FSType#

Bases: roboto.compat.StrEnum

File system type enum

Directory = 'directory'#

File = 'file'#

Link = 'link'#

A pointer to one version of another file, which may live under a different dataset, device, or org.

A link stores no object of its own. Its record’s uri is roboto://file/<target_file_id>?v=<version> and its size is 0; downloading it fetches the target at that version.

class roboto.domain.files.File(record, roboto_client=None, file_service=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])
file_service (Optional[roboto.storage.FileService])

add_topic(topic_name, df, timestamp_column=None, timestamp_unit=None)#

Create a Topic from a pandas DataFrame and associate it with this file.

If a topic with the same name already exists for this file, it will be updated with the new data and schema.

Parameters:

topic_name (str) – Name for the topic. Must be unique within this file.
df (pandas.DataFrame) – pandas DataFrame containing the data to ingest. Must include a timestamp column (either explicitly specified or automatically detectable).
timestamp_column (Optional[str]) – Name of the column to use as the timestamp. If not provided, the method will attempt to automatically detect a timestamp column by looking for the first column that is a timezone-aware timestamp type.
timestamp_unit (Optional[Union[str, roboto.time.TimeUnit]]) – Unit of the timestamp column values. Required when timestamp_column contains numeric values (int, float, decimal). Valid values include “s”, “ms”, “us”, “ns”. Not needed for datetime columns or when timestamp_column is not specified.

Returns:

The created or updated Topic instance.

Raises:

IngestionException – If the timestamp column cannot be determined, is not present in the DataFrame, has an invalid type, or if the timestamp unit is required but not provided.
ImportError – If pandas or pyarrow are not installed. Install with pip install roboto[ingestion] to use this feature.
RobotoIllegalArgumentException – This file is associated with a device or with the org itself, not with a dataset; only a dataset’s files hold topics.
RobotoUnauthorizedException – If the caller lacks permission to create topics or upload files to this file’s dataset.

Return type:

roboto.domain.topics.Topic

Notes

Requires installing this package using the roboto[ingestion] extra
Topic names are unique within a file
Schema and statistics are automatically inferred from the DataFrame

Examples

Create a topic with explicit timestamp column and unit:

>>> import pandas as pd
>>> from roboto import File
>>> file = File.from_id("file_abc123")
>>> df = pd.DataFrame(
...     {
...         "timestamp": [1763947309.4198897, 1763947316.7686195, 1763947335.0095527],
...         "temperature": [20.5, 21.0, 20.8],
...         "humidity": [45.2, 46.1, 45.8],
...     }
... )
>>> topic = file.add_topic(
...     topic_name="sensor_data", df=df, timestamp_column="timestamp", timestamp_unit="s"
... )
>>> print(f"Created topic: {topic.name}")
Created topic: sensor_data

Create a topic with automatic timestamp detection:

>>> import pandas as pd
>>> from roboto import File
>>> file = File.from_id("file_abc123")
>>> df = pd.DataFrame(
...     {
...         "ts": pd.date_range("2025-11-24", periods=3, freq="1s", tz="UTC"),
...         "velocity": [10.5, 11.2, 10.8],
...         "acceleration": [0.5, 0.3, -0.2],
...     }
... )
>>> topic = file.add_topic("motion_data", df)

Retrieve the data back

>>> retrieved_df = topic.get_data_as_df()
>>> print(f"Retrieved {len(retrieved_df)} rows")
Retrieved 3 rows

Add derived data as a new topic to the same file, using the original topic’s timestamp index:

>>> import pandas as pd
>>> from roboto import File
>>> file = File.from_id("file_abc123")
>>> # Get existing topic data as DataFrame
>>> original_topic = file.get_topic("sensor_data")
>>> original_df = original_topic.get_data_as_df()
>>> # Create derived data
>>> derived_df = pd.DataFrame(
...     {
...         "temp_category": original_df["temperature"].apply(lambda x: "hot" if x > 25 else "not_hot"),
...     },
...     index=original_df.index,
... )
>>> derived_topic = file.add_topic(
...     "temperature_categories",
...     derived_df,
... )

property association: roboto.association.Association#

The dataset, device, or org this file is associated with.

Every file has exactly one association, inferred from the prefix of its association ID. Read file.association.association_type to branch on it.

Return type:: roboto.association.Association

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.

Valid only for a file associated with a dataset; files associated with a device or with the org itself have no dataset. Prefer association, which works for every file.

Raises:: RobotoIllegalArgumentException – This file is not associated with a 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]

property device_id: str | None#

Identifier of the device that generated this data.

Returns the optional identifier of the device that generated the data contained within this file. Can be None if the file was not generated by a device.

Return type:: Optional[str]

download(local_path, print_progress=True)#

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.

For a link, downloads the version of the target file that the link pins.

Parameters:

local_path (pathlib.Path) – Local filesystem path where the file should be saved.
print_progress (bool) – Whether to show a progress bar during download.

Raises:

RobotoNotFoundException – This file is a link whose target, at the pinned version, no longer exists.
RobotoUnauthorizedException – Caller lacks permission to download the file, or a link’s target.
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}")

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:

File

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:

File

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'

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

Return type:

str

For a link, the URL is for the version of the target file that the link pins.

Returns:

Signed URL string that provides temporary access to the file.

Raises:

RobotoNotFoundException – This file is a link whose target, at the pinned version, no longer exists.
RobotoUnauthorizedException – Caller lacks permission to access the file, or a link’s target.

Parameters:

override_content_type (Optional[str])
override_content_disposition (Optional[str])

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

roboto.domain.topics.Topic

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, device_id=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.
device_id (Optional[str]) – Optional identifier of the device that generated this data.
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:

File

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

property is_link: bool#

Whether this file is a link to one version of another file.

A link sits at its own path under its own dataset, device, or org, and stores no object. download() and get_signed_url() fetch the target at the version the link pins.

Return type:: bool

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:

File

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 the root of its association’s files.

Uses forward slashes as separators regardless of the operating system. This path uniquely identifies the file among the files of its dataset, device, or org.

Return type:: str

rename_file(file_id, new_path)#

Rename this file to a new path within its dataset, device, or org.

Changes the relative path of the file among the files of its association. 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, relative to the root of its association’s files.

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:

roboto.domain.files.record.FileRecord

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'

set_device_id(device_id)#

Set the device ID for this file.

Parameters:

device_id (str) – The device ID to set for this file.

Returns:

Updated File instance with the new device ID.

Raises:

RobotoUnauthorizedException – Caller lacks permission to update the file.
RobotoDeviceNotFoundException – The specified device ID does not exist.

Return type:

File

Examples

>>> file = File.from_id("file_abc123")
>>> updated_file = file.set_device_id("device_xyz789")

set_timeline_offset(unix_epoch_offset_ns, *, topic=None, topic_name=None, timeline_source=None, timeline_source_name=None)#

Calibrate this file’s timeline to Unix-epoch wall-clock, optionally scoped to a topic and/or source.

Contract:

The offset is added to stored partition time to produce session wall-clock: session_time_ns = stored_time_ns + unix_epoch_offset_ns.
topic / topic_name scopes the update to a single topic in this file; timeline_source / timeline_source_name scopes it to a single source. With no selectors, the offset applies to every timeline on the file.

Use set_timeline_offsets() to send several offsets in one atomic request.

Parameters:

unix_epoch_offset_ns (int) – Offset to apply, in nanoseconds.
topic (Optional[roboto.domain.topics.Topic]) – Topic to scope the update to. Mutually exclusive with topic_name.
topic_name (Optional[str]) – Topic name to scope the update to (e.g. "/imu/raw"). Mutually exclusive with topic.
timeline_source (Optional[roboto.domain.topics.TimelineSourceRecord]) – Source record to scope the update to. Mutually exclusive with timeline_source_name.
timeline_source_name (Optional[str]) – Source name to scope the update to (e.g. "header.stamp"). Mutually exclusive with timeline_source.

Returns:

The updated TimelineExtentRecord objects returned by the server.

Return type:

list[roboto.domain.topics.TimelineExtentRecord]

Examples

Apply a file-wide offset:

>>> file = File.from_id("file_abc123")
>>> file.set_timeline_offset(1_700_000_000_000_000_000)

Apply an offset to a single topic by name:

>>> file.set_timeline_offset(1_700_000_000_000_000_000, topic_name="/imu/raw")

Apply an offset to a specific source on a topic:

>>> file.set_timeline_offset(
...     500_000_000,
...     topic_name="data",
...     timeline_source_name="ts",
... )

set_timeline_offsets(offsets)#

Apply multiple timeline offsets to this file in one atomic request.

Each entry carries a unix_epoch_offset_ns and optional selectors (topic_name, timeline_source_id, timeline_source_name) that narrow where the offset is applied. An entry with no selectors targets every timeline on the file.

Use set_timeline_offset() for the single-offset convenience form.

Parameters:: offsets (list[roboto.domain.topics.TimelineOffsetEntry]) – Offset entries to apply, each with its own selectors.
Returns:: The updated TimelineExtentRecord objects returned by the server.
Return type:: list[roboto.domain.topics.TimelineExtentRecord]

Examples

Apply per-topic offsets in a single request:

>>> from roboto.domain.topics import TimelineOffsetEntry
>>> file = File.from_id("file_abc123")
>>> file.set_timeline_offsets(
...     [
...         TimelineOffsetEntry(unix_epoch_offset_ns=1_700_000_000_000_000_000, topic_name="/imu/raw"),
...         TimelineOffsetEntry(unix_epoch_offset_ns=1_700_000_000_000_000_000, topic_name="/camera/image"),
...     ]
... )

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, device_id=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.
device_id (Optional[Union[str, roboto.sentinels.NotSetType]]) – New device ID for the file. Use NotSet to leave unchanged.

Returns:

Updated File instance with the new properties.

Raises:

RobotoUnauthorizedException – Caller lacks permission to update the file.

Return type:

File

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.domain.files.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#

Name of the bucket holding this file’s object.

Raises:: RobotoIllegalArgumentException – This record is a link, which stores no object.
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 is_link: bool#

Whether this record is a link to another file rather than a file with an object of its own.

Return type:: bool

property key: str#

Key of this file’s object within bucket.

Raises:: RobotoIllegalArgumentException – This record is a link, which stores no object.
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.domain.files.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.domain.files.FileStatus#

Bases: roboto.compat.StrEnum

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.domain.files.FileStorageType#

Bases: roboto.compat.StrEnum

Enumeration of file storage types in the Roboto platform.

Storage type indicates how the file was added to the platform and affects access patterns and permissions. This information is used internally for credential management and access control.

S3Directory = 'directory'#: This node is a virtual directory.

S3Imported = 'imported'#

File was imported from a read-only customer-managed S3 bucket.

These files remain in the customer’s bucket and are accessed using customer-provided credentials. The customer retains full control over the file storage and access permissions.

S3Uploaded = 'uploaded'#

File was uploaded to a Roboto-managed or customer read/write bucket.

These files were explicitly uploaded through the Roboto platform to either a Roboto-managed bucket or a customer’s bring-your-own read/write bucket. Access is managed through Roboto’s credential system.

class roboto.domain.files.FileSystem(association, roboto_client=None, file_service=None, org_id=None)#

The files and directories under one association: a dataset, a device, or the org itself.

A FileSystem holds that association’s directory tree and the operations on it: listing, uploading, downloading, renaming, and deleting files, and creating and renaming directories. Datasets, devices, and orgs each return one as files, files, and files.

Every relative path a method takes or returns is relative to the root of the association’s tree, which files of other associations do not share.

Parameters:

association (roboto.association.Association)
roboto_client (Optional[roboto.http.RobotoClient])
file_service (Optional[roboto.storage.FileService])
org_id (Optional[str])

property association: roboto.association.Association#

The dataset, device, or org whose files this object works on.

Return type:: roboto.association.Association

create_directory(name, error_if_exists=False, create_intermediate_dirs=False, parent_path=None, origination=None)#

Create a directory among the association’s files.

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 at the root of the association’s files.
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:

roboto.domain.files.record.DirectoryRecord

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> directory = device.files.create_directory("calib")
>>> print(directory.relative_path)
calib

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

create_link(target, relative_path)#

Put a link to another file at relative_path among the association’s files.

A link lets one file, such as a URDF in the org’s own files, appear in many devices’ files without being copied. It pins one version of its target: passing a File pins that file’s version, and passing a file ID pins the target’s current version. Later versions of the target do not move the link; create the link again at the same path to re-point it, which adds a version to the link unless it already pins that target version. Missing parent directories are created. Downloading the link, or asking it for a signed URL, fetches the pinned version of the target.

Parameters:

target (Union[roboto.domain.files.file.File, str]) – The file to link to, or its ID. It must be a file in the same org, not a link or a directory.
relative_path (str) – Where the link sits, relative to the root of the association’s files.

Returns:

The link, whose is_link is True.

Raises:

RobotoConflictException – A file or a directory already occupies relative_path. The reverse is refused too: uploading a file to a link’s path is a conflict until the link is deleted.
RobotoInvalidRequestException – The target is a link or a directory, is in another org, or does not exist at the version to pin.
RobotoUnauthorizedException – The caller cannot edit the association’s files or view the target.

Return type:

roboto.domain.files.file.File

Examples

>>> from roboto.domain import devices, orgs
>>> urdf = orgs.Org.from_id("og_abc123").files.get_file_by_path("urdf/lemi/lemi.urdf")
>>> device = devices.Device.from_id("lemi-01")
>>> link = device.files.create_link(urdf, "urdf/lemi.urdf")
>>> link.download(pathlib.Path("/tmp/lemi.urdf"))

delete_files(include_patterns=None, exclude_patterns=None)#

Delete the association’s files that match the given patterns.

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 or empty, all files are considered for deletion. An empty list is treated as no filter (all files), not as “include nothing”.
exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude from deletion. Takes precedence over include patterns. If None or empty, no files are excluded.

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

>>> from roboto.domain import orgs
>>> org = orgs.Org.from_id("og_abc123")
>>> org.files.delete_files(include_patterns=["**/*.png"], exclude_patterns=["**/back_camera/**"])

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

Download the association’s files to a local directory.

Downloads files that match the specified patterns to the given local directory. The files’ directory structure is preserved in the download location. If the output directory doesn’t exist, it will be created. Files are found with list_files(), which does not return links yet, so no link is downloaded; download one with download().

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 or empty, all files are downloaded. An empty list is treated as no filter (all files), not as “include nothing”.
exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude from download. Takes precedence over include patterns. If None or empty, no files are excluded.
print_progress (bool) – Whether to show a progress bar during download.

Returns:

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

Raises:

RobotoIllegalArgumentException – A selected file’s path resolves outside out_path; nothing is downloaded.
RobotoUnauthorizedException – Caller lacks permission to download files.

Return type:

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

Notes

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

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> downloaded = device.files.download_files(pathlib.Path("/tmp/lemi-01"), include_patterns=["calib/**"])
>>> print(f"Downloaded {len(downloaded)} files")
Downloaded 2 files

get_file_by_path(relative_path, version_id=None)#

Get a File instance for the association’s file at the specified path.

Parameters:

relative_path (Union[str, pathlib.Path]) – Path of the file relative to the root of the association’s files.
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 – The association has no file at the given path.
RobotoUnauthorizedException – Caller lacks permission to access the file.

Return type:

roboto.domain.files.file.File

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> file = device.files.get_file_by_path("manifest.json")
>>> print(file.file_id)
fl_xyz789

>>> old_file = device.files.get_file_by_path("manifest.json", version_id=1)
>>> print(old_file.version)
1

list_directories()#

Yield every directory among the association’s files, at any depth.

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> for directory in device.files.list_directories():
...     print(directory.relative_path)
calib
urdf

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

list_files(include_patterns=None, exclude_patterns=None)#

List the association’s files with optional pattern-based filtering.

Returns all of the association’s 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 or empty, all files are considered. An empty list is treated as no filter (all files), not as “include nothing”.
exclude_patterns (Optional[list[str]]) – List of gitignore-style patterns for files to exclude. Takes precedence over include patterns. If None or empty, no files are excluded.

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.File, None, None]

Notes

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

Files appear in this list shortly after their upload completes, not instantly.

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> for file in device.files.list_files():
...     print(file.relative_path)
manifest.json
calib/front_cam.yaml

>>> for file in device.files.list_files(include_patterns=["calib/**"], exclude_patterns=["**/*.bak"]):
...     print(file.relative_path)
calib/front_cam.yaml

rename_directory(old_path, new_path)#

Rename or move a directory among the association’s files.

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

Parameters:

old_path (str) – Current relative path of the directory (e.g. "logs/session1").
new_path (str) – Target relative path of the directory (e.g. "session1" to move up one level).

Returns:

Updated DirectoryRecord reflecting the new path.

Raises:

RobotoNotFoundException – No directory exists at old_path.
RobotoInvalidRequestException – new_path conflicts with an existing node or contains a cycle.

Return type:

roboto.domain.files.record.DirectoryRecord

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> device.files.rename_directory("calib/front", "front_calib")

rename_file(file_id, new_path)#

Rename or move a file among the association’s files.

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

The file’s storage URI is unchanged; only its relative path changes.

Parameters:

file_id (str) – ID of the file to rename or move.
new_path (str) – Target relative path for the file (e.g. "file.bag" to move to the root, or "other_dir/file.bag" to move into an existing directory).

Returns:

Updated FileRecord reflecting the new path.

Raises:

RobotoNotFoundException – No file with file_id exists.
RobotoInvalidRequestException – new_path conflicts with an existing file, the parent directory does not exist, or the move would create a cycle.

Return type:

roboto.domain.files.record.FileRecord

Examples

>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> record = device.files.rename_file("fl_xyz789", "manifest.json")
>>> record.relative_path
'manifest.json'

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

Upload all files and directories recursively from the specified directory path.

Use include_patterns and exclude_patterns to control what files and directories are uploaded, and delete_after_upload to clean up your local filesystem after the uploads succeed.

Parameters:

directory_path (pathlib.Path) – Local directory whose contents are uploaded, keeping its layout.
include_patterns (Optional[list[str]]) – gitignore-style patterns for files to include. If None, every file is included.
exclude_patterns (Optional[list[str]]) – gitignore-style patterns for files to exclude. Takes precedence over include_patterns.
delete_after_upload (bool) – If True, each uploaded local file is deleted once the uploads succeed.
max_batch_size (int) – Maximum number of files per upload transaction.
print_progress (bool) – Whether to display an upload progress bar.
device_id (Optional[str]) – Optional identifier of the device that generated this data.

Return type:

None

Notes

Both pattern lists follow the gitignore pattern format described in https://git-scm.com/docs/gitignore#_pattern_format.

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> device.files.upload_directory(
...     pathlib.Path("/path/to/calibration"),
...     exclude_patterns=["**/*.log"],
... )

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

Upload a single file associated with association.

Parameters:

file_path (pathlib.Path) – Local file to upload.
file_destination_path (Optional[str]) – Destination path among the association’s files. Defaults to the file’s own name at the root.
print_progress (bool) – Whether to display an upload progress bar.
device_id (Optional[str]) – Optional identifier of the device that generated this data.

Returns:

The uploaded file. Its record is fetched from the platform the first time it is read.

Raises:

RobotoInternalException – The upload reported success without reporting a file ID.

Return type:

roboto.domain.files.file.File

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> device.files.upload_file(pathlib.Path("/path/to/manifest.json"))

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

Upload multiple files associated with association.

Parameters:

files (collections.abc.Iterable[pathlib.Path]) – Local files to upload.
file_destination_paths (collections.abc.Mapping[pathlib.Path, str]) – Mapping from local path to destination path among the association’s files. Files not in the mapping upload to the root under their own name.
max_batch_size (int) – Maximum number of files per upload transaction.
print_progress (bool) – Whether to display an upload progress bar.
device_id (Optional[str]) – Optional identifier of the device that generated this data.

Returns:

Mapping from each uploaded local path to the ID of the file record it created.

Return type:

dict[pathlib.Path, str]

Examples

>>> import pathlib
>>> from roboto.domain import devices
>>> device = devices.Device.from_id("lemi-01")
>>> file_ids = device.files.upload_files(
...     [pathlib.Path("/path/to/front_cam.yaml")],
...     file_destination_paths={pathlib.Path("/path/to/front_cam.yaml"): "calib/front_cam.yaml"},
... )
>>> file_ids[pathlib.Path("/path/to/front_cam.yaml")]
'fl_0123456789abcdef'

class roboto.domain.files.FileTag(*args, **kwds)#

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.

AssociationId = 'association_id'#: Tag containing the ID of the dataset, device, or org a file is associated with.

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.

Deprecated in favour of AssociationId, which names a file’s dataset, device, or org alike. The platform still sets it on its own server-side copies of dataset files.

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

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

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.domain.files.IngestionStatus#

Bases: roboto.compat.StrEnum

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.domain.files.LazyLookupFile(hydrate_fn)#

Bases: roboto.domain.files.file.File

A File subclass that defers instantiation (hydration) of the real File until any non‐internal attribute is first accessed.

This is useful for scenarios where we know how to dereference a File (e.g., by ID), and we want to return a handle in case the caller wants to work with it, but we don’t want to pay the cost of dereferencing it unless necessary.

Parameters:: hydrate_fn (Callable[[], roboto.domain.files.file.File])

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

Bases: pydantic.BaseModel

Request payload for listing the files associated with a dataset, an org, or a device.

Supports gitignore-style patterns for flexible file selection and pagination. Despite the name, the same body lists the files of any association type.

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.

limit: int | None = None#: Maximum number of files to return per page.

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

sort_by: str | None = None#: Field to sort results by. Defaults to ‘created’.

sort_direction: str | None = None#: Sort direction (‘ASC’ or ‘DESC’). Defaults to ‘DESC’.

class roboto.domain.files.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.domain.files.RenameDirectoryRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for renaming a directory among the files of one association.

Changes the path of a directory and all its contained files. This updates the logical organization without moving actual file content.

Parameters:: data (Any)

new_path: str#: New path for the directory.

old_path: str#: Current path of the directory to rename.

class roboto.domain.files.RenameFileRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for renaming a file within its dataset.

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

Parameters:: data (Any)

association_id: str#: ID of the dataset containing the file to rename.

new_path: str#: New relative path for the file within the dataset.

class roboto.domain.files.SignedUrlResponse(/, **data)#

Bases: pydantic.BaseModel

Response containing a signed URL for direct file access.

Provides a time-limited URL that allows direct access to file content without requiring Roboto authentication. Used for file downloads and integration with external systems.

Parameters:: data (Any)

url: str#: Signed URL that provides temporary direct access to the file.

class roboto.domain.files.UpdateFileRecordRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for updating file record properties.

Used to modify file metadata, description, and ingestion status. Only specified fields are updated; others remain unchanged. Uses NotSet sentinel values to distinguish between explicit None values and fields that should not be modified.

Parameters:: data (Any)

description: str | roboto.sentinels.NotSetType | None#: New description for the file, or NotSet to leave unchanged.

device_id: str | roboto.sentinels.NotSetType | None#: New device ID for the file, or NotSet to leave unchanged.

ingestion_complete: Literal[True] | roboto.sentinels.NotSetType#: Set to True to mark file as fully ingested, or NotSet to leave unchanged.

metadata_changeset: roboto.updates.MetadataChangeset | roboto.sentinels.NotSetType#: Metadata changes to apply (add, update, or remove fields/tags), or NotSet to leave unchanged.

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

roboto.domain.files.is_directory(record)#

Parameters:: record (Union[FileRecord, DirectoryRecord])
Return type:: TypeGuard[DirectoryRecord]

roboto.domain.files.is_file(record)#

Parameters:: record (Union[FileRecord, DirectoryRecord])
Return type:: TypeGuard[FileRecord]