ros_sugar.core._action_registry#

What a running stack can be asked to do, by name.

At launch an action is a Python callable a recipe already holds. At runtime a caller has only a string, and no way of knowing which component owns the thing it names. This registry is the translation between the two: it is built once from the components the Launcher knows about, and every runtime reference is resolved against it.

A reference is "owner/name" – the component’s node name and one of the things it exposes:

registry = SystemActionRegistry.from_components([planner, driver])
registry.get("driver/move_to_unblock")          # a @component_action method
registry.get("planner/plan_path")               # its main action server
registry.get("planner/save_plan_to_file")       # an additional service
registry.get("controller/track_vision_target")  # an additional action server

Resolution stops here: turning an entry into something callable needs the Monitor’s clients, so it lives on the Monitor. This module stays ROS free.

Everything a component exposes is registered, including the lifecycle actions every component inherits: a mission routine legitimately wants driver/restart. A consumer that must not offer some of them filters for itself, on whatever it considers dangerous.

NOTE: unrelated to :class:ros_sugar.robot.ActionRegistry, which is a registry of factories producing Actions for robot plugins.

Module Contents#

Classes#

RegisteredAction

One thing the running stack can be asked to do.

RegisteredEvent

One condition the running stack can be asked to watch for.

SystemActionRegistry

Every action the running stack exposes, addressable by name.

API#

class ros_sugar.core._action_registry.RegisteredAction#

Bases: ros_sugar.config.base_attrs.BaseAttrs

One thing the running stack can be asked to do.

Parameters:

  • ref – How a caller names it, as “owner/name”

  • owner – Node name of the component that runs it, or “monitor”

  • name – The short name within that owner. Never contains a slash, so a reference always splits cleanly

  • kind – Which resolution path applies, one of the module constants. The Monitor dispatches on this

  • description – What it does, for a caller listing what is available

  • schema – Its tool schema, whole, for a caller that builds tool calls from it. A component method has one when @component_action(description=...) was given a dict, and none when it was described in prose; a plugin’s action always has one, since the plugin’s own registry makes a whole schema of however it was described

  • signature – Its call signature, with the bound instance dropped

  • interface_type – Name of the action or service type, for the kinds that have one

  • server_name – The ROS name to address, for the kinds that have one. Needed because a server’s name is not derivable from the reference

  • in_process – False when the owner runs as its own process, which is the case a caller cannot reach by holding the object

asdict(filter: Optional[Callable] = None) → Dict#
to_dict() → Dict#
asdict_explicit() → Dict#
from_dict(dict_obj: Dict) → None#
from_file(file_path: str, nested_root_name: Union[str, None] = None, get_common: bool = False) → bool#
to_json() → Union[str, bytes, bytearray]#
from_json(json_obj: Union[str, bytes, bytearray]) → None#
has_attribute(attr_name: str) → bool#
get_attribute_type(attr_name: str) → Optional[type]#
update_value(attr_name: str, attr_value: Any) → bool#
classmethod get_fields_info(class_object) → Dict[str, Dict[str, Any]]#
class ros_sugar.core._action_registry.RegisteredEvent#

Bases: ros_sugar.config.base_attrs.BaseAttrs

One condition the running stack can be asked to watch for.

Contributed by a plugin, which offers it as a factory building a fresh Event. Unlike an action it is not run: it is registered, together with the actions it is to trigger.

Parameters:

  • ref – How a caller names it, as “owner/name”

  • owner – Id of the plugin that offers it

  • name – The short name within that owner

  • description – What it watches for

  • signature – What its factory takes, such as a threshold

asdict(filter: Optional[Callable] = None) → Dict#
to_dict() → Dict#
asdict_explicit() → Dict#
from_dict(dict_obj: Dict) → None#
from_file(file_path: str, nested_root_name: Union[str, None] = None, get_common: bool = False) → bool#
to_json() → Union[str, bytes, bytearray]#
from_json(json_obj: Union[str, bytes, bytearray]) → None#
has_attribute(attr_name: str) → bool#
get_attribute_type(attr_name: str) → Optional[type]#
update_value(attr_name: str, attr_value: Any) → bool#
classmethod get_fields_info(class_object) → Dict[str, Dict[str, Any]]#
class ros_sugar.core._action_registry.SystemActionRegistry(actions: Optional[Iterable[ros_sugar.core._action_registry.RegisteredAction]] = None)#

