roboto.action_runtime.action_input.input_resolver
Module Contents
ActionInputResolver
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_service roboto.ActionInputResolver.from_env()
Parameters
roboto_client Optional[roboto.roboto_search Optional[roboto.Return type
ActionInputResolver.resolve_input_spec()
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.Input specification containing data selectors. See InvocationInput for more detail.
download boolIf True, download all resolved files to local disk. Defaults to False.
download_path Optional[pathlib.Directory path where files should be downloaded. If not provided and download=True, a temporary directory will be created. Ignored if download=False.
Returns
ActionInputRecord containing:
- 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.
Usage
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 emptyResolve 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 pathsResolve 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