`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]#

Get Enum member equal to the given value

Parameters:: __value (str) – Reference value
Returns:: Enum member
Return type:: StrEnum | None

classmethod values() → List[str]#: Get all enum values

capitalize()#

Return a capitalized version of the string.

More specifically, make the first character have upper case and the rest lower case.

casefold()#: Return a version of the string suitable for caseless comparisons.

center()#

Return a centered string of length width.

Padding is done using the specified fill character (default is a space).

count()#

S.count(sub[, start[, end]]) -> int

Return the number of non-overlapping occurrences of substring sub in string S[start:end]. Optional arguments start and end are interpreted as in slice notation.

encode()#

Encode the string using the codec registered for encoding.

encoding The encoding in which to encode the string. errors The error handling scheme to use for encoding errors. The default is ‘strict’ meaning that encoding errors raise a UnicodeEncodeError. Other possible values are ‘ignore’, ‘replace’ and ‘xmlcharrefreplace’ as well as any other name registered with codecs.register_error that can handle UnicodeEncodeErrors.

endswith()#

S.endswith(suffix[, start[, end]]) -> bool

Return True if S ends with the specified suffix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. suffix can also be a tuple of strings to try.

expandtabs()#

Return a copy where all tab characters are expanded using spaces.

If tabsize is not given, a tab size of 8 characters is assumed.

find()#

S.find(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

format()#

S.format(*args, **kwargs) -> str

Return a formatted version of S, using substitutions from args and kwargs. The substitutions are identified by braces (‘{’ and ‘}’).

format_map()#

S.format_map(mapping) -> str

Return a formatted version of S, using substitutions from mapping. The substitutions are identified by braces (‘{’ and ‘}’).

index()#

S.index(sub[, start[, end]]) -> int

Return the lowest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

isalnum()#

Return True if the string is an alpha-numeric string, False otherwise.

A string is alpha-numeric if all characters in the string are alpha-numeric and there is at least one character in the string.

isalpha()#

Return True if the string is an alphabetic string, False otherwise.

A string is alphabetic if all characters in the string are alphabetic and there is at least one character in the string.

isascii()#

Return True if all characters in the string are ASCII, False otherwise.

ASCII characters have code points in the range U+0000-U+007F. Empty string is ASCII too.

isdecimal()#

Return True if the string is a decimal string, False otherwise.

A string is a decimal string if all characters in the string are decimal and there is at least one character in the string.

isdigit()#

Return True if the string is a digit string, False otherwise.

A string is a digit string if all characters in the string are digits and there is at least one character in the string.

isidentifier()#

Return True if the string is a valid Python identifier, False otherwise.

Call keyword.iskeyword(s) to test whether string s is a reserved identifier, such as “def” or “class”.

islower()#

Return True if the string is a lowercase string, False otherwise.

A string is lowercase if all cased characters in the string are lowercase and there is at least one cased character in the string.

isnumeric()#

Return True if the string is a numeric string, False otherwise.

A string is numeric if all characters in the string are numeric and there is at least one character in the string.

isprintable()#

Return True if all characters in the string are printable, False otherwise.

A character is printable if repr() may use it in its output.

isspace()#

Return True if the string is a whitespace string, False otherwise.

A string is whitespace if all characters in the string are whitespace and there is at least one character in the string.

istitle()#

Return True if the string is a title-cased string, False otherwise.

In a title-cased string, upper- and title-case characters may only follow uncased characters and lowercase characters only cased ones.

isupper()#

Return True if the string is an uppercase string, False otherwise.

A string is uppercase if all cased characters in the string are uppercase and there is at least one cased character in the string.

join()#

Concatenate any number of strings.

The string whose method is called is inserted in between each given string. The result is returned as a new string.

Example: ‘.’.join([‘ab’, ‘pq’, ‘rs’]) -> ‘ab.pq.rs’

ljust()#

Return a left-justified string of length width.

Padding is done using the specified fill character (default is a space).

lower()#: Return a copy of the string converted to lowercase.

lstrip()#

Return a copy of the string with leading whitespace removed.

If chars is given and not None, remove characters in chars instead.

partition()#

Partition the string into three parts using the given separator.

This will search for the separator in the string. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing the original string and two empty strings.

removeprefix()#

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

removesuffix()#

Return a str with the given suffix string removed if present.

If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.

replace()#

Return a copy with all occurrences of substring old replaced by new.

count Maximum number of occurrences to replace. -1 (the default value) means replace all occurrences.

If the optional argument count is given, only the first count occurrences are replaced.

rfind()#

S.rfind(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Return -1 on failure.

rindex()#

S.rindex(sub[, start[, end]]) -> int

Return the highest index in S where substring sub is found, such that sub is contained within S[start:end]. Optional arguments start and end are interpreted as in slice notation.

Raises ValueError when the substring is not found.

rjust()#

Return a right-justified string of length width.

Padding is done using the specified fill character (default is a space).

rpartition()#

Partition the string into three parts using the given separator.

This will search for the separator in the string, starting at the end. If the separator is found, returns a 3-tuple containing the part before the separator, the separator itself, and the part after it.

If the separator is not found, returns a 3-tuple containing two empty strings and the original string.

rsplit()#

Return a list of the substrings in the string, using sep as the separator string.

sep The separator used to split the string.

When set to None (the default value), will split on any whitespace
character (including \n \r \t \f and spaces) and will discard
empty strings from the result.

maxsplit Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the end of the string and works to the front.

rstrip()#

Return a copy of the string with trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

split()#

Return a list of the substrings in the string, using sep as the separator string.

sep The separator used to split the string.

When set to None (the default value), will split on any whitespace
character (including \n \r \t \f and spaces) and will discard
empty strings from the result.

maxsplit Maximum number of splits. -1 (the default value) means no limit.

Splitting starts at the front of the string and works to the end.

Note, str.split() is mainly useful for data that has been intentionally delimited. With natural text that includes punctuation, consider using the regular expression module.

splitlines()#

Return a list of the lines in the string, breaking at line boundaries.

Line breaks are not included in the resulting list unless keepends is given and true.

startswith()#

S.startswith(prefix[, start[, end]]) -> bool

Return True if S starts with the specified prefix, False otherwise. With optional start, test S beginning at that position. With optional end, stop comparing S at that position. prefix can also be a tuple of strings to try.

strip()#

Return a copy of the string with leading and trailing whitespace removed.

If chars is given and not None, remove characters in chars instead.

swapcase()#: Convert uppercase characters to lowercase and lowercase characters to uppercase.

title()#

Return a version of the string where each word is titlecased.

More specifically, words start with uppercased characters and all remaining cased characters have lower case.

translate()#

Replace each character in the string using the given translation table.

table Translation table, which must be a mapping of Unicode ordinals to Unicode ordinals, strings, or None.

The table must implement lookup/indexing via getitem, for instance a dictionary or list. If this operation raises LookupError, the character is left untouched. Characters mapped to None are deleted.

upper()#: Return a copy of the string converted to uppercase.

zfill()#

Pad a numeric string with zeros on the left, to fill a field of the given width.

The string is never truncated.

name()#: The name of the Enum member.

value()#: The value of the Enum member.

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]])#: Used to set the missing (dynamic) argument type in derived pre-built actions

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

Get all the required input topic to execute this action

Returns:: Required topics
Return type:: List[Topic]

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

Replaces an input topic with a new topic if the old topic exists in the required input topics

Parameters:

old_topic (Topic) – Old topic
new_topic (Topic) – New topic

property executable#

Get the action callable

Returns:: description
Return type:: type

property parent_component#

Getter of parent component class name if it is a component action, else None

Returns:: description
Return type:: str | None

property action_name: str#

Getter of the action name Equals exact executable name if it is not a component action Equals method name in the component if it is a component action

Returns:: description
Return type:: str

property monitor_method: str#

Name of the Monitor method this system level action runs.

The action’s own name, unless it was named something else: two steps of one routine running the same Monitor method need names that tell them apart

Return type:: str

property component_action: bool#

component_action.

Return type:: bool

property description: Optional[str]#

Get the description of the action, used for openai function calling

Returns:: description
Return type:: Optional[str]

property json: str#

Property to get/set the event using a json

Returns:: Event description dictionary as json
Return type:: str

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

Get the ros launch action

Returns:: description
Return type:: OpaqueCoroutine | 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#

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.

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.

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.

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.

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]])#: Used to set the missing (dynamic) argument type in derived pre-built actions

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

Replaces an input topic with a new topic if the old topic exists in the required input topics

Parameters:

old_topic (Topic) – Old topic
new_topic (Topic) – New topic

property executable#

Get the action callable

Returns:: description
Return type:: type

property parent_component#

Getter of parent component class name if it is a component action, else None

Returns:: description
Return type:: str | None

property action_name: str#

Getter of the action name Equals exact executable name if it is not a component action Equals method name in the component if it is a component action

Returns:: description
Return type:: str

property monitor_method: str#

Name of the Monitor method this system level action runs.

The action’s own name, unless it was named something else: two steps of one routine running the same Monitor method need names that tell them apart

Return type:: str

property component_action: bool#

component_action.

Return type:: bool

property description: Optional[str]#

Get the description of the action, used for openai function calling

Returns:: description
Return type:: Optional[str]

property json: str#

Property to get/set the event using a json

Returns:: Event description dictionary as json
Return type:: str

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

Get the ros launch action

Returns:: description
Return type:: OpaqueCoroutine | OpaqueFunction

ros_sugar.core.action#

Module Contents#

Classes#

Functions#

API#

`ros_sugar.core.action`#