roboto
Submodules
- roboto.action_runtime
- roboto.ai
- roboto.analytics
- roboto.api_version
- roboto.association
- roboto.auth
- roboto.collection_utils
- roboto.config
- roboto.domain
- roboto.env
- roboto.exceptions
- roboto.experimental
- roboto.formats
- roboto.http
- roboto.image_registry
- roboto.logging
- roboto.notifications
- roboto.paths
- roboto.principal
- roboto.progress
- roboto.query
- roboto.regionalization
- roboto.roboto_search
- roboto.sentinels
- roboto.storage
- roboto.templating
- roboto.testing
- roboto.time
- roboto.types
- roboto.updates
- roboto.upload_agent
- roboto.uri
- roboto.version
- roboto.waiters
- roboto.warnings
Package Contents
AISummary
Bases: pydantic.BaseModel
A wire-transmissible representation of an AI summary
Parameters
data AnyAttributes
Accessibility
Bases: roboto.compat.StrEnum
Controls who can query for and invoke an action.
Accessibility levels determine the visibility and usability of actions within the Roboto platform. Actions can be private to an organization or published publicly in the Action Hub.
Future accessibility levels may include: “user” and/or “team”.
Action
A reusable function to process, transform or analyze data in Roboto.
Actions are containerized functions that can be invoked to process datasets, files, or other data sources within the Roboto platform. They encapsulate processing logic, dependencies, and compute requirements, making data processing workflows reproducible and scalable.
Actions can be created, updated, invoked, and managed through this class. They support parameterization, inheritance from other actions, and can be triggered automatically based on events or conditions.
An Action consists of:
- Container image and execution parameters
- Input/output specifications
- Compute requirements (CPU, memory, etc.)
- Parameters that can be customized at invocation time
- Metadata and tags for organization
Actions are owned by organizations and can have different accessibility levels (private to organization or public in the Action Hub).
Parameters
roboto_client Optional[roboto.Properties
Action.accessibility
The accessibility level of this action (Organization or ActionHub).
Action.compute_requirements
The compute requirements (CPU, memory) for running this action.
Action.container_parameters
The container parameters including image URI and execution settings.
Action.create()
Create a new action in the Roboto platform.
Creates a new action with the specified configuration. The action will be owned by the caller’s organization and can be invoked to process data.
Parameters
name strUnique name for the action within the organization.
compute_requirements Optional[roboto.CPU, memory, and other compute specifications.
container_parameters Optional[roboto.Container image URI, entrypoint, and environment variables.
description Optional[str]Detailed description of what the action does.
inherits Optional[roboto.Reference to another action to inherit configuration from.
metadata Optional[dict[str, Any]]Custom key-value metadata to associate with the action.
parameters Optional[list[roboto.List of parameters that can be provided at invocation time.
requires_downloaded_inputs Optional[bool]Whether input files should be downloaded before execution.
short_description Optional[str]Brief description (max 140 characters) for display purposes.
tags Optional[list[str]]List of tags for categorizing and searching actions.
timeout Optional[int]Maximum execution time in minutes before the action is terminated.
uri Optional[str]Container image URI if not inheriting from another action.
caller_org_id Optional[str]Organization ID to create the action in. Defaults to caller’s org.
roboto_client Optional[roboto.Roboto client instance. Uses default if not provided.
Returns
The newly created Action instance.
Raises
If short_description exceeds 140 characters or other validation errors occur.
If the request is malformed.
If the caller lacks permission to create actions.
Usage
Create a simple action:
action = Action.create(
name="hello_world", uri="ubuntu:latest", description="A simple hello world action"
)Create an action with parameters and compute requirements:
from roboto.domain.actions import ComputeRequirements, ActionParameter
action = Action.create(
name="data_processor",
uri="my-registry.com/processor:v1.0",
description="Processes sensor data with configurable parameters",
compute_requirements=ComputeRequirements(vCPU=4096, memory=8192),
parameters=[
ActionParameter(name="threshold", required=True, description="Processing threshold"),
ActionParameter(name="output_format", default="json", description="Output format"),
],
tags=["data-processing", "sensors"],
timeout=60,
)Create an action that inherits from another:
base_action = Action.from_name("base_processor", owner_org_id="roboto-public")
derived_action = Action.create(
name="custom_processor",
inherits=base_action.record.reference,
description="Custom processor based on base_processor",
metadata={"version": "2.0", "team": "data-science"},
)Action.delete()
Delete this action from the Roboto platform.
Permanently removes this action and all its versions. This operation cannot be undone.
Raises
If the action is not found.
If the caller lacks permission to delete the action.
Return type
Usage
Delete an action:
action = Action.from_name("old_action")
action.delete()Action.from_name()
Load an existing action by name.
Retrieves an action from the Roboto platform by its name and optionally a specific version digest. Action names are unique within an organization, so a name + org_id combination always provides a fully qualified reference to a specific action.
Parameters
name strName of the action to retrieve. Must be unique within the organization.
digest Optional[str]Specific version digest of the action. If not provided, returns the latest version.
owner_org_id Optional[str]Organization ID that owns the action. If not provided, searches in the caller’s organization.
roboto_client Optional[roboto.Roboto client instance. Uses default if not provided.
Returns
The Action instance.
Raises
If the action is not found.
If the caller lacks permission to access the action.
Usage
Load the latest version of an action:
action = Action.from_name("data_processor")Load a specific version of an action:
action = Action.from_name("data_processor", digest="abc123def456")Load an action from another organization:
action = Action.from_name("public_processor", owner_org_id="roboto-public")Properties
Action.inherits_from
Reference to another action this action inherits configuration from.
Action.invoke()
Invokes this action using any inputs and options provided.
Executes this action with the specified parameters and returns an Invocation object that can be used to track progress and retrieve results.
Parameters
invocation_source roboto.Manual, trigger, etc.
data_source_type Optional[roboto.If set, should equal Dataset for backward compatibility.
data_source_id Optional[str]If set, should be a dataset ID for backward compatibility.
input_data Optional[Union[list[str], roboto.Either a list of file name patterns, or an InvocationInput specification.
upload_destination Optional[roboto.Default upload destination (e.g. dataset) for files written to the invocation’s output directory.
compute_requirement_overrides Optional[roboto.Overrides for the action’s default compute requirements (e.g. vCPU)
container_parameter_overrides Optional[roboto.Overrides for the action’s default container parameters (e.g. entrypoint)
idempotency_id Optional[str]Unique ID to ensure an invocation is run exactly once.
invocation_source_id Optional[str]ID of the trigger or manual operator performing the invocation.
parameter_values Optional[dict[str, Any]]Action parameter values.
timeout Optional[int]Action timeout in minutes.
caller_org_id Optional[str]Org ID of the caller.
Returns
An Invocation object that can be used to track the invocation’s progress.
Raises
Invalid method parameters or combinations.
Incorrectly formed request.
The caller is not authorized to invoke this action.
Usage
Basic invocation with a dataset:
from roboto import Action, InvocationSource
action = Action.from_name("ros_ingestion", owner_org_id="roboto-public")
iv = action.invoke(
invocation_source=InvocationSource.Manual,
data_source_id="ds_12345",
data_source_type=InvocationDataSourceType.Dataset,
input_data=["**/*.bag"],
upload_destination=InvocationUploadDestination.dataset("ds_12345"),
)
iv.wait_for_terminal_status()Invocation with compute requirement overrides:
from roboto import Action, InvocationSource, ComputeRequirements
action = Action.from_name("image_processing", owner_org_id="roboto-public")
compute_reqs = ComputeRequirements(vCPU=4096, memory=8192)
iv = action.invoke(
invocation_source=InvocationSource.Manual,
compute_requirement_overrides=compute_reqs,
parameter_values={"threshold": 0.75},
)
status = iv.wait_for_terminal_status()
print(status)
# 'COMPLETED'Properties
Action.metadata
Custom metadata key-value pairs associated with this action.
Action.modified
The timestamp when this action was last modified.
Action.parameters
The list of parameters that can be provided when invoking this action.
Action.published
The timestamp when this action was published to the Action Hub, if applicable.
Action.query()
Query actions with optional filtering and pagination.
Searches for actions based on the provided query specification. Can search within an organization or across the public Action Hub.
Parameters
spec Optional[roboto.Query specification with filters, sorting, and pagination. If not provided, returns all accessible actions.
accessibility roboto.Whether to search organization actions or public Action Hub. Defaults to Organization.
owner_org_id Optional[str]Organization ID to search within. If not provided, searches in the caller’s organization.
roboto_client Optional[roboto.Roboto client instance. Uses default if not provided.
Yields
Action instances matching the query criteria.
Raises
ValueErrorIf the query specification contains unknown fields.
If the caller lacks permission to query actions.
Return type
Usage
Query all actions in your organization:
for action in Action.query():
print(f"Action: {action.name}")Query actions with specific tags:
from roboto.query import QuerySpecification
spec = QuerySpecification().where("tags").contains("ml")
for action in Action.query(spec):
print(f"ML Action: {action.name}")Query public actions in the Action Hub:
from roboto.domain.actions import Accessibility
spec = QuerySpecification().where("name").contains("ros")
for action in Action.query(spec, accessibility=Accessibility.ActionHub):
print(f"Public ROS Action: {action.name}")Query with pagination:
spec = QuerySpecification().limit(10).order_by("created", ascending=False)
recent_actions = list(Action.query(spec))
print(f"Found {len(recent_actions)} recent actions")Properties
Action.record
The underlying action record containing all action data.
Action.requires_downloaded_inputs
Whether input files should be downloaded before executing this action.
Action.set_accessibility()
Set the accessibility level of this action.
Changes whether this action is private to the organization or published to the public Action Hub.
Parameters
accessibility roboto.The new accessibility level (Organization or ActionHub).
Returns
This Action instance with updated accessibility.
Raises
If the caller lacks permission to modify the action.
Usage
Make an action public in the Action Hub:
from roboto.domain.actions import Accessibility
action = Action.from_name("my_action")
action.set_accessibility(Accessibility.ActionHub)Make an action private to the organization:
action.set_accessibility(Accessibility.Organization)Properties
Action.short_description
A brief description of the action (max 140 characters) for display purposes.
Action.tags
The list of tags associated with this action for categorization.
Action.timeout
The maximum execution time in minutes before the action is terminated.
Action.to_dict()
Convert this action to a dictionary representation.
Returns
Dictionary containing all action data in JSON-serializable format.
Usage
Get action as dictionary:
action = Action.from_name("my_action")
action_dict = action.to_dict()
print(action_dict["name"])
# 'my_action'Action.update()
Update this action with new configuration.
Updates the action with the provided changes. Only specified parameters will be modified; others remain unchanged. This creates a new version of the action.
Parameters
compute_requirements Optional[Union[roboto.New compute requirements (CPU, memory).
container_parameters Optional[Union[roboto.New container parameters (image, entrypoint, etc.).
description Optional[Union[str, roboto.New detailed description.
inherits Optional[Union[roboto.New action reference to inherit from.
metadata_changeset Union[roboto.Changes to apply to metadata (add, remove, update keys).
parameter_changeset Union[roboto.Changes to apply to parameters (add, remove, update).
short_description Optional[Union[str, roboto.New brief description (max 140 characters).
timeout Optional[Union[int, roboto.New maximum execution time in minutes.
uri Optional[Union[str, roboto.New container image URI.
requires_downloaded_inputs Union[bool, roboto.Whether to download input files before execution.
Returns
This Action instance with updated configuration.
Raises
If short_description exceeds 140 characters or other validation errors occur.
If the caller lacks permission to update the action.
Usage
Update action description and timeout:
action = Action.from_name("my_action")
action.update(description="Updated description of what this action does", timeout=45)Update compute requirements:
from roboto.domain.actions import ComputeRequirements
action.update(compute_requirements=ComputeRequirements(vCPU=4096, memory=8192))Add metadata using changeset:
from roboto.updates import MetadataChangeset
changeset = MetadataChangeset().set("version", "2.0").set("team", "ml")
action.update(metadata_changeset=changeset)Properties
ActionParameter
Bases: pydantic.BaseModel
A parameter that can be provided to an Action at invocation time.
Action parameters allow customization of action behavior without modifying the action itself. Parameters can be required or optional, have default values, and include descriptions for documentation.
Parameters are validated when an action is invoked, ensuring that required parameters are provided and that values conform to expected types.
Parameters
data AnyAttributes
ActionParameter.default
Default value applied for parameter if it is not required and no value is given at invocation.
Accepts any default value, but coerced to a string.
ActionParameter.description
Human-readable description of the parameter.
ActionParameter.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
ActionParameter.required
Whether this parameter is required at invocation time.
ActionParameter.validate_default()
Parameters
v Optional[Any]Return type
ActionParameterChangeset
Bases: pydantic.BaseModel
A changeset used to modify Action parameters.
Parameters
data AnyActionParameterChangeset.Builder
ActionParameterChangeset.Builder.build()
Return type
ActionParameterChangeset.Builder.put_parameter()
Parameters
parameter ActionParameterReturn type
ActionParameterChangeset.Builder.remove_parameter()
Parameters
parameter_name strReturn type
ActionParameterChangeset.is_empty()
Return type
Attributes
ActionParameterChangeset.put_parameters
Parameters to add or update.
ActionParameterChangeset.remove_parameters
Names of parameters to remove.
ActionProvenance
ActionRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of an action.
Attributes
ActionRecord.accessibility
ActionRecord.compute_digest()
Return type
Attributes
ActionRecord.compute_requirements
ActionRecord.container_parameters
ActionRecord.created
ActionRecord.created_by
ActionRecord.description
ActionRecord.digest
ActionRecord.inherits
ActionRecord.metadata
ActionRecord.modified
ActionRecord.modified_by
ActionRecord.name
ActionRecord.org_id
ActionRecord.parameters
ActionRecord.published
Properties
Attributes
ActionRecord.requires_downloaded_inputs
ActionRecord.serialize_metadata()
Parameters
metadata dict[str, Any]Return type
ActionReference
ActionRuntime
Bases: InvocationContext
Deprecated. Use InvocationContext instead.
AddMessagePathRepresentationRequest
Bases: BaseAddRepresentationRequest
Request to associate a message path with a representation.
Creates a link between a specific message path and a data representation, enabling efficient access to individual fields within topic data.
Parameters
data AnyAddMessagePathRequest
Bases: pydantic.BaseModel
Request to add a new message path to a topic.
Defines a new message path within a topic’s schema, specifying its data type, canonical type, and initial metadata. Used during topic creation or when extending an existing topic’s schema.
Parameters
data AnyAttributes
AddMessagePathRequest.canonical_data_type
Normalized Roboto data type that enables specialized platform features for maps, images, timestamps, and other data.
AddMessagePathRequest.data_type
Native data type as it appears in the original data source (e.g., “float32”, “geometry_msgs/Pose”). Used for display purposes.
AddMessagePathRequest.message_path
Dot-delimited path to the attribute (e.g., “pose.position.x”).
AddMessagePathRequest.metadata
Initial key-value pairs to associate with the message path.
AddMessagePathRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
AddMessagePathRequest.path_in_schema
List of path components representing the field’s location in the original data schema. Unlike message_path, which assumes dots separate path parts implying nested data, this preserves the exact path from the source data for accurate attribute access.
AgentThread
An interactive AI agent session within the Roboto platform.
An AgentThread is a conversational interface with Roboto’s AI assistant, enabling users to ask questions, request data analysis, and interact with their robotics data through natural language. Sessions maintain conversation history and support streaming responses for real-time interaction.
The primary control-flow primitives are run() (drive the session forward with auto-dispatch of client-side tools) and events() (observe events as the agent generates without taking any actions).
Usage
Fire-and-forget with client-side tools:
from roboto.ai import AgentThread, client_tool
@client_tool
def remember(fact: str) -> str:
"""Store a fact in long-term memory."""
...
session = AgentThread.start("Remember my favorite color is blue.", client_tools=[remember])
session.run()Observing events as they happen:
session = AgentThread.start("Explain machine learning.")
for event in session.events():
if isinstance(event, AgentTextDeltaEvent):
print(event.text, end="", flush=True)Parameters
roboto_client Optional[roboto.client_tools Optional[collections.Properties
AgentThread.client_tool_names
Names of client-side tools registered on this session with callbacks.
AgentThread.events()
Yield events from the agent as they are generated.
Polls the session and yields AgentEvent objects as new content arrives. Does not auto-dispatch client-side tools — if the session reaches AgentThreadStatus.CLIENT_TOOL_TURN, the generator returns and the caller is expected to call submit_client_tool_results() (and then call events() again to continue). For automatic dispatch, use run().
Parameters
tick floatPolling interval in seconds between checks for new content.
timeout Optional[float]Maximum time to wait in seconds. If None, waits indefinitely.
Yields
AgentEvent objects (AgentStartTextEvent, AgentTextDeltaEvent, AgentTextEndEvent, AgentToolUseEvent, AgentToolResultEvent, AgentErrorEvent) as they become available. Text events are scoped to a single message: an AgentTextEndEvent is emitted at the end of each message that carried text, so adjacent assistant messages produce separate start/end pairs.
Raises
TimeoutErrorIf timeout elapses before the session pauses.
Return type
Usage
Stream text output as it arrives:
for event in session.events():
if isinstance(event, AgentTextDeltaEvent):
print(event.text, end="", flush=True)AgentThread.fork()
Fork this session’s history up to a specific message into a new session owned by the caller.
Fork access mirrors read access: anyone who can read the source session can fork it. A ThreadVisibility.ORG-visible session can be forked by any member of its org; a ThreadVisibility.PRIVATE session can be forked only by its creator or a Roboto admin. The new session carries the source session’s org_id so tool calls resolve against the source org’s data, and is always created with ThreadVisibility.PRIVATE visibility, owned by the caller, regardless of the source’s visibility — so forks never appear in another member’s session list.
Parameters
message_sequence_num intHighest message sequence number (inclusive) to copy.
Returns
A new AgentThread instance for the forked session.
Raises
If the caller is not a member of the source session’s org, or the source session is ThreadVisibility.PRIVATE and the caller is neither its creator nor a Roboto admin.
If message_sequence_num is out of range or points at a message still generating.
AgentThread.from_id()
Retrieve an existing agent thread by its unique identifier.
Loads a previously created thread from the Roboto platform, allowing users to resume conversations and access message history.
Parameters
thread_id strUnique identifier for the thread. Accepts current ath_* identifiers as well as legacy ags_* and ch_* identifiers.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
load_messages boolWhether to load the thread’s messages. If False, the thread’s messages will be empty.
Returns
AgentThread instance representing the existing thread.
Raises
If the thread does not exist.
If the caller lacks permission to access the thread.
Usage
Resume an existing thread:
thread = AgentThread.from_id("ath_abc123")
print(f"Thread has {len(thread.messages)} messages")
# Thread has 5 messagesProperties
AgentThread.goals
Goals declared across this thread’s turns, oldest first.
Each entry is an AgentThreadGoalView SDK wrapper that delegates the goal-record fields and adds three resolved properties:
AgentThreadGoalView.achieve_tool_use— the LLM’s achieve-tool invocation, located byachieve_tool_use_id.AgentThreadGoalView.achieve_tool_result— the matching tool result block, if one was persisted.AgentThreadGoalView.result— the typed, per-goal-typeGoalResult(with the LLM’s submitted payload parsed into typed fields).
Returns an empty list when the thread declared no goals or when the underlying record was fetched without loading them — call refresh() if you expect goals to be present but the list is empty.
Wrappers are cached against the underlying goals-list identity: repeated reads return the same list (and the same per-goal wrappers) until a refresh installs a fresh goals list, at which point the cache is invalidated transparently. Callers may rely on identity comparison across reads against the same snapshot.
AgentThread.invoke_skill()
Manually invoke a skill into this thread.
Thin wrapper around send() that builds a single-element invoke_skills list and sends it with no user message — kept for SDK ergonomics on the common “invoke exactly one skill” case.
Parameters
skill_id strThe skill to invoke.
version Optional[int]Optional version number. Must exist on the skill (any version is invokable). If omitted, the latest (MAX(version)) version is used.
Returns
Self for method chaining.
Usage
Apply a skill’s latest version to the current turn:
thread.invoke_skill("sk_qa_review")Apply a specific version of the skill:
thread.invoke_skill("sk_qa_review", version=2)Properties
AgentThread.latest_message
The most recent message in the conversation, or None if no messages exist.
AgentThread.messages
Complete list of messages in the conversation in chronological order.
AgentThread.refresh()
Update the session with the latest messages and status.
Fetches any new messages or status changes from the server and updates the local session state.
Returns
Self for method chaining.
AgentThread.register_client_tool()
Register a client-side tool for auto-dispatch in subsequent turns.
The tool’s spec is not sent to the backend by this call; pass it via the client_tools= argument on send(), send_text(), or submit_client_tool_results() on the next outbound request.
Parameters
The ClientTool to register.
Returns
Self for method chaining.
AgentThread.run()
Drive the session forward until it is the user’s turn.
Polls the session, auto-dispatching any pending client-side tool invocations against the callbacks registered with this session (via start(), send(), or register_client_tool()). Returns once the session status is AgentThreadStatus.USER_TURN.
If the agent requests a client-side tool that has no registered callback, an error result is submitted automatically with a descriptive message so the agent can recover, and execution continues. If a registered callback raises, the exception is caught and also submitted as an error result.
Parameters
on_event Optional[OnEvent]Optional callback invoked for each AgentEvent as the agent generates (text deltas, tool uses, tool results, start/end markers). Use this for progress display or logging.
tick floatPolling interval in seconds between status checks.
timeout Optional[float]Total time budget in seconds across the whole loop. If None, waits indefinitely.
Returns
Self for method chaining.
Raises
TimeoutErrorIf the timeout budget is exhausted before the session reaches USER_TURN.
RuntimeErrorIf the session is in CLIENT_TOOL_TURN with no messages (i.e. a server state that should not be reachable), or if an unexpected AgentThreadStatus value is observed.
RobotoHttpExceptionPropagated from the underlying submit_client_tool_results() POST if the server rejects the submission (for example, a concurrent caller already answered the tool-use).
Usage
Fire-and-forget:
session = AgentThread.start("Remember my favorite color is blue.", client_tools=[remember])
session.run()With progress logging:
def log(event):
if isinstance(event, AgentToolUseEvent):
print(f"[tool-use] {event.name}({event.input})")
session.run(on_event=log)AgentThread.send()
Send a structured message to the session.
Parameters
message Optional[roboto.AgentMessage object containing the message content and metadata. Optional when at least one entry is provided in goals or invoke_skills; in that case the server synthesizes a minimal user message for the turn.
client_context Optional[roboto.Optional ClientViewingContext describing what the calling client is currently viewing when this message was composed. Informational only; see AgentThread.start() for full semantics.
client_tools Optional[collections.Optional client-side tools to add or update for this and subsequent turns. ClientTool callbacks are registered on the session for auto-dispatch.
analysis_scope Optional[roboto.Optional replacement AnalysisScope. When provided, overwrites the session’s current scope for all subsequent tool invocations. When None, the session’s existing scope (if any) is left untouched.
goals Optional[collections.Optional structured goals to declare for this turn. The agent runner enforces achievement of every declared goal before completing the turn.
invoke_skills Optional[collections.Optional sequence of InvokeSkillSpec to invoke one or more stored skills as part of this turn. For each entry the server fabricates a load_skill tool_use/tool_result pair after message (if any). When message and goals are both omitted, the fabricated pairs alone trigger the turn. Latest (MAX(version)) is used when version is omitted on an entry.
Returns
Self for method chaining.
Raises
If the message format is invalid.
If the caller lacks permission to send messages.
If the thread was started on another surface (see AgentThreadRecord.origin). Fork it to keep going.
AgentThread.send_text()
Send a text message to the session.
Convenience method for sending a simple text message without needing to construct an AgentMessage.
Parameters
text strText content to send to the assistant.
client_context Optional[roboto.Optional ClientViewingContext describing what the calling client is currently viewing.
client_tools Optional[collections.Optional client-side tools to add or update.
analysis_scope Optional[roboto.Optional replacement AnalysisScope; see send() for update semantics.
goals Optional[collections.Optional goals to declare for this turn. See send() for full semantics.
Returns
Self for method chaining.
Raises
If the text is empty or invalid.
If the caller lacks permission to send messages.
If the thread was started on another surface (see AgentThreadRecord.origin). Fork it to keep going.
AgentThread.set_pinned()
Pin or unpin this thread for the calling user.
A pin is a personal bookmark: it promotes the thread to the top of the caller’s chat sidebar and is invisible to everyone else. Anyone who may read the thread may pin it, including teammates on a thread whose visibility is ORG.
Setting the state the thread already has is accepted and changes nothing; re-pinning in particular does not move the thread back to the top of the pinned block.
Leaves pinned_at untouched on this instance’s record: only the thread listing and search endpoints populate that field.
Parameters
pinned boolTrue to pin, False to unpin.
Returns
Self for method chaining.
Raises
If the thread does not exist.
If the caller is not a member of the thread’s org, or the thread is private and the caller did not create it.
Usage
Pin a thread to the top of your sidebar:
thread.set_pinned(True)Unpin it again:
thread.set_pinned(False)AgentThread.set_visibility()
Re-scope who may read this thread.
Only the thread’s creator may call this. Roboto admins can read every thread but cannot re-scope one they did not create. Setting the visibility a thread already has is accepted and leaves it unchanged.
Parameters
ThreadVisibility.ORG to open the thread to every member of the thread’s organization, or ThreadVisibility.PRIVATE to restrict it back to its creator.
Returns
Self for method chaining.
Raises
If the thread does not exist.
If the caller is not a member of the thread’s org, or is a member but did not create the thread.
AgentThread.start()
Start a new agent session with an initial message.
Creates a new session and sends the initial message to begin the conversation. The AI assistant will process the message and generate a response, which can be driven to completion with run() or observed event-by-event with events().
Parameters
message Optional[Union[str, roboto.Initial message to start the conversation. Can be a text string, a single AgentMessage, or a sequence of AgentMessage objects for multi-turn initialization. Optional when at least one entry is provided in goals or invoke_skills; in those cases the server synthesizes a minimal user message — implicitly “achieve the goals” or “apply the invoked skills” — and the agent will work the turn from there.
client_context Optional[roboto.Optional ClientViewingContext describing what the calling client (e.g. the Web UI) is currently displaying when this session is started. Lets the agent resolve deictic references like “this dataset” without the user spelling them out. Informational only; does not gate authorization or scope tools — see analysis_scope for that.
system_prompt Optional[str]Optional system prompt to customize the AI assistant’s behavior for this conversation.
model_profile Optional[str]Optional model profile ID (e.g. “standard”, “advanced”). Defaults to the deployment’s default profile.
org_id Optional[str]Organization ID to create the session in. If None, uses the caller’s default organization.
client_tools Optional[collections.Optional list of client-side tools to make available to the agent. Accepts ClientTool instances (which include a callback for auto-dispatch) and bare ClientToolSpec objects (which describe the tool but require the caller to submit results manually).
analysis_scope Optional[roboto.Optional AnalysisScope for the session (e.g. a time window or topic-pattern filter). When provided, the scope is persisted on the session and delivered to every tool invocation on the server side. Individual tools opt in to honoring the scope as they are adopted.
goals Optional[collections.Optional structured goals to declare for the first turn. When provided, message may be omitted; the server will synthesize a minimal user message and the agent runner will enforce achievement of every declared goal before completing the turn.
invoke_skills Optional[collections.Optional sequence of InvokeSkillSpec to invoke one or more stored skills at session start, in order. For each entry the server fabricates a load_skill tool_use/tool_result pair after any seeded message; with no message and no goals, the fabricated pairs alone seed the conversation. Each skill must be visible to the caller (org skill or own private skill); the version must exist on the skill but is not required to be the latest one. Latest (MAX(version)) is used when version is omitted on an entry.
available_skills Optional[collections.Optional explicit set of skills the AI may auto-invoke during this session, replacing the registry it would otherwise derive from the caller’s skill subscriptions. None (the default) keeps the subscription-derived behavior; an empty list gives the AI no auto-invokable skills; a non-empty list of AvailableSkillSpec makes exactly those skill versions auto-invokable and ignores subscriptions and per-user ai_version pins. Each entry may reference any org skill or the caller’s own private skill (visibility only — no subscription needed), at any version. This is session configuration, not a turn trigger: it does not by itself satisfy the “needs a message, goal, or invoked skill” requirement. Distinct from invoke_skills, which seeds skill bodies into the opening transcript.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
AgentThread instance representing the newly created session.
Raises
If the message format is invalid.
If the caller lacks permission to create sessions.
Usage
Start and drive a session with client-side tools:
from roboto.ai import client_tool
@client_tool
def recall(query: str) -> str:
"""Search long-term memory for facts matching a query."""
...
session = AgentThread.start("What do you remember?", client_tools=[recall])
session.run()Properties
AgentThread.status
Current status of the thread.
AgentThread.submit_client_tool_results()
Submit results of client-side tool execution to resume the session.
On success the server has persisted every submitted tool_result and queued a new worker turn; the local record.status flips to ROBOTO_TURN to match. See the inline comment for why the next delta poll cannot communicate that transition on its own.
Parameters
results collections.Tool results from client-side execution.
client_tools Optional[collections.Optional updated client-side tools for the next invocation. ClientTool callbacks are registered on the session for auto-dispatch.
Returns
Self for method chaining.
AgentThread.submit_feedback()
Submit structured feedback on a specific assistant message in this session.
Persists categorized feedback so it can be reviewed by Roboto operators. Re-submitting from the same user on the same message overwrites sentiment, categories, and notes on the existing row rather than creating a duplicate; the feedback_id and original created/created_by are preserved.
Parameters
message_sequence_num intZero-indexed position of the assistant message being rated.
Overall rating direction.
categories Optional[list[roboto.Zero or more categories describing the feedback. Must match sentiment. FeedbackCategory.OTHER is always permitted but requires notes.
notes Optional[str]Free-text notes. Required when FeedbackCategory.OTHER is among the categories.
Returns
The persisted feedback as a UserFeedbackRecord. Admin triage columns (admin_label, admin_note, resolved, resolved_by, resolved_at) are intentionally not part of this shape — they are only visible through the admin API.
Raises
If the message is still generating, the sentiment/category combination is invalid, or OTHER is used without notes.
If the caller cannot access this session.
If message_sequence_num is out of range.
Properties
AgentThread.tasks
The thread’s task list, ordered by position.
Returns an empty list when the thread has no tasks or when the underlying record was fetched without loading them — call refresh() if you expect tasks but the list is empty.
AgentThread.transcript
Human-readable transcript of the entire conversation.
Returns a formatted string containing all messages in the conversation, with role indicators and message content clearly separated.
AgentThread.unregister_client_tool()
Remove a previously registered client-tool callback.
This only removes the local callback. The backend was told about the tool in StartAgentThreadRequest client_tools (or via a later send()) and may still emit tool_use events for it; once the callback is gone, run() will submit an error result for those invocations so the agent can recover. There is no server-side deregistration API.
The tool name remains recorded as a declared client tool on this session, so the dispatcher still treats it as client-side (and not as a server tool whose result the server will post).
Parameters
name strName of the client tool to unregister.
Returns
True if a callback was removed, False if no callback was registered under name.
Properties
AgentThread.visibility
Read scope for this thread: creator-only (PRIVATE) or open to the thread’s org (ORG).
Association
Bases: pydantic.BaseModel
Use to declare an association between two Roboto entities.
Parameters
data AnyAttributes
Association.URL_ENCODING_SEP
Association.association_type
association_type is the Roboto domain entity type of the association.
Association.association_version
association_version is the Roboto domain entity version of the association, if it exists.
Association.coalesce()
Parameters
associations Optional[collections.dataset_ids Optional[collections.file_ids Optional[collections.topic_ids Optional[collections.message_path_ids Optional[collections.throw_on_empty boolReturn type
Association.dataset()
Parameters
dataset_id strReturn type
Properties
Association.device()
Create an association with a device.
Parameters
universal_device_id strThe device’s Roboto-assigned dv_ ID (universal_device_id), not its customer-chosen device_id.
Return type
Usage
Association.device("dv_abc123")
# Association(association_id='dv_abc123', association_type=<AssociationType.Device: 'device'>, ...)Association.file()
Parameters
file_id strversion Optional[int]Properties
Association.from_id()
Infer the association type from the ID prefix.
Roboto IDs follow the pattern {prefix}_{random_chars} where the prefix indicates the entity type:
ds_→ Datasetfl_→ Filetp_→ Topicmp_→ MessagePathdv_→ Deviceog_→ Org
Parameters
association_id strA Roboto entity ID with a recognized prefix.
Returns
An Association with the inferred type.
Raises
If the ID prefix is not recognized.
Usage
Association.from_id("ds_abc123")
# Association(association_id='ds_abc123', association_type=AssociationType.Dataset)Association.from_url_encoded_value()
Reverse of Association::url_encode.
Parameters
encoded strReturn type
Association.group_by_type()
Parameters
associations collections.Return type
Properties
Association.msgpath()
Parameters
msgpath_id strReturn type
Association.org()
Create an association with an organization.
Parameters
org_id strThe organization’s ID.
Return type
Usage
Association.org("og_abc123")
# Association(association_id='og_abc123', association_type=<AssociationType.Org: 'org'>, ...)Attributes
Association.parent
The next level up in the hierarchy of this association. A message path’s parent is its topic, a topic’s parent is its file, and a file’s parent is its dataset.
The absense of a parent in an Association object doesn’t necessarily mean that a parent doesn’t exist; parents are only provided when they’re easily computable in the context of a given request.
Association.topic()
Parameters
topic_id strProperties
Association.url_encode()
Association encoded in a URL path segment ready format.
Return type
AssociationType
Bases: enum.Enum
AssociationType is the Roboto domain entity type of the association.
BatchRequest
BeginSignedUrlUploadRequest
Bases: pydantic.BaseModel
Request payload to begin a single file upload with a signed URL.
Used for simpler upload scenarios where a pre-signed URL is preferred over temporary credentials. The returned URL can be used directly for uploading the file content.
Parameters
data AnyAttributes
BeginSignedUrlUploadRequest.association
The entity this file will be associated with (e.g., dataset, topic).
BeginSignedUrlUploadRequest.file_path
Destination path for the file within the association.
BeginSignedUrlUploadRequest.origination
Optional description of the upload source.
BeginSignedUrlUploadResponse
Bases: pydantic.BaseModel
Response from beginning a single file upload.
Contains the upload ID for completing the transaction and a pre-signed URL that can be used to upload the file content directly.
Parameters
data AnyBeginUploadRequest
Bases: pydantic.BaseModel
Request payload to begin a batch file upload transaction.
Used to initiate a multi-file upload transaction for any association type (dataset, topic, etc.). Returns a transaction ID and upload mappings that specify where each file should be uploaded.
Parameters
data AnyAttributes
BeginUploadRequest.association
The entity these files will be associated with (e.g., dataset, topic).
BeginUploadRequest.device_id
Optional identifier of the device that generated this data.
BeginUploadRequest.origination
Description of the upload source (e.g., ‘roboto-sdk v1.0.0’).
BeginUploadRequest.resource_manifest
Dictionary mapping destination file paths to file sizes in bytes.
BeginUploadResponse
Bases: pydantic.BaseModel
Response from beginning a batch upload transaction.
Contains the transaction ID needed for subsequent progress reporting and completion calls, plus mappings from file paths to their upload URIs.
Parameters
data AnyCanonicalDataType
Bases: enum.Enum
Normalized data types used across different robotics frameworks.
Well-known and simplified data types that provide a common vocabulary for describing message path data types across different frameworks and technologies. These canonical types are primarily used for UI rendering decisions and cross-platform compatibility.
The canonical types abstract away framework-specific details while preserving the essential characteristics needed for data processing and visualization.
References
- ROS 1 field types: http://wiki.ros.org/msg
- ROS 2 field types: https://docs.ros.org/en/iron/Concepts/Basic/About-Interfaces.html#field-types
- uORB: https://docs.px4.io/main/en/middleware/uorb.html#adding-a-new-topic
Example mappings:
float32->CanonicalDataType.Numberuint8[]->CanonicalDataType.Arraysensor_msgs/Image->CanonicalDataType.Imagegeometry_msgs/Pose->CanonicalDataType.Objectstd_msgs/Header->CanonicalDataType.Objectstring->CanonicalDataType.Stringchar->CanonicalDataType.Stringbool->CanonicalDataType.Booleanbyte->CanonicalDataType.Byte
Attributes
CanonicalDataType.Boolean
CanonicalDataType.Byte
CanonicalDataType.Categorical
Data that can take a limited, fixed set of values. To be interpreted correctly by Roboto clients, a MessagePathRecord with this type must have a "categories" metadata key on the MessagePathRecord, which must be the ordered list of values that the Categorical can take.
For example, a signal that is logged as either “off” or “on” could be represented as a Categorical with the metadata "categories"=["off", "on"]. This allows Roboto to map the value “off” to 0 and “on” to 1 –each corresponding to their index position in the metadata array– and therefore visualize these state transitions as a plot.
The default visual representation of Categorical data will be the same as String data, but the Roboto visualizer will be capable of rendering Categorical data in a plot.
CanonicalDataType.Image
Special purpose type for data that can be rendered as an image.
CanonicalDataType.LatDegFloat
Geographic point in degrees. E.g. 47.6749387 (used in ULog ver_data_format >= 2)
CanonicalDataType.LatDegInt
Geographic point in degrees, expressed as an integer. E.g. 317534036 (used in ULog ver_data_format < 2)
CanonicalDataType.LonDegFloat
Geographic point in degrees. E.g. 9.1445274 (used in ULog ver_data_format >= 2)
CanonicalDataType.LonDegInt
Geographic point in degrees, expressed as an integer. E.g. 1199146398 (used in ULog ver_data_format < 2)
CanonicalDataType.Number
CanonicalDataType.NumberArray
CanonicalDataType.String
CanonicalDataType.Timestamp
Time elapsed since the Unix epoch, identifying a single instant on the time-line. Roboto clients will look for a "unit" metadata key on the MessagePath record, and will assume “ns” if none is found. If the timestamp is in a different unit, add the following metadata to the MessagePath record: { "unit": "s"|"ms"|"us"|"ns" } The unit must be a known value from TimeUnit.
ClientTool
A client-side tool with an execution callback.
Wraps a Python callable as a tool that the Roboto agent can request the client to execute. The tool’s JSON schema is inferred from the callable’s type hints; the tool description and per-parameter descriptions are taken from the function’s Google-style docstring unless passed explicitly.
Most callers build ClientTools via the client_tool() decorator or ClientTool.from_function() rather than instantiating this class directly.
Usage
Using the decorator — descriptions come from the docstring:
@client_tool
def remember(fact: str, tags: Optional[list[str]] = None) -> str:
"""Store a fact in long-term memory.
Args:
fact: A standalone sentence worth remembering.
tags: Optional tags for later retrieval.
"""
...Using Annotated[T, Field(...)] instead (takes precedence over the docstring):
from typing import Annotated
from pydantic import Field
@client_tool
def recall(
query: Annotated[str, Field(description="Substring to search for.")],
) -> str:
"""Search long-term memory."""
...Using the factory with explicit overrides:
tool = ClientTool.from_function(
my_fn,
name="store_fact",
description="Store a fact in long-term memory.",
)Parameters
fn collections.name strdescription strinput_schema dict[str, Any]ClientTool.from_function()
Build a ClientTool from a Python callable.
The tool’s name defaults to fn.__name__. The tool description defaults to the summary-and-body of fn’s docstring (everything before the first Google-style section header like Args: or Returns:). Per-parameter descriptions are pulled from the docstring’s Args: section, and can be overridden with typing.Annotated[T, pydantic.Field(description="...")] or param: T = pydantic.Field(description="...").
Parameters
fn collections.The callable to invoke when the tool is dispatched.
name Optional[str]Override for the tool name (default: fn.__name__).
description Optional[str]Override for the tool description (default: the docstring’s summary-and-body). Required if fn has no docstring.
input_schema Optional[dict[str, Any]]Override for the input JSON Schema (default: inferred from fn’s type hints and docstring).
Returns
A ClientTool wrapping the given callable.
Raises
ValueErrorIf the description cannot be resolved, or if input_schema is not provided and the signature cannot be automatically converted (e.g. uses *args or **kwargs).
Properties
ClientTool.spec
Declarative spec sent to the Roboto backend.
ClientToolSpec
Bases: pydantic.BaseModel
Declarative specification for a client-side tool.
Unlike AgentTool (which is an ABC with a __call__ method for server-side execution), ClientToolSpec is a plain data model. The backend includes it in the LLM’s tool list but never executes it — the client is responsible for execution and submitting the result.
Parameters
data AnyCollection
A higher-level container for grouping datasets together. Collections can also be used to group files from several distinct datasets together.
Parameters
roboto_client Optional[roboto.Collection.add_dataset()
Parameters
dataset_id strReturn type
Collection.add_event()
Parameters
event_id strReturn type
Collection.add_file()
Parameters
file_id strReturn type
Collection.add_session()
Parameters
session_id strReturn type
Collection.changes()
Yield this collection’s revision history, oldest change first.
Version 0 is a real collection version, so 0 is a bound like any other: from_version=0 starts at the beginning, and to_version=0 selects the empty range. Omit a bound to leave that end open.
Parameters
from_version Optional[int]to_version Optional[int]Return type
Collection.clear_custom_field()
Clear a single custom-field value on this collection to None.
Parameters
name strReturn type
Collection.clear_custom_fields()
Clear multiple custom-field values on this collection to None.
Parameters
names collections.Return type
Properties
Collection.create()
Parameters
description Optional[str]name Optional[str]resource_type roboto.resources Optional[list[roboto.dataset_ids Optional[collections.event_ids Optional[collections.file_ids Optional[collections.session_ids Optional[collections.tags Optional[list[str]]custom_fields Optional[dict[str, Any]]roboto_client Optional[roboto.caller_org_id Optional[str]Return type
Properties
Collection.custom_fields
Custom-field values defined on Collections in this org.
Every Ready CustomField defined for (org_id, Collection) appears as a key. Values that have not been set on this collection surface as None rather than being absent. Empty when no custom fields are defined for the org.
A Timestamp value is returned as an ISO 8601 string.
Collection.delete()
Properties
Collection.edit_access()
Parameters
Return type
Collection.from_id()
Parameters
Return type
Collection.get_access()
Return type
Collection.list_all()
Parameters
roboto_client Optional[roboto.owner_org_id Optional[str]sort_by Optional[str]sort_direction Optional[roboto.Return type
Properties
Collection.record
Collection.remove_dataset()
Parameters
dataset_id strReturn type
Collection.remove_event()
Parameters
event_id strReturn type
Collection.remove_file()
Parameters
file_id strReturn type
Collection.remove_session()
Parameters
session_id strReturn type
Properties
Collection.resource_count
Number of resources this collection holds.
Read from the collection record rather than from resources, so it is available in every content mode, including summary-only loads where the member list is empty. Counts every membership reference, including references to resources that have since been deleted.
Collection.set_custom_field()
Set a single custom-field value on this collection.
name must be the name of a Ready custom field for this collection’s org and the Collection entity type; value must satisfy the field’s declared type.
Parameters
name strvalue AnyReturn type
Collection.set_custom_fields()
Set or overwrite multiple custom-field values on this collection.
Each key must name a Ready custom field for this collection’s org and the Collection entity type; each value must satisfy the field’s declared type.
Parameters
fields dict[str, Any]Return type
Collection.update()
Parameters
add_resources Union[list[roboto.add_tags Union[list[str], roboto.description Optional[Union[roboto.name Optional[Union[roboto.remove_resources Union[list[roboto.remove_tags Union[list[str], roboto.custom_fields_changeset Optional[roboto.Return type
CollectionChangeRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of a collection change record
Parameters
data AnyAttributes
CollectionChangeRecord.applied
CollectionChangeRecord.applied_by
CollectionChangeRecord.change_set
CollectionChangeRecord.collection_id
CollectionChangeRecord.from_version
CollectionChangeRecord.to_version
CollectionChangeSet
Bases: pydantic.BaseModel
Changeset for updating a collection
Parameters
data AnyAttributes
CollectionChangeSet.added_resources
CollectionChangeSet.added_tags
CollectionChangeSet.field_changes
CollectionChangeSet.removed_resources
CollectionChangeSet.removed_tags
CollectionContentMode
CollectionRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of a collection
Parameters
data AnyAttributes
CollectionRecord.collection_id
CollectionRecord.created
CollectionRecord.created_by
CollectionRecord.custom_fields
Values for the custom fields defined on Collections in this org.
Every Ready custom field defined for (org_id, Collection) appears as a key; values that have not been set surface as None rather than being absent. Empty when no custom fields are defined for the org.
CollectionRecord.description
CollectionRecord.missing
CollectionRecord.name
CollectionRecord.org_id
CollectionRecord.resource_count
Number of resources the collection holds.
Maintained by the service as members are added and removed, so it is present in every content mode, including summary_only where resources is empty. Counts every membership reference, including references to resources that have since been deleted; hydrating the collection (content_mode=full) sorts those into missing, so the hydrated resources map can hold fewer entries than this count.
CollectionRecord.resource_type
CollectionRecord.resources
CollectionRecord.tags
CollectionRecord.updated
CollectionRecord.updated_by
CollectionRecord.version
CollectionResourceRef
Bases: pydantic.BaseModel
Reference to a collection resource
Parameters
data AnyAttributes
CollectionResourceRef.resource_id
CollectionResourceRef.resource_type
CollectionResourceRef.resource_version
CollectionResourceType
Comment
A comment attached to a Roboto platform entity.
Comments provide a way to add contextual information, feedback, or discussion to various Roboto platform resources including datasets, files, actions, invocations, triggers, and collections. Comments support @mention syntax using the format @[display_name](user_id) to notify specific users.
Comments are created through the create() class method and cannot be instantiated directly. They can be retrieved by entity, entity type, user, or organization using the various class methods provided.
Each comment tracks creation and modification metadata, including timestamps and user information. Comments can be updated or deleted by authorized users.
Parameters
roboto_client Optional[roboto.Properties
Comment.create()
Create a new comment on a Roboto platform entity.
Creates a comment attached to the specified entity. The comment text can include @mention syntax to notify users using the format @[display_name](user_id).
Parameters
comment_text strThe text content of the comment. May include @mention syntax to notify users.
entity_id strUnique identifier of the entity to attach the comment to.
entity_type roboto.Type of entity being commented on.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
caller_org_id Optional[str]Optional organization ID of the caller. If not provided, uses the organization from the client context.
Returns
A new Comment instance representing the created comment.
Raises
If the user lacks permission to comment on the specified entity.
If the specified entity does not exist.
If the provided arguments are invalid.
Usage
from roboto.domain import comments
# Create a comment on a dataset
comment = comments.Comment.create(
comment_text="This dataset looks good!",
entity_id="ds_1234567890abcdef",
entity_type=comments.CommentEntityType.Dataset,
)
print(comment.comment_id)
# cm_abcdef1234567890# Create a comment with user mentions
comment = comments.Comment.create(
comment_text="@[John Doe](john.doe@example.com) please review this",
entity_id="fl_9876543210fedcba",
entity_type=comments.CommentEntityType.File,
)Comment.delete_comment()
Delete this comment permanently.
Removes the comment from the platform. This action cannot be undone. Only the comment author or users with appropriate permissions can delete a comment.
Raises
If the user lacks permission to delete this comment.
If the comment no longer exists.
Return type
Usage
from roboto.domain import comments
comment = comments.Comment.from_id("cm_1234567890abcdef")
comment.delete_comment()
# # Comment is now permanently deletedComment.for_entity()
Retrieve all comments for a specific entity.
Fetches all comments attached to a particular entity, such as a dataset, file, action, invocation, trigger, or collection. Results are paginated and returned in chronological order.
Parameters
entity_type roboto.Type of entity to retrieve comments for.
entity_id strUnique identifier of the entity.
owner_org_id Optional[str]Optional organization ID that owns the entity. If not provided, uses the organization from the client context.
page_token Optional[str]Optional pagination token to retrieve the next page of results. Use None to start from the beginning.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
A tuple containing:
- A sequence of Comment instances for the entity
- An optional pagination token for the next page, or None if no more pages are available
Raises
If the entity does not exist.
If the user lacks permission to access the entity or its comments.
Usage
from roboto.domain import comments
# Get all comments for a dataset
comments_list, next_token = comments.Comment.for_entity(
entity_type=comments.CommentEntityType.Dataset, entity_id="ds_1234567890abcdef"
)
print(f"Found {len(comments_list)} comments")
# Found 5 comments# Paginate through comments
all_comments = []
page_token = None
while True:
comments_page, page_token = comments.Comment.for_entity(
entity_type=comments.CommentEntityType.File,
entity_id="fl_9876543210fedcba",
page_token=page_token,
)
all_comments.extend(comments_page)
if page_token is None:
breakComment.for_entity_type()
Retrieve all comments for a specific entity type.
Fetches all comments attached to entities of a particular type within an organization. For example, retrieve all comments on datasets or all comments on files. Results are paginated and returned in chronological order.
Parameters
entity_type roboto.Type of entities to retrieve comments for.
owner_org_id Optional[str]Optional organization ID to scope the search to. If not provided, uses the organization from the client context.
page_token Optional[str]Optional pagination token to retrieve the next page of results. Use None to start from the beginning.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
A tuple containing:
- A sequence of Comment instances for the entity type
- An optional pagination token for the next page, or None if no more pages are available
Raises
If the user lacks permission to access comments for the specified entity type.
Usage
from roboto.domain import comments
# Get all comments on datasets in the organization
dataset_comments, next_token = comments.Comment.for_entity_type(
entity_type=comments.CommentEntityType.Dataset
)
print(f"Found {len(dataset_comments)} dataset comments")
# Found 12 dataset comments# Get all comments on action invocations
invocation_comments, _ = comments.Comment.for_entity_type(
entity_type=comments.CommentEntityType.Invocation, owner_org_id="og_1234567890abcdef"
)Comment.for_user()
Retrieve all comments created by a specific user.
Fetches all comments authored by the specified user within an organization. Results are paginated and returned in chronological order.
Parameters
user_id strUnique identifier of the user whose comments to retrieve.
owner_org_id Optional[str]Optional organization ID to scope the search to. If not provided, uses the organization from the client context.
page_token Optional[str]Optional pagination token to retrieve the next page of results. Use None to start from the beginning.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
A tuple containing:
- A sequence of Comment instances created by the user
- An optional pagination token for the next page, or None if no more pages are available
Raises
If the user lacks permission to access comments by the specified user.
Usage
from roboto.domain import comments
# Get all comments by a specific user
user_comments, next_token = comments.Comment.for_user(user_id="john.doe@example.com")
print(f"User has created {len(user_comments)} comments")
# User has created 8 comments# Get comments by user in a specific organization
org_user_comments, _ = comments.Comment.for_user(
user_id="jane.smith@example.com", owner_org_id="og_1234567890abcdef"
)Comment.from_id()
Retrieve a comment by its unique identifier.
Fetches a specific comment using its comment ID. The caller must have permission to access the comment and its associated entity.
Parameters
comment_id strUnique identifier of the comment to retrieve.
owner_org_id Optional[str]Optional organization ID that owns the comment. If not provided, uses the organization from the client context.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
A Comment instance representing the retrieved comment.
Raises
If the comment does not exist or the caller lacks permission to access it.
If the user lacks permission to access the comment.
Usage
from roboto.domain import comments
# Retrieve a specific comment
comment = comments.Comment.from_id("cm_1234567890abcdef")
print(comment.record.comment_text)
# This is the comment text# Retrieve a comment from a specific organization
comment = comments.Comment.from_id(comment_id="cm_abcdef1234567890", owner_org_id="og_fedcba0987654321")Comment.recent_for_org()
Retrieve recent comments for an organization.
Fetches the most recently created or modified comments within an organization, across all entity types. Results are paginated and returned in reverse chronological order (most recent first).
Parameters
owner_org_id Optional[str]Optional organization ID to retrieve comments for. If not provided, uses the organization from the client context.
page_token Optional[str]Optional pagination token to retrieve the next page of results. Use None to start from the beginning.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
A tuple containing:
- A sequence of Comment instances ordered by recency
- An optional pagination token for the next page, or None if no more pages are available
Raises
If the user lacks permission to access comments in the organization.
Usage
from roboto.domain import comments
# Get recent comments in the organization
recent_comments, next_token = comments.Comment.recent_for_org()
print(f"Found {len(recent_comments)} recent comments")
# Found 10 recent comments# Get recent comments for a specific organization
org_comments, _ = comments.Comment.recent_for_org(owner_org_id="og_1234567890abcdef")
if org_comments:
print(f"Most recent comment: {org_comments[0].record.comment_text}")Properties
Comment.record
The underlying CommentRecord data structure.
Comment.update_comment()
Update the text content of this comment.
Modifies the comment text and updates the modification timestamp. The updated comment text can include @mention syntax to notify users. Only the comment author or users with appropriate permissions can update a comment.
Parameters
comment_text strNew text content for the comment. May include @mention syntax to notify users.
Returns
This Comment instance with updated content.
Raises
If the user lacks permission to update this comment.
If the comment no longer exists.
If the comment text is invalid.
Usage
from roboto.domain import comments
comment = comments.Comment.from_id("cm_1234567890abcdef")
updated_comment = comment.update_comment("Updated text content")
print(updated_comment.record.comment_text)
# Updated text content# Update with mentions
comment.update_comment("@[Jane Doe](jane.doe@example.com) please check this")CommentEntityType
Bases: roboto.compat.StrEnum
Enumeration of Roboto platform entities that support comments.
This enum defines the types of resources in the Roboto platform that can have comments attached to them. Each value corresponds to a specific domain entity type.
Attributes
CommentEntityType.Collection
Collections of related datasets or resources.
CommentRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of a comment.
This model represents the complete data structure of a comment as stored and transmitted by the Roboto platform API. It includes all metadata about the comment including creation/modification timestamps, user mentions, and the associated entity information.
Parameters
data AnyAttributes
CommentRecord.comment_text
The text content of the comment, may include @mention syntax.
CommentRecord.created
Timestamp when the comment was created.
Stored as datetime.datetime in Python but serialized as ISO 8601 string in UTC when transmitted over the API.
CommentRecord.mentions
List of user IDs mentioned in this comment using @mention syntax.
CommentRecord.modified
Timestamp when the comment was last modified.
Stored as datetime.datetime in Python but serialized as ISO 8601 string in UTC when transmitted over the API.
ComputeRequirements
Bases: pydantic.BaseModel
Compute requirements for an action invocation.
Parameters
data AnyAttributes
ComputeRequirements.memory
Container memory in MiB. Set to 1024 MiB by default.
The possible values depend on the CPU units chosen, with as little as 512 MiB and as much as 122,800 MiB (120 GiB).
ComputeRequirements.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
ComputeRequirements.storage
Container storage in GiB. Set to 21 GiB by default.
The minimum allowed value is 21 GiB, and the maximum allowed value is 200 GiB (for premium-tier orgs).
ComputeRequirements.vCPU
Container CPU units. Set to 512 by default.
1024 CPU units equal 1 vCPU.
Possible values: 256, 512, 1024, 2048, 4096, 8192, 16384.
ComputeRequirements.validate_storage_limit()
ComputeRequirements.validate_vcpu_mem_combination()
Return type
ContainerParameters
Bases: pydantic.BaseModel
Container parameters for an action invocation.
Parameters
data AnyAttributes
ContainerParameters.command
ContainerParameters.entry_point
ContainerParameters.env_vars
ContainerParameters.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
ContainerParameters.workdir
CreateActionRequest
Bases: pydantic.BaseModel
Request payload to create a new action.
Contains all the configuration needed to create a new action in the Roboto platform, including container settings, compute requirements, parameters, and metadata.
Parameters
data AnyAttributes
CreateActionRequest.compute_requirements
CPU, memory, and other compute specifications.
CreateActionRequest.container_parameters
Container image URI, entrypoint, and environment variables.
CreateActionRequest.description
Detailed description of what the action does.
CreateActionRequest.inherits
Reference to another action to inherit configuration from.
CreateActionRequest.metadata
Custom key-value metadata to associate with the action.
CreateActionRequest.parameters
List of parameters that can be provided at invocation time.
CreateActionRequest.requires_downloaded_inputs
Whether input files should be downloaded before execution.
CreateActionRequest.short_description
Brief description (max 140 characters) for display purposes.
CreateActionRequest.timeout
Maximum execution time in minutes before the action is terminated.
CreateActionRequest.uri
Container image URI if not inheriting from another action.
CreateCollectionRequest
Bases: pydantic.BaseModel
Request payload to create a collection
Parameters
data AnyAttributes
CreateCollectionRequest.custom_fields
Initial values for Ready custom fields on this collection.
Each key must be the name of a CustomField that is Ready for the caller’s org and the Collection entity type; each value must satisfy the field’s declared type. Names that are undefined or not Ready, and values that don’t match the field’s type, are rejected with a structured error.
CreateCollectionRequest.description
CreateCollectionRequest.name
CreateCollectionRequest.resource_type
CreateCollectionRequest.resources
CreateCollectionRequest.tags
CreateCommentRequest
Bases: pydantic.BaseModel
Request payload for creating a new comment.
This model defines the required information to create a comment on a Roboto platform entity.
Parameters
data AnyAttributes
CreateCommentRequest.comment_text
Text content of the comment, may include @mention syntax.
CreateCommentRequest.entity_id
Unique identifier of the entity to attach the comment to.
CreateCommentRequest.entity_type
Type of entity to attach the comment to.
CreateDatasetIfNotExistsRequest
Bases: pydantic.BaseModel
Request payload to create a dataset if no existing dataset matches the specified query.
Searches for existing datasets using the provided RoboQL query. If a matching dataset is found, returns that dataset. If no match is found, creates a new dataset with the specified properties and returns it.
Parameters
data AnyAttributes
CreateDatasetIfNotExistsRequest.create_request
CreateDatasetIfNotExistsRequest.match_roboql_query
CreateDatasetRequest
Bases: pydantic.BaseModel
Request payload for creating a new dataset.
Used to specify the initial properties of a dataset during creation, including optional metadata, tags, name, and description.
Attributes
CreateDatasetRequest.custom_fields
Initial values for Ready custom fields on this dataset.
Each key must be the name of a CustomField that is Ready for the caller’s org and the Dataset entity type; each value must satisfy the field’s declared type. Names that are undefined or not Ready, and values that don’t match the field’s type, are rejected with a structured error.
CreateDatasetRequest.description
Optional human-readable description of the dataset.
CreateDatasetRequest.device_id
Optional identifier of the device that generated this data.
CreateDatasetRequest.metadata
Key-value metadata pairs to associate with the dataset for discovery and search.
CreateDatasetRequest.name
Optional short name for the dataset (max 120 characters).
CreateDatasetRequest.tags
List of tags for dataset discovery and organization.
CreateDeviceRequest
Bases: pydantic.BaseModel
Request payload to create a new device.
This request is used to register a new device with the Roboto platform. The device will be associated with the specified organization and can subsequently be used for authentication and data operations.
Parameters
data AnyAttributes
CreateDeviceRequest.custom_fields
Initial values for Ready custom fields on this device.
Each key must be the name of a CustomField that is Ready for the caller’s org and the Device entity type; each value must satisfy the field’s declared type. Names that are undefined or not Ready, and values that don’t match the field’s type, are rejected with a structured error.
CreateDeviceRequest.device_id
A user-provided identifier for a device, which is unique within that device’s org.
CreateDeviceRequest.metadata
Key-value metadata pairs to associate with the device for discovery and search.
CreateDeviceRequest.org_id
The org to which this device belongs. If None, the device will be registered under the caller’s organization (if they belong to only one org) or an error will be raised if the caller belongs to multiple organizations.
CreateInvocationRequest
Bases: pydantic.BaseModel
Request payload to create a new action invocation.
Contains all the configuration needed to invoke an action, including input data specifications, parameter values, and execution overrides.
Parameters
data AnyAttributes
CreateInvocationRequest.compute_requirement_overrides
compute_requirement_overrides roboto.Optional overrides for CPU, memory, and other compute specifications.
CreateInvocationRequest.container_parameter_overrides
container_parameter_overrides roboto.Optional overrides for container image, entrypoint, and environment variables.
CreateInvocationRequest.data_source_id
ID of the data source providing input data.
CreateInvocationRequest.data_source_type
Type of the data source (e.g., Dataset).
CreateInvocationRequest.idempotency_id
Optional unique ID to ensure the invocation runs exactly once.
CreateInvocationRequest.input_data
List of file patterns for input data selection.
CreateInvocationRequest.invocation_source
Source of the invocation (Manual, Trigger, etc.).
CreateInvocationRequest.invocation_source_id
Optional ID of the entity that initiated the invocation.
CreateInvocationRequest.parameter_values
Optional parameter values to pass to the action.
CreateInvocationRequest.rich_input_data
Optional rich input data specification that supersedes the simple input_data patterns.
CreateInvocationRequest.upload_destination
upload_destination roboto.Optional destination for output files.
CreateTopicRequest
Bases: pydantic.BaseModel
Request to create a new topic in the Roboto platform.
Contains all the information needed to register a topic found within a source recording file, including its schema, temporal boundaries, and initial message paths.
Parameters
data AnyAttributes
CreateTopicRequest.association
CreateTopicRequest.end_time
CreateTopicRequest.message_count
CreateTopicRequest.message_paths
CreateTopicRequest.metadata
CreateTopicRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
CreateTopicRequest.schema_checksum
CreateTopicRequest.schema_name
CreateTopicRequest.start_time
CreateTopicRequest.topic_name
CreateTriggerRequest
Bases: pydantic.BaseModel
Request payload to create a new trigger.
Contains all the configuration needed to create a trigger that automatically invokes actions when specific conditions are met.
Parameters
data AnyAttributes
CreateTriggerRequest.action_digest
Optional specific version digest of the action to invoke. If not provided, uses the latest version.
CreateTriggerRequest.action_name
Name of the action to invoke when the trigger fires.
CreateTriggerRequest.action_owner_id
Organization ID that owns the target action. If not provided, searches in the caller’s organization.
CreateTriggerRequest.additional_inputs
Optional additional file patterns to include in action invocations beyond the required inputs.
CreateTriggerRequest.causes
List of events that can cause this trigger to be evaluated. If not provided, uses default causes.
CreateTriggerRequest.compute_requirement_overrides
Optional compute requirement overrides for action invocations.
CreateTriggerRequest.condition
Optional condition that must be met for the trigger to fire.
Can filter based on metadata, file properties, etc.
CreateTriggerRequest.container_parameter_overrides
Optional container parameter overrides for action invocations.
CreateTriggerRequest.enabled
Whether the trigger should be active immediately after creation.
CreateTriggerRequest.for_each
Granularity of execution - Dataset or DatasetFile.
CreateTriggerRequest.name
Unique name for the trigger (alphanumeric, hyphens, underscores only, max 256 characters).
CreateTriggerRequest.parameter_values
Parameter values to pass to the action when invoked.
CreateTriggerRequest.required_inputs
List of file patterns that must be present for the trigger to fire. Uses glob patterns like ‘**/*.bag’.
CreateTriggerRequest.service_user_id
Optional service user ID for authentication.
CreateTriggerRequest.timeout
Optional timeout override for action invocations in minutes.
CreateTriggerRequest.validate_additional_inputs()
Parameters
value Optional[list[str]]Return type
CreateTriggerRequest.validate_required_inputs()
Parameters
value list[str]Return type
CreateUserRequest
Bases: pydantic.BaseModel
Request payload to create a new user.
Parameters
data AnyAttributes
CreateUserRequest.default_notification_channels
Default notification channels to enable for the user.
CreateUserRequest.default_notification_types
Default notification types to enable for the user.
CreateUserRequest.is_service_user
Whether this is a service user for automated operations.
CreateUserRequest.is_system_user
Whether this is a system user for internal platform operations.
Dataset
Represents a dataset within the Roboto platform.
A dataset is a logical container for files organized in a directory structure. Datasets are the primary organizational unit in Roboto, typically containing files from a single robot activity such as a drone flight, autonomous vehicle mission, or sensor data collection session. However, datasets are versatile enough to serve as a general-purpose assembly of files.
Datasets provide functionality for:
- File upload and download operations
- Metadata and tag management
- File organization and directory operations
- Topic data access and analysis
- AI-powered content summarization
- Integration with automated workflows and triggers
Files within a dataset can be processed by actions, visualized in the web interface, and searched using the query system. Datasets inherit access permissions from their organization and can be shared with other users and systems.
The Dataset class serves as the primary interface for dataset operations in the Roboto SDK, providing methods for file management, metadata operations, and content analysis.
Parameters
roboto_client Optional[roboto.file_service Optional[roboto.content_mode Optional[roboto.Dataset.clear_custom_field()
Clear a single custom-field value on this dataset to None.
Parameters
name strReturn type
Dataset.clear_custom_fields()
Clear multiple custom-field values on this dataset to None.
Parameters
names collections.Return type
Dataset.create()
Create a new dataset in the Roboto platform.
Creates a new dataset with the specified properties and returns a Dataset instance for interacting with it. The dataset will be created in the caller’s organization unless a different organization is specified.
Parameters
description Optional[str]Optional human-readable description of the dataset.
metadata Optional[dict[str, Any]]Optional key-value metadata pairs to associate with the dataset.
name Optional[str]Optional short name for the dataset (max 120 characters).
tags Optional[list[str]]Optional list of tags for dataset discovery and organization.
device_id Optional[str]Optional identifier of the device that generated this data.
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields defined on Datasets in the caller’s org. Keys must match Ready field names; values must satisfy each field’s declared type.
caller_org_id Optional[str]Organization ID to create the dataset in. Required for multi-org users.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
create_device_if_missing boolIf True, and a device_id is provided that does not exist in the organization, a new device will be created automatically. If False, and a device_id is provided that does not exist, a RobotoDeviceNotFoundException will be raised.
Returns
Dataset instance representing the newly created dataset.
Raises
A device_id has been provided in this request, but was not found as a device registered with Roboto for the organization.
Invalid dataset parameters.
Caller lacks permission to create datasets.
Usage
dataset = Dataset.create(
name="Highway Test Session",
description="Autonomous vehicle highway driving test data",
tags=["highway", "autonomous", "test"],
metadata={"vehicle_id": "vehicle_001", "test_type": "highway"},
)
print(dataset.dataset_id)
# ds_abc123# Create minimal dataset
dataset = Dataset.create()
print(f"Created dataset: {dataset.dataset_id}")Dataset.create_directory()
Create a directory within the dataset.
Parameters
name strName of the directory to create.
error_if_exists boolIf True, raises an exception if the directory already exists.
parent_path Optional[pathlib.Path of the parent directory. If None, creates the directory in the root of the dataset.
origination Optional[str]Optional string describing the source or context of the directory creation.
create_intermediate_dirs boolIf True, creates intermediate directories in the path if they don’t exist. If False, requires all parent directories to already exist.
Raises
If the directory already exists and error_if_exists is True.
If the caller lacks permission to create the directory.
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.
Usage
Create a simple directory:
from roboto.domain import datasets
dataset = datasets.Dataset.from_id(...)
directory = dataset.create_directory("foo")
print(directory.relative_path)
# fooCreate a directory with intermediate directories:
directory = dataset.create_directory(
name="final",
parent_path=pathlib.Path("path/to/deep"),
create_intermediate_dirs=True,
)
print(directory.relative_path)
# path/to/deep/finalDataset.create_if_not_exists()
Create a dataset if no existing dataset matches the specified query.
Searches for existing datasets using the provided RoboQL query. If a matching dataset is found, returns that dataset. If no match is found, creates a new dataset with the specified properties and returns it.
Concurrent calls with the same match_roboql_query in one organization create one dataset between them: the service runs them one at a time from the search through the create. Calls with different queries are not serialized, even when both queries would match the same dataset.
The dataset created must match match_roboql_query. If name, tags or metadata describe a dataset the query does not match, every later call creates another one. When several datasets match, which one is returned is not defined unless the query ends with a SORT BY clause.
Parameters
match_roboql_query strRoboQL query string to search for existing datasets. If this query matches any dataset, that dataset will be returned instead of creating a new one.
description Optional[str]Optional human-readable description of the dataset.
metadata Optional[dict[str, Any]]Optional key-value metadata pairs to associate with the dataset.
name Optional[str]Optional short name for the dataset (max 120 characters).
tags Optional[list[str]]Optional list of tags for dataset discovery and organization.
device_id Optional[str]Optional identifier of the device that generated this data.
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields defined on Datasets in caller_org_id. Keys must match Ready field names; values must satisfy each field’s declared type. Ignored when an existing dataset matches match_roboql_query — the existing record is returned unchanged.
caller_org_id Optional[str]Organization ID to create the dataset in. Required for multi-org users.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
create_device_if_missing boolIf True, and a device_id is provided that does not exist in the organization, a new device will be created automatically. If False, and a device_id is provided that does not exist, a RobotoDeviceNotFoundException will be raised.
Returns
Dataset instance representing either the existing matched dataset or the newly created dataset.
Raises
A device_id has been provided in this request, but was not found as a device registered with Roboto for the organization.
Invalid dataset parameters or malformed RoboQL query.
Other calls with the same query kept this one waiting for more than 10 seconds, after the SDK’s own retries. Calling again is safe.
Caller lacks permission to create datasets or search existing ones.
Usage
Create a dataset only if no dataset with specific metadata exists:
dataset = Dataset.create_if_not_exists(
match_roboql_query="dataset.metadata.vehicle_id = 'vehicle_001'",
name="Vehicle 001 Test Session",
description="Test data for vehicle 001",
metadata={"vehicle_id": "vehicle_001", "test_type": "highway"},
tags=["vehicle_001", "highway"],
)
print(dataset.dataset_id)
# ds_abc123Create a dataset only if no dataset with specific tags exists:
dataset = Dataset.create_if_not_exists(
match_roboql_query="dataset.tags CONTAINS 'unique_session_id_xyz'",
name="Unique Test Session",
tags=["unique_session_id_xyz", "test"],
)
# If a dataset with tag 'unique_session_id_xyz' already exists,
# that dataset is returned instead of creating a new oneDataset.create_session()
Create a Session populated with files from this dataset.
By default, every file in the dataset is added to the new Session. include_patterns / exclude_patterns narrow that set using the same gitignore-style syntax as Dataset.list_files().
Parameters
name Optional[str]Short display name for the Session (max 120 characters).
device_ids Optional[collections.Devices to attach to the Session. Defaults to no devices.
include_patterns Optional[list[str]]Gitignore-style patterns selecting which files to include. Same syntax as Dataset.list_files().
exclude_patterns Optional[list[str]]Gitignore-style patterns selecting which files to exclude. Takes precedence over include_patterns.
description Optional[str]Optional description of the Session.
metadata Optional[dict[str, Any]]Optional initial metadata. Sessions are not filterable or sortable by metadata keys; for queryable structured attributes, define a custom field on the Session entity type.
tags Optional[collections.Optional initial tags. Sessions can be filtered by tag membership but are not sortable by tag.
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields defined on Sessions in this dataset’s org. Keys must match Ready field names; values must satisfy each field’s declared type.
Returns
The newly created Session.
Usage
dataset = Dataset.from_id("ds_abc123")
session = dataset.create_session("flight-2026-04-23-001")Notes
Convenience wrapper around Session.create() followed by Session.add_files(). A dataset may hold more files than one add request accepts: the files are sent in consecutive requests of at most MAX_FILES_AND_TOPICS_PER_REQUEST each. Every file goes in or none does: if any add request fails, or the platform refuses any one file, the partially populated Session is deleted before the exception propagates, so the call is safe to retry. If that cleanup fails too (e.g. a transient network error), the Session is left behind holding whatever files did go in; the cleanup failure is logged, and the original exception is what the caller sees.
Properties
Dataset.created
Timestamp when this dataset was created.
Returns the UTC datetime when this dataset was first created in the Roboto platform. This property is immutable.
Dataset.created_by
Identifier of the user who created this dataset.
Returns the identifier of the person or service which originally created this dataset in the Roboto platform.
Dataset.custom_fields
Custom-field values defined on Datasets in this org.
Every Ready CustomField defined for (org_id, Dataset) appears as a key. Values that have not been set on this dataset surface as None rather than being absent. Empty when no custom fields are defined for the org.
A Timestamp value is returned as an ISO 8601 string.
Dataset.dataset_id
Unique identifier for this dataset.
Returns the globally unique identifier assigned to this dataset when it was created. This ID is immutable and used to reference the dataset across the Roboto platform. It is always prefixed with ‘ds_’ to distinguish it from other Roboto resource IDs.
Dataset.delete()
Delete this dataset from the Roboto platform.
Permanently removes the dataset and all its associated files, metadata, and topics. This operation cannot be undone.
If a dataset’s files are hosted in Roboto managed S3 buckets or customer read/write bring-your-own-buckets, the files in this dataset will be deleted from S3 as well. For files hosted in customer read-only buckets, the files will not be deleted from S3, but the dataset record and all associated metadata will be deleted.
Raises
Dataset does not exist or has already been deleted.
Caller lacks permission to delete the dataset.
Return type
Usage
dataset = Dataset.from_id("ds_abc123")
dataset.delete()
# # Dataset and all its files are now permanently deletedDataset.delete_files()
Delete files from this dataset based on pattern matching.
Deletes files that match the specified include patterns while excluding those that match exclude patterns. Uses gitignore-style pattern matching for flexible file selection.
Parameters
include_patterns Optional[list[str]]List of gitignore-style patterns for files to include. If None 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
Caller lacks permission to delete files.
Return type
Notes
Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.
Usage
dataset = Dataset.from_id("ds_abc123")
# Delete all PNG files except those in back_camera directory
dataset.delete_files(include_patterns=["**/*.png"], exclude_patterns=["**/back_camera/**"])# Delete all log files
dataset.delete_files(include_patterns=["**/*.log"])Properties
Dataset.description
Human-readable description of this dataset.
Returns the optional description text that provides details about the dataset’s contents, purpose, or context. Can be None if no description was provided.
Dataset.device_id
Identifier of the device that generated this data.
Returns the optional identifier of the device that generated the data contained within this dataset. Can be None if the dataset was not generated by a device.
Dataset.download_files()
Download files from this dataset to a local directory.
Downloads files that match the specified patterns to the given local directory. The directory structure from the dataset is preserved in the download location. If the output directory doesn’t exist, it will be created.
Parameters
out_path pathlib.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 boolWhether to show a progress bar during download.
Returns
List of tuples containing (FileRecord, local_path) for each downloaded file.
Raises
Caller lacks permission to download files.
Notes
Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.
Usage
import pathlib
dataset = Dataset.from_id("ds_abc123")
downloaded = dataset.download_files(
pathlib.Path("/tmp/dataset_download"),
include_patterns=["**/*.bag"],
exclude_patterns=["**/test/**"],
)
print(f"Downloaded {len(downloaded)} files")
# Downloaded 5 files# Download all files
all_files = dataset.download_files(pathlib.Path("/tmp/all_files"))Properties
Dataset.files
The files associated with this dataset.
The file methods on Dataset delegate here, so dataset.upload_files(...) and dataset.files.upload_files(...) are the same call.
Dataset.from_id()
Create a Dataset instance from a dataset ID.
Retrieves dataset information from the Roboto platform using the provided dataset ID and returns a Dataset instance for interacting with it.
Parameters
dataset_id strUnique identifier for the dataset.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
Dataset instance representing the requested dataset.
Raises
Dataset with the given ID does not exist.
Caller lacks permission to access the dataset.
Usage
dataset = Dataset.from_id("ds_abc123")
print(dataset.name)
# 'Highway Test Session'
print(len(list(dataset.list_files())))
# 42Dataset.generate_summary()
Generate a new AI summary for this dataset.
Creates a new AI-generated summary that analyzes the dataset’s content, structure, and metadata. The summary generation is asynchronous and can be monitored through the returned StreamingAISummary object.
Returns
StreamingAISummary object that provides access to the summary as it is being generated. The summary starts in pending status and can be monitored for completion.
Raises
Caller lacks permission to generate summaries for this dataset.
Usage
Generate a summary and wait for completion:
dataset = Dataset.from_id("ds_abc123")
summary = dataset.generate_summary()
complete_text = summary.complete_text
print(complete_text)
# 'This dataset contains 42 files with sensor data from highway driving tests...'Generate a summary and stream the text as it’s generated:
dataset = Dataset.from_id("ds_abc123")
summary = dataset.generate_summary()
for text_chunk in summary.text_stream():
print(text_chunk, end="", flush=True)Check summary status without blocking:
dataset = Dataset.from_id("ds_abc123")
summary = dataset.generate_summary()
if summary.current and summary.current.status == AISummaryStatus.Complete:
print("Summary is ready!")Dataset.get_file_by_path()
Get a File instance for a file at the specified path in this dataset.
Retrieves a file by its relative path within the dataset. Optionally retrieves a specific version of the file.
Parameters
relative_path Union[str, pathlib.Path of the file relative to the dataset root.
version_id Optional[int]Specific version of the file to retrieve. If None, gets the latest version.
Returns
File instance representing the file at the specified path.
Raises
File at the given path does not exist in the dataset.
Caller lacks permission to access the file.
Usage
dataset = Dataset.from_id("ds_abc123")
file = dataset.get_file_by_path("logs/session1.bag")
print(file.file_id)
# file_xyz789# Get specific version
old_file = dataset.get_file_by_path("data/sensors.csv", version_id=1)
print(old_file.version)
# 1Dataset.get_metadata()
Return custom metadata associated with this dataset.
Returns a copy of the dataset’s metadata dictionary containing arbitrary key-value pairs for storing custom information. Supports nested structures and dot notation for accessing nested fields.
Return type
Dataset.get_sessions()
Iterate over Sessions that include at least one file from this dataset.
A Session may draw files from one or more datasets; this method yields every Session whose current-version file list intersects this dataset.
Yields
Each matching Session. Pagination is handled automatically.
Return type
Usage
dataset = Dataset.from_id("ds_abc123")
for session in dataset.get_sessions():
print(session.session_id, session.name)Dataset.get_summary()
Retrieve this dataset’s existing AI summary.
Returns the dataset’s current AI summary if one exists. Reading never generates a summary as a side effect: a dataset that has never been summarized raises RobotoNotFoundException rather than implicitly kicking off — and paying for — generation. Call generate_summary() to create one explicitly.
Returns
StreamingAISummary wrapping the dataset’s existing summary. If a generation kicked off elsewhere is still in flight, the returned summary is Pending; poll it via await_completion or text_stream.
Raises
This dataset has no AI summary yet. Call generate_summary() to create one.
Caller lacks permission to access summaries for this dataset.
Usage
Get the existing summary, generating one first if there is none:
from roboto.exceptions import RobotoNotFoundException
dataset = Dataset.from_id("ds_abc123")
try:
summary = dataset.get_summary()
except RobotoNotFoundException:
summary = dataset.generate_summary()
print(summary.complete_text)
# 'This dataset contains 42 files with sensor data from highway driving tests...'Check whether a summary exists without generating one:
from roboto.exceptions import RobotoNotFoundException
dataset = Dataset.from_id("ds_abc123")
try:
summary = dataset.get_summary()
print(summary.complete_text)
except RobotoNotFoundException:
print("No summary yet — call generate_summary() to create one.")Dataset.get_topic_time_bounds()
Get the earliest start and latest end across every topic in this dataset.
The same aggregate you would reach by folding start_time and end_time over get_topics(), computed server-side in one request instead of one per page of topics. Reach for it when you want the dataset’s time extent and not the topics themselves.
Returns
Bounds in nanoseconds since the Unix epoch. Both fields are None for a dataset whose files hold no topics, and either is None when no topic in the dataset carries that timestamp.
Raises
Dataset does not exist.
Caller lacks permission to access the dataset.
Usage
dataset = Dataset.from_id("ds_abc123")
bounds = dataset.get_topic_time_bounds()
print(bounds.start_time, bounds.end_time)
# 1722870127699468923 1722870187004821001Dataset.get_topics()
Get all topics associated with files in this dataset, with optional filtering.
Retrieves all topics that were extracted from files in this dataset during ingestion. If multiple files have topics with the same name (e.g., chunked files with the same schema), they are returned as separate topic objects.
Topics can be filtered by name using include/exclude patterns. Topics specified on both the inclusion and exclusion lists will be excluded.
Parameters
include Optional[collections.If provided, only topics with names in this sequence are yielded.
exclude Optional[collections.If provided, topics with names in this sequence are skipped. Takes precedence over include list.
Yields
Topic instances associated with files in this dataset, filtered according to the parameters.
Return type
Usage
dataset = Dataset.from_id("ds_abc123")
for topic in dataset.get_topics():
print(f"Topic: {topic.name}")
# Topic: /camera/image
# Topic: /imu/data
# Topic: /gps/fix# Only get camera topics
camera_topics = list(dataset.get_topics(include=["/camera/image", "/camera/info"]))
print(f"Found {len(camera_topics)} camera topics")# Exclude diagnostic topics
data_topics = list(dataset.get_topics(exclude=["/diagnostics"]))Dataset.get_topics_by_file()
Get all topics associated with a specific file in this dataset.
Retrieves all topics that were extracted from the specified file during ingestion. This is a convenience method that combines file lookup and topic retrieval.
Parameters
relative_path Union[str, pathlib.Path of the file relative to the dataset root.
Yields
Topic instances associated with the specified file.
Raises
File at the given path does not exist in the dataset.
Caller lacks permission to access the file or its topics.
Return type
Usage
dataset = Dataset.from_id("ds_abc123")
for topic in dataset.get_topics_by_file("logs/session1.bag"):
print(f"Topic: {topic.name}")
# Topic: /camera/image
# Topic: /imu/data
# Topic: /gps/fixDataset.list_directories()
Yield every directory in this dataset, at any depth.
Usage
dataset = Dataset.from_id("ds_abc123")
for directory in dataset.list_directories():
print(directory.relative_path)
# logs
# logs/session1Return type
Dataset.list_files()
List files in this dataset with optional pattern-based filtering.
Returns all files in the dataset that match the specified include patterns while excluding those that match exclude patterns. Uses gitignore-style pattern matching for flexible file selection.
Parameters
include_patterns Optional[list[str]]List of gitignore-style patterns for files to include. If None 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
Caller lacks permission to list files.
Return type
Notes
Pattern matching follows gitignore syntax. See https://git-scm.com/docs/gitignore for detailed pattern format documentation.
Usage
dataset = Dataset.from_id("ds_abc123")
for file in dataset.list_files():
print(file.relative_path)
# logs/session1.bag
# data/sensors.csv
# images/camera_001.jpg# List only image files, excluding back camera
for file in dataset.list_files(
include_patterns=["**/*.png", "**/*.jpg"], exclude_patterns=["**/back_camera/**"]
):
print(file.relative_path)
# images/front_camera_001.jpg
# images/side_camera_001.jpgProperties
Dataset.metadata
Custom metadata associated with this dataset.
Returns a copy of the dataset’s metadata dictionary containing arbitrary key-value pairs for storing custom information. Supports nested structures and dot notation for accessing nested fields.
Note: this attribute is kept for backward compatibility. Prefer get_metadata(), since metadata may need to be loaded on-demand from the server.
Dataset.modified
Timestamp when this dataset was last modified.
Returns the UTC datetime when this dataset was most recently updated. This includes changes to metadata, tags, description, or other properties.
Dataset.modified_by
Identifier of the user or service which last modified this dataset.
Returns the identifier of the person or service which most recently updated this dataset’s metadata, tags, description, or other properties.
Dataset.name
Human-readable name of this dataset.
Returns the optional display name for this dataset. Can be None if no name was provided during creation. For users whose organizations have their own idiomatic internal dataset IDs, it’s recommended to set the name to the organization’s internal dataset ID, since the Roboto dataset_id property is randomly generated.
Dataset.org_id
Organization identifier that owns this dataset.
Returns the unique identifier of the organization that owns and has primary access control over this dataset.
Dataset.put_metadata()
Add or update metadata fields for this dataset.
Sets each key-value pair in the provided dictionary as dataset metadata. If a key doesn’t exist, it will be created. If it exists, the value will be overwritten. Keys must be strings and dot notation is supported for nested keys.
Parameters
metadata dict[str, Any]Dictionary of metadata key-value pairs to add or update.
Raises
Caller lacks permission to update the dataset.
Return type
Usage
dataset = Dataset.from_id("ds_abc123")
dataset.put_metadata(
{
"vehicle_id": "vehicle_001",
"test_type": "highway_driving",
"weather.condition": "sunny",
"weather.temperature": 25,
}
)
print(dataset.metadata["vehicle_id"])
# 'vehicle_001'
print(dataset.metadata["weather"]["condition"])
# 'sunny'Dataset.put_tags()
Add or update tags for this dataset.
Adds each tag in the provided sequence to the dataset. If a tag already exists, it will not be duplicated. This operation replaces the current tag list with the provided tags.
Parameters
Sequence of tag strings to set on the dataset.
Raises
Caller lacks permission to update the dataset.
Return type
Usage
dataset = Dataset.from_id("ds_abc123")
dataset.put_tags(["highway", "autonomous", "test", "sunny"])
print(dataset.tags)
# ['highway', 'autonomous', 'test', 'sunny']Dataset.query()
Query datasets using a specification with filters and pagination.
Searches for datasets matching the provided query specification. Results are returned as a generator that automatically handles pagination, yielding Dataset instances as they are retrieved from the API.
Parameters
spec Optional[roboto.Query specification with filters, sorting, and pagination options. If None, returns all accessible datasets.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
owner_org_id Optional[str]Organization ID to scope the query. If None, uses caller’s org.
Yields
Dataset instances matching the query specification.
Raises
ValueErrorQuery specification references unknown dataset attributes.
Caller lacks permission to query datasets.
Return type
Usage
from roboto.query import Comparator, Condition, QuerySpecification
spec = QuerySpecification(
condition=Condition(field="name", comparator=Comparator.Contains, value="Roboto")
)
for dataset in Dataset.query(spec):
print(f"Found dataset: {dataset.name}")
# Found dataset: Roboto Test
# Found dataset: Other Roboto TestProperties
Dataset.record
Underlying data record for this dataset.
Returns the raw DatasetRecord that contains all the dataset’s data fields. This provides access to the complete dataset state as stored in the platform.
Dataset.refresh()
Refresh this dataset instance with the latest data from the platform.
Fetches the current state of the dataset from the Roboto platform and updates this instance’s data. Useful when the dataset may have been modified by other processes or users.
Returns
This Dataset instance with refreshed data.
Raises
Dataset no longer exists.
Caller lacks permission to access the dataset.
Usage
dataset = Dataset.from_id("ds_abc123")
# Dataset may have been updated by another process
refreshed_dataset = dataset.refresh()
print(f"Current file count: {len(list(refreshed_dataset.list_files()))}")Dataset.remove_metadata()
Remove each key in this sequence from dataset metadata if it exists. Keys must be strings. Dot notation is supported for nested keys.
Usage
from roboto.domain import datasets
dataset = datasets.Dataset(...)
dataset.remove_metadata(["foo", "baz.qux"])Parameters
metadata roboto.Return type
Dataset.remove_tags()
Remove each tag in this sequence if it exists
Parameters
Return type
Dataset.rename_directory()
Rename or move a directory within this dataset.
Both old_path and new_path are relative to the dataset root. Pass a new_path with fewer path components to move the directory up the tree, or a different leaf name at the same depth to rename in place.
Parameters
old_path strCurrent relative path of the directory (e.g. "logs/session1").
new_path strTarget relative path of the directory (e.g. "session1" to move up one level).
Returns
Updated DirectoryRecord reflecting the new path.
Raises
No directory exists at old_path.
new_path conflicts with an existing node or contains a cycle.
Usage
dataset = Dataset.from_id("ds_abc123")
dataset.rename_directory("logs/session1", "session1")Dataset.rename_file()
Rename or move a file within this dataset.
new_path is relative to the dataset root. Pass a path with fewer components to move the file up the tree, a different name at the same depth to rename in place, or a path under a different directory to move sideways.
The file’s storage URI is unchanged; only the logical location in the dataset hierarchy moves.
Parameters
file_id strID of the file to rename or move.
new_path strTarget relative path for the file within this dataset (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
No file with file_id exists.
new_path conflicts with an existing file, the parent directory does not exist, or the move would create a cycle.
Usage
dataset = Dataset.from_id("ds_abc123")
record = dataset.rename_file("file_xyz789", "file.bag")
record.relative_path
# 'file.bag'Dataset.set_custom_field()
Dataset.set_custom_fields()
Dataset.set_device_id()
Set the device ID for this dataset.
Parameters
device_id Optional[str]The device ID to set for this dataset. If None, the device association will be cleared.
create_device_if_missing boolIf True, and a device_id is provided that does not exist in the organization, a new device will be created automatically. If False, and a device_id is provided that does not exist, a RobotoDeviceNotFoundException will be raised.
Returns
This Dataset instance with refreshed data.
Raises
A device_id has been provided in this request, but was not found as a device registered with Roboto for the organization this dataset is being created in.
Dataset.set_summary()
Explicitly set the AI summary text for this dataset.
This method is intended to be used in cases where an action or other active component is able to generate a more specialized summary than Dataset::generate_summary would, and you want to make that summary canonical from the perspective of the UI and Dataset::get_summary.
Parameters
summary strThe summary text to set for this dataset. This text will be rendered as Markdown, and can include
roboto specialized// entity links for rich UI linking.
Returns
This Dataset instance for method chaining.
Properties
Dataset.tags
List of tags associated with this dataset.
Returns a copy of the list of string tags that have been applied to this dataset for categorization and filtering purposes.
Dataset.to_association()
Return type
Dataset.to_dict()
Convert this dataset to a dictionary representation.
Returns the dataset’s data as a JSON-serializable dictionary containing all dataset attributes and metadata.
Returns
Dictionary representation of the dataset data.
Usage
dataset = Dataset.from_id("ds_abc123")
dataset_dict = dataset.to_dict()
print(dataset_dict["name"])
# 'Highway Test Session'
print(dataset_dict["metadata"])
# {'vehicle_id': 'vehicle_001', 'test_type': 'highway'}Dataset.update()
Update this dataset’s properties.
Updates various properties of the dataset including name, description, and metadata. Only specified parameters are updated; others remain unchanged.
Parameters
description Optional[Union[str, roboto.New description for the dataset. Set to None to clear the description.
device_id Optional[Union[str, roboto.New device ID for the dataset. Set to None to clear the device association.
metadata_changeset Union[roboto.Metadata changes to apply (add, update, or remove fields/tags).
name Optional[Union[str, roboto.New name for the dataset. Set to None to clear the name.
create_device_if_missing boolIf True, and a device_id is provided that does not exist in the organization, a new device will be created automatically. If False, and a device_id is provided that does not exist, a RobotoDeviceNotFoundException will be raised.
custom_fields_changeset Optional[roboto.Changes to apply to Ready custom-field values on this dataset. Field names not referenced by the changeset are left unchanged.
Returns
Updated Dataset instance with the new properties.
Raises
A device_id has been provided in this request, but was not found as a device registered with Roboto for the organization.
Caller lacks permission to update the dataset.
Usage
dataset = Dataset.from_id("ds_abc123")
updated_dataset = dataset.update(
name="Updated Highway Test Session", description="Updated description with more details"
)
print(updated_dataset.name)
# 'Updated Highway Test Session'# Update with metadata changes
from roboto.updates import MetadataChangeset
changeset = MetadataChangeset(put_fields={"processed": True})
updated_dataset = dataset.update(metadata_changeset=changeset)# Clear the device association
updated_dataset = dataset.update(device_id=None)# Clear the description
updated_dataset = dataset.update(description=None)Dataset.upload_directory()
Uploads all files and directories recursively from the specified directory path. You can use include_patterns and exclude_patterns to control what files and directories are uploaded, and can use delete_after_upload to clean up your local filesystem after the uploads succeed.
Usage
from roboto import Dataset
dataset = Dataset(...)
dataset.upload_directory(
pathlib.Path("/path/to/directory"),
exclude_patterns=[
"__pycache__/",
"*.pyc",
"node_modules/",
"**/*.log",
],
)Notes
- Both include_patterns and exclude_patterns follow the ‘gitignore’ pattern format described in https://git-scm.com/docs/gitignore#_pattern_format.
- If both include_patterns and exclude_patterns are provided, files matching exclude_patterns will be excluded even if they match include_patterns.
Parameters
directory_path pathlib.include_patterns Optional[list[str]]exclude_patterns Optional[list[str]]delete_after_upload boolmax_batch_size intprint_progress booldevice_id Optional[str]Return type
Dataset.upload_file()
Upload a single file to the dataset. If file_destination_path is not provided, the file will be uploaded to the top-level of the dataset.
Parameters
file_path pathlib.Local file to upload.
file_destination_path Optional[str]Destination path within the dataset. Defaults to the file’s own name at the dataset’s top level.
print_progress boolWhether to display an upload progress bar.
device_id Optional[str]Optional identifier of the device that generated this data.
Returns
The file record the upload created.
Raises
The upload reported success without reporting a file ID.
Usage
from roboto.domain import datasets
dataset = datasets.Dataset(...)
dataset.upload_file(
pathlib.Path("/path/to/file.txt"),
file_destination_path="foo/bar.txt",
)Dataset.upload_files()
Upload multiple files to the dataset.
If file_destination_paths is not provided, files will be uploaded to the top-level of the dataset.
Parameters
files collections.Local files to upload.
file_destination_paths collections.Mapping from local path to destination path within the dataset. Files not in the mapping upload to the dataset’s top level under their own name.
max_batch_size intMaximum number of files per upload transaction.
print_progress boolWhether 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.
Usage
import pathlib
from roboto.domain import datasets
dataset = datasets.Dataset.from_id("ds_abc123")
file_ids = dataset.upload_files(
[pathlib.Path("/path/to/file.txt")],
file_destination_paths={
pathlib.Path("/path/to/file.txt"): "foo/bar.txt",
},
)
file_ids[pathlib.Path("/path/to/file.txt")]
# 'fl_0123456789abcdef'DatasetRecord
Bases: pydantic.BaseModel
Wire-transmissible representation of a dataset in the Roboto platform.
DatasetRecord contains all the metadata and properties associated with a dataset, including its identification, timestamps, metadata, tags, and organizational information. This is the data structure used for API communication and persistence.
DatasetRecord instances are typically created by the platform during dataset creation operations and are updated as datasets are modified. The Dataset domain class wraps DatasetRecord to provide a more convenient interface for dataset operations.
The record includes audit information (created/modified timestamps and users), organizational context, and user-defined metadata and tags for discovery and organization purposes.
Parameters
data AnyAttributes
DatasetRecord.administrator
Deprecated field maintained for backwards compatibility. Always defaults to ‘Roboto’.
DatasetRecord.created
Timestamp when this dataset was created in the Roboto platform.
DatasetRecord.custom_fields
Values for the custom fields defined on Datasets in this org.
Every Ready custom field defined for (org_id, Dataset) appears as a key — values that have not been set surface as None rather than being absent. Empty when no custom fields are defined for the org.
DatasetRecord.dataset_id
Unique identifier for this dataset within the Roboto platform.
DatasetRecord.description
Human-readable description of the dataset’s contents and purpose.
DatasetRecord.device_id
Optional identifier of the device that generated this dataset’s data.
DatasetRecord.metadata
User-defined key-value pairs for storing additional dataset information.
DatasetRecord.modified_by
User ID or service account that last modified this dataset.
DatasetRecord.name
A short name for this dataset. This may be an org-specific unique ID that’s more meaningful than the dataset_id, or a short summary of the dataset’s contents. If provided, must be 120 characters or less.
DatasetRecord.roboto_record_version
Internal version number for this record, automatically incremented on updates.
DatasetRecord.storage_ctx
Deprecated storage context field maintained for backwards compatibility with SDK versions prior to 0.10.0.
DatasetRecord.storage_location
Deprecated storage location field maintained for backwards compatibility. Always defaults to ‘S3’.
DatasetRecord.tags
List of tags for categorizing and discovering this dataset.
DeleteFileRequest
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 AnyAttributes
DeleteFileRequest.uri
Storage URI of the file to delete (e.g., ‘s3://bucket/path/to/file.bag’).
DeleteMessagePathRequest
Bases: pydantic.BaseModel
Request to delete a message path from a topic.
Removes a message path from a topic’s schema. This operation cannot be undone and will remove all associated data and metadata for the specified path.
Parameters
data AnyDevice
A device is a non-human entity that can interact with Roboto on behalf of an organization.
Devices represent robots, systems, or other non-human entities that need to authenticate and interact with the Roboto platform. Each device is uniquely identified by a device_id within its organization and can be assigned API tokens for secure authentication.
Common device types include:
- Robots that upload log data directly from their onboard software
- Automated upload stations that collect and transmit data from multiple sources
- Edge computing devices that process and forward data to Roboto
Devices are associated with Org entities and can create Token objects for authentication. The underlying data is stored in DeviceRecord objects for wire transmission.
Device IDs are typically meaningful identifiers like serial numbers, asset tags, or other organization-specific naming schemes that help identify the physical or logical entity in the real world.
Parameters
roboto_client Optional[roboto.Device.clear_custom_field()
Clear a single custom-field value on this device to None.
Parameters
name strReturn type
Device.clear_custom_fields()
Clear multiple custom-field values on this device to None.
Parameters
names collections.Return type
Device.create()
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.
Parameters
device_id strA user-provided identifier for the device, unique within the organization. This is typically a meaningful identifier like a serial number, asset tag, or other organization-specific naming scheme.
metadata Optional[dict[str, Any]]Optional key-value pairs to associate with the device for discovery and search. For example: {“model”: “mk2”, “serial_number”: “SN001234”}.
tags Optional[list[str]]Optional list of tags to associate with the device for discovery and organization. For example: [“production”, “warehouse-a”].
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields defined on Devices in the caller’s org. Keys must match Ready field names; values must satisfy each field’s declared type.
caller_org_id Optional[str]The organization ID to register the device under. If not specified and the caller belongs to only one organization, that organization will be used. Required if the caller belongs to multiple organizations.
roboto_client Optional[roboto.Optional RobotoClient instance for API communication. If not provided, the default client configuration will be used.
Returns
A Device instance representing the newly registered device.
Raises
If a device with the same device_id already exists in the specified organization.
If the caller lacks permission to create devices in the specified organization.
If the device_id is invalid or the organization ID is malformed.
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_001Register an upload station:
device = Device.create(device_id="upload_station_alpha")
print(f"Device org: {device.org_id}")
# Device org: og_xyz789Device.create_session()
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.
Parameters
name strName of the Session, unique within this Device (max 120 characters).
description Optional[str]Optional description of the Session.
metadata Optional[dict[str, Any]]Optional initial metadata. Sessions are not filterable or sortable by metadata keys; for queryable structured attributes, define a custom field on the Session entity type.
tags Optional[collections.Optional initial tags. Sessions can be filtered by tag membership but are not sortable by tag.
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields defined on Sessions in this Device’s org. Keys must match Ready field names; values must satisfy each field’s declared type.
anchor Optional[roboto.Optional wall-clock anchor, the real-world instant at which the declared data’s time 0 occurred. An int is nanoseconds since the Unix epoch; any other Time is read as to_epoch_nanoseconds() reads it (a datetime or ISO 8601 string is that instant; a float, Decimal, or numeric string is seconds since the epoch). It applies to every file entry that does not carry its own anchor_ns.
files Optional[collections.Files composing this Session, with the topics whose data each one carries. Every file must already be uploaded, and may appear at most once. Files can also be included after creation with add_file() or add_files().
Returns
The created Session.
Raises
TypeErrorIf anchor is not one of the Time types.
ValueErrorIf anchor is a boolean, a negative number (an int, float, Decimal, or numeric string), or a string that is neither a number of seconds nor an ISO 8601 timestamp. Raised before anything is sent to the platform.
OverflowErrorIf anchor is an infinite float, Decimal, or string, such as "inf". Raised before anything is sent to the platform.
pydantic.ValidationErrorIf name is empty or longer than 120 characters, anchor does not fall after the Unix epoch or is too large for a signed 64-bit integer of nanoseconds, a file appears in more than one entry, files declares more than MAX_FILES_AND_TOPICS_PER_REQUEST files and topics combined, or representations name one file in two storage formats. Raised while the request is being built, before anything is sent to the platform.
If the platform refuses the declaration, either because it contradicts data the platform already holds or because it carries a value the platform rejects, such as a custom_fields value that does not satisfy its field’s declared type. No Session, file attachment, topic, or time range is created; the topic identifiers and schema definitions the declaration resolved stay stored, and a resend reuses them.
If something the declaration was prepared against changed while it was being written. Nothing is created; resending is the fix.
If this Device is no longer registered, the file_id of a file entry, or of a representation one of its topics lists, does not name a file in this Device’s organization whose status is Available, or the declaration names something else that does not exist, such as a custom_fields key naming a custom field the organization does not define on Sessions. Nothing is created.
If the caller lacks permission to create Sessions on this Device, or to edit a file the declaration declares topics on, a file it anchors, or a file a listed representation names, or lacks topic edit access in this Device’s organization while the declaration states is_default_for_reads on a timeline source.
If the platform refuses the declaration under an error code this SDK release does not define. Carries the code and message the platform sent.
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),
)Device.create_sessions()
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.
Parameters
sessions collections.One declaration per Session to create. An empty sequence returns an empty response without contacting the platform.
Returns
A BatchResponse with one element per declaration, in request order, holding either the Session the declaration created or why the platform refused it. A declaration naming a Session this Device already holds yields that Session rather than a second one.
Raises
pydantic.ValidationErrorIf more than MAX_SESSIONS_PER_REQUEST declarations are given, the batch declares more than MAX_FILES_AND_TOPICS_PER_REQUEST files and topics combined, the same session name is declared more than once, or representations name one file in two storage formats. All are enforced when the request body is constructed, before anything is sent to the platform.
If the batch is malformed. Nothing is created.
If this Device is no longer registered, the file_id of a file entry, or of a representation one of its topics lists, does not name a file in this Device’s organization whose status is Available, or a Session a declaration reuses is deleted before the call completes. Nothing is created.
If the caller lacks permission to create Sessions on this Device, or to edit a file a declaration declares topics on, a file it anchors, or a file a listed representation names, or lacks topic edit access in this Device’s organization while a declaration states is_default_for_reads on a timeline source.
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
# []Device.create_token()
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.
Parameters
expiry_days intNumber of days until the token expires. Defaults to 366 days (1 year). Must be a positive integer.
name Optional[str]Human-readable name for the token. If not provided, defaults to “{org_id}_{device_id}” format.
description Optional[str]Optional description explaining the token’s purpose or usage context.
api_scopes Optional[collections.Optional set of API scopes to limit the token’s permissions. If not provided, the token will have full access to all APIs.
Returns
A tuple containing:
- Token: The Token object representing the created token
- str: The secret token value (only available at creation time)
Raises
If token creation fails or the secret is not returned by the server (this should never happen under normal circumstances).
If the caller lacks permission to create tokens for this device.
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_def789ghi012Properties
Device.created
The timestamp when this device was registered with Roboto.
Device.created_by
The user ID of the person who registered this device.
Device.custom_fields
Custom-field values defined on Devices in this org.
Every Ready CustomField defined for (org_id, Device) appears as a key. Values that have not been set on this device surface as None rather than being absent. Empty when no custom fields are defined for the org.
A Timestamp value is returned as an ISO 8601 string.
Device.delete()
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.
Parameters
keep_files boolMove the device’s files to the org root instead of deleting them. Requires permission to upload files to the org.
Raises
If the caller lacks permission to delete this device.
If the device has already been deleted or does not exist.
Return type
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 successfullyDelete a device but keep its calibrations and manifests in the org root:
device = Device.from_id("old_robot_001")
device.delete(keep_files=True)Properties
Device.device_id
This device’s ID. Device ID is a user-provided identifier for a device, which is unique within the device’s org.
Device.encoded_device_id
The device ID, URL-encoded. This is useful for constructing URLs to Roboto APIs which contain the device ID.
Device.files
The files associated with this device: its calibrations, part manifests, and the like.
These are the device’s own files, distinct from the dataset files whose device_id names this device as the one that recorded them.
Raises
The device has no universal_device_id, which only a Roboto deployment that predates device files returns.
Return type
Device.for_org()
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.
Parameters
org_id strThe organization ID to list devices for.
roboto_client Optional[roboto.Optional RobotoClient instance for API communication. If not provided, the default client configuration will be used.
Returns
A generator of Device objects. For organizations with many devices, this may involve multiple service calls, and the generator will yield results as they become available.
Raises
If the caller lacks permission to list devices in the specified organization.
If the specified organization does not exist.
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: 2Device.from_id()
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.
Parameters
device_id strThe device ID to look up. This is the user-provided identifier that was specified when the device was created.
roboto_client Optional[roboto.Optional RobotoClient instance for API communication. If not provided, the default client configuration will be used.
org_id Optional[str]The organization ID that owns the device. If not specified and the caller belongs to only one organization, that organization will be used. Required if the caller belongs to multiple organizations.
Returns
A Device object representing the specified device.
Raises
If the specified device is not registered with Roboto or does not exist in the specified organization.
If the caller lacks permission to access the device or the specified organization.
If the device_id or org_id parameters are malformed.
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_abc123Retrieve a device:
device = Device.from_id("upload_station_alpha")
print(f"Found device created by: {device.created_by}")
# Found device created by: user@example.comDevice.get_or_create()
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.
Parameters
device_id strA user-provided identifier for the device, unique within the organization.
metadata Optional[dict[str, Any]]Optional key-value pairs to associate with the device on first registration.
tags Optional[list[str]]Optional tags to associate with the device on first registration.
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields, applied on first registration.
caller_org_id Optional[str]The organization the device belongs to. Required if the caller belongs to multiple organizations.
roboto_client Optional[roboto.Optional RobotoClient instance for API communication.
Returns
The newly registered or pre-existing Device.
Raises
If the caller lacks permission to create devices in, or read devices from, the specified organization.
If the device_id is invalid or the organization ID is malformed.
If the device is deleted between the registration attempt and the lookup that follows it. Those are two calls rather than one atomic operation, so the race is possible, though unlikely.
Usage
device = Device.get_or_create(device_id="aloha_001")
device.device_id
# 'aloha_001'Device.list_sessions()
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)Return type
Properties
Device.metadata
Key-value metadata pairs associated with this device.
Device.modified
The timestamp when this device record was last modified.
Device.modified_by
The user ID of the person who last modified this device record.
Device.put_metadata()
Add or update metadata fields for this device.
Parameters
metadata dict[str, Any]Key-value pairs to add or update in the device’s metadata. Existing keys will be overwritten, new keys will be added.
Returns
Updated Device instance with the new metadata.
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.0Device.put_tags()
Add tags to this device.
Parameters
tags list[str]List of tags to add to the device. Duplicate tags will be ignored.
Returns
Updated Device instance with the new tags added.
Usage
device = Device.from_id("robot_001")
updated_device = device.put_tags(["production", "warehouse-c"])
print("production" in updated_device.tags)
# TrueProperties
Device.record
Underlying DeviceRecord for this device.
This is the wire representation used in API requests and may evolve over time; prefer the public Device API unless you need direct access to the record.
Device.remove_metadata()
Remove metadata fields from this device.
Parameters
keys list[str]List of metadata keys to remove from the device.
Returns
Updated Device instance with the specified metadata keys removed.
Usage
device = Device.from_id("robot_001")
updated_device = device.remove_metadata(["old_field", "deprecated_key"])Device.remove_tags()
Remove tags from this device.
Parameters
tags list[str]List of tags to remove from the device.
Returns
Updated Device instance with the specified tags removed.
Usage
device = Device.from_id("robot_001")
updated_device = device.remove_tags(["old_tag", "deprecated"])Device.set_custom_field()
Device.set_custom_fields()
Properties
Device.tokens()
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.
Returns
A sequence of Token objects representing all tokens created for this device. The sequence may be empty if no tokens have been created.
Raises
If the caller lacks permission to list tokens for this device.
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_ghi789jkl012Check 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 deviceDevice.update()
Update device properties using a structured request.
Parameters
UpdateDeviceRequest containing the changes to apply.
Returns
Updated Device instance with the changes applied.
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"]
)
)
)DeviceRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of a device.
This record contains all the essential information about a device that can be transmitted over the network. It includes metadata about when the device was created and modified, along with its organizational association.
Parameters
data AnyAttributes
DeviceRecord.custom_fields
Values for the custom fields defined on Devices in this org.
Every Ready custom field defined for (org_id, Device) appears as a key; values that have not been set surface as None rather than being absent. Empty when no custom fields are defined for the org.
DeviceRecord.device_id
A user-provided identifier for a device, which is unique within that device’s org.
DeviceRecord.metadata
Key-value metadata pairs associated with this device.
DeviceRecord.modified
Date/time when this device record was last modified.
EvaluateTriggersRequest
Bases: pydantic.BaseModel
Request payload to manually evaluate specific triggers.
Used to force evaluation of triggers outside of their normal automatic evaluation cycle. This is typically used for testing or debugging trigger behavior.
Parameters
data AnyAttributes
EvaluateTriggersRequest.trigger_evaluation_ids
Collection of trigger evaluation IDs to process.
Event
Represents an event within the Roboto platform.
An event is a time-anchored annotation that relates Roboto entities (datasets, files, topics, and message paths) to specific time periods. Events enable temporal analysis, data correlation, and annotation of activities across different data sources.
Events serve as temporal markers that can:
- Annotate specific time periods in your data
- Associate multiple entities (datasets, files, topics, message paths) with time ranges
- Enable time-based data retrieval and analysis
- Support metadata and tagging for organization and search
- Provide visual markers in timeline views and analysis tools
Events can represent instantaneous moments (point in time) or time ranges. They are particularly useful for marking significant occurrences like sensor anomalies, system events, behavioral patterns, or any other time-based phenomena in your data.
Events cannot be instantiated directly through the constructor. Use the class methods Event.create() to create new events or Event.from_id() to load existing events.
Parameters
roboto_client Optional[roboto.Event.clear_custom_field()
Clear a single custom-field value on this event to None.
Parameters
name strReturn type
Event.clear_custom_fields()
Clear multiple custom-field values on this event to None.
Parameters
names collections.Return type
Properties
Event.create()
Create a new event associated with at least one dataset, file, topic, or message path.
Creates a time-anchored event that can be associated with various Roboto entities. For instantaneous events (a point in time), only start_time is required. Otherwise, both start_time and end_time should be provided. These fields accept nanoseconds since the UNIX epoch, or any other compatible representations supported by to_epoch_nanoseconds().
Events must be associated with at least one entity. While associations, file_ids, topic_ids, dataset_ids and message_path_ids are all optional, at least one of them must contain a valid association for the event.
Parameters
name strHuman-readable name for the event. Required.
start_time roboto.Start timestamp of the event as nanoseconds since UNIX epoch, or any value convertible by to_epoch_nanoseconds().
end_time Optional[roboto.End timestamp of the event. If not provided, defaults to start_time for instantaneous events.
associations Optional[collections.Collection of Association objects linking the event to specific entities.
dataset_ids Optional[collections.Dataset IDs to associate the event with.
file_ids Optional[collections.File IDs to associate the event with.
topic_ids Optional[collections.Topic IDs to associate the event with.
message_path_ids Optional[collections.Message path IDs to associate the event with.
description Optional[str]Optional human-readable description of the event.
metadata Optional[dict[str, Any]]Key-value metadata for discovery and search.
tags Optional[list[str]]Tags for categorizing and searching the event.
display_options Optional[roboto.Visual display options such as color.
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields defined on Events in the caller’s org. Keys must match Ready field names; values must satisfy each field’s declared type.
caller_org_id Optional[str]Organization ID of the SDK caller. If not provided, uses the caller’s organization.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
Event instance with the provided attributes and associations.
Raises
Invalid parameters (e.g., start_time > end_time), or associations point to non-existent resources.
Caller lacks permission to access associated entities.
Usage
Create an event for a sensor anomaly on a specific topic:
from roboto.domain.events import Event
event = Event.create(
name="Temperature Spike",
start_time=1722870127699468923,
end_time=1722870127799468923,
description="Unusual temperature readings detected",
topic_ids=["tp_abc123"],
tags=["anomaly", "temperature"],
metadata={"severity": "high", "sensor_id": "temp_01"},
)Create an instantaneous event on a file:
event = Event.create(
name="System Boot",
start_time="1722870127.699468923", # String format also supported
file_ids=["fl_xyz789"],
tags=["system", "boot"],
)Create an event with display options:
from roboto.domain.events import EventDisplayOptions
event = Event.create(
name="Critical Alert",
start_time=1722870127699468923,
end_time=1722870127799468923,
dataset_ids=["ds_abc123"],
display_options=EventDisplayOptions(color="red"),
metadata={"alert_type": "critical", "component": "engine"},
)Properties
Event.created
Date and time when this event was created.
Event.custom_fields
Custom-field values defined on Events in this org.
Every Ready CustomField defined for (org_id, Event) appears as a key. Values that have not been set on this event surface as None rather than being absent. Empty when no custom fields are defined for the org.
A Timestamp value is returned as an ISO 8601 string.
Event.dataset_ids()
Get dataset IDs associated with this event.
Parameters
strict_associations boolIf True, only return datasets with direct associations. If False (default), also return datasets inferred from file and topic associations.
Returns
List of unique dataset IDs associated with this event.
Usage
Get all associated dataset IDs:
event = Event.from_id("ev_abc123")
dataset_ids = event.dataset_ids()
print(f"Associated with {len(dataset_ids)} datasets")Get only directly associated datasets:
strict_dataset_ids = event.dataset_ids(strict_associations=True)
print(f"Directly associated with {len(strict_dataset_ids)} datasets")Event.delete()
Delete this event permanently.
This operation cannot be undone. The event and all its associations will be permanently removed from the platform.
Raises
Caller lacks permission to delete this event.
Event has already been deleted or does not exist.
Return type
Usage
Delete an event:
event = Event.from_id("ev_abc123")
event.delete()
# Event is now permanently deletedConditional deletion:
event = Event.from_id("ev_abc123")
if "temporary" in event.tags:
event.delete()
print("Temporary event deleted")Event.delete_many()
Delete multiple events.
Authorization works just like delete(): you can delete any event you’re able to manage. If the list includes an event you’re not allowed to delete, the whole request is rejected and nothing is deleted. This operation cannot be undone. Event IDs that don’t exist are ignored, so the call is idempotent and safe to retry.
The bulk delete API caps each request at MAX_EVENTS_PER_DELETE_BATCH event IDs. This method removes that limit for callers by splitting event_ids into chunks of that size and sending one request per chunk. Because each chunk is its own request, deleting a very large number of events is not atomic: if a request fails partway through, earlier chunks stay deleted. The operation is idempotent, so retrying with the same IDs safely finishes the job.
Parameters
event_ids collections.IDs of the events to delete. May exceed the per-request limit; they are batched automatically.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Raises
Caller isn’t allowed to delete one of the requested events.
Return type
Usage
Delete several events at once:
from roboto.domain.events import Event
Event.delete_many(["ev_abc123", "ev_def456", "ev_ghi789"])Delete the events surfaced by a query:
events = list(Event.get_by_dataset("ds_abc123"))
Event.delete_many([event.event_id for event in events])Properties
Event.description
Optional human-readable description of the event.
Event.display_options
Display options for the event, such as color.
Event.file_ids()
Get file IDs associated with this event.
Parameters
strict_associations boolIf True, only return files with direct associations. If False (default), also return files inferred from topic and message path associations.
Returns
List of unique file IDs associated with this event.
Usage
Get all associated file IDs:
event = Event.from_id("ev_abc123")
file_ids = event.file_ids()
print(f"Associated with {len(file_ids)} files")Get only directly associated files:
strict_file_ids = event.file_ids(strict_associations=True)
for file_id in strict_file_ids:
print(f"Directly associated file: {file_id}")Event.from_id()
Load an existing event by its ID.
Parameters
event_id strUnique identifier of the event to retrieve.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
Event instance for the specified ID.
Raises
Event with the specified ID does not exist.
Caller lacks permission to access the event.
Usage
Load an event by ID:
event = Event.from_id("ev_abc123")
print(f"Event: {event.name}")
print(f"Created: {event.created}")Load and update an event:
event = Event.from_id("ev_abc123")
updated_event = event.set_description("Updated description")
print(f"New description: {updated_event.description}")Event.get_by_associations()
Retrieve all events associated with the provided associations.
Returns events that match any of the provided associations. Events that you don’t have access to will be filtered out of the response rather than raising an exception.
Parameters
associations collections.Collection of Association objects to query events for.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Yields
Event instances associated with any of the specified associations.
Return type
Usage
Query events for multiple associations:
from roboto import Association
associations = [Association.topic("tp_abc123"), Association.file("fl_xyz789")]
events = list(Event.get_by_associations(associations))
for event in events:
print(f"Event: {event.name}")Query events for a specific dataset and file combination:
associations = [Association.dataset("ds_abc123"), Association.file("fl_xyz789")]
events = list(Event.get_by_associations(associations))Event.get_by_dataset()
Retrieve all events associated with a specific dataset.
Returns events that are associated with the given dataset. By default, this includes events associated with the dataset itself, as well as events associated with any files or topics within that dataset. Use strict_associations=True to only return events with direct dataset associations.
Parameters
dataset_id strID of the dataset to query events for.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
strict_associations boolIf True, only return events with direct dataset associations. If False (default), also return events associated with files or topics within the dataset.
Yields
Event instances associated with the specified dataset.
Return type
Usage
Get all events for a dataset (including file and topic events):
events = list(Event.get_by_dataset("ds_abc123"))
for event in events:
print(f"Event: {event.name} at {event.start_time}")Get only events directly associated with the dataset:
strict_events = list(Event.get_by_dataset("ds_abc123", strict_associations=True))
print(f"Found {len(strict_events)} dataset-level events")Process events in batches:
for event in Event.get_by_dataset("ds_abc123"):
if "anomaly" in event.tags:
print(f"Anomaly event: {event.name}")Event.get_by_file()
Retrieve all events with a direct association to a specific file.
Parameters
file_id strID of the file to query events for.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Yields
Event instances directly associated with the specified file.
Return type
Usage
Get all events for a specific file:
events = list(Event.get_by_file("fl_xyz789"))
for event in events:
print(f"File event: {event.name}")Check if a file has any events:
file_events = list(Event.get_by_file("fl_xyz789"))
if file_events:
print(f"File has {len(file_events)} events")
else:
print("No events found for this file")Event.get_by_message_path()
Retrieve all events with a direct association to a specific message path.
Parameters
message_path_id strID of the message path to query events for.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Yields
Event instances directly associated with the specified message path.
Return type
Usage
Get all events for a specific message path:
events = list(Event.get_by_message_path("mp_abc123"))
for event in events:
print(f"Message path event: {event.name}")Find events within a time range for a message path:
events = Event.get_by_message_path("mp_abc123")
filtered_events = [event for event in events if event.start_time >= 1722870127699468923]Event.get_by_topic()
Retrieve all events with a direct association to a specific topic.
Parameters
topic_id strID of the topic to query events for.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Yields
Event instances directly associated with the specified topic.
Return type
Usage
Get all events for a specific topic:
events = list(Event.get_by_topic("tp_abc123"))
for event in events:
print(f"Topic event: {event.name}")Analyze event patterns for a topic:
events = list(Event.get_by_topic("tp_abc123"))
anomaly_events = [e for e in events if "anomaly" in e.tags]
print(f"Found {len(anomaly_events)} anomaly events")Event.get_data()
Iteratively yield records of the underlying topic data this event annotates.
An event can be associated with data at multiple resolutions:
- as an event on its containing dataset, file, and/or topic,
- but also directly with the message path (“signal data”)
A single event can also span signals that share a timeline, so it may annotate multiple topics in a file, or even multiple files in a dataset.
For now, getting the underlying signal data associated with an event only works for events that can be sourced to a single topic (extracted from one file, uploaded to one dataset). This means that the event must have been made on either a single file, a single topic, or one or many message paths within that topic.
If the event was made on a file, topic_name must be provided, and either or both of message_paths_include or message_paths_exclude may be provided, but are optional.
If the event was made on a topic, either or both of message_paths_include or message_paths_exclude may be provided, but are optional. topic_name, if provided in this instance, is ignored.
If the event was made on one or many message paths, each of those message paths must be found in the same topic. topic_name, message_paths_include, and message_paths_exclude, if provided in this instance, are ignored.
If the event is associated with data at multiple resolutions (e.g., two message paths, one topic, one file), this method will consider the lowest resolution associations first (message path), then topic, then file.
If message_paths_include or message_paths_exclude are defined, they should be dot notation paths that match attributes of individual data records. If a partial path is provided, it is treated as a wildcard, matching all subpaths.
For example, given topic data with the following interface:
{
"velocity": {
"x": <uint32>,
"y": <uint32>,
"z": <uint32>
}
}Calling get_data on an Event associated with that topic like:
event.get_data(message_paths_include=["velocity.x", "velocity.y"])
is expected to give the same output as:
event.get_data(message_paths_include=["velocity"], message_paths_exclude=["velocity.z"])
Parameters
message_paths_include Optional[collections.message_paths_exclude Optional[collections.topic_name Optional[str]topic_data_service Optional[roboto.cache_dir Union[str, pathlib.strict_associations boolReturn type
Event.get_data_as_df()
Return the underlying topic data this event annotates as a pandas DataFrame.
Collects all data from get_data() and returns it as a pandas DataFrame with the log time as the index. Requires installing this package using the roboto[analytics] extra.
Parameters
message_paths_include Optional[collections.Dot notation paths to include in the data.
message_paths_exclude Optional[collections.Dot notation paths to exclude from the data.
topic_name Optional[str]Required when event is associated with a file.
topic_data_service Optional[roboto.Service for accessing topic data.
cache_dir Union[str, pathlib.Directory for caching downloaded data.
strict_associations boolReturns
DataFrame containing the event’s underlying topic data, indexed by log time.
Raises
ImportErrorIf pandas is not installed (install with roboto[analytics]).
Invalid parameters or event associations.
Usage
Get event data as a DataFrame:
event = Event.from_id("ev_abc123")
df = event.get_data_as_df()
print(f"Data shape: {df.shape}")
print(df.head())Get specific message paths as DataFrame:
df = event.get_data_as_df(message_paths_include=["velocity.x", "velocity.y"])
print(df.columns.tolist())Analyze event data:
df = event.get_data_as_df()
print(f"Event duration: {df.index.max() - df.index.min()} ns")
print(f"Data points: {len(df)}")Event.message_path_ids()
Get message path IDs directly associated with this event.
Returns
List of unique message path IDs directly associated with this event.
Usage
Get message path IDs:
event = Event.from_id("ev_abc123")
msgpath_ids = event.message_path_ids()
print(f"Associated with {len(msgpath_ids)} message paths")Properties
Event.metadata
Key-value metadata associated with this event.
Event.modified
Date and time when this event was last modified.
Event.put_metadata()
Add or update metadata fields for this event.
Parameters
metadata dict[str, Any]Dictionary of key-value pairs to add or update.
Returns
Updated Event instance.
Usage
Add metadata to an event:
event = Event.from_id("ev_abc123")
updated_event = event.put_metadata({"severity": "high", "component": "engine", "alert_id": "alert_001"})
print(updated_event.metadata["severity"])
# 'high'Event.put_tags()
Replace all tags for this event.
Parameters
tags list[str]List of tags to set for this event.
Returns
Updated Event instance.
Usage
Set tags for an event:
event = Event.from_id("ev_abc123")
updated_event = event.put_tags(["anomaly", "critical", "engine"])
print(updated_event.tags)
# ['anomaly', 'critical', 'engine']Properties
Event.record
Underlying event record data.
Event.refresh()
Refresh this event’s data from the server.
Fetches the latest version of this event from the server, updating all properties to reflect any changes made by other processes.
Returns
This Event instance with refreshed data.
Usage
Refresh an event to get latest changes:
event = Event.from_id("ev_abc123")
# Event may have been updated by another process
refreshed_event = event.refresh()
print(f"Current description: {refreshed_event.description}")Event.remove_metadata()
Remove metadata fields from this event.
Parameters
metadata roboto.Sequence of metadata field names to remove. Supports dot notation for nested fields.
Returns
Updated Event instance.
Usage
Remove specific metadata fields:
event = Event.from_id("ev_abc123")
updated_event = event.remove_metadata(["severity", "temp_data.max"])
# Fields 'severity' and nested 'temp_data.max' are now removedEvent.remove_tags()
Remove specific tags from this event.
Parameters
Sequence of tag names to remove from this event.
Returns
Updated Event instance.
Usage
Remove specific tags:
event = Event.from_id("ev_abc123")
updated_event = event.remove_tags(["temporary", "draft"])
# Tags 'temporary' and 'draft' are now removedEvent.set_color()
Set the display color for this event.
Parameters
color Optional[str]CSS-compatible color value (e.g., “red”, “#ff0000”, “rgb(255,0,0)”). Use None to clear the color and use automatic coloring.
Returns
Updated Event instance.
Usage
Set event color to red:
event = Event.from_id("ev_abc123")
updated_event = event.set_color("red")
print(updated_event.color)
# 'red'Clear event color:
updated_event = event.set_color(None)
print(updated_event.color)
# NoneEvent.set_custom_field()
Event.set_custom_fields()
Event.set_description()
Set the description for this event.
Parameters
description Optional[str]New description for the event. Use None to clear the description.
Returns
Updated Event instance.
Usage
Set event description:
event = Event.from_id("ev_abc123")
updated_event = event.set_description("Updated event description")
print(updated_event.description)
# 'Updated event description'Clear event description:
updated_event = event.set_description(None)
print(updated_event.description)
# NoneEvent.set_name()
Set the name for this event.
Parameters
name strNew name for the event.
Returns
Updated Event instance.
Usage
Update event name:
event = Event.from_id("ev_abc123")
updated_event = event.set_name("Critical System Alert")
print(updated_event.name)
# 'Critical System Alert'Event.to_dict()
Convert this event to a dictionary representation.
Returns
Dictionary containing all event data in JSON-serializable format.
Usage
Convert event to dictionary:
event = Event.from_id("ev_abc123")
event_dict = event.to_dict()
print(event_dict["name"])
print(event_dict["start_time"])Event.topic_ids()
Get topic IDs associated with this event.
Parameters
strict_associations boolIf True, only return topics with direct associations. If False (default), also return topics inferred from message path associations.
Returns
List of unique topic IDs associated with this event.
Usage
Get all associated topic IDs:
event = Event.from_id("ev_abc123")
topic_ids = event.topic_ids()
print(f"Associated with {len(topic_ids)} topics")Get only directly associated topics:
strict_topic_ids = event.topic_ids(strict_associations=True)
for topic_id in strict_topic_ids:
print(f"Directly associated topic: {topic_id}")Event.update()
Update this event’s attributes.
Updates various properties of the event including name, time range, description, metadata, and display options. Only specified parameters are updated; others remain unchanged.
When provided, start_time and end_time should be integers representing nanoseconds since the UNIX epoch, or convertible to such integers by to_epoch_nanoseconds().
Parameters
name Union[str, roboto.New human-readable name for the event.
start_time Union[roboto.New start timestamp for the event.
end_time Union[roboto.New end timestamp for the event.
description Union[str, None, roboto.New description for the event. Set to None to clear existing description.
metadata_changeset Union[roboto.Changes to apply to the event’s metadata and tags.
display_options_changeset Union[roboto.Changes to apply to the event’s display options.
custom_fields_changeset Optional[roboto.Changes to apply to Ready custom-field values on this event. Field names not referenced by the changeset are left unchanged.
Returns
This Event instance with attributes updated accordingly.
Raises
ValueErrorIf start_time or end_time are negative.
If start_time > end_time.
Caller lacks permission to edit this event.
Usage
Update event name and description:
event = Event.from_id("ev_abc123")
updated_event = event.update(
name="Critical System Alert", description="Updated description with more details"
)Update event time range:
updated_event = event.update(start_time=1722870127699468923, end_time=1722870127799468923)Update metadata and display options:
from roboto.updates import MetadataChangeset
from roboto.domain.events import EventDisplayOptionsChangeset
updated_event = event.update(
metadata_changeset=MetadataChangeset(
put_fields={"severity": "high"}, put_tags=["critical", "urgent"]
),
display_options_changeset=EventDisplayOptionsChangeset(color="red"),
)EventDisplayOptions
Bases: pydantic.BaseModel
Display options for an event.
Parameters
data AnyAttributes
EventDisplayOptions.color
Display color for the event.
Used to visually distinguish events on a timeline, and optionally to signal semantic information about the event (e.g. “red” for events representing critical issues).
Any value that is permissible in CSS to define a valid color can be used here, encoded as a string. For instance, the following are all valid: “red”, “#ff0000”, “rgb(255 0 0)”.
EventDisplayOptions.has_options()
Checks whether any display options have been specified.
Return type
EventDisplayOptionsChangeset
Bases: pydantic.BaseModel
A set of changes to the display options of an event.
Parameters
data AnyEventDisplayOptionsChangeset.apply_to()
Applies this changeset to some existing display options.
Parameters
display_options EventDisplayOptionsReturn type
Attributes
EventDisplayOptionsChangeset.color
An update to an event’s color.
Use None to clear any previously set color value. On the Roboto website, the event will be displayed using an automatically selected color.
EventDisplayOptionsChangeset.has_changes()
Checks whether this changeset contains any changes.
Return type
Attributes
EventDisplayOptionsChangeset.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
EventRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of an event.
Parameters
data AnyAttributes
EventRecord.associations
Datasets, files, topics and message paths which this event pertains to.
EventRecord.custom_fields
Values for the custom fields defined on Events in this org.
Every Ready custom field defined for (org_id, Event) appears as a key; values that have not been set surface as None rather than being absent. Empty when no custom fields are defined for the org.
EventRecord.description
An optional human-readable description of the event.
EventRecord.display_options
Display options for the event, such as color.
EventRecord.end_time
The end time of the event, in nanoseconds since epoch (assumed Unix epoch). This can be equal to start_time if the event is discrete, but can never be less than start_time.
EventRecord.metadata
Key-value pairs to associate with this event for discovery and search.
EventRecord.name
A brief human-readable name for the event. Many events can have the same name. “Takeoff”, “Crash”, “CPU Spike”, “Bad Image Quality”, and “Unexpected Left” are a few potential examples.
EventRecord.start_time
The start time of the event, in nanoseconds since epoch (assumed Unix epoch).
ExecutableProvenance
ExecutorContainer
Bases: enum.Enum
Type of container running as part of an action invocation
ExperimentalWarning
Bases: Warning
Warning category for experimental APIs.
File
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
roboto_client Optional[roboto.file_service Optional[roboto.File.add_topic()
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 strName for the topic. Must be unique within this file.
df pandas.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.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
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.
ImportErrorIf pandas or pyarrow are not installed. Install with pip install roboto[ingestion] to use this feature.
This file is associated with a device or with the org itself, not with a dataset; only a dataset’s files hold topics.
If the caller lacks permission to create topics or upload files to this file’s dataset.
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_dataCreate 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 rowsAdd 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,
)Properties
File.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.
File.created
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.
File.created_by
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.
File.dataset_id
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
This file is not associated with a dataset.
Return type
File.declare_topic()
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.
Parameters
topic_name strTopic this File contributes data to. Topic names are unique within an org.
topic_schema roboto.Structure of the topic’s data.
timeline_sources collections.Timeline sources this File’s topic data carries, each with the bounds it spans in this File, stated in the File’s own timestamps.
data_range Optional[tuple[int, int]]The part of the File this topic’s data occupies, or None for the whole File.
anchor Optional[roboto.Optional wall-clock instant the data this declaration names was captured at: an int of nanoseconds since the Unix epoch, or any other Time, read as to_epoch_nanoseconds() reads it (a datetime or ISO 8601 string is that instant; a float, Decimal, or numeric string is seconds since the epoch). Must fall after the Unix epoch.
representations collections.The files a read of this topic’s data opens, each with how it holds that data: this File, when its own bytes are readable, and other files when the data is read from them, such as files converted out of it. Empty lists none, and reads of a topic with no representations return no rows.
Returns
The topic this declaration registered against.
Raises
TypeErrorIf anchor is not one of the Time types.
ValueErrorIf anchor is a boolean, a negative number (an int, float, Decimal, or numeric string), or a string that is neither a number of seconds nor an ISO 8601 timestamp. Raised before anything is sent to the platform.
OverflowErrorIf anchor is an infinite float, Decimal, or string, such as "inf". Raised before anything is sent to the platform.
pydantic.ValidationErrorIf these arguments do not form a valid FileTopicDeclaration, for instance two representations of the whole topic, or of one field, sharing a storage format, content format and transformations, or any timeline source but SchemaFieldSource beside a PARQUET representation, or if representations names one file in two storage formats. Enforced before anything is sent to the platform.
Whatever the platform refused this declaration with.
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)File.declare_topics()
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().
Parameters
topics collections.One declaration per topic and slice of this File. An empty sequence returns an empty response without contacting the platform.
Returns
One element per declaration, in request order, holding either the topic it registered against or why the platform refused it.
Raises
pydantic.ValidationErrorIf more than MAX_FILES_AND_TOPICS_PER_REQUEST topics are given, one topic is declared twice over the same slice, two topics anchor one slice at different instants, or representations name one file in two storage formats. These are enforced when the request body is constructed, before anything is sent to the platform; each declaration’s own rules are enforced earlier, when the caller builds it.
If this File no longer exists, or a representation names a file that does not exist in this File’s org or whose status is not Available. Nothing is registered.
If the caller lacks edit access to this File or to a file a listed representation names, or lacks topic edit access in this File’s org while a declaration states is_default_for_reads on a timeline source.
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],
),
],
)File.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
File does not exist or has already been deleted.
Caller lacks permission to delete the file.
Return type
Usage
file = File.from_id("file_abc123")
file.delete()
# # File is now permanently deletedProperties
File.description
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.
File.device_id
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.
File.download()
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.Local filesystem path where the file should be saved.
print_progress boolWhether to show a progress bar during download.
Raises
This file is a link whose target, at the pinned version, no longer exists.
Caller lacks permission to download the file, or a link’s target.
FileNotFoundErrorFile content is not available in storage.
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}")Properties
File.file_id
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.
File.from_id()
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 strUnique 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 client for API communication. If None, uses the default client.
Returns
File instance representing the requested file.
Raises
File with the given ID does not exist.
Caller lacks permission to access the file.
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)
# 1File.from_path_and_dataset_id()
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.Relative path of the file within the dataset.
dataset_id strID 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 client for API communication. If None, uses the default client.
Returns
File instance representing the requested file.
Raises
File at the given path does not exist in the dataset.
Caller lacks permission to access the file or dataset.
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'File.get_signed_url()
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
For a link, the URL is for the version of the target file that the link pins.
Parameters
override_content_type Optional[str]override_content_disposition Optional[str]Returns
Signed URL string that provides temporary access to the file.
Raises
This file is a link whose target, at the pinned version, no longer exists.
Caller lacks permission to access the file, or a link’s target.
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")File.get_topic()
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 strName of the topic to retrieve (e.g., “/camera/image”, “/imu/data”).
Returns
Topic instance for the specified topic name.
Raises
Topic with the given name does not exist in this file.
Caller lacks permission to access the topic.
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']}")File.get_topics()
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.If provided, only topics with names in this sequence are yielded.
exclude Optional[collections.If provided, topics with names in this sequence are skipped.
Yields
Topic instances associated with this file, filtered according to the parameters.
Return type
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"]))File.import_batch()
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.Sequence of import requests, each specifying file details and metadata.
roboto_client Optional[roboto.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
If any URI is not a valid S3 URI, if the batch exceeds 500 items, or if bucket integrations are not properly configured.
If the caller lacks upload permissions for target datasets or if buckets don’t belong to the caller’s organization.
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 filesFile.import_one()
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 strID of the dataset to import the file into.
relative_path strPath of the file relative to the dataset root (e.g., logs/session1.bag).
uri strURI 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 client for API communication. If None, uses the default client.
Returns
File object representing the imported file.
Raises
If the URI is not a valid URI or if the bucket integration is not properly configured.
If the specified object does not exist.
If the caller lacks upload permissions for the target dataset or if the bucket doesn’t belong to the caller’s 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.bagImport 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")Properties
File.ingestion_status
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.
File.is_link
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.
File.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
Caller lacks permission to update the 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.
Usage
file = File.from_id("file_abc123")
print(file.ingestion_status)
# IngestionStatus.NotIngested
updated_file = file.mark_ingested()
print(updated_file.ingestion_status)
# IngestionStatus.IngestedProperties
File.metadata
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.
File.modified
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.
File.modified_by
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.
File.org_id
Organization identifier that owns this file.
Returns the unique identifier of the organization that owns and has primary access control over this file.
File.put_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
Caller lacks permission to update the file.
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'File.put_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
Caller lacks permission to update the file.
Usage
file = File.from_id("file_abc123")
updated_file = file.put_tags(["sensor-data", "highway", "sunny"])
print(updated_file.tags)
# ['sensor-data', 'highway', 'sunny']File.query()
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 specification with filters, sorting, and pagination options. If None, returns all accessible files.
roboto_client Optional[roboto.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
ValueErrorQuery specification references unknown file attributes.
Caller lacks permission to query files.
Return type
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")Properties
File.record
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.
File.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
File no longer exists.
Caller lacks permission to access the file.
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}")Properties
File.relative_path
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.
File.rename_file()
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 strFile ID (currently unused, kept for API compatibility).
new_path strNew relative path for the file, relative to the root of its association’s files.
Returns
Updated FileRecord with the new path.
Raises
Caller lacks permission to rename the file.
New path is invalid or conflicts with existing file.
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'File.set_device_id()
Set the device ID for this file.
Parameters
device_id strThe device ID to set for this file.
Returns
Updated File instance with the new device ID.
Raises
Caller lacks permission to update the file.
The specified device ID does not exist.
Usage
file = File.from_id("file_abc123")
updated_file = file.set_device_id("device_xyz789")File.set_representations()
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:
- Remove a representation.
- 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
RepresentationDeclarationdescribes, so declaring the new one adds it beside the first. - 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.
Parameters
topics collections.One entry per topic and slice, at most MAX_FILES_AND_TOPICS_PER_REQUEST; split a larger set across several calls. An empty sequence returns without contacting the platform.
Raises
pydantic.ValidationErrortopics is longer than the cap, lists one topic and slice twice, or lists representations naming one file in two storage formats. Raised before any request is made. The rules for one topic’s own representations are enforced earlier, when the caller builds its TopicRepresentations.
This File does not exist, a topic is not declared on it, an entry’s data_range is not one the topic is declared over on it, or a representation names a file that does not exist in this File’s org or whose status is not Available. Nothing is written.
A topic declared with a timeline source other than SchemaFieldSource (MCAP log or publish time, MP4 presentation time) would get a PARQUET representation, a topic declared over a data_range would be left with a representation that cannot be read by row position and none that can covering the same fields, as transformations describes, or a representation’s field_path names no field of the schema the topic is declared under on this File. Nothing is written.
The caller cannot edit this File or a file a listed representation names.
Return type
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=[]),
]
)File.set_timeline_offset()
Calibrate this file’s timeline to Unix-epoch wall-clock, optionally scoped to a topic and/or source.
Contract:
- 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 adatetime, is the nanoseconds since the Unix epoch at which stored time 0 occurred. topic/topic_namescopes the update to a single topic in this file;timeline_source/timeline_source_namescopes 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
offset roboto.Offset to apply: an int of nanoseconds, or any other Time, read as to_epoch_nanoseconds() reads it (a datetime or ISO 8601 string is that instant; a float, Decimal, or numeric string is seconds). Must not be negative, and must fit in a signed 64-bit integer of nanoseconds.
topic Optional[roboto.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.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.
Raises
TypeErroroffset is not one of the Time types.
ValueErroroffset is a boolean, a negative number (an int, float, Decimal, or numeric string), or a string that is neither seconds nor ISO 8601; or both of a mutually exclusive pair of selectors are given. Raised before any request is made.
OverflowErroroffset is an infinite float, Decimal, or string, such as "inf". Raised before any request is made.
pydantic.ValidationErroroffset converts to a negative number of nanoseconds, or to more than a signed 64-bit integer holds. A subclass of ValueError, raised before any request is made.
The caller cannot edit this file.
The file carries no timeline data, or the selectors match none of it.
The offset would place the data it reaches, or a session time range declared over that data, before the Unix epoch or past the largest storable Unix-epoch nanosecond value. Nothing is written.
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",
)File.set_timeline_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.Offset entries to apply, each with its own selectors.
Returns
The updated TimelineExtentRecord objects returned by the server.
Raises
pydantic.ValidationErroroffsets is empty. Raised before any request is made.
The caller cannot edit this file.
The file carries no timeline data, or the selectors of every entry together match none of it.
An entry’s offset would place the data it reaches, or a session time range declared over that data, before the Unix epoch or past the largest storable Unix-epoch nanosecond value. The whole request is refused and nothing is written.
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"),
]
)Properties
File.tags
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.
File.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.
Usage
file = File.from_id("file_abc123")
association = file.to_association()
print(f"Association: {association.association_type}:{association.association_id}")
# Association: file:file_abc123File.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.
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'}File.update()
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.New description for the file. Use NotSet to leave unchanged.
metadata_changeset Union[roboto.Metadata changes to apply (add, update, or remove fields/tags). Use NotSet to leave metadata unchanged.
ingestion_complete Union[Literal[True], roboto.Set to True to mark the file as fully ingested. Use NotSet to leave ingestion status unchanged.
device_id Optional[Union[str, roboto.New device ID for the file. Use NotSet to leave unchanged.
Returns
Updated File instance with the new properties.
Raises
Caller lacks permission to update the file.
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)Properties
File.uri
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.
File.version
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.
FileRecord
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 AnyAttributes
FileRecord.association_id
Properties
FileRecord.bucket
Name of the bucket holding this file’s object.
Raises
This record is a link, which stores no object.
Return type
Attributes
FileRecord.created
FileRecord.created_by
FileRecord.description
FileRecord.device_id
FileRecord.file_id
FileRecord.ingestable
Whether this file is meant to be ingested: its path matched one of its org’s ingestion rules when this version was created or when the file was last renamed or moved, or it has since been partly or fully ingested. A file that is ingestable and ingestion_status not_ingested is awaiting ingestion.
FileRecord.ingestion_status
Properties
FileRecord.is_link
Whether this record is a link to another file rather than a file with an object of its own.
FileRecord.key
Key of this file’s object within bucket.
Raises
This record is a link, which stores no object.
Return type
Attributes
FileRecord.metadata
FileRecord.modified
FileRecord.modified_by
FileRecord.name
FileRecord.org_id
FileRecord.origination
FileRecord.parent_id
FileRecord.relative_path
FileRecord.size
FileRecord.status
FileRecord.storage_type
FileRecord.tags
FileRecord.upload_id
FileRecord.uri
FileRecord.version
FileRecordRequest
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 AnyAttributes
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.
Attributes
FileStatus.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.
FileStatus.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.
FileStatus.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.
FileSystem
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.roboto_client Optional[roboto.file_service Optional[roboto.org_id Optional[str]Properties
FileSystem.association
The dataset, device, or org whose files this object works on.
FileSystem.create_directory()
Create a directory among the association’s files.
Parameters
name strName of the directory to create.
error_if_exists boolIf True, raises an exception if the directory already exists.
parent_path Optional[pathlib.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 boolIf True, creates intermediate directories in the path if they don’t exist. If False, requires all parent directories to already exist.
Raises
If the directory already exists and error_if_exists is True.
If the caller lacks permission to create the directory.
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.
Usage
from roboto.domain import devices
device = devices.Device.from_id("rover-01")
directory = device.files.create_directory("calib")
print(directory.relative_path)
# calibdirectory = device.files.create_directory(
name="final",
parent_path=pathlib.Path("path/to/deep"),
create_intermediate_dirs=True,
)
print(directory.relative_path)
# path/to/deep/finalFileSystem.create_link()
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.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 strWhere the link sits, relative to the root of the association’s files.
Returns
The link, whose is_link is True.
Raises
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.
The target is a link or a directory, is in another org, or does not exist at the version to pin.
The caller cannot edit the association’s files or view 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"))FileSystem.delete_files()
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
Caller lacks permission to delete files.
Return type
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/**"])FileSystem.download_files()
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.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 boolWhether to show a progress bar during download.
Returns
List of tuples containing (FileRecord, local_path) for each downloaded file.
Raises
A selected file’s path resolves outside out_path; nothing is downloaded.
Caller lacks permission to download files.
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 filesFileSystem.get_file_by_path()
Get a File instance for the association’s file at the specified path.
Parameters
relative_path Union[str, pathlib.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
The association has no file at the given path.
Caller lacks permission to access the file.
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_xyz789old_file = device.files.get_file_by_path("manifest.json", version_id=1)
print(old_file.version)
# 1FileSystem.list_directories()
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
# urdfReturn type
FileSystem.list_files()
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
Caller lacks permission to list files.
Return type
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.yamlfor file in device.files.list_files(include_patterns=["calib/**"], exclude_patterns=["**/*.bak"]):
print(file.relative_path)
# calib/front_cam.yamlFileSystem.rename_directory()
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 strCurrent relative path of the directory (e.g. "logs/session1").
new_path strTarget relative path of the directory (e.g. "session1" to move up one level).
Returns
Updated DirectoryRecord reflecting the new path.
Raises
No directory exists at old_path.
new_path conflicts with an existing node or contains a cycle.
Usage
from roboto.domain import devices
device = devices.Device.from_id("rover-01")
device.files.rename_directory("calib/front", "front_calib")FileSystem.rename_file()
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 strID of the file to rename or move.
new_path strTarget 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
No file with file_id exists.
new_path conflicts with an existing file, the parent directory does not exist, or the move would create a cycle.
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'FileSystem.upload_directory()
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.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 boolIf True, each uploaded local file is deleted once the uploads succeed.
max_batch_size intMaximum number of files per upload transaction.
print_progress boolWhether to display an upload progress bar.
device_id Optional[str]Optional identifier of the device that generated this data.
Return type
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"],
)FileSystem.upload_file()
Upload a single file associated with association.
Parameters
file_path pathlib.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 boolWhether 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
The upload reported success without reporting a file ID.
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"))FileSystem.upload_files()
Upload multiple files associated with association.
Parameters
files collections.Local files to upload.
file_destination_paths collections.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 intMaximum number of files per upload transaction.
print_progress boolWhether 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.
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'FileTag
Bases: enum.Enum
Enumeration of system-defined file tag types.
These tags are used internally by the platform for indexing and organizing files. They are automatically applied during file operations and should not be manually modified by users.
Attributes
FileTag.AssociationId
Tag containing the ID of the dataset, device, or org a file is associated with.
FileTag.CommonPrefix
Tag containing the common path prefix for files in a batch operation.
FileTag.DatasetId
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.
FileTag.TransactionId
Tag containing the transaction ID for files uploaded in a batch.
FilesChangesetFileManager
This class is used to pre-write tags/metadata updates to files which haven’t been uploaded yet, but will be at the conclusion of an action, by virtue of being in that action’s output directory.
It uses a “file changeset” file to accumulate these pending updates during an action’s runtime, and then applies them automatically at the end of an action, after the action’s output directory has been uploaded.
The most common way to get access to this would be via roboto.action_runtime.InvocationContext.
FilesChangesetFileManager.put_fields()
Adds metadata key/value pairs to a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.
This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.file_changeset_manager
# This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 2, "trucks": 3})
# Actually there was a 3rd car I missed in the first pass, and a plane, let me fix that...
file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 3, "planes": 1})Parameters
relative_path strmetadata dict[str, Any]FilesChangesetFileManager.put_tags()
Adds tags to a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.
This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.file_changeset_manager
# This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
file_changeset_manager.put_tags("images/front0_raw_000734.jpg", ["cloudy", "rainy"]})Parameters
relative_path strtags list[str]FilesChangesetFileManager.remove_fields()
Removes metadata key/value pairs from a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime. You’ll generally only need this to remove values which were added by a previous call to put_fields().
This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.file_changeset_manager
# This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
file_changeset_manager.put_fields("images/front0_raw_000734.jpg", {"cars": 2, "trucks": 3})
# Whoops, actually I don't want to count those trucks...
file_changeset_manager.remove_fields("images/front0_raw_000734.jpg", ["trucks"])Parameters
relative_path strkeys list[str]FilesChangesetFileManager.remove_tags()
Removes tags from a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime. You’ll generally only need this to remove tags which were added by a previous call to put_tags().
This can be called multiple times throughout the runtime of an action, and will be accumulated accordingly. Order of calls matters.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.file_changeset_manager
# This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
file_changeset_manager.put_tags("images/front0_raw_000734.jpg", ["cloudy", "rainy"]})
# Actually this is just Seattle's aggressive mist, that's not really rainy...
file_changeset_manager.remove_tags("images/front0_raw_000734.jpg", ["rainy"]})Parameters
relative_path strtags list[str]FilesChangesetFileManager.set_description()
Sets the human-readable description of a to-be-uploaded file which is expected to be written to ${ROBOTO_OUTPUT_DIR}/relative_path by the end of the user portion of an action’s runtime.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()
file_changeset_manager = context.file_changeset_manager
# This would reference a file at ${ROBOTO_OUTPUT_DIR}/images/front0_raw_000734.jpg
file_changeset_manager.set_description("images/front0_raw_000734.jpg", "This image was over-exposed")Parameters
relative_path strdescription Optional[str]ImportFileRequest
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 AnyAttributes
ImportFileRequest.description
Optional human-readable description of the file.
ImportFileRequest.device_id
Optional identifier of the device that generated this data.
ImportFileRequest.metadata
Optional key-value metadata pairs to associate with the file.
ImportFileRequest.relative_path
Path of the file relative to the dataset root (e.g., logs/session1.bag).
ImportFileRequest.size
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.
ImportFileRequest.tags
Optional list of tags for file discovery and organization.
ImportFileRequest.uri
Storage URI where the file is located (e.g., s3://bucket/path/to/file.bag).
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.
Attributes
IngestionStatus.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.
IngestionStatus.NotIngested
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.
IngestionStatus.PartlyIngested
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.
Invocation
An instance of an execution of an action, initiated manually by a user or automatically by a trigger.
An Invocation represents a single execution of an Action with specific inputs, parameters, and configuration. It tracks the execution lifecycle from creation through completion, including status updates, logs, and results.
Invocations are created by calling Action.invoke() or through the UI. They cannot be created directly through the constructor. Each invocation has a unique ID and maintains a complete audit trail of its execution.
Key features:
- Status tracking (Queued, Running, Completed, Failed, etc.)
- Input data specification and parameter values
- Compute requirement and container parameter overrides
- Log collection and output file management
- Progress monitoring and result retrieval
Parameters
roboto_client Optional[roboto.Properties
Invocation.action
Provenance information about the action that was invoked.
Invocation.cancel()
Cancel this invocation if it is not already in a terminal status.
Attempts to cancel the invocation. If the invocation has already completed, failed, or reached another terminal status, this method has no effect.
Raises
If the invocation is not found.
If the caller lacks permission to cancel the invocation.
Return type
Usage
Cancel a running invocation:
invocation = Invocation.from_id("iv_12345")
if not invocation.reached_terminal_status:
invocation.cancel()Properties
Invocation.compute_requirements
The compute requirements (CPU, memory) used for this invocation.
Invocation.container_parameters
The container parameters used for this invocation.
Invocation.created
The timestamp when this invocation was created.
Invocation.current_status
The current status of this invocation (e.g., Queued, Running, Completed).
Invocation.data_source
The data source that provided input data for this invocation.
Invocation.executable
Provenance information about the executable (container) that was run.
Invocation.from_id()
Load an existing invocation by its ID.
Retrieves an invocation from the Roboto platform using its unique identifier.
Parameters
invocation_id strThe unique ID of the invocation to retrieve.
roboto_client Optional[roboto.Roboto client instance. Uses default if not provided.
Returns
The Invocation instance.
Raises
If the invocation is not found.
If the caller lacks permission to access the invocation.
Usage
Load an invocation and check its status:
invocation = Invocation.from_id("iv_12345")
print(f"Status: {invocation.current_status}")
print(f"Created: {invocation.created}")Invocation.get_logs()
Retrieve runtime STDOUT/STDERR logs generated during this invocation’s execution.
Fetches log records from the invocation’s container execution, with support for pagination to handle large log volumes.
Parameters
page_token Optional[str]Optional token for pagination. If provided, starts retrieving logs from that point.
Yields
LogRecord instances containing log messages and metadata.
Raises
If the invocation is not found.
If the caller lacks permission to access logs.
Return type
Properties
Invocation.input_data
The input data specification for this invocation, if any.
Invocation.is_queued_for_scheduling()
An invocation is queued for scheduling if:
1. its most recent status is “Queued” 3. and is not “Deadly”
Return type
Invocation.query()
Query invocations with optional filtering and pagination.
Searches for invocations based on the provided query specification. Can filter by status, action name, creation time, and other attributes.
Parameters
spec Optional[roboto.Query specification with filters, sorting, and pagination. If not provided, returns all accessible invocations.
owner_org_id Optional[str]Organization ID to search within. If not provided, searches in the caller’s organization.
roboto_client Optional[roboto.Roboto client instance. Uses default if not provided.
Yields
Invocation instances matching the query criteria.
Raises
ValueErrorIf the query specification contains unknown fields.
If the query filters or sorts on a field the invocations API does not accept.
If the caller lacks permission to query invocations.
Return type
Usage
Query all invocations:
for invocation in Invocation.query():
print(f"Invocation: {invocation.id}")Query invocations whose data source is a given dataset:
from roboto.query import Comparator, Condition, QuerySpecification
spec = QuerySpecification(
condition=Condition(
field="data_source_id",
comparator=Comparator.Equals,
value="ds_abc123",
)
)
for invocation in Invocation.query(spec):
print(invocation.id)Query completed invocations:
from roboto.domain.actions import InvocationStatus
spec = QuerySpecification(
condition=Condition(
field="last_status",
comparator=Comparator.Equals,
value=InvocationStatus.Completed.value,
)
)
completed = list(Invocation.query(spec))Query the ten most recent invocations. Neither limit nor max_results caps an invocation query, so take the first ten from the generator:
import itertools
from roboto.query import SortDirection
spec = QuerySpecification(sort_by="created", sort_direction=SortDirection.Descending)
recent = list(itertools.islice(Invocation.query(spec), 10))Properties
Invocation.reached_terminal_status
True if this invocation has reached a terminal status (Completed, Failed, etc.).
Invocation.record
The underlying invocation record containing all invocation data.
Invocation.refresh()
Return type
Invocation.set_container_image_digest()
This is an admin-only operation to memorialize the digest of the container image that was pulled in the course of invoking the action.
Parameters
digest strReturn type
Invocation.set_logs_location()
This is an admin-only operation to memorialize the base location where invocation logs are saved.
Use the “get_logs” or “stream_logs” methods to access invocation logs.
Parameters
Return type
Properties
Invocation.source
Provenance information about the source that initiated this invocation.
Invocation.status_log
The complete history of status changes for this invocation.
Invocation.stream_logs()
Parameters
last_read Optional[str]Return type
Properties
Invocation.to_dict()
Return type
Invocation.update_status()
Parameters
detail Optional[str]Return type
Properties
Invocation.upload_destination
The destination where output files from this invocation will be uploaded.
Invocation.wait_for_terminal_status()
Wait for the invocation to reach a terminal status.
Throws a TimeoutError if the timeout is reached.
Parameters
timeout floatThe maximum amount of time, in seconds, to wait for the invocation to reach a terminal status.
poll_interval roboto.The amount of time, in seconds, to wait between polling iterations.
Return type
InvocationContext
A utility for performing common lookups and other operations during a Roboto Action’s runtime.
The easiest and most common way to initialize this is:
from roboto import InvocationContext
context = InvocationContext.from_env()…which will inspect environment variables to initialize the InvocationContext.
If you want to test a script using InvocationContext in a setting such as a developer machine or unit test, and you don’t want to set environment variables to mirror Roboto’s remote execution environment, initialize InvocationContext directly:
import pathlib
from roboto import InvocationContext
context = InvocationContext(
dataset_id="ds_XXXXXXXXXXXX",
input_dir=pathlib.Path("/path/to/tmp/input/dir"),
invocation_id="iv_XXXXXXXXXXXX",
org_id="og_XXXXXXXXXXXX",
output_dir=pathlib.Path("/path/to/tmp/output/dir"),
)Parameters
dataset_id strinput_dir pathlib.invocation_id strorg_id stroutput_dir pathlib.input_data_manifest_file Optional[pathlib.parameters_file Optional[pathlib.secrets_file Optional[pathlib.roboto_client Optional[roboto.dry_run boollog_level Optional[str]Properties
InvocationContext.dataset
A Dataset instance for the dataset whose data this action is operating on, if any.
This resource will be lazily initialized the first time it is accessed. After the first call, the dataset will be cached.
This is particularly useful for adding tags or metadata to a dataset at runtime.
Raises
If the dataset does not exist.
If the dataset is not specified (e.g., when running locally, using scheduled triggers, or invoking via CLI with query-based input data).
Return type
Usage
Add tags/metadata to the dataset:
context.dataset.put_tags(["tagged_by_action"])
context.dataset.put_metadata({"voltage_spikes_seen": 693})InvocationContext.dataset_id
The ID of the dataset whose data this action is operating on.
InvocationContext.file_changeset_manager
A FilesChangesetFileManager which can be used to associate tags and metadata with the yet-to-be-uploaded files in this invocation’s output directory. In practice, you might use this like:
from roboto import InvocationContext
context = InvocationContext.from_env()
my_output_file = context.output_dir / "my_output_file.txt"
my_output_file.write_text("Hello World")
context.file_changeset_manager.put_tags(my_output_file.name, ["tagged_by_action"])
context.file_changeset_manager.put_fields(
my_output_file.name, {"roboto_proficiency": "extreme - I can annotate output files!"}
)This only works for files that have not yet been uploaded to Roboto. To tag existing files, you should instead use:
from roboto import InvocationContext
context = InvocationContext.from_env()
existing_file = context.dataset.get_file_by_path("some_file_that_already_exists.txt")
existing_file.put_tags(["tagged_by_action"])
existing_file.put_metadata({"roboto_proficiency": "also extreme - I can annotate input files!"})For more info, see the top-level docs on the FilesChangesetFileManager class.
InvocationContext.from_env()
Initialize an InvocationContext from values in environment variables. Will throw an exception if any required environment variables are not available.
All required environment variables will be available at runtime when an action is running in Roboto’s remote execution environment.
Usage
from roboto import InvocationContext
context = InvocationContext.from_env()Return type
InvocationContext.get_input()
Instance of ActionInput containing resolved references to input data.
InvocationContext.get_optional_parameter()
Retrieve the value of the action parameter with the given name, defaulting to default_value if the parameter is not set.
Parameters
name strThe name of the parameter to retrieve.
default_value Optional[str]The value to return if the parameter is not set. Defaults to None.
Returns
The parameter value, or default_value if not set. If the value is a secret URI, returns the resolved secret value.
Usage
import roboto
context = roboto.InvocationContext.from_env()
context.get_optional_parameter("model_version", "latest")
# "latest"InvocationContext.get_parameter()
Gets the value of the action parameter with the given name, raising an ActionRuntimeException if the parameter is not set.
Parameters
name strReturn type
InvocationContext.get_secret_parameter()
Gets the value of the secret action parameter with the given name.
Parameters
name strReturn type
Properties
InvocationContext.input_dir
The directory where the action’s input files are located.
InvocationContext.invocation
An Invocation object for the currently running action invocation.
This object will be lazy-initialized the first time it is accessed, which might result in a RobotoNotFoundException if the invocation does not exist. After the first call, the invocation will be cached.
InvocationContext.invocation_id
The ID of the currently running action invocation.
InvocationContext.log_level
The log level for the action invocation.
Returns
The log level constant (e.g., logging.DEBUG, logging.INFO, logging.WARNING, logging.ERROR) if set, or logging.INFO if no log level was specified.
InvocationContext.org
An Org object for the org which invoked the currently running action.
This object will be lazy-initialized the first time it is accessed, which might result in a RobotoNotFoundException if the org does not exist. After the first call, the org will be cached.
InvocationContext.org_id
The ID of the org which invoked the currently running action.
InvocationContext.output_dir
The directory where the action’s output files are expected. After the user portion of the action runtime concludes (i.e. when their container exits with a 0 exit code), every file in this directory will be uploaded to the dataset associated with this action invocation.
InvocationContext.roboto_client
The RobotoClient instance used by this action runtime.
InvocationDataSource
Bases: pydantic.BaseModel
Abstracted data source that can be provided to an invocation.
Represents a source of input data for action invocations. The data source type determines how the ID should be interpreted (e.g., as a dataset ID).
Parameters
data AnyAttributes
InvocationDataSource.data_source_id
The ID of the data source. For Dataset type, this is a dataset ID.
InvocationDataSource.data_source_type
The type of data source (currently only Dataset).
InvocationDataSource.is_unspecified()
Check if this data source is unspecified.
Returns
True if this is an unspecified data source, False otherwise.
InvocationDataSource.unspecified()
Returns a special value indicating that no invocation source is specified.
Returns
An InvocationDataSource instance representing an unspecified data source.
InvocationDataSourceType
Bases: enum.Enum
Source of data for an action’s input binding.
Defines the type of data source that provides input data to an action invocation. Currently supports datasets, with potential for future expansion to other data source types.
Attributes
InvocationDataSourceType.Dataset
InvocationProvenance
Bases: pydantic.BaseModel
Provenance information for an invocation
Parameters
data AnyAttributes
InvocationProvenance.executable
The underlying executable (e.g., Docker image) that was run.
InvocationRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of an invocation.
Parameters
data AnyAttributes
InvocationRecord.compute_requirements
InvocationRecord.container_parameters
InvocationRecord.created
InvocationRecord.data_source
InvocationRecord.duration
InvocationRecord.idempotency_id
InvocationRecord.input_data
InvocationRecord.invocation_id
InvocationRecord.last_heartbeat
InvocationRecord.last_status
InvocationRecord.org_id
InvocationRecord.parameter_values
InvocationRecord.provenance
InvocationRecord.rich_input_data
InvocationRecord.status
InvocationRecord.timeout
InvocationRecord.upload_destination
InvocationSource
InvocationStatus
Bases: int, enum.Enum
Invocation status enum
Attributes
InvocationStatus.Cancelled
InvocationStatus.Completed
InvocationStatus.Deadly
InvocationStatus.Downloading
InvocationStatus.Failed
InvocationStatus.Processing
InvocationStatus.Queued
InvocationStatus.Scheduled
InvocationStatus.Uploading
InvocationStatus.can_transition_to()
Parameters
other InvocationStatusReturn type
InvocationStatus.from_value()
Parameters
v Union[int, str]Return type
InvocationStatus.is_running()
Return type
InvocationStatus.is_terminal()
Return type
InvocationStatus.next()
Return type
InvocationStatusRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of an invocation status.
Parameters
data AnyAttributes
InvocationStatusRecord.detail
InvocationStatusRecord.status
InvocationStatusRecord.timestamp
InvocationStatusRecord.to_presentable_dict()
Return type
InvocationUploadDestination
Bases: pydantic.BaseModel
Default destination to which invocation outputs - if any - should be uploaded.
Specifies where files generated during action execution should be stored. Actions can write files to their output directory, and this destination determines where those files are uploaded after execution completes.
Parameters
data AnyInvocationUploadDestination.dataset()
Create a dataset upload destination with the given ID.
Parameters
dataset_id strThe ID of the dataset where outputs should be uploaded.
Returns
An InvocationUploadDestination configured for the specified dataset.
Attributes
InvocationUploadDestination.destination_id
Optional identifier for the upload destination. In the case of a dataset, it would be the dataset ID.
InvocationUploadDestination.destination_type
Type of upload destination. By default, outputs are uploaded to a dataset.
Properties
InvocationUploadDestination.is_dataset
True if this is a dataset destination with a dataset ID, False otherwise.
Returns
True if this destination is configured for a dataset with a valid ID.
InvocationUploadDestination.is_unknown
True if the upload destination is not of a supported type, False otherwise.
InvocationUploadDestination.pre_validate_destination_type()
Parameters
value AnyReturn type
LogRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of a log record.
Parameters
data AnyAttributes
LogRecord.log
LogRecord.partial_id
LogRecord.process
LogRecord.timestamp
LogsLocation
MessagePath
Represents a message path within a topic in the Roboto platform.
A message path defines a specific field or signal within a topic’s data schema, using dot notation to specify nested attributes. Message paths enable fine-grained access to individual data elements within time-series robotics data, supporting operations like statistical analysis, data filtering, and visualization.
Each message path has an associated data type (both native and canonical), metadata, and statistical information computed from the underlying data. Message paths are the fundamental building blocks for data analysis in Roboto, allowing users to work with specific signals or measurements from complex robotics data structures.
Message paths support temporal filtering, data export to various formats including pandas DataFrames, and integration with the broader Roboto analytics ecosystem. They provide efficient access to time-series data while maintaining the semantic structure of the original robotics messages.
The MessagePath class serves as the primary interface for accessing individual data signals within topics, providing methods for data retrieval, statistical analysis, and metadata management.
Parameters
roboto_client Optional[roboto.topic_data_service Optional[roboto.Attributes
MessagePath.DELIMITER
Properties
MessagePath.canonical_data_type
Canonical Roboto data type corresponding to the native data type.
MessagePath.count
Number of data points available for this message path.
MessagePath.created
Timestamp when this message path was created.
MessagePath.created_by
Identifier of the user or system that created this message path.
MessagePath.data_type
Native data type for this message path, e.g. ‘float32’
MessagePath.from_id()
Retrieve a message path by its unique identifier.
Fetches a message path record from the Roboto platform using its unique ID. This is useful when you have a message path identifier from another operation.
Parameters
message_path_id strUnique identifier for the message path.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
topic_data_service Optional[roboto.Service for accessing topic data. If None, creates a default instance.
Returns
MessagePath instance representing the requested message path.
Raises
Message path with the given ID does not exist.
Caller lacks permission to access the message path.
Usage
message_path = MessagePath.from_id("mp_abc123")
print(message_path.path)
# 'angular_velocity.x'
print(message_path.canonical_data_type)
# CanonicalDataType.NumberMessagePath.get_data()
Return data for this specific message path.
Retrieves and yields data records containing only the values for this message path, with optional temporal filtering. This provides a focused view of a single signal or field within the broader topic data.
Parameters
start_time Optional[roboto.Start time (inclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
end_time Optional[roboto.End time (exclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
cache_dir Union[str, pathlib.Directory where topic data will be downloaded if necessary. Defaults to DEFAULT_CACHE_DIR.
Yields
Dictionary records containing the log_time and the value for this message path.
Return type
Notes
For each example below, assume the following is a sample datum record that can be found in this message path’s associated topic:
{
"angular_velocity": {
"x": <uint32>,
"y": <uint32>,
"z": <uint32>
},
"orientation": {
"x": <uint32>,
"y": <uint32>,
"z": <uint32>,
"w": <uint32>
}
}Usage
Print all data for a specific message path:
topic = Topic.from_name_and_file("/imu/data", "file_abc123")
angular_velocity_x = topic.get_message_path("angular_velocity.x")
for record in angular_velocity_x.get_data():
print(f"Time: {record['log_time']}, Value: {record['angular_velocity']['x']}")Get data within a time range:
for record in angular_velocity_x.get_data(start_time=1722870127699468923, end_time=1722870127799468923):
print(record)Collect data into a dataframe (requires installing the roboto[analytics] extra):
df = angular_velocity_x.get_data_as_df()
import math
assert math.isclose(angular_velocity_x.mean, df[angular_velocity_x.path].mean())MessagePath.get_data_as_df()
Return this message path’s data as a pandas DataFrame.
Retrieves message path data and converts it to a pandas DataFrame for analysis and visualization. The DataFrame is indexed by log time and contains a column for this message path’s values.
Parameters
start_time Optional[roboto.Start time (inclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
end_time Optional[roboto.End time (exclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
cache_dir Union[str, pathlib.Directory where topic data will be downloaded if necessary. Defaults to DEFAULT_CACHE_DIR.
Returns
pandas DataFrame containing the message path data, indexed by log time.
Raises
ImportErrorpandas is not installed. Install with roboto[analytics] extra.
Notes
Requires installing this package using the roboto[analytics] extra.
Usage
topic = Topic.from_name_and_file("/imu/data", "file_abc123")
angular_velocity_x = topic.get_message_path("angular_velocity.x")
df = angular_velocity_x.get_data_as_df()
print(df.head())
# angular_velocity.x
# log_time
# 1722870127699468923 0.1
# 1722870127699468924 0.15
print(f"Mean: {df[angular_velocity_x.path].mean()}")
# Mean: 0.125Properties
MessagePath.max
Maximum value observed for this message path.
MessagePath.mean
Mean (average) value for this message path.
MessagePath.median
Median value for this message path.
MessagePath.message_path_id
Unique identifier for this message path.
MessagePath.metadata
Metadata dictionary associated with this message path.
MessagePath.min
Minimum value observed for this message path.
MessagePath.modified
Timestamp when this message path was last modified.
MessagePath.modified_by
Identifier of the user or system that last modified this message path.
MessagePath.p25
25th percentile of the values observed for this message path.
MessagePath.p75
75th percentile of the values observed for this message path.
MessagePath.p95
95th percentile of the values observed for this message path.
MessagePath.p99
99th percentile of the values observed for this message path.
MessagePath.parents()
Get parent paths for a message path.
Given a path_in_schema (list of path components), returns a list of its parent paths ordered from most specific to least specific.
Parameters
path_in_schema list[str]List of path components (e.g., [“pose”, “pose”, “position”, “x”]).
Returns
List of parent paths in dot notation, ordered from most to least specific.
Raises
TypeErrorIf a string is passed instead of a list. This method previously accepted a dot-delimited string; passing a string now would silently iterate over its characters and produce wrong results.
Usage
path_in_schema = ["pose", "pose", "position", "x"]
MessagePath.parents(path_in_schema)
# ['pose.pose.position', 'pose.pose', 'pose']# Single level path has no parents
MessagePath.parents(["velocity"])
# []Properties
MessagePath.path
Dot-delimited path to the attribute (e.g., ‘pose.position.x’).
MessagePath.record
Underlying MessagePathRecord for this message path.
MessagePath.stddev
Standard deviation of the values observed for this message path.
MessagePath.to_association()
Convert this message path to an Association object.
Creates an Association object that can be used to reference this message path in other parts of the Roboto platform.
Returns
Association object representing this message path.
Usage
message_path = MessagePath.from_id("mp_abc123")
association = message_path.to_association()
print(association.association_type)
# AssociationType.MessagePath
print(association.association_id)
# mp_abc123Properties
MessagePath.topic_id
Unique identifier of the topic containing this message path.
MessagePathChangeset
Bases: pydantic.BaseModel
Changeset for batch operations on topic message paths.
Defines a collection of add, delete, and update operations to be applied to a topic’s message paths in a single atomic operation. Useful for making multiple schema changes efficiently.
Parameters
data AnyMessagePathChangeset.check_replace_all_correctness()
Return type
MessagePathChangeset.from_replacement_message_paths()
Create a changeset that replaces all existing message paths.
Creates a changeset that will replace all existing message paths on a topic with the provided set of message paths. This is useful for completely redefining a topic’s schema.
Parameters
message_paths collections.Sequence of message path requests to replace existing paths.
Returns
MessagePathChangeset configured to replace all existing message paths.
Usage
from roboto.domain.topics import AddMessagePathRequest, CanonicalDataType
new_paths = [
AddMessagePathRequest(
message_path="velocity.x", data_type="float32", canonical_data_type=CanonicalDataType.Number
)
]
changeset = MessagePathChangeset.from_replacement_message_paths(new_paths)MessagePathChangeset.has_changes()
Check whether the changeset contains any actual changes.
Returns
True if the changeset contains operations that would modify the topic’s message paths.
Attributes
MessagePathChangeset.message_paths_to_add
Message paths to add to a topic.
MessagePathChangeset.message_paths_to_delete
Message paths to delete from a topic.
MessagePathChangeset.message_paths_to_update
Message paths to update on a topic.
MessagePathChangeset.replace_all
Flag indicating whether this changeset should replace all message paths on a topic.
It assumes that the replacement message paths will be provided via message_paths_to_add. Rather than setting this flag directly, use appropriate class methods such as from_replacement_message_paths.
MessagePathMetadataWellKnown
Bases: roboto.compat.StrEnum
Well-known metadata key names (with well-known semantics) that may be set in metadata.
These are most often set by Roboto’s first-party ingestion actions and used by Roboto clients.
Attributes
MessagePathMetadataWellKnown.Categories
An ordered list of values that a Categorical can take.
Usage
"categories"=["off", "on"]"categories"=["left", "up", "right", "down"]
MessagePathMetadataWellKnown.ColumnName
The original name or path to this field in the source data schema. May differ from message_path if character substitutions were applied to conform to naming requirements.
Notes
- Use of this metadata field is soft-deprecated as of SDK v0.24.0.
- Prefer use of
source_pathandpath_in_schemainstead. Those attributes are now first-class fields onMessagePathRecordand can be specified viaAddMessagePathRequest.
MessagePathRecord
Bases: pydantic.BaseModel
Record representing a message path within a topic.
Defines a specific field or signal within a topic’s data schema, including its data type, metadata, and statistical information. Message paths use dot notation to specify nested attributes within complex message structures.
Message paths are the fundamental units for accessing individual data elements within time-series robotics data, enabling fine-grained analysis and visualization of specific signals or measurements.
Parameters
data AnyAttributes
MessagePathRecord.canonical_data_type
Normalized data type, used primarily internally by the Roboto Platform.
MessagePathRecord.created
MessagePathRecord.created_by
MessagePathRecord.data_type
‘Native’/framework-specific data type of the attribute at this path. E.g. “float32”, “uint8[]”, “geometry_msgs/Pose”, “string”.
MessagePathRecord.message_path
Dot-delimited path to the attribute within the datum record.
MessagePathRecord.message_path_id
MessagePathRecord.metadata
Key-value pairs to associate with this metadata for discovery and search, e.g. { ‘min’: ‘0.71’, ‘max’: ’1.77 }
MessagePathRecord.modified
MessagePathRecord.modified_by
MessagePathRecord.org_id
This message path’s organization ID, which is the organization ID of the containing topic.
MessagePathRecord.parents()
Logical message path ancestors of this path.
Usage
Given a deeply nested field root.sub_obj_1.sub_obj_2.leaf_field:
field = "root.sub_obj_1.sub_obj_2.leaf_field"
record = MessagePathRecord(message_path=field) # other fields omitted for brevity
print(record.parents())
# ['root.sub_obj_1.sub_obj_2', 'root.sub_obj_1', 'root']Parameters
delimiter strReturn type
Attributes
MessagePathRecord.path_in_schema
List of path components representing the field’s location in the original data schema. Unlike message_path, which must conform to Roboto-specific naming requirements and assumes dots separated path parts imply nested data, this preserves the exact path from the source data for accurate attribute access. This is expected to be the split representation of source_path.
MessagePathRecord.representations
Zero to many Representations of this MessagePath.
MessagePathRecord.source_path
The original name of this field in the source data schema. May differ from message_path if character substitutions were applied to conform to naming requirements.
This is the preferred field to use when specifying message_path_include or message_path_exclude to the get_data or get_data_as_df methods of Topic and Event.
MessagePathRecord.to_field_selection()
Translate this record into the FieldSelection the format decoders accept.
Return type
Attributes
MessagePathRecord.topic_id
MessagePathStatistic
Bases: enum.Enum
Statistics computed by Roboto in our standard ingestion actions.
Which of these a given message path actually carries depends on the ingestion path that produced it, so treat every one as optional: read them with metadata.get(...) or through the corresponding MessagePath property, both of which yield None when the statistic was never written, and never assume a missing value means zero. Indexing metadata directly raises KeyError for a statistic that was never written.
Attributes
MessagePathStatistic.Count
MessagePathStatistic.Max
MessagePathStatistic.Mean
MessagePathStatistic.Median
MessagePathStatistic.Min
MessagePathStatistic.P25
MessagePathStatistic.P75
MessagePathStatistic.P95
MessagePathStatistic.P99
MessagePathStatistic.Stddev
QueryDatasetFilesRequest
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 AnyAttributes
QueryDatasetFilesRequest.exclude_patterns
List of gitignore-style patterns for files to exclude from results.
QueryDatasetFilesRequest.include_patterns
List of gitignore-style patterns for files to include in results.
QueryDatasetFilesRequest.page_token
Token for retrieving the next page of results in paginated queries.
QueryDatasetFilesRequest.sort_by
Field to sort results by. Defaults to ‘created’.
QueryDatasetFilesRequest.sort_direction
Sort direction (‘ASC’ or ‘DESC’). Defaults to ‘DESC’.
QueryDatasetsRequest
Bases: pydantic.BaseModel
Request payload for querying datasets with filters.
Used to search for datasets based on various criteria such as metadata, tags, and other dataset properties. The filters are applied server-side to efficiently return matching datasets.
Parameters
data AnyQueryFilesRequest
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 AnyQueryTriggersRequest
Bases: pydantic.BaseModel
Request payload to query triggers with filters.
Used to search for triggers based on various criteria such as name, status, or other attributes.
Parameters
data AnyReportUploadProgressRequest
Bases: pydantic.BaseModel
Request payload for reporting file upload progress.
Used to notify the platform about the completion status of individual files within a batch upload transaction. This enables progress tracking and partial completion handling for large file uploads.
Parameters
data AnyAttributes
ReportUploadProgressRequest.manifest_items
List of file URIs that have completed upload.
RepresentationRecord
Bases: pydantic.BaseModel
Record representing a data representation for topic content.
A representation is a pointer to processed topic data stored in a specific format and location. Representations enable efficient access to topic data by providing multiple storage formats optimized for different use cases.
Most message paths within a topic point to the same representation (e.g., an MCAP or Parquet file containing all topic data). However, some message paths may have multiple representations for analytics or preview formats.
Representations are versioned and associated with specific files or storage locations through the association field.
Parameters
data AnyAttributes
RepresentationRecord.association
Identifier and entity type with which this Representation is associated. E.g., a file, a database.
RepresentationRecord.created
RepresentationRecord.format
Content format descriptor for this representation. For image topics: the image encoding (e.g. “jpeg”, “png”) for simplified representations, or the ROS schema name (e.g. “sensor_msgs/Image”) for original/passthrough representations. None for non-image topics or legacy representations.
RepresentationRecord.modified
RepresentationRecord.representation_id
RepresentationRecord.storage_format
RepresentationRecord.topic_id
RepresentationRecord.transformations
Ordered list of transformation descriptors applied to produce this representation. Empty for original/passthrough representations.
Each entry is a "<kind>:<param>" string where <kind> is a TransformationKind member. Construct entries via TransformationKind.with_param() and parse them via TransformationKind.parse() to keep the vocabulary centralized.
Example: ["downsample:0.5", "encode:jpeg"]
RepresentationRecord.version
RepresentationStorageFormat
Bases: enum.Enum
Supported storage formats for topic data representations.
Defines the available formats for storing and accessing topic data within the Roboto platform. Each format has different characteristics and use cases.
RobotoClient
A client for making HTTP requests against Roboto service
Parameters
endpoint strauth_decorator Optional[roboto.http_client_kwargs Optional[dict[str, Any]]RobotoClient.defaulted()
Parameters
client Optional[RobotoClient]Return type
RobotoClient.delete()
Parameters
path ApiRelativePathcaller_org_id Optional[str]data Anyheaders Optional[dict[str, str]]idempotent boolowner_org_id Optional[str]query Optional[dict[str, Any]]retry_wait_fn Optional[roboto.timeout roboto.Return type
Properties
RobotoClient.for_profile()
Parameters
profile strReturn type
RobotoClient.from_config()
Parameters
config roboto.Return type
RobotoClient.from_env()
Return type
Properties
RobotoClient.get()
Parameters
path ApiRelativePathcaller_org_id Optional[str]headers Optional[dict[str, str]]idempotent boolowner_org_id Optional[str]query Optional[dict[str, Any]]retry_wait_fn Optional[roboto.timeout roboto.Return type
Properties
RobotoClient.http_client
RobotoClient.patch()
Parameters
path ApiRelativePathcaller_org_id Optional[str]data Anyheaders Optional[dict[str, str]]idempotent boolowner_org_id Optional[str]query Optional[dict[str, Any]]retry_wait_fn Optional[roboto.timeout roboto.Return type
RobotoClient.post()
Parameters
path ApiRelativePathcaller_org_id Optional[str]data Anyheaders Optional[dict[str, str]]idempotent boolowner_org_id Optional[str]query Optional[dict[str, Any]]retry_wait_fn Optional[roboto.timeout roboto.Return type
RobotoClient.put()
Parameters
path ApiRelativePathcaller_org_id Optional[str]data Anyheaders Optional[dict[str, str]]idempotent boolowner_org_id Optional[str]query Optional[dict[str, Any]]retry_wait_fn Optional[roboto.timeout roboto.Return type
RobotoConfig
Bases: pydantic.BaseModel
RobotoConfig captures an api_key and endpoint required to programmatically interact with Roboto. Multiple profiles can be configured if desired.
Parameters
data AnyAttributes
RobotoConfig.api_key
RobotoConfig.cache_dir
RobotoConfig.default_http_timeout
RobotoConfig.endpoint
RobotoConfig.from_env()
Build a config from the access token in the environment, or else from a profile in the Roboto config file.
A token in ROBOTO_API_KEY or ROBOTO_BEARER_TOKEN takes precedence, and is sent to the endpoint in ROBOTO_SERVICE_ENDPOINT, or to https://api.roboto.ai when that is unset. Without a token, the config comes from the file at ROBOTO_CONFIG_FILE, or ~/.roboto/config.json when that is unset. Either way, ROBOTO_CACHE_DIR and ROBOTO_DEFAULT_HTTP_TIMEOUT, when set, override the profile’s cache_dir and default_http_timeout.
Parameters
profile_override Optional[str]Config file profile to read, ahead of ROBOTO_PROFILE and the file’s default profile.
env Optional[roboto.Roboto environment variables to read. Defaults to the process’s own, through RobotoEnv.default().
Raises
FileNotFoundErrorNo access token is set and the config file doesn’t exist.
OSErrorThe config file exists but can’t be read.
ValueErrorThe config file isn’t a JSON object, or has no usable profile by the chosen name.
Return type
RobotoConfig.get_cache_dir()
Return type
Attributes
RobotoConfig.org_id
Organization to act in when the caller names none and ROBOTO_ORG_ID is unset. The CLI and from_env() read it. Only a member of several organizations needs it. roboto setup saves it to the config file profile; a config built from ROBOTO_API_KEY or ROBOTO_BEARER_TOKEN has none.
RobotoEnv
Bases: pydantic_settings.BaseSettings
Enivronment within an action invocation runtime
Parameters
_case_sensitive bool | None_nested_model_default_partial_update bool | None_env_prefix str | None_env_prefix_target pydantic_settings._env_file pydantic_settings._env_file_encoding str | None_env_ignore_empty bool | None_env_nested_delimiter str | None_env_nested_max_split int | None_env_parse_none_str str | None_env_parse_enums bool | None_cli_prog_name str | None_cli_parse_args bool | list[str] | tuple[str, ._cli_settings_source pydantic_settings._cli_parse_none_str str | None_cli_hide_none_type bool | None_cli_avoid_json bool | None_cli_enforce_required bool | None_cli_use_class_docs_for_groups bool | None_cli_show_env_vars bool | None_cli_exit_on_error bool | None_cli_prefix str | None_cli_flag_prefix_char str | None_cli_implicit_flags bool | Literal['dual', 'toggle'] | None_cli_ignore_unknown_args bool | None_cli_kebab_case bool | Literal['all', 'no_enums'] | None_cli_shortcuts collections._secrets_dir pydantic_settings._build_sources tuple[tuple[pydantic_settings.values AnyAttributes
RobotoEnv.action_inputs_manifest_file
RobotoEnv.action_parameters_file
RobotoEnv.action_runtime_config_dir
RobotoEnv.action_timeout
RobotoEnv.api_key
RobotoEnv.cache_dir
RobotoEnv.config_file
An override for the location of a roboto config file. If not provided, the default ~/.roboto/config.json will be used (subject to RobotoConfig’s implementation)
RobotoEnv.dataset_id
RobotoEnv.dataset_metadata_changeset_file
RobotoEnv.default()
Attributes
RobotoEnv.default_http_timeout
Give up on Roboto Platform HTTP requests that take longer than this many seconds to complete. If the request is known to be idempotent, it will be automatically retried. Set to None to wait indefinitely.
RobotoEnv.dry_run
Flag that action developers should use to gate side effects during local invocation.
When set to True, actions should skip operations that have side effects, such as: - Uploading files to Roboto datasets - Modifying metadata - Making external API calls that are not idempotent
This enables safe local testing and development without affecting production resources.
RobotoEnv.file_metadata_changeset_file
RobotoEnv.get_env_var()
Parameters
var_name strdefault_value Optional[str]Return type
Attributes
RobotoEnv.input_dir
RobotoEnv.invocation_id
RobotoEnv.log_level
RobotoEnv.org_id
RobotoEnv.output_dir
RobotoEnv.profile
The profile name to use if getting RobotoConfig from a config file.
RobotoEnv.roboto_env
RobotoEnv.roboto_service_endpoint
A Roboto Service API endpoint to send requests to, typically https://api.roboto.ai
RobotoEnv.roboto_service_url
Deprecated, use roboto_service_endpoint instead. Left here until 0.3.3 is released so we can migrate existing actions to use the new env var.
RobotoRegion
Bases: roboto.compat.StrEnum
The geographic region of a Roboto resource. Used when configuring org-level default behavior for data storage, in order to ensure that data is close to your users.
RobotoSearch
A high-level interface for querying the Roboto data platform.
In most cases, using this class should be as simple as:
from roboto import RobotoSearch
rs = RobotoSearch()
for dataset in rs.find_datasets(...):
...Parameters
query_client Optional[roboto.RobotoSearch.find_collections()
Parameters
query Optional[roboto.timeout_seconds floatReturn type
RobotoSearch.find_datasets()
Parameters
Return type
RobotoSearch.find_devices()
Yield Device objects matching query, one at a time.
Results stream lazily as you iterate; timeout_seconds bounds how long iteration waits for results before stopping.
Usage
from roboto import RobotoSearch
searcher = RobotoSearch()
for device in searcher.find_devices("tags CONTAINS 'warehouse'"):
print(device.device_id)Parameters
query Optional[roboto.timeout_seconds floatReturn type
RobotoSearch.find_events()
Parameters
query Optional[roboto.timeout_seconds floatReturn type
RobotoSearch.find_files()
Parameters
query Optional[roboto.timeout_seconds floatReturn type
RobotoSearch.find_message_paths()
Parameters
query Optional[roboto.timeout_seconds floatReturn type
RobotoSearch.find_sessions()
Yield Session objects matching query, one at a time.
Submits query against the structured-query API targeting Sessions and lazily materializes each row into a Session instance bound to the caller’s RobotoClient. Iteration drives server-side pagination under the hood; timeout_seconds bounds the total wall-clock time spent waiting for query results before iteration stops.
Filterable fields:
session_id(aliasid).name.min_timestamp_ns(aliasstart_time) — inclusive lower bound of the session’s recorded time window.max_timestamp_ns(aliasend_time) — inclusive upper bound of the session’s recorded time window.duration— synthetic numeric field equal tomax_timestamp_ns - min_timestamp_ns; accepts integer nanoseconds only.dataset.dataset_id(aliasdataset.id) — matches sessions that include at least one file from the given dataset.=/!=only.device.device_id(aliasdevice.id) — matches sessions attached to the given device.=/!=only.collection.collection_id(aliascollection.id) — matches sessions that are a member of the given collection.=/!=only.metric.<name>(aliasmetrics.<name>) — matches sessions by a session metric named<name>; dots are part of the metric name (e.g.metric.cpu.load.max). Accepts value and existence comparators. The value comparators=,!=,>,>=,<,<=require a numeric value, and only match sessions that have the metric and whose value satisfies the comparison. The presence comparators take no value:IS_NOT_NULL/EXISTSmatch sessions that have the metric (any value);IS_NULL/NOT_EXISTSmatch sessions that lack it.
The four time-window fields accept any shape roboto.time.Time permits — integer epoch nanoseconds, float / Decimal / <sec>.<nsec> string seconds, ISO8601 strings, or a datetime (read as UTC when it carries no timezone) — and the server normalizes the value to epoch nanoseconds before the comparison runs.
Sortable fields: session_id, min_timestamp_ns, and duration.
Parameters
query Optional[roboto.timeout_seconds floatReturn type
RobotoSearch.find_topics()
Usage
import matplotlib.pyplot as plt
from roboto import RobotoSearch
searcher = RobotoSearch()
for topic in searcher.find_topics("msgpaths[cpuload.load].max > 0.9"):
df = topic.get_data_as_df(message_paths_include=["cpuload.load"])
plt.plot(df.index, df["cpuload.load"], label=topic.topic_id)
plt.legend()
plt.show()Parameters
query Optional[roboto.timeout_seconds floatReturn type
RobotoSearch.for_roboto_client()
Parameters
roboto_client roboto.org_id Optional[str]Return type
RobotoSearch.from_env()
Create a RobotoSearch instance configured from environment variables.
Reads authentication credentials and endpoint configuration from environment variables ($ROBOTO_API_KEY/$ROBOTO_BEARER_TOKEN, $ROBOTO_SERVICE_ENDPOINT) or the config file at $ROBOTO_CONFIG_FILE (default: ~/.roboto/config.json). If using the config file, $ROBOTO_PROFILE can be used to select a profile from the config.
$ROBOTO_ORG_ID can be used to set the organization ID to query. When it is unset, the organization queried is the org_id of the config file profile in use, which roboto setup saves. Naming an organization should only be necessary if you belong to multiple organizations.
Returns
A configured RobotoSearch instance ready to query the Roboto platform.
Usage
import roboto
roboto_search = roboto.RobotoSearch.from_env()
for dataset in roboto_search.find_datasets():
print(dataset.name)SchemaFieldRecord
Bases: pydantic.BaseModel
A single field within a topic schema.
One entry per unique field path within a schema; field paths are deduplicated across topics that share the schema.
Parameters
data AnyAttributes
SchemaFieldRecord.canonical_data_type
Normalized data type used for cross-framework compatibility and UI rendering decisions.
SchemaFieldRecord.created
SchemaFieldRecord.created_by
SchemaFieldRecord.data_type
Native, framework-specific data type of the field. E.g. “float32”, “uint8[]”, “geometry_msgs/Pose”.
SchemaFieldRecord.field_id
SchemaFieldRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SchemaFieldRecord.modified
SchemaFieldRecord.modified_by
SchemaFieldRecord.name
Human-readable display name of the field (typically the final component of path_in_schema).
SchemaFieldRecord.org_id
SchemaFieldRecord.path_in_schema
Path components locating this field in the source data schema. Each component is a schema-native attribute name, in order from the schema root to the leaf.
SchemaFieldRecord.schema_id
SchemaFieldRecord.unit
Optional unit of the field’s values (e.g., "ns", "m/s"). None if the field is unitless or unknown.
Session
An operational time window of a Device.
A Session is a drone flight, a vehicle drive, a robot arm test run: some contiguous activity in the real world. It groups the recordings, logs, and other data produced during that window. Because a Session is bounded by the activity rather than by the recordings, it can span many files or cover just a slice of one. Each file it includes can be narrowed to a sub-window of that file.
The Session’s aggregate bounds, min_timestamp_ns and max_timestamp_ns in Unix-epoch nanoseconds, span every file the Session includes. Roboto recomputes them whenever the Session’s files or the anchors of their data change, and each method of this class that makes such a change returns with the updated bounds.
A Session can reference one or many devices: a single drone for a solo mission, or all of the drones in a formation flight. Use attach_to_device() and detach_from_device() to change which devices it references.
How to create a Session:
Session.create()accepts zero, one, or many devices, and does not require a name.create_session()creates one named Session on a device, optionally declaring its files and topics in the same call.create_sessions()creates many such Sessions on a device in one call.create_session()creates a Session for an existing Dataset, inferring the devices involved and pre-populating files from the Dataset.
Once created, include files with add_file() or add_files().
Usage
Create a Session for a drone flight, include a recording, and list its topics:
from roboto.experimental.sessions import Session
session = Session.create(name="flight-2026-04-23-001", device_ids=["robot-abc"])
session.add_file("fl_0123456789abcdef")
for topic in session.list_topics():
print(topic.name)Parameters
roboto_client Optional[roboto.Session.add_file()
Include a single file in this Session, with whatever topic data it carries.
The singular form of add_files(), taking the fields of one SessionFile as separate arguments. That class documents what each field means; add_files() documents what the platform does with them.
Parameters
file Union[roboto.A File or a file ID.
data_range Optional[roboto.Slice of the file this Session holds, or None for the whole file.
min_file_timestamp_ns Optional[int]Optional lower bound of the part of the file to include, in the file’s own timestamps. Must be paired with max_file_timestamp_ns.
max_file_timestamp_ns Optional[int]Optional upper bound paired with min_file_timestamp_ns.
anchor Optional[roboto.Optional wall-clock instant at which time 0 of the data added here occurred: an int of nanoseconds since the Unix epoch, or any other Time, read as to_epoch_nanoseconds() reads it (a datetime or ISO 8601 string is that instant; a float, Decimal, or numeric string is seconds since the epoch). Must fall after the Unix epoch.
topics Optional[collections.Topics this file contributes data to, over the part of the file that carries them. Each lists the files a read of its data opens in representations.
Returns
The file’s place in this Session, as list_files() reports it.
Raises
TypeErroranchor is not one of the Time types.
ValueErroranchor is a boolean, a string that is neither seconds nor ISO 8601, or a negative number (an int, float, Decimal, or numeric string); rejected client-side, before any request is made.
OverflowErroranchor is infinite, such as float("inf"); rejected client-side, before any request is made.
pydantic.ValidationErrorThe arguments break a rule SessionFile enforces, such as an anchor at or before the Unix epoch or a time window with only one of its two bounds, or the representations listed in topics name one file in two storage formats; rejected client-side, before any request is made. The rules for one topic’s own representations are enforced earlier, when the caller builds its TopicDeclaration.
With the anchor covering it added, the file’s data or the window stated here would fall before the Unix epoch or past the largest storable Unix-epoch nanosecond value, or the anchor would move a window another Session declared over the same data there. Anchor the data at the instant it was recorded.
Whatever else the platform refused this file with.
Usage
Include a whole file:
session.add_file("fl_0123456789abcdef")Include only a sub-window of a file:
session.add_file(
"fl_0123456789abcdef",
min_file_timestamp_ns=0,
max_file_timestamp_ns=60_000_000_000,
)Session.add_files()
Include the given files in this Session, with whatever topic data they carry.
Each entry states one file’s place in this Session, in the same terms a SessionDeclaration states the files of a Session declared whole, so a Session composed file by file can say everything a declared one says.
The platform decides every refusal before adding anything, so an entry it refuses leaves the others added, while a failure it did not anticipate, such as a timeout, adds none of them. Resending converges on the same composition rather than duplicating it. The platform then recomputes this Session’s aggregate bounds across every file it includes, and this instance reflects the new min_timestamp_ns / max_timestamp_ns on return.
Parameters
files collections.Files to include in the Session, each appearing exactly once and listing all of its topics; SessionFile documents what one entry states, including how the window it names survives re-anchoring the file. An empty sequence returns an empty response without contacting the platform.
Returns
One element per entry, in request order, holding either the file’s place in this Session or why the platform refused it.
Raises
pydantic.ValidationErrorThe sequence names a file more than once, declares more than MAX_FILES_AND_TOPICS_PER_REQUEST files and topics between them, or lists representations naming one file in two storage formats; rejected client-side, before any request is made.
A file an entry names, or a file one of its topics’ representations names, does not exist in this Session’s org or has a status other than Available. Nothing is added.
The caller lacks permission to manage Sessions in the org that owns this Session, cannot edit a file an entry declares topics on, a file it anchors, or a file a listed representation names, or lacks topic edit access in that org while an entry states is_default_for_reads on a timeline source. Nothing is added.
Usage
from roboto.experimental.sessions import SessionFile
added = session.add_files(
[
SessionFile(file_id="fl_aaa"),
SessionFile(
file_id="fl_bbb",
min_file_timestamp_ns=0,
max_file_timestamp_ns=60_000_000_000,
),
]
)
print([view.file_id for view in added.succeeded])Session.attach_to_device()
Attach a Device to this Session as a subject.
A Session may have many device attachments. For example, a formation flight where multiple drones operate within a single activity window.
Parameters
device_id strID of the Device to add as a subject of this Session.
Raises
The Device does not exist in this Session’s org, or the Session no longer exists.
Return type
Usage
session.attach_to_device("wingman")
list(session.list_devices())
# ['lead', 'wingman']Session.clear_custom_field()
Clear a single custom-field value on this session to None.
Parameters
name strReturn type
Session.clear_custom_fields()
Clear multiple custom-field values on this session to None.
Parameters
names collections.Return type
Session.clear_unix_offset()
Return this Session’s data to an offset of 0.
Removes the wall-clock anchor from all of this Session’s topic data, so the Session’s bounds return to their stored values, read as nanoseconds since the Unix epoch with nothing added.
Clearing reaches this Session’s data and no more, exactly the data set_unix_offset() writes.
Any anchoring state can be cleared, and repeating the call changes nothing: clearing a Session that carries no anchor, or has no topic data at all, is a successful no-op, and a Session made of slices anchored at several different instants is still cleared, each slice moving back by its own offset.
Returns
This Session, refreshed with recomputed aggregate bounds.
Raises
Returning the data to an offset of 0 would start it, or a time range declared over it, before the Unix epoch. Data starts there when its own timestamps are negative; a range starts there when it begins earlier than its data’s anchor.
A concurrent writer added files to the Session while the clear was being applied; retry the call.
The Session no longer exists.
The caller lacks permission to manage Sessions in the org that owns this Session.
Usage
The recomputed bounds return to the Session’s stored values:
session.min_timestamp_ns
# 1700000000250000000
session = session.clear_unix_offset()
session.min_timestamp_ns
# 250000000Session.complete()
Mark this Session complete: declare that all of its files have been added.
Once it is complete, the platform announces the session.ingested platform event, which triggers can subscribe to, as soon as every ingestable file in the Session is ingested. A file is ingestable when its path matched one of the org’s ingestion rules when its current version was created, or it has since been partly or fully ingested. Files that are not ingestable never delay the announcement.
Adding a file to a complete Session puts it back in progress; mark it complete again once the new files are added. If its ingestable files change while it stays complete (a new version, a file removed or deleted), it is announced again once they are all ingested. Marking a complete Session complete changes nothing.
Returns
This Session, refreshed from the server response.
Usage
session.add_file("fl_0123456789abcdef")
session = session.complete()
session.status is SessionStatus.Complete
# TrueProperties
Session.completes_at
When Roboto will mark this Session complete unless another file is added first, per its completion_policy.
None while the Session is complete, has no completion policy, or has no files.
Session.completion_policy
When Roboto marks this Session complete on its own, or None if it is marked complete only by complete().
Session.create()
Create a new Session, optionally associating it with one or more devices.
Every call creates a new Session. For the common single-device case, prefer create_session(), which identifies the Session by name so a resend converges on the Session it already created. To add devices to an existing Session later, see attach_to_device().
Parameters
name Optional[str]Optional short name for the Session (max 120 characters).
device_ids collections.Devices to associate with the Session at creation. Empty (the default) creates a Session with no associated devices.
description Optional[str]Optional description of the Session.
metadata Optional[dict[str, Any]]Optional initial metadata. Sessions are not filterable or sortable by metadata keys; for queryable structured attributes, define a custom field on the Session entity type.
tags Optional[collections.Optional initial tags. Sessions can be filtered by tag membership but are not sortable by tag.
custom_fields Optional[dict[str, Any]]Optional initial values for Ready custom fields defined on Sessions in the caller’s org. Keys must match Ready field names; values must satisfy each field’s declared type.
caller_org_id Optional[str]Caller’s org scope. Required when the caller belongs to multiple orgs.
roboto_client Optional[roboto.Optional RobotoClient; defaults to the ambient one.
completion_policy Union[roboto.When Roboto marks the Session complete on its own. Left unset, the Session gets the org’s default completion policy, if an org admin has set one, and completion_policy on the returned Session shows the policy applied. None means the Session is marked complete only by complete(), whatever the org’s default. Change it later with update().
Returns
The created Session.
Raises
A device in device_ids does not exist in the caller’s org. No Session is created.
Usage
from roboto.experimental.sessions import Session
session = Session.create(
name="flight-2026-04-23-001",
device_ids=["robot-a", "robot-b"],
description="formation flight #4",
metadata={"pilot": "alice"},
tags=["pre-flight-check"],
)Session.create_if_not_exists()
Return the first Session matching a RoboQL query, creating one when none matches.
Concurrent calls with the same query create one Session between them, so a process handling one file at a time (such as an S3 event handler) can use it to put each file into the Session it belongs to.
The Session created must match match_roboql_query. If name, tags and the other arguments describe a Session the query does not match, every later call creates another one.
When several Sessions match, which one is returned is not defined unless the query ends with a SORT BY clause, e.g. name = 'flight-0042' SORT BY session_id.
A matching Session is returned as it is: the arguments below apply only to a Session this call creates, so completion_policy and the rest are ignored on a match. Use update() to change a matched Session. The Session returned may be complete; adding a file to it puts it back in progress. To match only Sessions in progress, add AND status = 'in_progress' to the query.
Parameters
match_roboql_query strRoboQL query over Sessions, e.g. name = 'flight-0042'.
name Optional[str]Name of the Session to create when none matches.
device_ids collections.Devices to associate with a created Session.
description Optional[str]Description of a created Session.
metadata Optional[dict[str, Any]]Metadata of a created Session.
tags Optional[collections.Tags of a created Session.
custom_fields Optional[dict[str, Any]]Custom-field values of a created Session.
completion_policy Union[roboto.Completion policy of a created Session; see create().
caller_org_id Optional[str]Caller’s org scope. Required when the caller belongs to multiple orgs.
roboto_client Optional[roboto.Optional RobotoClient; defaults to the ambient one.
Returns
The matching or created Session.
Raises
Other calls with the same query kept this one waiting for more than 10 seconds, after the SDK’s own retries. Calling again is safe.
Usage
from roboto.experimental.sessions import CompletionPolicy, Session
session = Session.create_if_not_exists(
"name = 'flight-0042'",
name="flight-0042",
completion_policy=CompletionPolicy(inactivity_minutes=15),
)
session.add_file("fl_0123456789abcdef")Properties
Session.created
UTC timestamp when this Session was created.
Session.created_by
Identifier of the user or service which created this Session.
Session.custom_fields
Custom-field values defined on Sessions in this org.
Every Ready CustomField for the org appears as a key. Values that have not been set on this session surface as None rather than being absent. Empty when no custom fields are defined for the org.
A Timestamp value is returned as an ISO 8601 string.
Session.delete()
Delete this Session. The files it included and the devices attached to it are not deleted.
Return type
Properties
Session.description
Optional description of this Session.
Session.detach_from_device()
Remove a Device from this Session’s subjects.
Parameters
device_id strID of the Device to remove as a subject of this Session.
Return type
Session.for_dataset()
Iterate Sessions whose composition includes any file in the given dataset.
Parameters
dataset_id strDataset whose sessions to list.
roboto_client Optional[roboto.Optional RobotoClient; defaults to the ambient one.
Yields
Sessions, one at a time, following pagination automatically.
Return type
Usage
from roboto.experimental.sessions import Session
for session in Session.for_dataset("ds_abc"):
print(session.session_id, session.name)Session.for_org()
Iterate all Sessions visible to the caller’s org.
Parameters
org_id Optional[str]Caller’s org scope. Required when the caller belongs to multiple orgs.
roboto_client Optional[roboto.Optional RobotoClient; defaults to the ambient one.
Yields
Sessions, one at a time, following pagination automatically.
Return type
Session.from_id()
Load a Session by ID.
Parameters
session_id strSession primary key.
roboto_client Optional[roboto.Optional RobotoClient; defaults to the ambient one.
Returns
The Session.
Raises
No session with this ID exists.
The caller lacks view access to the org that owns the session.
Usage
from roboto.experimental.sessions import Session
session = Session.from_id("se_abc123")
session.name
# 'flight-2026-04-23-001'Session.get_topic()
Return the named Topic, scoped to this Session.
The returned Topic is scoped to this Session’s associated files and defaults its read window to this Session’s aggregate bounds, so get_data* reads just this Session’s data without an explicit window.
Parameters
topic_name strExact name of the topic to retrieve (e.g. "/camera/image").
Returns
The matching Topic.
Raises
No topic with topic_name is reachable from this Session (the topic is absent from the org, or this Session holds none of its data).
The caller lacks permission to view Sessions in the org that owns this Session.
Usage
topic = session.get_topic("/camera/image")
for timestamp, record in topic.get_data():
print(timestamp, record)Properties
Session.ingestion_count
How many times this Session has been announced ingested.
Session.ingestion_status()
Return where this Session stands in ingestion, with the files it is still waiting on.
Returns
The current status. It does not refresh this Session instance.
Usage
status = session.ingestion_status()
status.state
# <SessionIngestionState.Processing: 'processing'>
[f.relative_path for f in status.pending_files]
# ['flight_002.mcap']Session.ingestion_summaries()
Where each of many Sessions stands in ingestion, in counts, a hundred Sessions per request.
Cheaper than ingestion_status() on each Session for finding which of many are stuck: a summary has no file lists, only counts, including how many waiting files failed to ingest.
Parameters
session_ids collections.Sessions in the caller’s org. Others are left out of the result.
org_id Optional[str]Caller’s org scope. Required when the caller belongs to multiple orgs.
roboto_client Optional[roboto.Optional RobotoClient; defaults to the ambient one.
Returns
One summary per Session found, in request order.
Usage
summaries = Session.ingestion_summaries(["se_abc123", "se_def456"])
[s.session_id for s in summaries if s.failed_file_count]
# ['se_def456']Session.list_devices()
Iterate the device IDs attached as subjects of this Session, paginated.
Return type
Session.list_files()
Iterate the files this Session includes, following pagination automatically.
Yields
SessionFileView entries, each carrying the part of the file this Session holds (the optional data_range slice and min_wall_clock_timestamp_ns / max_wall_clock_timestamp_ns window), the unix_epoch_offset_ns the platform added to reach that window, and display fields of the file itself (name, dataset, tags, size, …).
Return type
Session.list_metrics()
Return all metrics published to this Session.
Returns
List of Metric instances for this Session.
Usage
metrics = session.list_metrics()
for m in metrics:
print(m.name, m.value)Session.list_topics()
Iterate the topics reachable from this Session, following pagination.
A topic is yielded only when this Session holds some of its data: a time span of the topic (TimelineExtentRecord) on one of the Session’s files, inside the slice the Session holds of that file and overlapping the time window it holds. Each topic is yielded once however many files and partitions carry it, ordered by name with topic_id as a deterministic tiebreaker.
A yielded Topic is scoped to this Session’s files and defaults its read window to this Session’s aggregate bounds, so get_data() (and the other get_data* methods) read just this Session’s data without an explicit window.
Yields
Topic instances.
Return type
Usage
for topic in session.list_topics():
for timestamp, record in topic.get_data():
print(topic.name, timestamp, record)Properties
Session.max_timestamp_ns
Latest time covered by this Session, in Unix-epoch nanoseconds.
None while the Session includes no files, or only files added without a time window whose topic data has no time span registered yet.
Session.metadata
User-supplied metadata attached to this Session.
Sessions are not filterable or sortable by metadata keys. For queryable structured attributes on a Session, define a custom field on the Session entity type.
Session.min_timestamp_ns
Earliest time covered by this Session, in Unix-epoch nanoseconds.
None while the Session includes no files, or only files added without a time window whose topic data has no time span registered yet.
Session.modified
UTC timestamp when this Session was last modified.
Session.modified_by
Identifier of the user or service which last modified this Session.
Session.publish_metrics()
Record metric values for this Session in a single network call.
Convenience wrapper around publish() that supplies this Session’s session_id and org_id. Republishing a metric under the same name replaces its previous value for this Session.
If a metric definition does not already exist for a given name it is created automatically.
Parameters
metrics list[roboto.Metric names and numeric values to record.
device_id Union[roboto.Device to associate with each published value, or None to opt out. When omitted, the server infers a device from this Session’s attached devices: the call succeeds only if exactly one device is associated and is rejected when zero or more than one are.
Returns
One element per metric entry, in request order, holding either the recorded Metric or why the platform refused it.
Raises
device_id was omitted and this Session has zero or more than one attached devices.
Usage
Let the server infer the device from this Session’s single attached device:
from roboto.domain.metrics import MetricEntry
published = session.publish_metrics(
[
MetricEntry(name="cpu.usage_max", value=87.2),
MetricEntry(name="memory.peak_mb", value=2048.0),
]
)
len(published.succeeded)
# 2Attach to an explicit device, overriding inference:
session.publish_metrics(
[MetricEntry(name="cpu.usage_max", value=87.2)],
device_id="robot01",
)Session.put_metadata()
Add or update metadata fields on this Session.
Parameters
metadata dict[str, Any]Field-to-value map. Existing fields are overwritten; fields not in this map are left unchanged.
Returns
This Session, refreshed from the server response.
Usage
session.put_metadata({"weather": "clear", "pilot": "alice"})Session.put_tags()
Add tags to this Session.
Tags already present on the Session are not duplicated.
Parameters
Tags to add.
Returns
This Session, refreshed from the server response.
Usage
session.put_tags(["pre-flight-check", "training"])Properties
Session.record
Underlying data record for this Session.
Session.refresh()
Re-read this Session from the platform, replacing every property backed by its record.
Call it when something other than this instance changed the Session: the aggregate bounds min_timestamp_ns and max_timestamp_ns are recomputed whenever a Session’s composition or anchoring changes, including by another caller.
Returns
This Session.
Session.remove_file()
Remove a single file from this Session.
The singular form of remove_files().
Parameters
file Union[roboto.A File or a file ID.
Returns
The ID of the removed file.
Raises
This Session does not hold the file.
Session.remove_files()
Remove the given files from this Session.
A file this Session does not hold is reported as its own element rather than failing the call, while a failure the platform did not anticipate, such as a timeout, removes none of the files. The platform then recomputes this Session’s aggregate bounds across the files that remain, and this instance reflects the new min_timestamp_ns / max_timestamp_ns on return.
Parameters
files collections.Files to remove, each a File or a file ID. An empty sequence returns an empty response without contacting the platform.
Returns
One element per named file, in request order, holding either its ID or why the platform refused to remove it.
Raises
pydantic.ValidationErrorThe sequence names a file more than once, or names more than MAX_FILES_AND_TOPICS_PER_REQUEST files; rejected client-side, before any request is made.
Session.remove_metadata()
Remove metadata keys from this Session.
Parameters
metadata roboto.Metadata keys to remove. Dot notation addresses nested keys ("weather.condition").
Returns
This Session, refreshed from the server response.
Usage
session.remove_metadata(["pilot", "weather.condition"])Session.remove_tags()
Remove the given tags from this Session.
Parameters
Tags to remove. Tags not present on the Session are silently ignored.
Returns
This Session, refreshed from the server response.
Usage
session.remove_tags(["training"])Properties
Session.session_id
Globally unique identifier assigned to this Session on creation.
Session.set_custom_field()
Session.set_custom_fields()
Session.set_unix_offset()
Anchor this Session’s data to wall-clock time.
anchor becomes the wall-clock instant of stored time 0 for all of this Session’s topic data, and the Session’s aggregate bounds are recomputed to reflect it.
This write reaches this Session’s data and no more. Where several Sessions share one file, each owning a slice of it, it anchors the slices this Session holds and leaves the file’s other slices at whatever instant they were given. Data another Session also holds is shared, not copied, so that Session reads the same anchor.
An anchor exists only when a caller supplies one, either as the data is added (the anchor argument of add_file(), or anchor_ns on a SessionFile) or through this method. Until then, the data carries an offset of 0 and its stored timestamps are read as nanoseconds since the Unix epoch. Applying an anchor overwrites whatever anchor the data carried before; applying the one it already carries changes nothing, so repeating the call succeeds. An anchor survives re-ingest: redeclaring a slice without supplying an anchor preserves the one it already had.
Parameters
anchor roboto.Wall-clock instant of stored time 0: an int of nanoseconds since the Unix epoch, or any other Time, read as to_epoch_nanoseconds() reads it (a datetime or ISO 8601 string is that instant; a float, Decimal, or numeric string is seconds since the epoch). Must fall after the Unix epoch: zero is not an anchor (use clear_unix_offset() to return the Session to an offset of 0), and earlier instants are rejected.
Returns
This Session, refreshed with recomputed aggregate bounds.
Raises
TypeErroranchor is not one of the Time types.
ValueErroranchor is a boolean, a string that is neither seconds nor ISO 8601, zero, before the Unix epoch, or too large for a signed 64-bit integer of nanoseconds; rejected client-side, before any request is made. A range refusal is raised as pydantic.ValidationError, a subclass of ValueError.
OverflowErroranchor is infinite, such as float("inf"); rejected client-side, before any request is made.
Any of three cases: the Session has no topic data to anchor; the Session’s own data already carries several distinct anchors, and the server will not pick one of them to move everything from (anchor less than a whole Session at a time instead, either one topic with set_unix_offset() on a Topic from get_topic() or list_topics(), or one whole file with set_timeline_offset()); or the anchor would move the Session’s data, or a time range declared over it, before the Unix epoch or past the largest storable Unix-epoch nanosecond value; anchor the data at the instant it was recorded.
A concurrent writer added files to the Session while the anchor was being applied; retry the call.
The Session no longer exists.
The caller lacks permission to manage Sessions in the org that owns this Session.
Usage
The recomputed bounds are the offset plus the Session’s stored values:
session.min_timestamp_ns
# 250000000
session = session.set_unix_offset(1_700_000_000_000_000_000)
session.min_timestamp_ns
# 1700000000250000000The same anchor given as a datetime:
import datetime
session = session.set_unix_offset(
datetime.datetime(2023, 11, 14, 22, 13, 20, tzinfo=datetime.timezone.utc)
)
session.min_timestamp_ns
# 1700000000250000000Session.skip_waiting_for()
Stop waiting for these files to be ingested, e.g. ones whose ingestion failed.
A complete Session is announced ingested once its other ingestable files are. The skip covers each file’s current upload: uploading it again makes the Session wait for it again, while editing its tags or metadata does not. Skipped files stay in the Session and are listed in SessionIngestionStatus.skipped_files.
Parameters
file_ids collections.Files in this Session.
Returns
Where this Session stands in ingestion afterwards, and the requested files that are not in it.
Usage
response = session.skip_waiting_for(["fl_0123456789abcdef"])
response.ingestion.state
# <SessionIngestionState.Ingested: 'ingested'>
response.not_in_session
# []Properties
Session.status
Whether this Session is in progress or complete (see complete()).
Session.update()
Update mutable Session fields.
Fields left at the NotSet default are preserved; for nullable fields (description, name, completion_policy), pass None to clear.
Parameters
description Optional[Union[str, roboto.New description for the Session. Set to None to clear the description. Leave at the default to leave the description unchanged.
metadata_changeset Union[roboto.Tag and metadata changes to apply (put/remove tags and fields). See put_tags(), remove_tags(), put_metadata(), and remove_metadata() for shorthand helpers.
name Optional[Union[str, roboto.New name for the Session. Set to None to clear the name. Leave at the default to leave the name unchanged.
custom_fields_changeset Optional[roboto.Changes to apply to Ready custom-field values on this session. Field names not referenced by the changeset are left unchanged.
completion_policy Union[roboto.When Roboto marks the Session complete on its own. None removes the policy, so the Session is marked complete only by complete(). On a Session in progress that has files, a new policy counts its inactivity from now. A complete Session stays complete, and the policy applies after a file is added to it.
Returns
This Session, refreshed from the server response.
Usage
session.update(description="formation flight #4", name="flight-2026-04-23-001")Let Roboto mark a Session complete 30 minutes after its last file is added:
from roboto.experimental.sessions import CompletionPolicy
session.update(completion_policy=CompletionPolicy(inactivity_minutes=30))Session.wait_until_ingested()
Block until every ingestable file in this Session is ingested and Roboto has announced it ingested.
The wait ends once Roboto records the announcement that fires session.ingested, which raises ingestion_count. A Session already announced ingested since it was last marked complete returns at once.
A Session in progress is waited for when it will complete on its own: it has a completion_policy and files, so completes_at is set. The wait then includes the policy’s inactivity. A Session in progress that will not complete on its own is refused, as only complete() would complete it.
The default timeout of 10 minutes may be too short for a Session of large recordings. Pass a longer one, e.g. timeout=2 * 60 * 60 for two hours.
Parameters
timeout floatMaximum seconds to wait.
poll_interval intSeconds between status checks.
Returns
This Session, refreshed from the server.
Raises
RuntimeErrorThis Session is in progress and will not complete on its own: it has no completion policy, or it has one but no files.
timeout elapsed first. The message says where the Session stands and lists up to 10 of the files it is still waiting for. It is also a built-in TimeoutError.
Usage
session = session.complete().wait_until_ingested(timeout=2 * 60 * 60)SessionFile
Bases: FileDeclaration
One already-uploaded file that belongs to a session, and the topic data it carries.
Adds to FileDeclaration the window of the file’s data this session holds. Everything else the entry states is true of the file whichever session, if any, names it.
The same entry states a file’s place in a session however that session is composed: inside a SessionDeclaration that creates the session whole, or handed to add_files() afterwards. An entry with no topics attaches the file without registering topic data; topics can be declared for the same file later, through this entry again or through declare_topics().
A topic listed here takes its anchor from this entry’s anchor_ns, which anchors everything the entry declares. A FileTopicDeclaration carrying an anchor_ns of its own is rejected here; state that anchor on the entry instead.
Time window (min_file_timestamp_ns and max_file_timestamp_ns):
- Values are nanoseconds as the file’s own data carries them, measured the same way as the
min_file_timestamp_nsa timeline source such asSchemaFieldSourcedeclares. A slice of a shared file whose timestamp column restarts at 0 states bounds from 0. - Set both or set neither; a window with only one bound is rejected. The window is the closed interval
[min_file_timestamp_ns, max_file_timestamp_ns], both endpoints included. - The window this entry gives the session is the smallest one enclosing these bounds and the bounds of every timeline source its topics declare. State them when the file’s topics are not declared here, or when what belongs to the session runs past the declared topic data. Leave both unset for a window spanning whatever the entry’s topic data spans, or, on an entry declaring no topics, the file’s whole window.
- The anchor covering this entry is added to the stored window, so re-anchoring the file moves the window along with the data it names. The bounds may be negative, but with the anchor added the window must lie between the Unix epoch and the largest storable Unix-epoch nanosecond value (
2**63 - 1). The platform refuses an entry whose window would fall outside that span, an entry whose anchor would move a window another session declared over the same data outside it, and a later re-anchoring that would move this window outside it. The platform reports the window back in wall clock, onmin_wall_clock_timestamp_nsandmax_wall_clock_timestamp_ns. - Several sessions can share one file, each stating its own window. A session takes on the file’s data that overlaps its window, and its own time bounds span the windows of all the files it holds; a read scoped to the session covers those bounds unless it names a window of its own.
- The window this entry gives the session also trims a read of this file. A read scoped to this session returns only the rows of the file inside that window, however wide a window the read itself names, so a session holding part of a shared file reads back that part and not the whole file.
Data range (data_range) inside a session:
- The range decides which of the file’s partitions this session admits. It admits a partition only when every one of its positions sits inside it; a partition reaching past either end is left out whole, never trimmed.
- A range that cuts through a partition the file has already registered is refused by the platform rather than accepted to admit nothing of that partition.
Parameters
data AnyAttributes
SessionFile.max_file_timestamp_ns
Upper bound of the time window this entry states, in the file’s own timestamps.
SessionFile.min_file_timestamp_ns
Lower bound of the time window this entry states, in the file’s own timestamps.
SessionFileRecord
Bases: pydantic.BaseModel
Wire-format row for one file a Session holds, and the part of the file it holds.
Time window contract (min_wall_clock_timestamp_ns and max_wall_clock_timestamp_ns):
- Set together or both
None; a window with only one bound is rejected on write. - When both are
None, the Session holds the file’s whole recorded time window. - When both are set,
min_wall_clock_timestamp_ns <= max_wall_clock_timestamp_ns. Consumers iterating session data must keep only the file’s data inside the closed interval[min_wall_clock_timestamp_ns, max_wall_clock_timestamp_ns]. - Values are nanoseconds since the Unix epoch, measured the same way as the parent Session’s own bounds. A caller states this window in the file’s own timestamps, on
SessionFile; the platform adds the anchor covering the data the window names and reports the sum here, alongside theunix_epoch_offset_nsit added.
Data range contract (data_range):
Nonemeans the Session holds the whole file.(start, end):startis the first covered position;endis one past the last, with0 <= start < end. Values are in the file’s own units: stored-row positions (counted from 0), or nanoseconds of media time for video.- Used when one file is shared by several sessions; the range names the slice of the file that belongs to this session.
Parameters
data AnyAttributes
SessionFileRecord.created
When this file was added to the session.
SessionFileRecord.created_by
User ID or service account that added this file to the session.
SessionFileRecord.data_range
The slice of the file the Session holds, as (start, end) in the file’s own units, or None when it holds the whole file. start is the first covered position; end is one past the last.
SessionFileRecord.max_wall_clock_timestamp_ns
Upper bound (inclusive) of the part of the file the Session holds, in Unix-epoch nanoseconds. None means the Session holds the file up to the end of its recorded time window; paired with min_wall_clock_timestamp_ns.
SessionFileRecord.min_wall_clock_timestamp_ns
Lower bound (inclusive) of the part of the file the Session holds, in Unix-epoch nanoseconds. None means the Session holds the file from the beginning of its recorded time window; paired with max_wall_clock_timestamp_ns.
SessionFileRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SessionFileRecord.modified
When this file’s place in the session was last modified.
SessionFileRecord.modified_by
User ID or service account that last modified this file’s place in the session.
SessionFileRecord.unix_epoch_offset_ns
Wall-clock instant of stored time 0 for the file’s data the Session holds, in nanoseconds since the Unix epoch: what the platform added to the file’s own timestamps to reach min_wall_clock_timestamp_ns and max_wall_clock_timestamp_ns, and what to subtract to read any other instant back in the file’s own timestamps. None when that data includes nothing registered, and when it sits at more than one instant, which leaves no single offset to report.
SessionFileView
Bases: pydantic.BaseModel
One row of the GET /v1/sessions/id/<session_id>/files response: a file’s place in a Session joined with display fields of the file itself.
These fields come from the session’s composition: file_id, the optional time window min_wall_clock_timestamp_ns / max_wall_clock_timestamp_ns in Unix-epoch nanoseconds, the optional data_range slice (the window and the slice both under the contracts documented on SessionFileRecord), and the unix_epoch_offset_ns the platform added to reach that window. Every other field is read from the file itself when the files are listed, and describes the file rather than its place in the session: created is when the file was created, not when it joined the session. None of those fields is part of a write.
Parameters
data AnyAttributes
SessionFileView.data_range
The slice of the file the Session holds, as (start, end) in the file’s own units, or None when it holds the whole file. start is the first covered position; end is one past the last.
SessionFileView.ingestable
Whether the file is meant to be ingested: its path matched one of its org’s ingestion rules when its current version was created, or it has since been partly or fully ingested.
SessionFileView.ingestion_status
How much of the contributing file has been ingested.
SessionFileView.max_wall_clock_timestamp_ns
Upper bound (inclusive) of the part of the file the Session holds, in Unix-epoch nanoseconds. None means the Session holds the file up to the end of its recorded time window; paired with min_wall_clock_timestamp_ns.
SessionFileView.min_wall_clock_timestamp_ns
Lower bound (inclusive) of the part of the file the Session holds, in Unix-epoch nanoseconds. None means the Session holds the file from the beginning of its recorded time window; paired with max_wall_clock_timestamp_ns.
SessionFileView.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SessionFileView.origination
Provenance of the file, e.g. an invocation id or upload source.
SessionFileView.unix_epoch_offset_ns
Wall-clock instant of stored time 0 for the file’s data the Session holds, in nanoseconds since the Unix epoch: what the platform added to the file’s own timestamps to reach min_wall_clock_timestamp_ns and max_wall_clock_timestamp_ns, and what to subtract to read any other instant back in the file’s own timestamps. None when that data includes nothing registered, and when it sits at more than one instant, which leaves no single offset to report.
SessionRecord
Bases: pydantic.BaseModel
Wire-format row for a session: an operational time window of a Device such as a drone flight, a vehicle drive, or a robot run.
A Session unifies the recordings and auxiliary data produced during its window; it may span many files or cover only a slice of one.
min_timestamp_ns and max_timestamp_ns span every file the Session holds: each file supplies the time window stated for it or, without one, the time span of the data the Session takes from it. The platform recomputes them in the same write as any change to the Session’s files or to the anchors of their data, so the row never disagrees with its contents.
Parameters
data AnyAttributes
SessionRecord.completed_at
When the session was last marked complete. None if it never was. Adding a file to a complete session puts it back in progress and keeps this value, so check status to tell whether the session is complete now.
SessionRecord.completed_by
User ID or service account that last marked the session complete. None if it never was.
SessionRecord.completes_at
When Roboto will mark the session complete unless another file is added first. None while the session is complete, has no completion policy, or has no files.
SessionRecord.completion_policy
When Roboto marks the session complete on its own. None: the session is marked complete only by request.
SessionRecord.custom_fields
Values for the custom fields defined on Sessions in this org.
Every Ready custom field defined for (org_id, Session) appears as a key; values that have not been set surface as None rather than being absent. Empty when no custom fields are defined for the org.
SessionRecord.ingested_at
When the session was last announced ingested. None until the first announcement.
SessionRecord.ingestion_count
How many times the session has been announced ingested. Each announcement fires a session.ingested event. A session is announced again when it is put back in progress and marked complete again, or when its ingestable files change while it is complete, once they are all ingested again. A file uploaded again, removed, or deleted changes them; editing a file’s tags, metadata, or description does not.
SessionRecord.max_timestamp_ns
Latest time covered by the Session, in Unix-epoch nanoseconds. None while none of its files supplies a time: the Session holds no files, or only files added without a time window whose topic data has no time span registered yet.
SessionRecord.metadata
User-supplied metadata.
Sessions cannot be filtered or sorted by metadata keys; for queryable structured attributes, define a custom field on the Session entity type.
SessionRecord.min_timestamp_ns
Earliest time covered by the Session, in Unix-epoch nanoseconds. None while none of its files supplies a time: the Session holds no files, or only files added without a time window whose topic data has no time span registered yet.
SessionRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SessionRecord.modified_by
User ID or service account that last modified the Session.
SessionRecord.name
A short, human-readable name for the Session. If provided, must be 120 characters or less.
SessionRecord.status
Whether the session is in progress or complete. Adding a file to a complete session puts it back in progress.
SessionRecord.tags
User-supplied tags.
Sessions can be filtered by tag membership (e.g., tags CONTAINS '<tag>') but are not sortable by tag.
SetActionAccessibilityRequest
Bases: pydantic.BaseModel
Request payload to set action accessibility.
Used to change whether an action is private to the organization or published publicly in the Action Hub.
Parameters
data AnyAttributes
SetActionAccessibilityRequest.accessibility
The new accessibility level (Organization or ActionHub).
SetActionAccessibilityRequest.digest
Specific version of Action. If not specified, the latest version’s accessibility will be updated.
SetActionAccessibilityRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SetContainerInfoRequest
Bases: pydantic.BaseModel
Request to set container information for an invocation.
Used internally by the Roboto platform to record container details after the action image has been pulled and inspected.
Parameters
data AnyAttributes
SetContainerInfoRequest.image_digest
The digest of the container image that was pulled.
SetDefaultRepresentationRequest
Bases: BaseAddRepresentationRequest
Request to set the default representation for a topic.
Designates a specific representation as the default for accessing topic data. The default representation is used when no specific representation is requested for data access operations.
Parameters
data AnyAttributes
SetDefaultRepresentationRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
SetLogsLocationRequest
Bases: pydantic.BaseModel
Request to set the location where invocation logs are stored.
Used internally by the Roboto platform to record where log files are saved for later retrieval.
Parameters
data AnySourceProvenance
Bases: pydantic.BaseModel
Provenance information for an invocation source
Parameters
data AnyAttributes
SourceProvenance.source_id
SourceProvenance.source_type
TimelineExtentRecord
Bases: pydantic.BaseModel
Min/max timestamp bounds for one topic partition measured against one timeline source.
Written by ingest when a partition’s timestamps are summarized for a given source (e.g., a schema timestamp field, or message log/publish time).
Stored timestamps come through verbatim from the data source: they may be absolute nanoseconds since the Unix epoch, or partition-relative (e.g., monotonic from zero). unix_epoch_offset_ns is the calibration that projects stored values onto Unix-epoch wall-clock: session_time_ns = stored_time_ns + unix_epoch_offset_ns. A value of 0 means the stored timestamps are already absolute Unix-epoch ns, or that no calibration has been applied yet.
Parameters
data AnyAttributes
TimelineExtentRecord.created
TimelineExtentRecord.created_by
TimelineExtentRecord.max_timestamp
Largest stored timestamp in this extent, in nanoseconds. Absolute or partition-relative per the source.
TimelineExtentRecord.min_timestamp
Smallest stored timestamp in this extent, in nanoseconds. Absolute or partition-relative per the source.
TimelineExtentRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
TimelineExtentRecord.modified
TimelineExtentRecord.modified_by
TimelineExtentRecord.org_id
TimelineExtentRecord.timeline_extent_id
TimelineExtentRecord.timeline_source_id
ID of the timeline source these bounds are measured against.
TimelineExtentRecord.topic_part_id
ID of the topic partition these bounds apply to.
TimelineExtentRecord.unix_epoch_offset_ns
Nanoseconds to add to each stored timestamp to obtain Unix-epoch wall-clock time: session_time_ns = stored_time_ns + unix_epoch_offset_ns. 0 when stored timestamps are already absolute Unix-epoch ns, or when no calibration has been recorded for this partition/source pair.
TimelineSourceKind
Discriminator for how a TimelineSourceRecord derives its timestamps.
"schema_field" points at a timestamp field inside the schema (field_id is set). "message_log_time" and "message_publish_time" point at the message envelope’s log or publish timestamp respectively (field_id is None).
TimelineSourceRecord
Bases: pydantic.BaseModel
A registered timeline source for a schema.
A timeline source either points at a timestamp field inside the schema (source="schema_field", field_id set) or at the message envelope’s log or publish timestamp (source in {"message_log_time", "message_publish_time"}, field_id is None). Timeline sources are scoped to a schema, not a topic, so topics that share a schema share their timeline sources.
Parameters
data AnyAttributes
TimelineSourceRecord.created
TimelineSourceRecord.created_by
TimelineSourceRecord.field_id
ID of the schema field supplying timestamps. Set when source == "schema_field"; otherwise None.
TimelineSourceRecord.is_default
Whether this timeline source is the default for its schema when no source is specified explicitly.
TimelineSourceRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
TimelineSourceRecord.modified
TimelineSourceRecord.modified_by
TimelineSourceRecord.org_id
TimelineSourceRecord.schema_id
ID of the schema this timeline source is registered against.
TimelineSourceRecord.source
Where timestamps come from: a schema field ("schema_field"), or the message envelope’s log or publish timestamp ("message_log_time" / "message_publish_time").
TimelineSourceRecord.timeline_source_id
Topic
Represents a topic within the Roboto platform.
A topic is a sequence of structured time-series data linked to a source file, typically containing sensor readings, robot state information, or other timestamped data streams. Topics are fundamental building blocks for data analysis in robotics, providing organized access to time-synchronized data from various sources like ROS bags, MCAP files, or other structured data formats.
Each topic follows a defined schema where message paths represent the individual fields or signals within that schema. Topics enable efficient querying, filtering, and analysis of time-series data, supporting operations like temporal slicing, field selection, and data export to various formats including pandas DataFrames.
Topics are associated with files and inherit access permissions from their parent dataset. They provide the primary interface for accessing ingested robotics data in the Roboto platform, supporting both programmatic access through the SDK and visualization in the web interface.
The Topic class serves as the main interface for topic operations in the Roboto SDK, providing methods for data retrieval, message path management, metadata operations, and schema management.
Parameters
roboto_client Optional[roboto.topic_data_service Optional[roboto.Topic.add_message_path()
Add a new message path to this topic.
Creates a new message path within this topic, defining a specific field or signal that can be extracted from the topic’s data. Message paths use dot notation to specify nested attributes within the topic’s schema.
Parameters
message_path strDot-delimited path to the attribute (e.g., “pose.position.x”).
data_type strNative data type of the attribute as it appears in the original data source (e.g., “float32”, “uint8[]”, “geometry_msgs/Pose”). Used primarily for display purposes and should match the robot’s runtime language or schema definitions.
canonical_data_type roboto.Normalized Roboto data type that enables specialized platform features for maps, images, timestamps, and other data with special interpretations.
path_in_schema Optional[list[str]]List of path components representing the field’s location in the source data schema. Unlike message_path, which assumes dots separate path parts implying nested data, this preserves the exact path from the source data for accurate attribute access.
metadata Optional[dict[str, Any]]Additional metadata to associate with the message path.
Returns
MessagePathRecord representing the newly created message path.
Raises
Message path already exists for this topic.
Caller lacks permission to modify the topic.
Usage
from roboto.domain.topics import CanonicalDataType
topic = Topic.from_id("topic_xyz789")
message_path = topic.add_message_path(
message_path="pose.position.x",
data_type="float64",
canonical_data_type=CanonicalDataType.Number,
metadata={"unit": "meters"},
)
print(message_path.message_path)
# pose.position.xTopic.add_message_path_representation()
Add a representation for a specific message path.
Associates a message path with a data representation, enabling efficient access to specific fields within the topic data. Representations can be in different storage formats like MCAP or Parquet.
Parameters
message_path_id strUnique identifier of the message path.
association roboto.Association pointing to the representation data.
storage_format roboto.Format of the representation data.
version intVersion number of the representation.
format Optional[str]Content format descriptor (e.g. “jpeg”, “sensor_msgs/Image”).
transformations Optional[list[str]]Transformation descriptors applied (e.g. [“downsample:0.5”]).
Returns
RepresentationRecord representing the newly created representation.
Raises
Message path with the given ID does not exist.
Caller lacks permission to modify the topic.
Usage
from roboto.association import Association
from roboto.domain.topics import RepresentationStorageFormat
topic = Topic.from_id("topic_xyz789")
representation = topic.add_message_path_representation(
message_path_id="mp_123",
association=Association.file("file_repr_456"),
storage_format=RepresentationStorageFormat.MCAP,
version=1,
)
print(representation.representation_id)
# repr_789Properties
Topic.association
Association linking this topic to its source entity (typically a file).
Topic.create()
Create a new topic associated with a file.
Creates a new topic record in the Roboto platform, associating it with the specified file and defining its schema and temporal boundaries. This method is typically used during data ingestion to register topics found in robotics data files.
Parameters
file_id strUnique identifier of the file this topic is associated with.
topic_name strName of the topic (e.g., “/camera/image”, “/imu/data”).
end_time Optional[int]End time of the topic data in nanoseconds since UNIX epoch.
message_count Optional[int]Total number of messages in this topic.
metadata Optional[collections.Additional metadata to associate with the topic.
schema_checksum Optional[str]Checksum of the topic’s message schema for validation.
schema_name Optional[str]Name of the message schema (e.g., “sensor_msgs/Image”).
start_time Optional[int]Start time of the topic data in nanoseconds since UNIX epoch.
message_paths Optional[collections.Message paths to create along with the topic.
caller_org_id Optional[str]Organization ID to create the topic in. Required for multi-org users.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
Topic instance representing the newly created topic.
Raises
Invalid topic parameters.
Caller lacks permission to create topics.
Usage
Create a basic topic for camera data:
topic = Topic.create(
file_id="file_abc123",
topic_name="/camera/image",
schema_name="sensor_msgs/Image",
start_time=1722870127699468923,
end_time=1722870127799468923,
message_count=100,
)
print(topic.topic_id)
# topic_xyz789Create a topic with metadata and message paths:
from roboto.domain.topics import AddMessagePathRequest, CanonicalDataType
message_paths = [
AddMessagePathRequest(
message_path="header.stamp.sec",
data_type="uint32",
canonical_data_type=CanonicalDataType.Timestamp,
)
]
topic = Topic.create(
file_id="file_abc123",
topic_name="/imu/data",
schema_name="sensor_msgs/Imu",
metadata={"sensor_type": "IMU", "frequency": 100},
message_paths=message_paths,
)Topic.create_from_df()
Create a Topic from a pandas DataFrame and associate it with a file.
If a topic with the same name already exists for the specified file, it will be updated with the new data and schema.
Parameters
file_id strID of the file to associate this topic with.
dataset_id strID of the dataset containing the file.
topic_name strName for the topic. Must be unique within the file.
df pandas.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.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.
caller_org_id Optional[str]Organization ID of the caller. If not provided, uses the default from the client context.
roboto_client Optional[roboto.Roboto client instance. If not provided, uses the default client.
Returns
The created or updated Topic instance.
Raises
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.
ImportErrorIf pandas or pyarrow are not installed. Install with pip install roboto[ingestion] to use this feature.
If the caller lacks permission to create topics or upload files to the specified dataset.
Notes
- For most use cases, prefer
File.add_topic()instead
Usage
Create a topic when you have file and dataset IDs:
import pandas as pd
from roboto.domain.topics import Topic
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 = Topic.create_from_df(
file_id="file_abc123",
dataset_id="ds_xyz789",
topic_name="sensor_data",
df=df,
timestamp_column="timestamp",
timestamp_unit="s",
)
print(f"Created topic: {topic.name}")
# Created topic: sensor_dataUsing File.add_topic() is typically more convenient:
from roboto import File
file = File.from_id("file_abc123")
topic = file.add_topic("sensor_data", df, timestamp_column="timestamp", timestamp_unit="s")Properties
Topic.created
Timestamp when this topic was created in the Roboto platform.
Topic.created_by
Identifier of the user or system that created this topic.
Topic.dataset_id
Unique identifier of the dataset containing this topic, if applicable.
Topic.default_representation
Default representation used for accessing this topic’s data.
Topic.delete()
Delete this topic from the Roboto platform.
Permanently removes this topic and all its associated message paths and representations from the platform. This operation cannot be undone.
Raises
Topic does not exist or has already been deleted.
Caller lacks permission to delete the topic.
Return type
Usage
topic = Topic.from_id("topic_xyz789")
topic.delete()
# # Topic and all its data are now permanently deletedTopic.from_id()
Retrieve a topic by its unique identifier.
Fetches a topic record from the Roboto platform using its unique topic ID. This is the most direct way to access a specific topic when you know its identifier.
Parameters
topic_id strUnique identifier for the topic.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
Topic instance representing the requested topic.
Raises
Topic with the given ID does not exist.
Caller lacks permission to access the topic.
Usage
topic = Topic.from_id("topic_xyz789")
print(topic.name)
# '/camera/image'
print(topic.message_count)
# 100Topic.from_name_and_file()
Retrieve a topic by its name and associated file.
Fetches a topic record using its name and the file it’s associated with. This is useful when you know the topic name (e.g., “/camera/image”) and the file containing the topic data.
Parameters
topic_name strName of the topic to retrieve.
file_id strUnique identifier of the file containing the topic.
owner_org_id Optional[str]Organization ID to scope the search. If None, uses caller’s org.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
Topic instance representing the requested topic.
Raises
Topic with the given name does not exist in the specified file.
Caller lacks permission to access the topic.
Usage
topic = Topic.from_name_and_file(topic_name="/camera/image", file_id="file_abc123")
print(topic.topic_id)
# topic_xyz789
print(len(topic.message_paths))
# 5Topic.get_by_dataset()
List all topics associated with files in a dataset.
Retrieves all topics from files within the specified dataset. If multiple files contain topics with the same name (e.g., chunked files with the same schema), they are returned as separate topic objects.
Parameters
dataset_id strUnique identifier of the dataset to search.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Yields
Topic instances associated with files in the dataset.
Raises
Dataset with the given ID does not exist.
Caller lacks permission to access the dataset.
Return type
Usage
for topic in Topic.get_by_dataset("ds_abc123"):
print(f"Topic: {topic.name} (File: {topic.file_id})")
# Topic: /camera/image (File: file_001)
# Topic: /imu/data (File: file_001)
# Topic: /camera/image (File: file_002)
# Topic: /imu/data (File: file_002)# Count topics by name
from collections import Counter
topic_names = [topic.name for topic in Topic.get_by_dataset("ds_abc123")]
print(Counter(topic_names))
# Counter({'/camera/image': 2, '/imu/data': 2})Topic.get_by_file()
List all topics associated with a specific file.
Retrieves all topics contained within the specified file. This is useful for exploring the structure of robotics data files and understanding what data streams are available.
Parameters
file_id strUnique identifier of the file to search.
owner_org_id Optional[str]Organization ID to scope the search. If None, uses caller’s org.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Yields
Topic instances associated with the specified file.
Raises
File with the given ID does not exist.
Caller lacks permission to access the file.
Return type
Usage
for topic in Topic.get_by_file("file_abc123"):
print(f"Topic: {topic.name} ({topic.message_count} messages)")
# Topic: /camera/image (150 messages)
# Topic: /imu/data (1500 messages)
# Topic: /gps/fix (50 messages)# Get topics with specific schema
camera_topics = [topic for topic in Topic.get_by_file("file_abc123") if "camera" in topic.name]Topic.get_data()
Return this topic’s underlying data.
Retrieves and yields data records from this topic, with optional filtering by message paths and time range. Each yielded datum is a dictionary that matches this topic’s schema.
Parameters
message_paths_include Optional[collections.Dot notation paths that match attributes of individual data records to include. If None, all paths are included.
message_paths_exclude Optional[collections.Dot notation paths that match attributes of individual data records to exclude. If None, no paths are excluded.
start_time Optional[roboto.Start time (inclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
end_time Optional[roboto.End time (exclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
cache_dir Union[str, pathlib.Directory where topic data will be downloaded if necessary. Defaults to DEFAULT_CACHE_DIR.
representation_selector roboto.Criteria for selecting among multiple representations. Defaults to RepresentationSelector.raw() — original, untransformed data. Pass a RepresentationSelector to request a specific content format or transformation pipeline.
Yields
Timestamp and dictionary records that match this topic’s schema, filtered according to the parameters.
Return type
Notes
For each example below, assume the following is a sample datum record that can be found in this topic:
{“angular_velocity”: {
“x”: <uint32>, “y”: <uint32>, “z”: <uint32>
}, “orientation”: { “x”: <uint32>, “y”: <uint32>, “z”: <uint32>, “w”: <uint32> }
}
Usage
Print all data to stdout:
topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data():
print(timestamp, record)Only include the “angular_velocity” sub-object, but filter out its “y” property:
topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data(
message_paths_include=["angular_velocity"],
message_paths_exclude=["angular_velocity.y"],
):
...Only include data between two timestamps:
topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data(
start_time=1722870127699468923,
end_time=1722870127699468924,
):
...Collect all topic data into a dataframe (requires installing the roboto[analytics] extra):
topic = Topic.from_name_and_file(...)
df = topic.get_data_as_df()Get the JPEG-encoded version of image data:
topic = Topic.from_name_and_file(...)
for timestamp, record in topic.get_data(
representation_selector=RepresentationSelector(content_format="jpeg"),
):
...Topic.get_data_as_df()
Return this topic’s underlying data as a pandas DataFrame.
Retrieves topic data and converts it to a pandas DataFrame for analysis and visualization. The DataFrame is indexed by log time and contains columns for each message path in the topic data.
Parameters
message_paths_include Optional[collections.Dot notation paths that match attributes of individual data records to include. If None, all paths are included.
message_paths_exclude Optional[collections.Dot notation paths that match attributes of individual data records to exclude. If None, no paths are excluded.
start_time Optional[roboto.Start time (inclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
end_time Optional[roboto.End time (exclusive) as nanoseconds since UNIX epoch or convertible to such by to_epoch_nanoseconds().
cache_dir Union[str, pathlib.Directory where topic data will be downloaded if necessary. Defaults to DEFAULT_CACHE_DIR.
representation_selector roboto.Criteria for selecting among multiple representations. Defaults to RepresentationSelector.raw() — original, untransformed data. Pass a RepresentationSelector to request a specific content format or transformation pipeline.
Returns
pandas DataFrame containing the topic data, indexed by log time.
Raises
ImportErrorpandas is not installed. Install with roboto[analytics] extra.
Notes
Requires installing this package using the roboto[analytics] extra.
An array-typed message path (for example a float32[3] acceleration) becomes a single column whose values are lists. Individual array elements are not addressable on their own — neither as separate DataFrame columns nor as message paths in message_paths_include / message_paths_exclude. Filter by the array’s path and unpack the column with numpy, as in the example below. np.stack requires every row to have the same length; a variable-length array path (for example a point cloud) must instead be processed row-wise, e.g. with df[col].map(...).
Usage
topic = Topic.from_name_and_file("/imu/data", "file_abc123")
df = topic.get_data_as_df()
print(df.head())
# angular_velocity.x angular_velocity.y ...
# log_time
# 1722870127699468923 0.1 0.2 ...
# 1722870127699468924 0.15 0.25 ...# Filter specific message paths
df_filtered = topic.get_data_as_df(
message_paths_include=["angular_velocity"], message_paths_exclude=["angular_velocity.z"]
)
print(df_filtered.columns.tolist())
# ['angular_velocity.x', 'angular_velocity.y']# Unpack a fixed-length array-typed path (a list-valued column) into a numpy array
import numpy as np
df_accel = topic.get_data_as_df(message_paths_include=["acceleration"])
xyz = np.stack(df_accel["acceleration"].to_numpy()) # shape (N, 3)Topic.get_message_path()
Get a specific message path from this topic.
Retrieves a MessagePath object for the specified path, enabling access to individual fields or signals within the topic’s data schema.
Parameters
message_path strDot-delimited path to the desired attribute (e.g., “pose.position.x”).
Returns
MessagePath instance for the specified path.
Raises
ValueErrorNo message path with the given name exists in this topic.
Usage
topic = Topic.from_name_and_file("/imu/data", "file_abc123")
angular_vel_x = topic.get_message_path("angular_velocity.x")
print(angular_vel_x.canonical_data_type)
# CanonicalDataType.Number# Access message path statistics
print(angular_vel_x.mean)
# 0.125
print(angular_vel_x.std_dev)
# 0.05Topic.get_schema()
Retrieve the schema for this topic, if one exists.
Returns
A TopicSchema describing the message structure of this topic, or None if this topic has no schema.
Raises
schema_id references a schema that no longer exists.
Usage
topic = Topic.from_id("topic_xyz789")
schema = topic.get_schema()
if schema is not None:
for field in schema.fields:
print(field.name, field.data_type)Topic.get_time_bounds_by_association()
Get the earliest start and latest end across every topic of a file or dataset.
The same aggregate you would reach by folding start_time and end_time over the topics of that file or dataset, computed server-side in one request instead of one per page of topics.
Parameters
association roboto.The file or dataset whose topics are aggregated, e.g. Association.file("file_abc123").
owner_org_id Optional[str]Organization ID to scope the lookup. If None, uses caller’s org.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
Bounds in nanoseconds since the Unix epoch. Both fields are None when the association holds no topics, and either is None when no topic of the association carries that timestamp.
Raises
Caller lacks permission to access the file or dataset.
Usage
from roboto.association import Association
bounds = Topic.get_time_bounds_by_association(Association.file("file_abc123"))
print(bounds.start_time, bounds.end_time)
# 1722870127699468923 1722870187004821001Properties
Topic.message_count
Total number of messages in this topic.
Topic.message_paths
Sequence of message path records defining the topic’s schema.
Topic.metadata
Metadata dictionary associated with this topic.
Topic.modified
Timestamp when this topic was last modified.
Topic.modified_by
Identifier of the user or system that last modified this topic.
Topic.record
Topic representation in the Roboto database.
This property is on the path to deprecation. All TopicRecord attributes are accessible directly using a Topic instance.
Topic.refresh()
Refresh this topic instance with the latest data from the platform.
Fetches the current state of the topic from the Roboto platform and updates this instance’s data. Useful when the topic may have been modified by other processes or users.
Usage
topic = Topic.from_id("topic_xyz789")
# Topic may have been updated by another process
topic.refresh()
print(f"Current message count: {topic.message_count}")Return type
Properties
Topic.schema_checksum
Checksum of the topic’s message schema for validation.
Topic.schema_id
ID of the schema for this topic.
None if the topic has no schema, or if the schema has not yet been populated.
Topic.schema_name
Name of the message schema (e.g., ‘sensor_msgs/Image’).
Topic.set_default_representation()
Set the default representation for this topic.
Designates a specific representation as the default for this topic, which will be used when accessing topic data without specifying a particular representation.
Parameters
association roboto.Association pointing to the representation data.
storage_format roboto.Format of the representation data.
version intVersion number of the representation.
format Optional[str]Content format descriptor (e.g. “jpeg”, “sensor_msgs/Image”).
transformations Optional[list[str]]Transformation descriptors applied (e.g. [“downsample:0.5”]).
Returns
RepresentationRecord representing the newly set default representation.
Raises
Specified representation does not exist.
Caller lacks permission to modify the topic.
Usage
from roboto.association import Association
from roboto.domain.topics import RepresentationStorageFormat
topic = Topic.from_id("topic_xyz789")
default_repr = topic.set_default_representation(
association=Association.file("file_repr_456"),
storage_format=RepresentationStorageFormat.MCAP,
version=2,
)
print(topic.default_representation.representation_id)
# repr_789Properties
Topic.start_time
Start time of the topic data in nanoseconds since UNIX epoch.
Topic.to_association()
Convert this topic to an Association object.
Creates an Association object that can be used to reference this topic in other parts of the Roboto platform.
Returns
Association object representing this topic.
Usage
topic = Topic.from_id("topic_xyz789")
association = topic.to_association()
print(association.association_type)
# AssociationType.Topic
print(association.association_id)
# topic_xyz789Topic.update()
Updates a topic’s attributes and (optionally) its message paths.
Parameters
schema_name Union[Optional[str], roboto.topic schema name. Setting to None clears the attribute.
schema_checksum Union[Optional[str], roboto.topic schema checksum. Setting to None clears the attribute.
start_time Union[Optional[int], roboto.topic data start time, in epoch nanoseconds. Must be non-negative. Setting to None clears the attribute.
end_time Union[Optional[int], roboto.topic data end time, in epoch nanoseconds. Must be non-negative, and greater than start_time. Setting to None clears the attribute.
message_count Union[int, roboto.number of messages recorded for this topic. Must be non-negative.
metadata_changeset Union[roboto.a set of changes to apply to the topic’s metadata
message_path_changeset Union[roboto.a set of additions, deletions or updates to this topic’s message paths. Updating or deleting non-existent message paths has no effect. Attempting to (re-)add existing message paths raises RobotoConflictException, unless the changeset’s replace_all flag is set to True
Returns
this Topic object with any updates applied
Raises
if any method argument has an invalid value, e.g. a negative message_count
if, as part of the update, an attempt is made to add an already extant message path, and to this topic, and replace_all is not toggled on the message_path_changeset
Topic.update_message_path()
Update the metadata and attributes of a message path.
Modifies an existing message path within this topic, allowing updates to its metadata, data type, and canonical data type. This is useful for correcting or enhancing message path definitions after initial creation.
Parameters
message_path strName of the message path to update (e.g., “pose.position.x”).
metadata_changeset Union[roboto.Metadata changeset to apply to any existing metadata.
data_type Union[str, roboto.Native (application-specific) message path data type.
canonical_data_type Union[roboto.Canonical Roboto data type corresponding to the native data type.
path_in_schema Union[list[str], roboto.Returns
MessagePath instance representing the updated message path.
Raises
No message path with the given name exists for this topic.
Caller lacks permission to modify the topic.
Usage
from roboto.updates import TaglessMetadataChangeset
from roboto.domain.topics import CanonicalDataType
topic = Topic.from_id("topic_xyz789")
# Update metadata for a message path
changeset = TaglessMetadataChangeset(put_fields={"unit": "meters"})
updated_path = topic.update_message_path(message_path="pose.position.x", metadata_changeset=changeset)
print(updated_path.metadata["unit"])
# meters# Update data type and canonical type
updated_path = topic.update_message_path(
message_path="velocity", data_type="float64", canonical_data_type=CanonicalDataType.Number
)TopicIdentityRecord
Bases: pydantic.BaseModel
A durable identity for a topic.
Within an organization, topic names are unique: data logged under the same topic name in different files shares a single identity record.
Parameters
data AnyAttributes
TopicIdentityRecord.created
TopicIdentityRecord.created_by
TopicIdentityRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
TopicIdentityRecord.modified
TopicIdentityRecord.modified_by
TopicIdentityRecord.name
Human-readable topic name (e.g., "/camera/image_raw"). Unique within an organization.
TopicIdentityRecord.org_id
TopicPartitionRecord
Bases: pydantic.BaseModel
One file’s data for a topic.
Pairs a topic identity with a file and carries the facts that vary from file to file: the schema the file’s messages follow (schema_id), the device that produced them, and the data_range locating them inside the file, for formats that pack several slices of data into one shared file. A partition references a file, not a specific version; reads always resolve to the current version.
Parameters
data AnyAttributes
TopicPartitionRecord.created
TopicPartitionRecord.created_by
TopicPartitionRecord.data_range
The slice of the file this partition’s data occupies, as (start, end), or None for the whole file.
start alone identifies the partition within its (topic, file) pair, since a slice’s starting position is stable across re-ingest: re-declaring a slice that begins at the same position updates the existing partition instead of adding a second, overlapping one.
TopicPartitionRecord.device_id
ID of the device that produced this partition’s data, if known.
TopicPartitionRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
TopicPartitionRecord.modified
TopicPartitionRecord.modified_by
TopicPartitionRecord.org_id
TopicPartitionRecord.topic_part_id
TopicRecord
Bases: pydantic.BaseModel
Record representing a topic in the Roboto platform.
A topic is a collection of timestamped data records that share a common name and association (typically a file). Topics represent logical data streams from robotics systems, such as sensor readings, robot state information, or other time-series data.
Data from the same file with the same topic name are considered part of the same topic. Data from different files or with different topic names belong to separate topics, even if they have similar schemas.
When source files are chunked by time or size but represent the same logical data collection, they will produce multiple topic records for the same “logical topic” (same name and schema) across those chunks.
Parameters
data AnyAttributes
TopicRecord.association
Identifier and entity type with which this Topic is associated. E.g., a file, a dataset.
TopicRecord.created
TopicRecord.created_by
TopicRecord.default_representation
Default Representation for this Topic. Assume that if a MessagePath is not more specifically associated with a Representation, it should use this one.
TopicRecord.end_time
Timestamp of oldest message in topic, in nanoseconds since epoch (assumed Unix epoch).
TopicRecord.message_count
TopicRecord.message_paths
Zero to many MessagePathRecords associated with this TopicSource.
TopicRecord.modified
TopicRecord.modified_by
TopicRecord.org_id
TopicRecord.schema_checksum
Checksum of topic schema. May be None if topic does not have a known/named schema.
TopicRecord.schema_id
ID of the schema record for this topic. May be None if the topic has no schema, or if the schema record has not yet been populated.
TopicRecord.schema_name
Type of messages in topic. E.g., “sensor_msgs/PointCloud2”. May be None if topic does not have a known/named schema.
TopicRecord.start_time
Timestamp of earliest message in topic, in nanoseconds since epoch (assumed Unix epoch).
TopicRecord.topic_id
TopicRecord.topic_name
TopicSchema
Describes the field structure of a topic’s messages.
A topic schema is identified by a name (e.g., "sensor_msgs/Imu") and a content-based checksum deterministically derived from its fields. Schemas are deduplicated within an organization: topics whose fields share the same names, paths, and data types reference the same schema.
Use from_id() when you already know the schema_id. Topic.get_schema() retrieves the schema associated with a specific topic.
Usage
Retrieve a schema and inspect its fields:
from roboto.domain.topics import TopicSchema
schema = TopicSchema.from_id("ts_abc123")
print(schema.name, schema.checksum)
for field in schema.fields:
print(field.path_in_schema, field.data_type)Parameters
fields list[roboto.roboto_client roboto.Properties
TopicSchema.fields
Field definitions belonging to this schema.
TopicSchema.from_id()
Retrieve a schema by its ID.
Parameters
schema_id strUnique identifier of the schema to retrieve.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
Returns
A TopicSchema for the given schema_id.
Raises
No schema with this ID exists.
Usage
from roboto.domain.topics import TopicSchema
schema = TopicSchema.from_id("ts_abc123")
for field in schema.fields:
print(field.path_in_schema, field.data_type)Properties
TopicSchema.name
Informational label for the schema (e.g. "sensor_msgs/Imu"). Not part of identity; may be None.
TopicSchema.record
Underlying schema record.
TopicSchemaRecord
Bases: pydantic.BaseModel
A content-addressed topic schema.
Within an organization, two schemas with identical fields share a single record (identified by a deterministic checksum of the fields). name is a mutable, informational label (last-writer-wins) and is not part of the schema’s identity.
Parameters
data AnyAttributes
TopicSchemaRecord.checksum
Deterministic checksum computed over the schema’s fields; identical schemas share a checksum.
TopicSchemaRecord.created
TopicSchemaRecord.created_by
TopicSchemaRecord.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
TopicSchemaRecord.modified
TopicSchemaRecord.modified_by
TopicSchemaRecord.name
Informational label for the schema (e.g., "sensor_msgs/PointCloud2"). Not part of identity.
TopicSchemaRecord.org_id
Trigger
A rule that automatically invokes an action when specific events or conditions occur.
Triggers enable automated data processing workflows by monitoring for specific events (like new datasets being created) and automatically invoking actions when conditions are met. They eliminate the need for manual intervention in routine data processing tasks.
Triggers can be configured to:
- Monitor for new datasets, files, or other data sources
- Apply conditional logic to determine when to execute
- Specify input data patterns and action parameters
- Override compute requirements and container parameters
- Execute actions for each matching item or in batch
A trigger consists of:
- Target action to invoke
- Input data requirements and patterns
- Execution conditions and causes
- Parameter values and overrides
- Scheduling and execution settings
Parameters
roboto_client Optional[roboto.Properties
Trigger.condition
Trigger.create()
Create a new trigger that automatically invokes an action when conditions are met.
Creates a trigger that monitors for specific events (like new datasets or files) and automatically invokes the specified action when the trigger conditions are satisfied. This enables automated data processing workflows.
Parameters
name strUnique name for the trigger within the organization.
action_name strName of the action to invoke when the trigger fires.
required_inputs list[str]List of file patterns that must be present for the trigger to fire. Uses glob patterns like “**/*.bag” or “data/*.csv”.
Granularity of execution - Dataset creates one invocation per dataset, DatasetFile creates one invocation per matching file.
enabled boolWhether the trigger should be active immediately after creation.
action_digest Optional[str]Specific version digest of the action to invoke. If not provided, uses the latest version.
action_owner_id Optional[str]Organization ID that owns the target action. If not provided, searches in the caller’s organization.
additional_inputs Optional[list[str]]Optional additional file patterns to include in invocations.
causes Optional[list[roboto.List of events that can cause this trigger to be evaluated. If not provided, uses default causes.
compute_requirement_overrides Optional[roboto.Optional compute requirement overrides for action invocations.
condition Optional[roboto.Optional condition that must be met for the trigger to fire. Can filter based on metadata, file properties, etc.
container_parameter_overrides Optional[roboto.Optional container parameter overrides for action invocations.
parameter_values Optional[dict[str, Any]]Parameter values to pass to the action when invoked.
service_user_id Optional[str]Optional service user ID for authentication.
timeout Optional[int]Optional timeout override for action invocations in minutes.
caller_org_id Optional[str]Organization ID to create the trigger in. Defaults to caller’s org.
roboto_client Optional[roboto.Roboto client instance. Uses default if not provided.
Returns
The newly created Trigger instance.
Raises
If the trigger configuration is invalid.
If the request is malformed.
If the caller lacks permission to create triggers.
Usage
Create a simple trigger for ROS bag files:
from roboto.domain.actions import Trigger, TriggerForEachPrimitive
trigger = Trigger.create(
name="auto_process_bags",
action_name="ros_ingestion",
required_inputs=["**/*.bag"],
for_each=TriggerForEachPrimitive.Dataset,
)Create a conditional trigger with parameters:
from roboto.query import Condition
condition = Condition("metadata.sensor_type").equals("lidar")
trigger = Trigger.create(
name="lidar_processing",
action_name="lidar_processor",
required_inputs=["**/*.pcd"],
for_each=TriggerForEachPrimitive.Dataset,
condition=condition,
parameter_values={"resolution": "high", "filter": "statistical"},
)Create a trigger with compute overrides:
from roboto.domain.actions import ComputeRequirements
trigger = Trigger.create(
name="heavy_processing",
action_name="ml_inference",
required_inputs=["**/*.jpg", "**/*.png"],
for_each=TriggerForEachPrimitive.DatasetFile,
compute_requirement_overrides=ComputeRequirements(vCPU=8192, memory=16384),
)Trigger.delete()
Trigger.disable()
Trigger.enable()
Properties
Trigger.for_each
Trigger.from_name()
Parameters
Return type
Trigger.get_action()
Return type
Trigger.get_evaluations()
Parameters
limit Optional[int]page_token Optional[str]Return type
Trigger.get_evaluations_for_dataset()
Get all trigger evaluations for a specific dataset.
Retrieves the history of trigger evaluations that were performed for a given dataset, including successful invocations and failed attempts.
Parameters
dataset_id strThe ID of the dataset to get evaluations for.
owner_org_id Optional[str]Organization ID that owns the dataset. If not provided, searches in the caller’s organization.
roboto_client Optional[roboto.Roboto client instance. Uses default if not provided.
Yields
TriggerEvaluationRecord instances for the dataset.
Raises
If the dataset is not found.
If the caller lacks permission to access evaluations.
Return type
Usage
Get all evaluations for a dataset:
for evaluation in Trigger.get_evaluations_for_dataset("ds_12345"):
print(f"Trigger: {evaluation.trigger_name}, Status: {evaluation.status}")Check if any triggers succeeded for a dataset:
from roboto.domain.actions import TriggerEvaluationStatus
evaluations = list(Trigger.get_evaluations_for_dataset("ds_12345"))
successful = [e for e in evaluations if e.status == TriggerEvaluationStatus.Succeeded]
print(f"Found {len(successful)} successful trigger evaluations")Trigger.get_invocations()
Return type
Trigger.invoke()
Parameters
idempotency_id Optional[str]input_data_override Optional[list[str]]upload_destination Optional[roboto.Return type
Trigger.latest_evaluation()
Return type
Trigger.query()
Parameters
spec Optional[roboto.owner_org_id Optional[str]roboto_client Optional[roboto.Return type
Properties
Trigger.record
Trigger.to_dict()
Return type
Properties
Trigger.update()
Parameters
action_name Union[str, roboto.action_owner_id Union[str, roboto.action_digest Optional[Union[str, roboto.additional_inputs Optional[Union[list[str], roboto.causes Union[list[roboto.compute_requirement_overrides Optional[Union[roboto.container_parameter_overrides Optional[Union[roboto.condition Optional[Union[roboto.enabled Union[bool, roboto.parameter_values Optional[Union[dict[str, Any], roboto.required_inputs Union[list[str], roboto.timeout Optional[Union[int, roboto.Return type
Trigger.wait_for_evaluations_to_complete()
Wait for all evaluations for this trigger to complete.
Throws a TimeoutError if the timeout is reached.
Parameters
timeout floatThe maximum amount of time, in seconds, to wait for the evaluations to complete.
poll_interval roboto.The amount of time, in seconds, to wait between polling iterations.
Return type
TriggerEvaluationCause
Bases: enum.Enum
The cause of a TriggerEvaluationRecord is the reason why the trigger was selected for evaluation.
Represents the specific event that caused a trigger to be evaluated for potential execution. Different causes may result in different trigger behavior or input data selection.
Attributes
TriggerEvaluationCause.DatasetMetadataUpdate
Trigger evaluation caused by changes to dataset metadata.
TriggerEvaluationCause.FileIngest
Trigger evaluation caused by files being ingested into a dataset.
TriggerEvaluationCause.FileMetadataUpdate
Trigger evaluation caused by file metadata or tag updates.
TriggerEvaluationCause.FileUpload
Trigger evaluation caused by new files being uploaded to a dataset.
TriggerEvaluationCause.RecurringSchedule
Trigger evaluation caused by a recurring schedule.
This cause is used internally by the Roboto system, to track the evaluation history of scheduled triggers. It should not be used when creating or updating triggers, and doing so will result in an error.
To create a trigger that invokes an action on a recurring schedule, use ScheduledTrigger.
TriggerEvaluationOutcome
Bases: enum.Enum
The outcome of a TriggerEvaluationRecord is the result of the evaluation. A trigger can either invoke its associated action (one or many times) or be skipped. If skipped, a skip reason is provided.
TriggerEvaluationOutcomeReason
Bases: enum.Enum
Context for why a trigger evaluation has its TriggerEvaluationOutcome
Attributes
TriggerEvaluationOutcomeReason.AlreadyRun
This trigger has already run its associated action for this dataset and/or file.
TriggerEvaluationOutcomeReason.ConditionNotMet
The trigger’s condition is not met.
TriggerEvaluationOutcomeReason.NoMatchingFiles
In the case of a dataset trigger, there is no subset of files that, combined, match ALL of the trigger’s required inputs.
In the case of a file trigger, there are no files that match ANY of the trigger’s required inputs.
TriggerEvaluationOutcomeReason.TriggerDisabled
The trigger is disabled.
TriggerEvaluationRecord
Bases: pydantic.BaseModel
Record of a point-in-time evaluation of whether to invoke an action associated with a trigger for a data source.
Parameters
data AnyAttributes
TriggerEvaluationRecord.cause
TriggerEvaluationRecord.data_constraint
TriggerEvaluationRecord.data_source
TriggerEvaluationRecord.evaluation_end
TriggerEvaluationRecord.evaluation_start
TriggerEvaluationRecord.outcome
TriggerEvaluationRecord.outcome_reason
TriggerEvaluationRecord.status
TriggerEvaluationRecord.status_detail
TriggerEvaluationRecord.trigger_evaluation_id
TriggerEvaluationRecord.trigger_id
TriggerEvaluationStatus
Bases: enum.Enum
When a trigger is selected for evaluation, a trigger evaluation record is created with a status of Pending. The evaluation can either run to completion (regardless of its outcome), in which case the status is Evaluated, or hit an unexpected exception, in which case the status is Failed.
TriggerEvaluationsSummaryResponse
Bases: pydantic.BaseModel
Response containing summary information about trigger evaluations.
Provides high-level statistics about trigger evaluation status, useful for monitoring and debugging trigger performance.
Parameters
data AnyAttributes
TriggerForEachPrimitive
Bases: roboto.compat.StrEnum
Defines the granularity at which a trigger executes.
Determines whether the trigger creates one invocation per dataset or one invocation per file within datasets that match the trigger conditions.
TriggerRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of a trigger.
Contains all the configuration and metadata for a trigger, including the target action, input requirements, conditions, and execution settings.
This is the underlying data structure used by the Trigger domain class to store and transmit trigger information.
Parameters
data AnyAttributes
TriggerRecord.action
Reference to the action that should be invoked.
TriggerRecord.additional_inputs
Optional additional file patterns to include.
TriggerRecord.causes
List of events that can cause this trigger to be evaluated.
TriggerRecord.compute_requirement_overrides
compute_requirement_overrides roboto.Optional compute requirement overrides.
TriggerRecord.condition
Optional condition that must be met for trigger to fire.
TriggerRecord.container_parameter_overrides
container_parameter_overrides roboto.Optional container parameter overrides.
TriggerRecord.for_each
Granularity of trigger execution (Dataset or DatasetFile).
TriggerRecord.parameter_values
Parameter values to pass to the action.
TriggerRecord.required_inputs
File patterns that must be present for trigger to fire.
TriggerRecord.validate_additional_inputs()
Parameters
value Optional[list[str]]Return type
TriggerRecord.validate_required_inputs()
Parameters
value list[str]Return type
UpdateActionRequest
Bases: pydantic.BaseModel
Request payload to update an action.
Contains the changes to apply to an existing action. Only specified fields will be updated; others remain unchanged. Uses NotSet sentinel values to distinguish between explicit None values and unspecified fields.
Parameters
data AnyAttributes
UpdateActionRequest.compute_requirements
compute_requirements roboto.New compute requirements (CPU, memory).
UpdateActionRequest.container_parameters
container_parameters roboto.New container parameters (image, entrypoint, etc.).
UpdateActionRequest.description
New detailed description.
UpdateActionRequest.inherits
New action reference to inherit from.
UpdateActionRequest.metadata_changeset
Changes to apply to metadata (add, remove, update keys).
UpdateActionRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateActionRequest.parameter_changeset
parameter_changeset roboto.Changes to apply to parameters (add, remove, update).
UpdateActionRequest.requires_downloaded_inputs
Whether to download input files before execution.
UpdateActionRequest.short_description
New brief description (max 140 characters).
UpdateActionRequest.timeout
New maximum execution time in minutes.
UpdateActionRequest.validate_uri()
UpdateCollectionRequest
Bases: pydantic.BaseModel
Request payload to update a collection
Parameters
data AnyAttributes
UpdateCollectionRequest.add_resources
add_resources list[roboto.UpdateCollectionRequest.add_tags
UpdateCollectionRequest.custom_fields_changeset
Changes to apply to Ready custom-field values on this collection.
Each referenced field name must be a Ready custom field for this collection’s org and the Collection entity type; each set_fields value must satisfy the field’s declared type. Names that are undefined or not Ready are rejected with a structured error. Field names not mentioned by the changeset are left unchanged.
UpdateCollectionRequest.description
UpdateCollectionRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateCollectionRequest.name
UpdateCollectionRequest.remove_resources
remove_resources list[roboto.UpdateCollectionRequest.remove_tags
UpdateCommentRequest
Bases: pydantic.BaseModel
Request payload for updating an existing comment.
This model defines the information that can be modified when updating a comment.
Parameters
data AnyAttributes
UpdateCommentRequest.comment_text
Updated text content of the comment, may include @mention syntax.
UpdateDatasetRequest
Bases: pydantic.BaseModel
Request payload for updating dataset properties.
Used to modify dataset metadata, description, name, and other properties. Only specified fields will be updated; others remain unchanged.
Parameters
data AnyAttributes
UpdateDatasetRequest.custom_fields_changeset
Changes to apply to Ready custom-field values on this dataset.
Each referenced field name must be a Ready custom field for this dataset’s org and the Dataset entity type; each set_fields value must satisfy the field’s declared type. Names that are undefined or not Ready are rejected with a structured error. Field names not mentioned by the changeset are left unchanged.
UpdateDatasetRequest.description
New description for the dataset. Set to None to clear the description.
UpdateDatasetRequest.device_id
New device ID for the dataset. Set to None to clear the device association.
UpdateDatasetRequest.metadata_changeset
Metadata changes to apply (add, update, or remove fields/tags).
UpdateDatasetRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateDatasetRequest.name
name Annotated[str, pydantic.New name for the dataset (max 120 characters). Set to None to clear the name.
UpdateFileRecordRequest
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 AnyAttributes
UpdateFileRecordRequest.description
New description for the file, or NotSet to leave unchanged.
UpdateFileRecordRequest.device_id
New device ID for the file, or NotSet to leave unchanged.
UpdateFileRecordRequest.ingestion_complete
Set to True to mark file as fully ingested, or NotSet to leave unchanged.
UpdateFileRecordRequest.metadata_changeset
Metadata changes to apply (add, update, or remove fields/tags), or NotSet to leave unchanged.
UpdateFileRecordRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateInvocationStatus
Bases: pydantic.BaseModel
Request payload to update an invocation’s status.
Used to record status changes during invocation execution, such as transitioning from Queued to Running to Completed.
Parameters
data AnyAttributes
UpdateInvocationStatus.status
The new status for the invocation.
UpdateMessagePathRequest
Bases: pydantic.BaseModel
Request to update an existing message path within a topic.
Allows modification of message path attributes including metadata, data type, and canonical data type. Used to correct or enhance message path definitions after initial creation.
Parameters
data AnyAttributes
UpdateMessagePathRequest.canonical_data_type
Canonical Roboto data type for the data under this message path (optional).
Note: updating this attribute should be done with care, as it affects Roboto’s ability to interpret and visualize the data.
UpdateMessagePathRequest.data_type
Native data type for the data under this message path (optional).
UpdateMessagePathRequest.has_updates()
Check whether this request would result in any message path modifications.
Returns
True if the request contains changes that would modify the message path.
Attributes
UpdateMessagePathRequest.metadata_changeset
A set of changes to the message path’s metadata (optional).
UpdateMessagePathRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateMessagePathRequest.path_in_schema
List of path components representing the field’s location in the source data schema (optional).
For nested fields like ‘position.x’, this would be [‘position’, ‘x’].
UpdateTopicRequest
Bases: pydantic.BaseModel
Request to update an existing topic’s properties.
Allows modification of topic attributes including temporal boundaries, message count, schema information, metadata, and message paths.
Parameters
data AnyAttributes
UpdateTopicRequest.end_time
UpdateTopicRequest.message_count
UpdateTopicRequest.message_path_changeset
UpdateTopicRequest.metadata_changeset
UpdateTopicRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateTopicRequest.schema_checksum
UpdateTopicRequest.schema_name
UpdateTopicRequest.start_time
UpdateTriggerRequest
Bases: pydantic.BaseModel
Request payload to update an existing trigger.
Contains the changes to apply to a trigger. Only specified fields will be updated; others remain unchanged. Uses NotSet sentinel values to distinguish between explicit None values and unspecified fields.
Parameters
data AnyAttributes
UpdateTriggerRequest.action_digest
New specific version digest of the action.
UpdateTriggerRequest.action_name
New action name to invoke.
UpdateTriggerRequest.action_owner_id
New organization ID that owns the target action.
UpdateTriggerRequest.additional_inputs
New additional file patterns to include.
UpdateTriggerRequest.causes
causes list[roboto.New list of events that can cause trigger evaluation.
UpdateTriggerRequest.compute_requirement_overrides
compute_requirement_overrides roboto.New compute requirement overrides.
UpdateTriggerRequest.condition
New condition that must be met for trigger to fire.
UpdateTriggerRequest.container_parameter_overrides
container_parameter_overrides roboto.New container parameter overrides.
UpdateTriggerRequest.enabled
New enabled status for the trigger.
UpdateTriggerRequest.for_each
for_each roboto.New execution granularity (Dataset or DatasetFile).
UpdateTriggerRequest.model_config
model_config #Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
UpdateTriggerRequest.parameter_values
New parameter values to pass to the action.
UpdateTriggerRequest.required_inputs
New list of required file patterns.
UpdateTriggerRequest.timeout
New timeout override for action invocations.
UpdateUserRequest
Bases: pydantic.BaseModel
Request payload to update an existing user.
Parameters
data AnyAttributes
UpdateUserRequest.notification_channels_enabled
Updated notification channel preferences.
UpdateUserRequest.notification_types_enabled
Updated notification type preferences.
UpdateUserRequest.picture_url
Updated URL to the user’s profile picture.
UpdateUserRequest.validate_non_empty_string()
Parameters
value Optional[str]info pydantic.Return type
UploadDestinationType
Bases: enum.Enum
Type of upload destination for invocation outputs.
Defines where files generated by action invocations should be uploaded. Currently supports datasets as the primary destination type.
Attributes
UploadDestinationType.Dataset
Outputs will be uploaded to a dataset. This is the default.
UploadDestinationType.Unknown
The output destination is unknown.
This destination type exists for compatibility between different versions of the Roboto SDK and the Roboto service backend. It should not be used directly in action invocation requests. If you encounter it in an SDK response, consider upgrading to the latest available SDK version.
User
Represents an individual who has access to the Roboto platform.
Users are the fundamental identity entities in Roboto. They can be members of organizations, create and manage datasets and files within those organizations, and execute actions. Users are created during the signup process and cannot be instantiated directly by SDK users.
User IDs are globally unique across the entire Roboto platform and typically correspond to email addresses for human users or service identifiers for automated users.
Parameters
roboto_client Optional[roboto.User.create()
Create a new user in Roboto.
This API is only used by the Roboto platform itself as part of the signup process. Any other caller will receive an Unauthorized response from the Roboto service.
Parameters
User creation request containing user details.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
A new User instance representing the created user.
Raises
The caller is not authorized to create users.
The request contains invalid data.
Usage
Create a new service user:
from roboto import CreateUserRequest, User
request = CreateUserRequest(user_id="service@example.com", name="Service User", is_service_user=True)
user = User.create(request) # Only works for platform itselfUser.delete()
Delete this user from Roboto.
Permanently removes the user and all associated data. This action cannot be undone.
Raises
The caller is not authorized to delete this user.
The user no longer exists.
Return type
Usage
Delete the current user:
from roboto import User
user = User.for_self()
user.delete() # Permanently removes the userUser.for_self()
Retrieve the current authenticated user.
Returns the User object for the currently authenticated user based on the authentication credentials in the provided or default client.
Parameters
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
User instance representing the authenticated user.
Raises
No valid authentication credentials provided.
Usage
Get the current user:
from roboto import User
current_user = User.for_self()
print(f"Current user: {current_user.name}")User.from_id()
Load an existing user by their unique user ID.
User IDs are globally unique across the Roboto platform and typically correspond to email addresses for human users.
Parameters
user_id strUnique identifier for the user to retrieve.
roboto_client Optional[roboto.Optional Roboto client instance. If not provided, uses the default client.
Returns
User instance for the specified user ID.
Raises
No user exists with the specified ID.
The caller is not authorized to access this user.
Usage
Load a user by email:
from roboto import User
user = User.from_id("alice@example.com")
print(f"User name: {user.name}")Properties
User.name
Human-readable display name for this user.
Returns
The user’s display name, or None if not set.
User.record
Access the underlying user record.
Provides access to the raw UserRecord containing all user data fields. This is useful for accessing fields not exposed as properties.
Returns
The underlying UserRecord instance.
User.to_dict()
Convert this user to a dictionary representation.
Returns a JSON-serializable dictionary containing all user data.
Returns
Dictionary representation of the user.
Usage
Export user data:
from roboto import User
user = User.for_self()
user_data = user.to_dict()
print(f"User created: {user_data.get('created')}")User.update()
Parameters
name Optional[str]picture_url Optional[str]notification_channels_enabled Optional[dict[roboto.notification_types_enabled Optional[dict[roboto.Return type
Properties
User.user_id
Unique identifier for this user.
User IDs are globally unique across the Roboto platform and typically correspond to email addresses for human users.
Returns
The user’s unique identifier.
UserRecord
Bases: pydantic.BaseModel
A wire-transmissible representation of a user.
Parameters
data AnyUserRecord.is_comment_mentions_enabled()
Return type
UserRecord.is_email_notifications_enabled()
Return type
Attributes
UserRecord.is_service_user
Whether this is a service user for automated operations.
Service users can be used to perform actions on behalf of customers. For example, a service user can be associated with a Trigger, which will then invoke its corresponding Action as the service user.
UserRecord.is_system_user
Whether this is a system user for internal platform operations.
UserRecord.notification_channels_enabled
Mapping of notification channels to their enabled/disabled status.
UserRecord.notification_types_enabled
Mapping of notification types to their enabled/disabled status.
client_tool()
Decorator that converts a function into a ClientTool.
Usable bare (@client_tool) or with keyword overrides (@client_tool(description="...")). See ClientTool.from_function() for how descriptions are resolved.
Usage
Bare — infers everything from the function, including per-parameter descriptions from the docstring’s Args: section:
@client_tool
def remember(fact: str) -> str:
"""Store a fact in long-term memory.
Args:
fact: A standalone sentence worth remembering.
"""
...With overrides:
@client_tool(name="store_fact", description="Persist a fact.")
def _store(fact: str) -> str: ...experimental()
Mark a class, function, or method as experimental.
Experimental APIs may be incomplete, subject to change, or removed without notice.
Prepends a .. warning:: notice to the target’s docstring, so documentation rendered from it and help() show the target as experimental, then returns the target itself.
Can be used in three ways:
@experimental
def my_function(): ...
@experimental("Custom message about this API.")
def my_function(): ...
@experimental(message="Custom message about this API.")
def my_function(): ...Parameters
targetThe object to mark when applied as @experimental, or the notice text when called as @experimental("...").
messageThe notice text. Defaults to “<qualified name of the target> is experimental and may change or be removed without notice.”
Raises
TypeErrorThe target is not callable, as when the decorator is placed above @property or @classmethod and not below it.