roboto.action_runtime.action_input#
Submodules#
Package Contents#
- class roboto.action_runtime.action_input.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:
record (ActionInputRecord)
roboto_client (roboto.http.RobotoClient)
- Return type:
- 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 fromself.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:
- topics: collections.abc.Sequence[roboto.domain.topics.Topic] = []#
Topics passed as input data to an action invocation.
- class roboto.action_runtime.action_input.ActionInputRecord(/, **data)#
Bases:
pydantic.BaseModel
Serializable representation of an
ActionInput
.- Parameters:
data (Any)
- files: collections.abc.Sequence[tuple[roboto.domain.files.FileRecord, pathlib.Path | None]] = None#
- topics: collections.abc.Sequence[roboto.domain.topics.TopicRecord] = None#
- class roboto.action_runtime.action_input.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_resolver (roboto.action_runtime.action_input.file_resolver.InputFileResolver)
topic_resolver (roboto.action_runtime.action_input.topic_resolver.InputTopicResolver)
file_downloader (roboto.domain.files.FileDownloader)
- file_downloader#
- classmethod from_env(roboto_client=None, roboto_search=None)#
- Parameters:
roboto_client (Optional[roboto.http.RobotoClient])
roboto_search (Optional[roboto.roboto_search.RobotoSearch])
- Return type:
- 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
- roboto.action_runtime.action_input.DEFAULT_INPUT_FILE#
- class roboto.action_runtime.action_input.InputFileResolver(roboto_client=None, roboto_search=None)#
Resolves file selectors to file entities, if available.
- Parameters:
roboto_client (Optional[roboto.http.RobotoClient])
roboto_search (Optional[roboto.roboto_search.RobotoSearch])
- resolve(selector)#
- Parameters:
selector (roboto.domain.actions.FileSelector)
- Return type:
list[roboto.domain.files.File]
- resolve_all(file_selectors)#
- Parameters:
file_selectors (collections.abc.Sequence[roboto.domain.actions.FileSelector])
- Return type:
list[roboto.domain.files.File]
- roboto_client = None#
- roboto_search = None#
- class roboto.action_runtime.action_input.InputTopicResolver(roboto_client=None, roboto_search=None, download=False)#
Resolves topic selectors to topic entities, if available.
- Parameters:
roboto_client (Optional[roboto.http.RobotoClient])
roboto_search (Optional[roboto.roboto_search.RobotoSearch])
download (bool)
- download = False#
- resolve(topic_selector)#
- Parameters:
topic_selector (roboto.domain.actions.DataSelector)
- Return type:
- resolve_all(topic_selectors)#
- Parameters:
topic_selectors (collections.abc.Sequence[roboto.domain.actions.DataSelector])
- Return type:
- roboto_client = None#
- roboto_search = None#