roboto.action_runtime.action_input.input_resolver#

Module Contents#

class roboto.action_runtime.action_input.input_resolver.ActionInputResolver(file_resolver, session_resolver, topic_resolver, file_service)#

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:
classmethod from_env(roboto_client=None, roboto_search=None)#
Parameters:
Return type:

ActionInputResolver

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

Resolve the input spec’s file, session, and topic selectors to the entities they match.

When the spec asks for files, sessions, or topics and none match, that category comes back empty with a logged warning rather than an error.

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.

  • sessions: List of SessionRecord instances.

  • 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