ros_sugar.core.monitor#

Monitor

Module Contents#

Classes#

Monitor

Monitor is a ROS2 Node (not Lifecycle) responsible of monitoring the status of the stack (rest of the running nodes) and managing requests/responses from the Orchestrator.

Functions#

unroutable_arguments

Why an action’s arguments cannot reach its component, or None.

API#

ros_sugar.core.monitor.unroutable_arguments(action) → Optional[str]#

Why an action’s arguments cannot reach its component, or None.

An action of a component is run by that component, so its arguments are sent to it. ROS messages, arrays and bytes all travel, so what is left is a value written into the recipe that has no representation to send at all: an open socket, a file handle, an instance of a class only this process knows.

Only the values the recipe wrote are checked. An argument read from a topic is some field of a ROS message, and every one of those can be sent.

Worth catching before launch rather than at dispatch, because the step that carries it may be a fallback or an on_abort: one that runs only when something has already gone wrong.

Parameters:

action – The action to check, a component action

Return type:

Optional[str]

class ros_sugar.core.monitor.Monitor(components_names: List[str], events_actions: Optional[Dict[ros_sugar.core.event.Event, List[ros_sugar.core.action.Action]]] = None, events_to_emit: Optional[List[ros_sugar.core.event.Event]] = None, config: Optional[ros_sugar.config.BaseConfig] = None, services_components: Optional[List[ros_sugar.core.component.BaseComponent]] = None, action_servers_components: Optional[List[ros_sugar.core.component.BaseComponent]] = None, activate_on_start: Optional[List[str]] = None, activation_timeout: Optional[float] = None, activation_attempt_time: float = 1.0, component_name: Optional[str] = None, action_registry: Optional[ros_sugar.core._action_registry.SystemActionRegistry] = None, **_)#

Bases: rclpy.node.Node

Monitor is a ROS2 Node (not Lifecycle) responsible of monitoring the status of the stack (rest of the running nodes) and managing requests/responses from the Orchestrator.

Note

When launching the stack using the Launcher, the user is not required to configure the Monitor. The Launcher will configure and launch its own Monitor internally.

Main Functionalities:

  • Creates Subscribers to registered Events. The Monitor is configured to declare an InternalEvent back to the Launcher so the corresponding Action can be executed (see source implementation in launch_actions.py)

  • Creates Subscribers to all registered Components health status topics

  • Creates clients for all components main services and main action servers

  • Creates service clients to components reconfiguration services to handle actions sent from the Launcher

add_internal_event_action_pair(event_id: str, action: ros_sugar.core.action.Action) → None#

Adds an internal event action pair to the monitor configuration.

Parameters:

  • event_id (str) – ID of the event to be monitored

  • action (Action) – Action to be executed on event trigger

set_static_transforms(transforms) → None#

Static transforms to broadcast once this node is up.

The Monitor is constructed while the launch description is still being built, before rclpy_init_node, so it has no clock to stamp with and no publisher to send on yet. They are held here and sent from activate. /tf_static is latched, so subscribers that come up afterwards still receive them.

Parameters:

transforms – TransformStamped list, stamped on broadcast

rclpy_init_node(*args, **kwargs)#

To init the node with rclpy and activate default services

activate()#

Activate all subscribers/publishers/etc…

watch_and_activate_component(component_name: str) → None#

Poll for a respawned component and drive it back to the active state via direct lifecycle service calls once it is reachable.

Called by the Launcher after a process-level respawn. Bypasses the LifecycleTransition launch action to avoid duplicate ChangeState dispatch from stale LifecycleEventManager instances left behind by the crashed process.

configure_component(component: ros_sugar.core.component.BaseComponent, new_config: Union[object, str], keep_alive: bool) → ros_sugar.utils.ActionReturnType#

Configure a given component from config instance or config file Creates and send the request to the component service

Parameters:

  • component (BaseComponent) – Component to configure

  • config (object | str) – Config instance or path to config file

  • keep_alive (bool) – To keep the component running while configuring

  • executor (ROS Executor, optional) – Used to spin the monitor node until the service response is received, defaults to None

update_parameter(component: Union[ros_sugar.core.component.BaseComponent, str], param_name: str, new_value: Any, keep_alive: bool = True) → ros_sugar.utils.ActionReturnType#

Sends a ChangeParameter service request to given component

Parameters:

  • component (Union[BaseComponent, str]) – description

  • param_name (str) – description

  • new_value (Any) – description

  • keep_alive (bool, optional) – description, defaults to True

update_parameters(component: Union[ros_sugar.core.component.BaseComponent, str], params_names: List[str], new_values: List, keep_alive: bool = True, **_) → ros_sugar.utils.ActionReturnType#

Sends a ChangeParameters service request to given component

Parameters:

  • component (BaseComponent) – description

  • params_names (List[str]) – description

  • new_values (List) – description

  • keep_alive (bool, optional) – description, defaults to True

