roboto.action_runtime.action_input.input_resolver#

Module Contents#

class roboto.action_runtime.action_input.input_resolver.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
type roboto.action_runtime.action_input.input_resolver.OptionallyDownloadedFile = tuple[File, pathlib.Path | None]#
roboto.action_runtime.action_input.input_resolver.log#
On this page