roboto.action_runtime#

Submodules#

Package Contents#

class roboto.action_runtime.ActionInput#

Resolved references to input data an Action was given to operate on.

To use, access via get_input().

Example

From within an Action, list files passed as input and check their size:

>>> context = InvocationContext.from_env()
>>> action_input = context.get_input()
>>> for file, local_path in action_input.files:
>>>     print(f"{file.file_id} is {local_path.stat().st_size} bytes")
files: collections.abc.Sequence[tuple[roboto.domain.files.File, pathlib.Path | None]] = []#

Files passed as input data to an action invocation.

A file is represented as a tuple of (File, Optional[Path]) where: - File exposes metadata about the file and useful file operations - Optional[Path] is the local file path if the file has been downloaded

classmethod from_record(record, roboto_client)#

Create an ActionInput instance from its serialized representation.

Parameters:
Return type:

ActionInput

get_topics_by_name(topic_name)#

Return any topics in this ActionInput that have the provided name.

Parameters:

topic_name (str) – Topic name to look for.

Returns:

A list of matching Topic instances from self.topics. If no topics have the provided name, the list will be empty. Otherwise, there will be one or more topics in the list, depending on the topic selectors provided to the action invocation.

Return type:

list[roboto.domain.topics.Topic]

topics: collections.abc.Sequence[roboto.domain.topics.Topic] = []#

Topics passed as input data to an action invocation.

class roboto.action_runtime.ActionInputResolver(file_resolver, topic_resolver, file_downloader)#

Resolves the action invocation input spec to concrete Roboto entities.

The entities are packaged together in an ActionInput instance, which is available to action code via ActionRuntime.

Parameters:
file_downloader#
classmethod from_env(roboto_client=None, roboto_search=None)#
Parameters:
Return type:

ActionInputResolver

input_file_resolver#
input_topic_resolver#
resolve_input_spec(input_spec, download=False, download_path=None)#

This method takes an InvocationInput containing data selectors (e.g., for files and topics) and resolves them to concrete entities.

Optionally downloads files to a local path.

Parameters:

  • input_spec (roboto.domain.actions.InvocationInput) – Input specification containing data selectors. See InvocationInput for more detail.

  • download (bool) – If True, download all resolved files to local disk. Defaults to False.

  • download_path (Optional[pathlib.Path]) – Directory path where files should be downloaded. If not provided and download=True, a temporary directory will be created. Ignored if download=False.

Returns:

  • files: List of (FileRecord, Optional[Path]) tuples. Path is None if download=False, otherwise contains the local path where the file was downloaded.

  • topics: List of TopicRecord instances.

Return type:

ActionInputRecord containing

Examples

Resolve files using a RoboQL query without downloading:

>>> input_spec = InvocationInput.file_query('dataset_id = "ds_abc123" AND path LIKE "%.mcap"')
>>> result = resolver.resolve_input_spec(input_spec)
>>> # result.files contains (FileRecord, None) tuples
>>> # result.topics is empty

Resolve and download files to a specific directory:

>>> input_spec = InvocationInput.file_query('dataset_id = "ds_abc123" AND path LIKE "%.mcap"')
>>> result = resolver.resolve_input_spec(input_spec, download=True, download_path=Path("/tmp/data"))
>>> # result.files contains (FileRecord, Path) tuples with local paths

Resolve both files and topics:

>>> input_spec = InvocationInput(
...     files=FileSelector(query='dataset_id = "ds_abc123" AND path LIKE "%.mcap"'),
...     topics=DataSelector(names=["battery_status", "gps"])
... )
>>> result = resolver.resolve_input_spec(input_spec)
>>> # result.files contains file records
>>> # result.topics contains topic records
class roboto.action_runtime.ActionRuntime(*args, **kwargs)#

Bases: InvocationContext

Deprecated. Use InvocationContext instead.

exception roboto.action_runtime.ActionRuntimeException#

Bases: Exception

Base class for all exceptions raised by the action_runtime submodule.

class roboto.action_runtime.ExitCode#

Bases: enum.IntEnum

Defined exit codes used by the action runtime. Exception codes are adapted from /usr/include/sysexits.h

ConfigurationError = 78#

From /usr/include/sysexits.h: > EX_CONFIG 78 /* configuration error */

InternalError = 70#

From /usr/include/sysexits.h: > EX_SOFTWARE – An internal software error has been detected. > This should be limited to non-operating system related > errors as possible.

Success = 0#
UsageError = 64#

From /usr/include/sysexits.h: > EX_USAGE – The command was used incorrectly, e.g., with > the wrong number of arguments, a bad flag, a bad > syntax in a parameter, or whatever.

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

put_fields(relative_path, metadata)#

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

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

Examples

>>> from roboto import 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 (str)

  • metadata (dict[str, Any])

