roboto.domain.devices

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

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

Register a new device with the Roboto platform.

Creates a new device entity that can authenticate and interact with Roboto on behalf of the specified organization. The device_id must be unique within the organization.

Usage

Register a robot device:

device = Device.create(device_id="robot_001", caller_org_id="og_abc123")
print(f"Registered device: {device.device_id}")
# Registered device: robot_001

Register an upload station:

device = Device.create(device_id="upload_station_alpha")
print(f"Device org: {device.org_id}")
# Device org: og_xyz789

Create one Session on this Device, optionally with its files, topics, and schemas.

The one-session form of create_sessions(), taking a single declaration’s fields as arguments and sharing its semantics: the Session, its file attachments, its topics, and its time ranges are created together or not at all, and every file the declaration names must already be uploaded. name identifies the Session within this Device, so resending the same call is safe; the platform reuses the Session already registered under that name instead of creating a second one. Arguments left at their defaults are left out of the request, so a call that reuses an existing Session never overwrites attributes it does not name. To create a Session with no name, or one spanning several Devices, use create().

A declaration the platform refuses raises here. Only create_sessions() reports a refusal instead of raising it, because only a batch has positions to trace refusals back to.

Declared times are stored exactly as given, in each file’s own timestamps, and read as nanoseconds since the Unix epoch; the platform never invents a wall-clock time. To place the Session at the wall-clock time it happened, supply anchor, or call set_unix_offset() later.

Usage

Create a Session and add a file to it:

device = Device.from_id("robot_001", org_id="og_abc123")
session = device.create_session(name="2024-05-01_morning_run")
session.add_file("fl_0123456789abcdef")

Create a Session placed at the wall-clock time it was recorded:

import datetime
session = device.create_session(
    name="2024-05-01_morning_run",
    anchor=datetime.datetime(2024, 5, 1, 9, 30, tzinfo=datetime.timezone.utc),
)

Create many Sessions on this Device, each with its files, topics, and schemas, in one call.

Each call accepts up to MAX_SESSIONS_PER_REQUEST declarations, one per Session (e.g. the episodes of a LeRobot dataset), and up to MAX_FILES_AND_TOPICS_PER_REQUEST files and topics combined, counted across every declaration; split anything larger across several calls. Every file a declaration names must already be uploaded; upload_files() returns the file IDs it creates, and files from any number of datasets may appear in one batch.

The platform decides which declarations to refuse before writing anything, then writes the rest together. A declaration’s Session, file attachments, topics, and time ranges are created together or not at all, and a declaration the platform refuses leaves the others written as if it were absent. None of the declarations is written when a failure the platform did not anticipate, such as a timeout, interrupts the call, or when a Session a declaration reuses is deleted before the call completes, which raises RobotoNotFoundException. Each declaration is written as it would be had the ones before it been sent as calls of their own: a later declaration anchoring data an earlier one holds moves the earlier Session’s time range with it. A refused declaration, or a call that fails, still leaves behind the topic identifiers and schema definitions it resolved, which a resend reuses.

Check failed before treating the batch as done. Each entry there is the RobotoDomainException the platform refused a declaration with, so isinstance tells the reasons apart; a refusal under an error code this SDK release does not define arrives as RobotoUnrecognizedErrorException.

Resending the same call is safe. This Device plus each declaration’s name identifies the Session the declaration creates or reuses, so a resend fills in only what is missing rather than duplicating what an earlier attempt created.

Declared times are stored exactly as given, in each file’s own timestamps, and read as nanoseconds since the Unix epoch; the platform never invents a wall-clock time. A recording whose timestamps start at 0 therefore sits at the epoch until it is anchored. To place a Session at the wall-clock time it happened, supply anchor_ns, or call set_unix_offset() later.

Usage

Register two chunks of one recording as a single Session, each chunk’s topic data read from the chunk itself:

import pathlib
from roboto.domain.datasets import Dataset
from roboto.domain.devices import Device
from roboto.domain.topics import CanonicalDataType, RepresentationStorageFormat
from roboto.experimental.ingest import (
    Field,
    McapLogTimeSource,
    RepresentationDeclaration,
    Schema,
    TopicDeclaration,
)
from roboto.experimental.sessions import SessionDeclaration, SessionFile
imu_schema = Schema(
    name="sensor_msgs/msg/Imu",
    fields=[
        Field(
            name="angular_velocity_x",
            data_type="float64",
            canonical_data_type=CanonicalDataType.Number,
        ),
    ],
)
dataset = Dataset.from_id("ds_0123456789ab")
device = Device.from_id("robot_001")
chunks = [pathlib.Path("recording/chunk_0000.mcap"), pathlib.Path("recording/chunk_0001.mcap")]
file_ids = dataset.upload_files(chunks)
batch = device.create_sessions(
    [
        SessionDeclaration(
            name="morning_drive",
            files=[
                SessionFile(
                    file_id=file_ids[chunks[0]],
                    topics=[
                        TopicDeclaration(
                            topic_name="/imu",
                            topic_schema=imu_schema,
                            timeline_sources=[
                                McapLogTimeSource(
                                    min_file_timestamp_ns=1_785_974_400_000_000_000,
                                    max_file_timestamp_ns=1_785_974_404_000_000_000,
                                ),
                            ],
                            representations=[
                                RepresentationDeclaration(
                                    file_id=file_ids[chunks[0]],
                                    storage_format=RepresentationStorageFormat.MCAP,
                                ),
                            ],
                        ),
                    ],
                ),
                SessionFile(
                    file_id=file_ids[chunks[1]],
                    topics=[
                        TopicDeclaration(
                            topic_name="/imu",
                            topic_schema=imu_schema,
                            timeline_sources=[
                                McapLogTimeSource(
                                    min_file_timestamp_ns=1_785_974_404_000_000_000,
                                    max_file_timestamp_ns=1_785_974_408_000_000_000,
                                ),
                            ],
                            representations=[
                                RepresentationDeclaration(
                                    file_id=file_ids[chunks[1]],
                                    storage_format=RepresentationStorageFormat.MCAP,
                                ),
                            ],
                        ),
                    ],
                ),
            ],
        ),
    ],
)
batch.failed
# []