get_action_client(action_name: str, action_type: type) → ros_sugar.base_clients.ActionClientHandler#

Helper method to get a ros action client handler for the provided service name/type

Parameters:

  • action_name (str) – Action name

  • action_type (type) – Action type (ROS2 action)

Returns:

Action client handler

Return type:

base_clients.ActionClientHandler

get_component_action_client(component_name: str) → ros_sugar.base_clients.ActionClientHandler#

Client for a component’s main action server

Parameters:

component_name – Node name of the component

Raises:

KeyError – If that component has no main action server, naming the ones that do

Return type:

base_clients.ActionClientHandler

send_srv_request(srv_request_msg: Any = None, srv_name: Optional[str] = None, srv_type: Optional[type] = None, **_) → ros_sugar.utils.ActionReturnType#

Action to send a ROS2 service request during runtime

Parameters:

  • srv_name (str) – Service name

  • srv_type (type) – Service type (ROS2 service)

  • srv_request_msg (Any) – Service request message

Return type:

ActionReturnType

send_action_goal(action_request_msg: Any = None, action_name: Optional[str] = None, action_type: Optional[type] = None, **_) → ros_sugar.utils.ActionReturnType#

Action to send a ROS2 action goal during runtime

Parameters:

  • action_name (str) – ROS2 action name

  • action_type (type) – ROS2 action type

  • action_request_msg (Any) – ROS2 action goal message

Return type:

ActionReturnType

send_component_action_goal(component_name: str, action_request_msg: Any = None, **_) → Tuple[bool, str]#

Action to send a ROS2 action goal during runtime

Parameters:

  • action_name (str) – ROS2 action name

  • action_type (type) – ROS2 action type

  • action_request_msg (Any) – ROS2 action goal message

publish_message(topic: ros_sugar.io.topic.Topic, msg: Any, publish_rate: Optional[float] = None, publish_period: Optional[float] = None, **_) → ros_sugar.utils.ActionReturnType#

Action to publish a message to a given topic

Parameters:

  • topic (Topic) – Published topic

  • msg (Any) – Published message

  • publish_rate (Optional[float], optional) – Publishing rate, if None the message is published once, defaults to None

  • publish_period (Optional[float], optional) – Publishing period, if none and rate is given the message is published forever, defaults to None

Return type:

ActionReturnType

register_external_topic(topic: ros_sugar.io.topic.Topic) → None#

Register a topic fed by a robot plugin feedback bus rather than a ROS subscription.

Events referencing this topic are still tracked and evaluated normally;

Meth:

_activate_event_monitoring simply skips creating a ROS subscription for it. Must be called before the Monitor is activated.

Parameters:

topic – The synthetic feedback topic (Feedback.as_topic()).

feed_external_topic(topic_name: str, msg: Any) → None#

Inject a decoded non-ROS message into the event blackboard.

Called by a robot plugin HOST when it decodes a telemetry packet, so that Events and Conditions over plugin feedback topics are evaluated exactly as if a ROS message had arrived. Event topics declared with use_plugin and served by this feedback receive it under their own name too.

Parameters:

  • topic_name – Synthetic topic name (the feedback’s bus channel).

  • msg – The decoded ROS message instance.

set_plugins(plugins: Dict[str, Any]) → None#

Hand the Monitor the recipe’s plugins, by id, so events on topics declared with use_plugin are fed from what the plugin serves. Must be called before the Monitor is activated.

NOTE: An attribute rather than a constructor argument, as downstream packages override Launcher._init_monitor_node to build their own

Parameters:

plugins – The attached plugins, by id

host_routines(routines: List[ros_sugar.core.routine.Routine]) → None#

Host routines that no event triggers, once the Monitor is active.

For routines that are only ever started by name: from the UI, from the runtime API or from another routine. A routine an event also triggers is hosted once. Must be called before activation

Parameters:

routines (List[Routine]) – The routines to host

get_topics_snapshot() → Dict[str, Any]#

The latest message seen on every topic the Monitor is watching.

Read by a routine when it enters a step, so a step acts on what is true when it runs rather than on what was true when the routine was triggered, which may be minutes earlier.

Return type:

Dict[str, Any]

wait(duration: float, **_) → ros_sugar.utils.ActionReturnType#

Do nothing for a while, so a routine can dwell between steps.

A step that waits has no other spelling. Every other action settles as soon as it is called, so “wait here for thirty seconds” cannot be expressed as a timeout on one of them: the timeout would never be reached. Waiting on a condition is different and needs no help, since a step’s success condition already holds it open.

NOTE: this holds one of the dispatch pool’s workers while it waits. It is the seconds-to-minutes dwell of a mission, not a scheduler. A wait stopped with its step, by an abort, a pause or its routine being removed, returns within one slice and gives its worker back; waiting out the time used to keep that worker from every other action.

Parameters:

duration – Seconds to wait

Return type:

ActionReturnType

