`ros_sugar.core.routine`#

Routine: an ordered sequence of monitored steps with a published cursor

Module Contents#

Classes#

`RoutineStatus`	Routine lifecycle status.
`Routine`	An ordered sequence of Action steps, run to completion or to a failure.

API#

class ros_sugar.core.routine.RoutineStatus#

Bases: ros_sugar.config.StrEnum

Routine lifecycle status.

Published on the cursor topic and readable through the host’s get_routine_state.

is_terminal() → bool#

Whether the routine reach an “end” lifecycle status and cannot transition further

Return type:: bool

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()#

An ordered sequence of Action steps, run to completion or to a failure.

A Routine is a procedure that can be declared in a Recipe implicitly or added at runtime through the ‘Monitor’: ‘detect, then pre-grasp, then close, then lift’. A routine gives each step its own success test and retry policy, and publishes where it has got to.

A step is an ordinary Action. The success, timeout and max_retries attributes decide whether a step worked; on_fail and fallback, which only a routine reads, decide what the sequence does when it did not.

pick = Routine(
    "pick_object",
    steps=[
        Action(perception.detect_object,
               success=perception_out.msg.object_found.is_true(),
               timeout=5.0),
        Action(arm.move_to_pregrasp,
               success=arm_state.msg.at_pregrasp.is_true(),
               timeout=10.0, cancel_method=arm.stop),
        Action(gripper.close, name="grasp",
               success=gripper_state.msg.closed.is_true(),
               timeout=3.0, max_retries=2,
               on_fail="fallback", fallback=gripper.reopen),
        Action(arm.lift, success=arm_state.msg.at_lift.is_true()),
    ],
    on_abort=safety.open_gripper_and_home,
)
launcher.on(pick_requested, pick)

Triggering: launcher.on(pick_requested, pick), as with any action.

A routine reports that it started, not that it succeeded. Steps are driven by callbacks rather than by a parked thread, so triggering one returns as soon as the first step is dispatched. The outcome arrives later, through on_complete / on_abort and the published cursor.

Pausing preempts the step in flight, and on_pause makes that safe. Preempting a step only stops what the step itself runs, not what it set in motion: on_pause is where a routine undoes that, such as stopping a robot a navigation step had driving. Resuming re-enters the step, once the on_pause actions are done.

Parameters:

name – Routine name, used in the cursor and its topic
steps – Ordered steps. Each is an Action or a plain callable, and they must end up with unique names
on_complete – Action run when the last step succeeds
on_abort – Action run when the routine fails or is aborted
on_pause – Action, or actions run in order, when the routine is paused. One that fails is logged and the next one still runs
description – What the routine is for, in plain words, for whoever lists the routines available and has to choose one: an operator, or an LLM planning with them

classmethod from_spec(spec: Dict, resolve: Callable[[Dict], ros_sugar.core.action.Action]) → ros_sugar.core.routine.Routine#

Build a routine from a plain dictionary, as sent over a service.

Only the shape is checked here. Turning a step into something callable needs a registry and clients, so that is the host’s job (Monitor) and arrives as resolve.

Routine.from_spec(
    {"name": "mission", "steps": [{"ref": "planner/main_action"}]},
    monitor._action_from_spec,
)

Parameters:

spec – {name, steps, on_complete, on_abort, on_pause, description}, where each step is whatever resolve understands, and on_pause is one of them or a list
resolve – Turns one step dictionary into an Action

Raises:

ValueError – If the spec names no routine or carries no steps

Return type:

Routine

set_host(host) → None#

Parameters:: host – A node exposing add_runtime_event_listener

set_state_publisher(publish: Optional[Callable[[str], None]]) → None#

Install the callable used to publish the cursor.

Injected by the host rather than created here, so the driver itself has no ROS dependency and can be tested without a node.

Parameters:: publish – Called with the cursor as a JSON string

actions() → List[ros_sugar.core.action.Action]#

Every action this routine can run: its steps, their fallbacks and its terminal actions.

What a host has to reach to register, verify or tear down a routine: anything in here can be dispatched, so all of it has to be accounted for, not only the steps.

Return type:: List[Action]

stop_watching() → None#

Stop monitoring the success conditions of every action of the routine.

Called when the routine is removed, so the conditions of a routine that is gone are not evaluated on every message of their topics.

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

Topics needed by the routine, which is every topic its steps need.

Reported up so that the host subscribes to all of them, and the steps can resolve their topic arguments whenever they are reached.

Return type:: List[Topic]

property state: Dict#

Current lifecycle state of the routine

Return type:: Dict

property latest_step_messages: Dict[str, str]#

What each step that finished in this run returned, by step name.

A step its fallback stood in for holds the fallback’s message, where step_messages has both. A step that was preempted, by a pause or an abort, has not finished and is not here. Cleared when the routine starts.

Return type:: Dict[str, str]

step_messages(index: Optional[int] = None) → Union[List[Dict], Dict]#

Every step that finished in this run, in the order it finished.

Each entry is {"step", "succeeded", "message", "fallback"}, where fallback marks the result of the step’s fallback rather than of the step itself. Cleared when the routine starts.

Parameters:: index – Only this entry, counting from the end when negative
Raises:: IndexError – If there is no entry at index
Return type:: Union[List[Dict], Dict]

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

Stop at the current step without ending the routine.

The step in flight is preempted, then the on_pause actions run, and resume() runs the step again from the start: a step is the smallest thing a routine can be positioned at.

Return type:: ActionReturnType

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

Re-enter the step the routine was paused at.

While the on_pause actions are still running the resume waits for them, rather than starting the step again alongside what is undoing it

Return type:: ActionReturnType

abort(reason: str = 'aborted by request', run_on_abort: bool = True, **_) → ros_sugar.utils.ActionReturnType#

End the routine now, preempting whatever it is running and running on_abort

Parameters:

reason – Recorded in the cursor as abort_reason, and logged
run_on_abort – Dispatch the on_abort action. A host tearing the routine down passes False: an action dispatched into a node that is going away cannot report back, and would outlive the node it calls

Return type:

ActionReturnType

halt_in_flight() → None#

Preempt whatever the routine has running right now.

That is the step, the fallback recovering it, an on_pause action or a terminal action: every one of them is dispatched the same way and can outlive the routine, so every one of them has to be stoppable. Used by pause and abort, and by a host taking the routine down.

ros_sugar.core.routine#

Module Contents#

Classes#

API#

`ros_sugar.core.routine`#