Create an authentication token for this device.

Generates a new API token that can be used to authenticate requests made on behalf of this device. The token secret is returned only once and cannot be retrieved again, so it must be stored securely by the caller.

Usage

Create a token with default settings:

device = Device.from_id("robot_001", org_id="og_abc123")
token, secret = device.create_token()
print(f"Token created: {token.token_id}")
print(f"Secret (save this!): {secret}")
# Token created: to_abc123def456
# Secret (save this!): robo_pat_abc123def456...

Create a token with custom expiry and description:

token, secret = device.create_token(
    expiry_days=30, name="Monthly Upload Token", description="Token for automated monthly data uploads"
)
print(f"Token expires in 30 days: {token.token_id}")
# Token expires in 30 days: to_def789ghi012

Delete this device from the Roboto platform.

Permanently removes this device and all associated tokens. This action cannot be undone. Any tokens created for this device will be immediately invalidated.

The device’s own files (files) are deleted with it, every version of each, shortly after this call returns. With keep_files=True they move to the org root instead, under devices/<universal_device_id>/, keeping their file IDs and every version, so they stay reachable through org.files. Links among the device’s files are deleted in both cases, never moved; their targets are left alone.

Usage

Delete a device after confirming its identity:

device = Device.from_id("old_robot_001")
print(f"Deleting device: {device.device_id}")
device.delete()
print("Device deleted successfully")
# Deleting device: old_robot_001
# Device deleted successfully

Delete a device but keep its calibrations and manifests in the org root:

device = Device.from_id("old_robot_001")
device.delete(keep_files=True)

List all devices registered for a given organization.

Retrieves all devices that belong to the specified organization. For organizations with large numbers of devices, this method uses pagination and yields results as they become available from the API.

Usage

List all devices in an organization:

for device in Device.for_org("og_abc123"):
    print(f"Device: {device.device_id} (created: {device.created})")
# Device: robot_001 (created: 2024-01-15 10:30:00)
# Device: upload_station_beta (created: 2024-01-17 09:15:00)

Count devices in an organization:

device_count = sum(1 for _ in Device.for_org("og_abc123"))
print(f"Total devices: {device_count}")
# Total devices: 2

Retrieve a device by its device ID.

Looks up and returns a Device instance for the specified device_id. The device_id must be unique within the organization scope.

Usage

Retrieve a device by ID with explicit organization:

device = Device.from_id("robot_001", org_id="og_abc123")
print(f"Device: {device.device_id} in org {device.org_id}")
# Device: robot_001 in org og_abc123

Retrieve a device:

device = Device.from_id("upload_station_alpha")
print(f"Found device created by: {device.created_by}")
# Found device created by: user@example.com

Register a device, or return the existing one if device_id is already taken.

metadata, tags, and custom_fields are applied only by the call that registers the device; a device that is already registered comes back unchanged.

Usage

device = Device.get_or_create(device_id="aloha_001")
device.device_id
# 'aloha_001'

Iterate all Sessions attached to this Device.

Yields results as they are returned from the server, paginating transparently.

Usage

Print the name of every Session for a Device:

device = Device.from_id("robot_001", org_id="og_abc123")
for session in device.list_sessions():
    print(session.name)

Add or update metadata fields for this device.

Usage

device = Device.from_id("robot_001")
updated_device = device.put_metadata({"firmware_version": "2.1.0", "location": "warehouse-b"})
print(updated_device.metadata["firmware_version"])
# 2.1.0

Add tags to this device.

Usage

device = Device.from_id("robot_001")
updated_device = device.put_tags(["production", "warehouse-c"])
print("production" in updated_device.tags)
# True

Remove metadata fields from this device.

Usage

device = Device.from_id("robot_001")
updated_device = device.remove_metadata(["old_field", "deprecated_key"])

Remove tags from this device.

Usage

device = Device.from_id("robot_001")
updated_device = device.remove_tags(["old_tag", "deprecated"])

Set a single custom-field value on this device.

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

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

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

Retrieve all authentication tokens associated with this device.

Returns a list of all tokens that have been created for this device, including both active and expired tokens. The token secrets are not included in the response as they are only available at creation time.

Usage

List all tokens for a device:

device = Device.from_id("robot_001")
tokens = device.tokens()
for token in tokens:
    print(f"Token: {token.token_id}")
# Token: to_abc123def456
# Token: to_ghi789jkl012

Check if device has any tokens:

device = Device.from_id("new_robot")
if device.tokens():
    print("Device has tokens")
else:
    print("No tokens found for device")
# No tokens found for device

Update device properties using a structured request.

Usage

from roboto.updates import MetadataChangeset
device = Device.from_id("robot_001")
updated_device = device.update(
    UpdateDeviceRequest(
        metadata_changeset=MetadataChangeset(
            put_fields={"version": "2.0"}, put_tags=["updated"], remove_tags=["old"]
        )
    )
)