roboto.domain.files

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.

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

Usage

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

Register one topic this File contributes data to, without naming a Session.

The singular form of declare_topics(), taking the fields of one FileTopicDeclaration as separate arguments. That class documents what each field means; declare_topics() documents what the platform does with it.

Usage

from roboto.domain.files import File
from roboto.domain.topics import CanonicalDataType, RepresentationStorageFormat
from roboto.experimental.ingest import Field, RepresentationDeclaration, Schema, SchemaFieldSource
timestamp = Field(
    name="timestamp",
    data_type="float64",
    canonical_data_type=CanonicalDataType.Timestamp,
    unit="s",
)
file = File.from_id("fl_0123456789ab")
topic = file.declare_topic(
    topic_name="observation.state",
    topic_schema=Schema(
        name="observation.state",
        fields=[timestamp, Field(name="observation.state", data_type="float32")],
    ),
    timeline_sources=[
        SchemaFieldSource(
            field_path=["timestamp"],
            min_file_timestamp_ns=0,
            max_file_timestamp_ns=4_000,
        )
    ],
    representations=[
        RepresentationDeclaration(
            file_id=file.file_id,
            storage_format=RepresentationStorageFormat.PARQUET,
        )
    ],
)
print(topic.topic_id)

Register the topic data this File carries, without naming a Session.

One call states everything the platform needs to serve this File’s topic data: for each topic, the structure of its rows, the timeline sources those rows carry with the bounds they span in this File, and, when the File packs its data into slices, which slice the topic occupies. The platform does not open the File when topics are declared on it, so this declaration is all it knows about the File’s contents.

The platform applies each declaration on its own: one it refuses leaves the others registered, and the response says what became of each. Nothing about the call involves a Session, so declarations on the Files of one recording can run concurrently, and a File’s topic data can be registered before the Session holding it exists. A Session takes that data on by attaching the File, through add_file() or a SessionFile carrying no topics. A Session already holding the File takes on what this call declares before the call returns, with its time bounds recomputed to cover the newly declared data.

Resending the same call is safe: the platform identifies the data a declaration registers by the topic plus the slice of the File that declaration names, so a resend converges on what the first attempt registered rather than duplicating it, and a corrected redeclaration replaces what it corrects.

A declaration states its bounds in the File’s own timestamps, read as nanoseconds since the Unix epoch; the platform never invents a wall-clock time. Data whose timestamps start at 0 therefore sits at the epoch until it is anchored. To place it at the wall-clock time it was captured, supply anchor_ns, which anchors the whole slice it names rather than the one topic declaring it, so the topics sharing a slice must agree on it.

The topic data belongs to this File: its bounds, anchors and slices are stated against it, and a Session holding this File holds the data. What makes the data readable is each topic’s representations: the files a read opens to get it, each decoded in the storage format its representation states. A file’s name and extension are not used. A topic lists this File when its own bytes are readable, and other files when the data is read from them, such as the per-topic MCAPs converted out of a PX4 ULog. A topic with no representations is still registered and still counts toward the bounds of the Sessions holding the File, but reads of it return no rows. RepresentationDeclaration states what a representation’s file must hold and when a read can decode an MCAP representation’s file.

Redeclaring a topic adds the representations listed to the ones it has, each taking the place of the stored ones it matches, as representations describes. To remove a representation, or to replace a topic’s representations outright, use set_representations().

Usage

Register the two topics a LeRobot episode file carries, each read from the file’s own bytes:

from roboto.domain.files import File
from roboto.domain.topics import CanonicalDataType, RepresentationStorageFormat
from roboto.experimental.ingest import (
    Field,
    FileTopicDeclaration,
    RepresentationDeclaration,
    Schema,
    SchemaFieldSource,
)
timestamp = Field(
    name="timestamp",
    data_type="float64",
    canonical_data_type=CanonicalDataType.Timestamp,
    unit="s",
)
file = File.from_id("fl_0123456789ab")
from_file = RepresentationDeclaration(
    file_id=file.file_id,
    storage_format=RepresentationStorageFormat.PARQUET,
)
registered = file.declare_topics(
    [
        FileTopicDeclaration(
            topic_name="observation.state",
            topic_schema=Schema(
                name="observation.state",
                fields=[timestamp, Field(name="observation.state", data_type="float32")],
            ),
            timeline_sources=[
                SchemaFieldSource(
                    field_path=["timestamp"],
                    min_file_timestamp_ns=0,
                    max_file_timestamp_ns=4_000,
                )
            ],
            representations=[from_file],
        ),
        FileTopicDeclaration(
            topic_name="action",
            topic_schema=Schema(
                name="action",
                fields=[timestamp, Field(name="action", data_type="float32")],
            ),
            timeline_sources=[
                SchemaFieldSource(
                    field_path=["timestamp"],
                    min_file_timestamp_ns=0,
                    max_file_timestamp_ns=4_000,
                )
            ],
            representations=[from_file],
        ),
    ],
)
print([topic.topic_id for topic in registered.succeeded])

