`ros_sugar.core.action`#

Actions: fire-and-forget by default, closed-loop when told what success means

Module Contents#

Classes#

`ActionOutcome`	Why a run ended.
`Action`	An executable method, optionally monitored until it verifiably worked.
`ActionServerGoal`	A step that sends a goal to a ROS action server and waits for its result.

Functions#

`current_attempt_is_live`	Whether the attempt the calling dispatch worker runs is still wanted.
`bind_monitored_actions`	Give any monitored actions among `actions` a handle to their host node.

API#

ros_sugar.core.action.current_attempt_is_live() → bool#

Whether the attempt the calling dispatch worker runs is still wanted.

For code that runs inside an action’s dispatch without holding the action, such as a Monitor method used as a routine step. A halt, a timeout or a run started over each end the attempt, and the worker should stop with it

Returns:: True outside any dispatch, where there is no attempt to ask about
Return type:: bool

class ros_sugar.core.action.ActionOutcome#

Bases: ros_sugar.config.StrEnum

Why a run ended.

The action contract carries only (bool, str), so a caller that has to tell a preemption apart from a genuine failure - a Routine choosing what to do with a step - reads the outcome alongside the result. A StrEnum, so each member compares and serializes as its plain string value.

classmethod get_enum(__value: str) → Optional[ros_sugar.config.robot.T]#

classmethod values() → List[str]#

capitalize()#

casefold()#

center()#

count()#

encode()#

endswith()#

expandtabs()#

find()#

format()#

format_map()#

index()#

isalnum()#

isalpha()#

isascii()#

isdecimal()#

isdigit()#

isidentifier()#

islower()#

isnumeric()#

isprintable()#

isspace()#

istitle()#

isupper()#

join()#

ljust()#

lower()#

lstrip()#

partition()#

removeprefix()#

removesuffix()#

replace()#

rfind()#

rindex()#

rjust()#

rpartition()#

rsplit()#

rstrip()#

split()#

splitlines()#

startswith()#

strip()#

swapcase()#

title()#

translate()#

upper()#

zfill()#

name()#

value()#

ros_sugar.core.action.bind_monitored_actions(actions: Iterable, host) → None#

Give any monitored actions among actions a handle to their host node.

The host is the node that owns the action, and is what a “monitored” action asks to start watching its success condition. Watching starts on the first dispatch, so a success topic is never subscribed by a host whose action is never triggered.

Parameters:

actions – The actions registered on the host, monitored or not. Entries may be single actions or lists of actions
host – The node monitoring these actions, exposing add_event_listener

class ros_sugar.core.action.Action(method: Callable, args: Optional[Union[Tuple, List, Any]] = None, kwargs: Optional[Dict] = None, success: Optional[Union[ros_sugar.condition.Condition, ros_sugar.io.Topic]] = None, timeout: Optional[float] = None, on_timeout: str = 'retry', max_retries: int = 0, retry_delay: float = 0.0, cancel_method: Optional[Callable] = None, on_fail: str = 'abort', fallback: Optional[Union[ros_sugar.core.base_action.BaseAction, Callable]] = None, name: Optional[str] = None, description: Optional[str] = None)#

Bases: ros_sugar.core.base_action.BaseAction

An executable method, optionally monitored until it verifiably worked.

In its plain form an action is fire and forget: it dispatches a method and nothing afterwards can tell whether the method actually worked.

Action(my_component.start)

Passing any of the monitoring parameters activates monitoring: the action dispatches, waits for a verdict, then re-dispatches while the verdict is negative and the retry budget allows.

grasp = Action(
    gripper.close,
    success=gripper_state.msg.closed.is_true(),
    timeout=3.0,
    max_retries=3,
)
launcher.on(grasp_requested, grasp)

The verdict comes from one of two sources:

A success condition (success), a Condition on live topic data such as gripper_state.msg.closed.is_true(). World state is authoritative: if the condition becomes true the action succeeded, even if the dispatched method reported otherwise.
The return value, when no success condition is given. The success half of the action’s (bool, str) result decides the verdict. A raised exception, or a return that does not follow the contract, is a failure.

There are two ways to run one. Calling it executes the action and returns its (bool, str) result - a monitored action blocks until its outcome is decided. start() runs the same logic without blocking anything, reporting the outcome to a callback instead; this is what Routine uses to sequence steps without parking a worker thread per active routine.

Parameters:

method – Method to dispatch
args – Positional arguments, may contain topic.msg.x expressions
kwargs – Keyword arguments, may contain topic.msg.x expressions
success – Condition proving the action worked. If omitted, the return value of the method is used instead. Activates monitoring
timeout – Seconds to wait for the verdict on each attempt. Activates monitoring
on_timeout – What a timeout means, one of “fail”, “succeed” or “retry”
max_retries – Number of re-dispatches, so the total number of attempts is max_retries + 1. Activates monitoring
retry_delay – Seconds to wait between attempts. Activates monitoring
cancel_method – Called by halt() to preempt a run in flight, for an action that can be told to stop, such as an arm motion. Activates monitoring
on_fail – Only read when this action is a step of a Routine: what the routine does with the step once its retries are spent, one of “abort”, “skip” or “fallback”
fallback – Only read when this action is a step of a Routine: the recovery action run when on_fail="fallback"
name – Name this action is known by, defaulting to the method name. This is what a routine’s cursor reports
description – Optional action description

property is_monitored: bool#

Whether this action watches its own outcome.

True as soon as any part of a watch or retry policy is declared: success, timeout, max_retries or cancel_method.

NOTE: not retry_delay. It only delays a retry, so with no retries to delay it can never take effect, and arming a watch loop for it would monitor an action that has nothing to monitor.

Return type:: bool

property success_event: Optional[ros_sugar.core.event.Event]#

Event monitored to detect success, None in return value mode.

NOTE: this is deliberately a separate event rather than extra topics on the triggering event. An ‘on any’ event fires only once all of its involved topics have data, so a success topic added there would hold the action back until the state it is meant to bring about had already been published at least once.

Return type:: Optional[Event]

property running: bool#

Whether a run is in flight

Return type:: bool

classmethod coerce(action: Union[ros_sugar.core.base_action.BaseAction, Callable, None], owner: str) → Optional[ros_sugar.core.action.Action]#

Normalize whatever was declared into an Action.

Anything a Routine runs, and any fallback, goes through the same dispatch machinery, so there is one implementation of ‘dispatch, wait for a verdict, retry’ rather than one per kind of thing being run.

Parameters:

action – An Action, a BaseAction or a plain callable
owner – Name used in error messages

Return type:

Optional[Action]

classmethod from_base_action(action: ros_sugar.core.base_action.BaseAction, **policy) → ros_sugar.core.action.Action#

Lift a BaseAction into a full Action with the given policy.

The method, arguments and routing flags are carried over, so an action built by internal machinery can be used wherever a full Action is expected without being redeclared.

Parameters:

action – The action to lift
policy – Monitoring policy, as accepted by the constructor

Return type:

Action

set_host(host) → None#

Set the node that will monitor this action’s success condition.

Parameters:: host – A node exposing add_event_listener

stop_watching() → None#

Ask the host to stop monitoring the success event.

For an action that will not be triggered again, such as a step of a routine being removed. Triggering it again watches the event anew.

start(on_done: Callable[[ros_sugar.utils.ActionReturnType, ros_sugar.core.action.ActionOutcome], None], **kwargs) → None#

Dispatch and watch the outcome without blocking the caller.

The whole watch and retry policy runs on dispatch workers and timer threads, and on_done is called exactly once, from whichever of those threads settles the run. Nothing is parked in the meantime, which is what lets a Routine hold many steps in flight without holding a thread per step.

Parameters:

on_done – Called once with the (success, message) result and the outcome that produced it, an ActionOutcome
kwargs – Passed to the dispatched method, typically topics

halt() → ros_sugar.utils.ActionReturnType#

Preempt a run in flight.

Stops the watch and retry loop, and invokes cancel_method if one was given so the action can also be told to stop acting.

NOTE: a dispatched call that is already executing cannot be interrupted. cancel_method is the only way to affect it, which is why an action used in a preemptible context should provide one.

Returns:: (success, message), where the message reports what was halted
Return type:: ActionReturnType

set_success(**_) → None#

Settle the current attempt as successful.

Registered as the action of success_event, so it is called by the host when the success condition holds. A condition that holds while no attempt is in flight is ignored: nothing this action did brought it about, so it cannot be credited to it.

get_feedback() → Optional[Dict]#

Progress of the attempt in flight, or None when there is none.

Read by a Routine to put the active step’s progress in its cursor.

set_feedback_sink(sink: Optional[Callable[[], None]]) → None#

Install a zero-arg callable to ping when feedback arrives.

No-op unless the action actually produces feedback.

property dictionary: Dict#

Serialized form, extending the base payload with the monitoring policy

Return type:: Dict

classmethod deserialize_action(serialized_action_dict: Dict, deserialized_method: Callable, cancel_method: Optional[Callable] = None, fallback: Optional[ros_sugar.core.action.Action] = None) → ros_sugar.core.action.Action#

Reconstruct an Action from serialized action data

Parameters:

serialized_action_dict – Serialized action data
deserialized_method – Deserialized action method
cancel_method – Resolved cancel method, overriding the serialized name. Needed when the executable is not a bound method, so there is no owner to resolve that name against
fallback – Resolved fallback action. dictionary cannot carry one

Return type:

Action

set_required_runtime_arguments(kwargs: Dict[str, Union[str, ros_sugar.io.supported_types.SupportedType, Type]])#

get_required_topics() → List[ros_sugar.io.Topic]#

replace_input_topic(old_topic: ros_sugar.io.Topic, new_topic: ros_sugar.io.Topic)#

property executable#

property parent_component#

property action_name: str#

property monitor_method: str#

property component_action: bool#

property description: Optional[str]#

property json: str#

launch_action(monitor_node=None) → Union[ros_sugar.core.base_action.OpaqueCoroutine, ros_sugar.core.base_action.OpaqueFunction]#

class ros_sugar.core.action.ActionServerGoal(*, component: Optional[str] = None, server_name: Optional[str] = None, server_type: Optional[type] = None, goal: Optional[Any] = None, success: Optional[Union[ros_sugar.condition.Condition, ros_sugar.io.Topic]] = None, success_grace: float = 1.0, timeout: Optional[float] = None, on_timeout: str = 'fail', max_retries: int = 0, retry_delay: float = 0.0, on_fail: str = 'abort', fallback: Optional[Union[ros_sugar.core.base_action.BaseAction, Callable]] = None, name: Optional[str] = None, description: Optional[str] = None)#

Bases: ros_sugar.core.action.Action

A step that sends a goal to a ROS action server and waits for its result.

Unlike a method step, the dispatch outlives the call: the goal runs on the server and its result is the verdict. So this kind reads the server’s own outcome instead of needing a success topic, cancels natively on halt(), and reports the server’s feedback into a routine’s cursor.

Routine("patrol", steps=[
    ActionServerGoal(component="planner", goal=goal_msg, timeout=120.0),
    ActionServerGoal(component="planner", goal=other_goal, on_fail="skip"),
])

The client is resolved from the host at dispatch, not held here: steps are built in a recipe or from JSON, both before any node exists.

Parameters:

component – Node name of a component whose main action server to drive
server_name – Action server name, when not naming a component
server_type – Action type, required with server_name
goal – A ready Goal message, a dict of goal fields, or None to take one from the call arguments
success – Condition proving the goal did what was wanted. When given it decides the verdict: met at any point during execution, or within success_grace of the goal returning, means success. The server’s own outcome then only bounds the window
success_grace – Seconds to keep checking success after the goal returns, for a condition topic that lags the server

property target: str#: Readable name of the server this step drives

get_required_topics() → List[ros_sugar.io.Topic]#

Topics the step reads, including the success condition’s.

A routine reports these so its host subscribes them before the step runs; without that the condition would never see any data.

set_feedback_sink(sink: Optional[Callable[[], None]]) → None#: Install the callable pinged when the server sends feedback

get_feedback() → Optional[Dict]#: Progress of the goal in flight, for a routine’s cursor

property dictionary: Dict#: Serialized form, adding what identifies the server and the goal

property is_monitored: bool#

property success_event: Optional[ros_sugar.core.event.Event]#

property running: bool#

classmethod coerce(action: Union[ros_sugar.core.base_action.BaseAction, Callable, None], owner: str) → Optional[ros_sugar.core.action.Action]#

classmethod from_base_action(action: ros_sugar.core.base_action.BaseAction, **policy) → ros_sugar.core.action.Action#

set_host(host) → None#

stop_watching() → None#

start(on_done: Callable[[ros_sugar.utils.ActionReturnType, ros_sugar.core.action.ActionOutcome], None], **kwargs) → None#

halt() → ros_sugar.utils.ActionReturnType#

set_success(**_) → None#

classmethod deserialize_action(serialized_action_dict: Dict, deserialized_method: Callable, cancel_method: Optional[Callable] = None, fallback: Optional[ros_sugar.core.action.Action] = None) → ros_sugar.core.action.Action#

set_required_runtime_arguments(kwargs: Dict[str, Union[str, ros_sugar.io.supported_types.SupportedType, Type]])#

replace_input_topic(old_topic: ros_sugar.io.Topic, new_topic: ros_sugar.io.Topic)#

property executable#

property parent_component#

property action_name: str#

property monitor_method: str#

property component_action: bool#

property description: Optional[str]#

property json: str#

launch_action(monitor_node=None) → Union[ros_sugar.core.base_action.OpaqueCoroutine, ros_sugar.core.base_action.OpaqueFunction]#

ros_sugar.core.action#

Module Contents#

Classes#

Functions#

API#

`ros_sugar.core.action`#