Every action the running stack exposes, addressable by name.

Parameters:

actions – Entries to start with, defaults to empty

static parse_ref(ref: str) → Tuple[str, str]#

Split “owner/name” into its two halves.

Parameters:

ref – The reference to split. A leading slash is tolerated, since a caller used to ROS topic names will write one

Raises:

ValueError – If it is not exactly one owner and one name

Return type:

Tuple[str, str]

classmethod normalize(ref: str) → str#

The canonical spelling of a reference, so lookups agree

static short_name(server_name: str, owner: str) → str#

Reduce a ROS server name to something a reference can carry.

Components name their entry points inconsistently: some prefix the node name, some do not. Both have to end up addressable, and a reference holds exactly one name, so what is left of the path is flattened.

add(action: ros_sugar.core._action_registry.RegisteredAction, interface: Any = None) → None#

Register one action.

Parameters:

  • action – What to register

  • interface – The live action or service class, for the kinds that are reached through a client rather than by method name

Raises:

ValueError – If the reference is taken by a different action. Silently keeping one of them would make which one a caller reaches a matter of ordering

interface_for(ref: str) → Any#

The live action or service class behind a reference.

Returns:

The class, or None for a kind that is reached by method name

add_event_factory(event: ros_sugar.core._action_registry.RegisteredEvent, factory: Callable) → None#

Register one condition a plugin offers.

Parameters:

  • event – What to register

  • factory – Builds a fresh Event, given whatever its signature asks for

Raises:

ValueError – If the reference is taken by a different event

event_factory_for(ref: str) → Optional[Callable]#

The factory behind an event reference, or None if it is unknown

events(owner: Optional[str] = None) → List[ros_sugar.core._action_registry.RegisteredEvent]#

Every registered event, sorted by reference, optionally one owner’s

property events_dictionary: List[Dict]#

Serialized form, for a caller asking what can be watched for

get_event(ref: str) → ros_sugar.core._action_registry.RegisteredEvent#

Look one up.

Raises:

KeyError – If unknown, naming what is registered

remove_event(ref: str) → ros_sugar.core._action_registry.RegisteredEvent#

Drop one condition and its factory.

Returns:

The entry that was removed

Raises:

KeyError – If unknown, as get_event does

get(ref: str) → ros_sugar.core._action_registry.RegisteredAction#

Look one up.

Raises:

KeyError – If unknown, naming what the owner does offer, or the known owners when the owner itself is unknown

remove(ref: str) → ros_sugar.core._action_registry.RegisteredAction#

Drop one action, and the live class or factory kept beside it.

Something built from the entry before, such as an event’s action, keeps what it was built with; only new lookups stop finding it.

Returns:

The entry that was removed

Raises:

KeyError – If unknown, as get does

owners() → List[str]#

Every owner that has at least one action, sorted

refs(owner: Optional[str] = None) → List[str]#

Every reference, sorted, optionally only one owner’s

list(owner: Optional[str] = None, kind: Optional[str] = None) → List[ros_sugar.core._action_registry.RegisteredAction]#

Every action, sorted by reference, narrowed by owner and/or kind

property dictionary: List[Dict]#

Serialized form, for a caller asking what is available

classmethod from_components(components: Iterable, monitor_methods: Optional[Iterable[str]] = None, monitor_class: Optional[type] = None, out_of_process: Optional[Iterable[str]] = None, plugins: Optional[Iterable] = None) → ros_sugar.core._action_registry.SystemActionRegistry#

Build the registry from what the Launcher knows.

Parameters:

  • components – The components in the stack

  • monitor_methods – Names of Monitor methods a runtime caller may use. An allowlist, not introspection: the Monitor holds lifecycle power over every component, and most of its methods take Python objects no JSON payload could carry

  • monitor_class – Class to read those methods’ signatures off. The registry is built before the Monitor exists, so this is the class rather than an instance

  • out_of_process – Node names launched as their own process, which is what makes an owner unreachable by holding its object

  • plugins – The plugins attached to the recipe, robot and sensor alike. What each contributes is registered under its plugin id

Return type:

SystemActionRegistry