Register a topic over the slice of a shared file that holds one episode, anchored at the instant that episode was recorded:

registered = file.declare_topics(
    [
        FileTopicDeclaration(
            topic_name="observation.state",
            topic_schema=Schema(
                name="observation.state",
                fields=[timestamp, Field(name="observation.state", data_type="float32")],
            ),
            timeline_sources=[
                SchemaFieldSource(
                    field_path=["timestamp"],
                    min_file_timestamp_ns=0,
                    max_file_timestamp_ns=4_000,
                )
            ],
            data_range=(0, 80),
            anchor_ns=1_785_974_400_000_000_000,
            representations=[from_file],
        ),
    ],
)

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.

Usage

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

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.

Usage

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

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.

Usage

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

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.

Usage

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'

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.

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

Usage

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

Usage

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

Usage

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

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.

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

Usage

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

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.

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

Usage

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

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.

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.

Usage

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

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.

Usage

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'

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.

Usage

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

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.

Usage

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

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.

Usage

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

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.

Usage

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 the device ID for this file.

Usage

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

Replace the representations the named topics’ data on this File is read from.

This File is the one the topics were declared on, through declare_topics() or a Session. Each topic listed ends up with exactly the representations listed, over the part of this File its entry’s data_range names; its other representations there are removed, whether they cover the whole topic or one field of it. Topics and slices not listed keep theirs. Nothing else about the topics changes: their schemas, timeline sources, bounds, anchors and slices stay as declared, and so do the time bounds of every Session holding this File, which do not depend on which files the data is read from.

Use it for what redeclaring a topic cannot do:

1. Remove a representation.
2. Replace a representation with one that names another file and differs from it in what it covers, its storage format, its content format or its transformations. Those four identify a representation, as RepresentationDeclaration describes, so declaring the new one adds it beside the first.
3. Stop a topic being read at all, by listing no representations for it.

The platform checks every entry before writing any, and one refused entry refuses the whole call.

Usage

Replace a camera topic’s representation re-encoded as JPEG with one downsampled and re-encoded as PNG, keeping the untransformed one that names the recording, and stop reading a debug topic at all:

from roboto.domain.files import File
from roboto.domain.topics import RepresentationStorageFormat
from roboto.experimental.ingest import RepresentationDeclaration, TopicRepresentations
recording = File.from_id("fl_recording_0412_mcap")
recording.set_representations(
    [
        TopicRepresentations(
            topic_name="/camera/front/image_raw",
            representations=[
                RepresentationDeclaration(
                    file_id=recording.file_id,
                    storage_format=RepresentationStorageFormat.MCAP,
                ),
                RepresentationDeclaration(
                    file_id="fl_front_png",
                    storage_format=RepresentationStorageFormat.MCAP,
                    content_format="png",
                    transformations=["downsample:0.5", "encode:png"],
                ),
            ],
        ),
        TopicRepresentations(topic_name="/debug/raw_dump", representations=[]),
    ]
)

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

Contract:

1. The offset, in nanoseconds, is added to stored partition time to produce session wall-clock: session_time_ns = stored_time_ns + offset_ns. An offset given as an instant, such as a datetime, is the nanoseconds since the Unix epoch at which stored time 0 occurred.
2. 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.

Usage

Apply a file-wide offset:

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

Apply the same offset as a datetime, the instant stored time 0 occurred:

import datetime
file.set_timeline_offset(datetime.datetime(2023, 11, 14, 22, 13, 20, tzinfo=datetime.timezone.utc))

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

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.

Usage

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

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.

Usage

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

Convert this file to a dictionary representation.

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

Usage

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 this file’s properties.

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

Usage

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)

Create a directory among the association’s files.

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-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

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.

Usage

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

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.

Notes

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

Usage

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

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

Notes

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

Usage

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

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

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-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

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

Usage

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

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.

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.

Usage

from roboto.domain import devices
device = devices.Device.from_id("rover-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 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.

Usage

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

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.

Usage

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

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.

Notes

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

Usage

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

Upload a single file associated with association.

Usage

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

Upload multiple files associated with association.

Usage

import pathlib
from roboto.domain import devices
device = devices.Device.from_id("rover-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'