add_event(event: ros_sugar.core.event.Event, actions: Union[ros_sugar.core.action.Action, ros_sugar.core.routine.Routine, List], event_id: Optional[str] = None, **_) → ros_sugar.utils.ActionReturnType#

Start watching an event the recipe did not declare.

Everything add_runtime_event_listener skips, because that one exists only to let an action watch its own success condition: the actions are bound to this node, the topics their arguments read are subscribed alongside the condition’s, and an action based event gets its timer.

Parameters:

  • event – What to watch for

  • actions – What to do when it fires, one or several

  • event_id – Name to remove it by later, defaults to the event’s id

Return type:

ActionReturnType

remove_event(event_id: str, **_) → ros_sugar.utils.ActionReturnType#

Stop watching an event that was added at runtime.

Its subscriptions stay: a topic is subscribed once and shared by every event reading it, so dropping one here would blind the others.

Parameters:

event_id – The name it was registered under

Return type:

ActionReturnType

add_routine(routine: ros_sugar.core.routine.Routine, replace: bool = False, **_) → ros_sugar.utils.ActionReturnType#

Take ownership of a routine that the recipe did not declare.

Parameters:

  • routine – The routine to host

  • replace – Replace one already registered under this name

Return type:

ActionReturnType

remove_routine(routine_name: str, force: bool = False, **_) → ros_sugar.utils.ActionReturnType#

Drop a routine and take its cursor topic down.

Parameters:

  • routine_name – Name it was registered under

  • force – Remove it even if it is running, aborting it first. Without this a running routine is kept, because removing one mid-step would leave whatever it started running with nothing watching it. The on_abort action is not run: the routine, its cursor and its success watches are going away, so an action dispatched now would report into nothing

Return type:

ActionReturnType

set_action_registry(registry: ros_sugar.core._action_registry.SystemActionRegistry) → None#

Install what the stack can be asked to do by name.

For the Launcher, when the Monitor that an override of its _init_monitor_node installed was built without the registry. Must happen before activation, which is when anything is first resolved against it

Parameters:

registry (SystemActionRegistry) – The stack’s registry

list_actions(**_) → ros_sugar.utils.ActionReturnType#

Every action addressable by name, as JSON

get_routines() → List[Dict]#

Every registered routine: what it is for, and where it has got to.

Each entry is the routine’s state (its name, steps, status and so on) plus its description, which is None if it was given none. Plain data, copied, so a caller can read it without reaching into the routines

Return type:

List[Dict]

list_routines(**_) → ros_sugar.utils.ActionReturnType#

Every registered routine, as get_routines gives it, as JSON

list_plugin_events(**_) → ros_sugar.utils.ActionReturnType#

Every condition a plugin offers, as JSON.

What an event spec’s ‘ref’ may name, as opposed to list_events, which is what is being watched for right now

list_events(**_) → ros_sugar.utils.ActionReturnType#

Every event registered at runtime, as JSON

start_routine(routine_name: str, **_) → ros_sugar.utils.ActionReturnType#

Start a routine by name

Parameters:

routine_name – Name the routine was declared with

Return type:

ActionReturnType

pause_routine(routine_name: str, **_) → ros_sugar.utils.ActionReturnType#

Pause a running routine, preempting the step in flight

Parameters:

routine_name – Name the routine was declared with

Return type:

ActionReturnType

resume_routine(routine_name: str, **_) → ros_sugar.utils.ActionReturnType#

Resume a paused routine, re-entering the step it stopped at

Parameters:

routine_name – Name the routine was declared with

Return type:

ActionReturnType

abort_routine(routine_name: str, reason: str = 'aborted by request', **_) → ros_sugar.utils.ActionReturnType#

End a routine now, preempting the step in flight and running its on_abort

Parameters:

  • routine_name – Name the routine was declared with

  • reason – Recorded in the cursor and logged

Return type:

ActionReturnType

get_routine_state(routine_name: str, **_) → ros_sugar.utils.ActionReturnType#

Where a routine has got to, as JSON

Parameters:

routine_name – Name the routine was declared with

Return type:

ActionReturnType

destroy_node()#

Abort the routines still running before the node goes away.

A step waiting on a goal would otherwise keep its dispatch worker, which the interpreter waits for at exit, until the step times out, and only then cancel the goal through a client that no longer exists.

add_runtime_event_listener(event: ros_sugar.core.event.Event) → None#

Start monitoring an event that was not known at activation.

Unlike the events wired up in _activate_event_monitoring, this is called while the Monitor is already running, from a worker thread. Used by a monitored Action to begin watching its success condition on first dispatch. A topic subscribed here is released by remove_runtime_event_listener once nothing reads it anymore.

Parameters:

event (Event) – Event to start monitoring

remove_runtime_event_listener(event: ros_sugar.core.event.Event) → None#

Stop monitoring an event added with add_runtime_event_listener.

A topic that was subscribed for runtime events is released along with the last event reading it, unless a registered routine still needs it. Topics subscribed at activation are kept.

Parameters:

event (Event) – Event to stop monitoring