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:
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]

sessions: collections.abc.Sequence[roboto.experimental.sessions.Session] = []#

Sessions passed as input data to an action invocation.

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#
sessions: collections.abc.Sequence[roboto.experimental.sessions.SessionRecord] = None#
topics: collections.abc.Sequence[roboto.domain.topics.TopicRecord] = None#
class roboto.action_runtime.action_input.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
roboto.action_runtime.action_input.DEFAULT_INPUT_FILE#
class roboto.action_runtime.action_input.InputFileResolver(roboto_client=None, roboto_search=None)#

Looks up the files an action invocation runs on, from the file selectors declared as its inputs.

A selector names files by ID, by name, by path within a dataset, or with a RoboQL query; every field it populates is looked up and the matches are returned together. resolve_all returns each file once, while resolve repeats a file that matches on more than one field.

Parameters:
resolve(selector)#

Looks up the files one selector names.

A dataset_id narrows the lookup only alongside paths; every other field ignores it, with a warning.

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#
class roboto.action_runtime.action_input.InputSessionResolver(roboto_client=None, roboto_search=None)#

Looks up the sessions an action invocation runs on, from the session selectors declared as its inputs.

A selector names sessions by ID, by name, or with a RoboQL query; every field it populates is looked up and the matches are returned together. resolve_all returns each session once, while resolve repeats a session that matches on more than one field.

Parameters:
resolve(session_selector)#

Looks up the sessions one selector names. A dataset_id on the selector is ignored, with a warning.

Parameters:

session_selector (roboto.domain.actions.DataSelector)

Return type:

list[roboto.experimental.sessions.Session]

resolve_all(session_selectors)#
Parameters:

session_selectors (collections.abc.Sequence[roboto.domain.actions.DataSelector])

Return type:

list[roboto.experimental.sessions.Session]

roboto_client = None#
class roboto.action_runtime.action_input.InputTopicResolver(roboto_client=None, roboto_search=None)#

Looks up the topics an action invocation runs on, from the topic selectors declared as its inputs.

A selector names topics by ID, by name, or with a RoboQL query; every field it populates is looked up and the matches are returned together. resolve_all returns each topic once, while resolve repeats a topic that matches on more than one field.

Parameters:
resolve(topic_selector)#

Looks up the topics one selector names. A dataset_id on the selector is ignored, with a warning.

Parameters:

topic_selector (roboto.domain.actions.DataSelector)

Return type:

list[roboto.domain.topics.Topic]

resolve_all(topic_selectors)#
Parameters:

topic_selectors (collections.abc.Sequence[roboto.domain.actions.DataSelector])

Return type:

list[roboto.domain.topics.Topic]

roboto_client = None#
On this page