put_tags(relative_path, tags)#

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

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

Examples

>>> from roboto import 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 (str)

  • tags (list[str])

remove_fields(relative_path, keys)#

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

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

Examples

>>> from roboto import 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 (str)

  • keys (list[str])

remove_tags(relative_path, tags)#

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

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

Examples

>>> from roboto import 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 (str)

  • tags (list[str])

set_description(relative_path, description)#

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

Examples

>>> from roboto import 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 (str)

  • description (Optional[str])

class roboto.action_runtime.InvocationContext(dataset_id, input_dir, invocation_id, org_id, output_dir, input_data_manifest_file=None, parameters_file=None, secrets_file=None, roboto_client=None)#

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/local/input/dir"),
...     invocation_id="iv_XXXXXXXXXXXX",
...     org_id="og_XXXXXXXXXXXX",
...     output_dir=pathlib.Path("/path/to/local/output/dir"),
... )
Parameters:

  • dataset_id (str)

  • input_dir (pathlib.Path)

  • invocation_id (str)

  • org_id (str)

  • output_dir (pathlib.Path)

  • input_data_manifest_file (Optional[pathlib.Path])

  • parameters_file (Optional[pathlib.Path])

  • secrets_file (Optional[pathlib.Path])

  • roboto_client (Optional[roboto.http.RobotoClient])

property dataset: roboto.domain.datasets.Dataset#

A Dataset object for the dataset whose data this action is operating on.

This object will be lazy-initialized the first time it is accessed, which might result in a RobotoNotFoundException if the dataset does not exist. After the first call, the dataset will be cached.

This is particularly useful for adding tags or metadata to a dataset at runtime, for example:

>>> from roboto import InvocationContext
>>> context = InvocationContext.from_env()
>>> context.dataset.put_tags(["tagged_by_action"])
>>> context.dataset.put_metadata({"voltage_spikes_seen": 693})
Return type:

roboto.domain.datasets.Dataset

property dataset_id: str#

The ID of the dataset whose data this action is operating on.

Return type:

str

property file_changeset_manager: roboto.action_runtime.file_changeset.FilesChangesetFileManager#

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.

Return type:

roboto.action_runtime.file_changeset.FilesChangesetFileManager

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

Example

>>> from roboto import InvocationContext
>>> context = InvocationContext.from_env()
Return type:

InvocationContext

get_input()#

Instance of ActionInput containing resolved references to input data.

Return type:

roboto.action_runtime.action_input.ActionInput

get_optional_parameter(name, default_value=None)#

Retrieve the value of the action parameter with the given name, defaulting to default_value if the parameter is not set.

Parameters:

  • name (str) – The 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.

Return type:

Optional[str]

Examples

>>> import roboto
>>> context = roboto.InvocationContext.from_env()
>>> context.get_optional_parameter("model_version", "latest")
"latest"
get_parameter(name)#

Gets the value of the action parameter with the given name, raising an ActionRuntimeException if the parameter is not set.

Parameters:

name (str)

Return type:

str

get_secret_parameter(name)#

Gets the value of the secret action parameter with the given name.

Parameters:

name (str)

Return type:

str

property input_dir: pathlib.Path#

The directory where the action’s input files are located.

Return type:

pathlib.Path

property invocation: roboto.domain.actions.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.

Return type:

roboto.domain.actions.Invocation

property invocation_id: str#

The ID of the currently running action invocation.

Return type:

str

property org: roboto.domain.orgs.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.

Return type:

roboto.domain.orgs.Org

property org_id: str#

The ID of the org which invoked the currently running action.

Return type:

str

property output_dir: pathlib.Path#

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.

Return type:

pathlib.Path

property roboto_client: roboto.http.RobotoClient#

The RobotoClient instance used by this action runtime.

Return type:

roboto.http.RobotoClient

exception roboto.action_runtime.PrepareEnvException(exit_code, reason)#

Bases: ActionRuntimeException

Base class for all exceptions raised by the action_runtime submodule.

Parameters:
exit_code: roboto.action_runtime.exit_codes.ExitCode#
reason: str#
roboto.action_runtime.prepare_invocation_environment(action_parameters, provided_parameter_values, parameters_values_file, secrets_file, requires_downloaded_inputs, input_data, input_download_dir, inputs_data_manifest_file, dataset_metadata_changeset_path, org_id, roboto_client, roboto_search)#
Responsibilities:

  1. Merge supplied parameter values with action parameter defaults

  2. Resolve secret parameter values (if any)

  3. Write provided parameters/values and resolved secrets to relevant files

  4. Resolve input data (files/topics) and optionally download data

  5. Write input data manifest

  6. Create dataset_metadata_changeset_file

Raises:

PrepareEnvException – If any of the preparation steps fail.

Parameters:
Return type:

None

On this page