`kompass.components.mission`#

Mission manager: run a sequence of navigation goals as one action.

A multi-waypoint mission is a sequence with per-step policy: drive somewhere, wait, drive somewhere else, and decide what a failure or a timeout means. That is what a Sugarcoat Routine is, so this component owns no sequencing of its own. It translates a MultiGoalPlanPath goal into a routine specification, registers it on the Monitor over the runtime API, and reports the routine’s cursor back as action feedback.

Module Contents#

Classes#

`MissionManagerConfig`	Mission manager configuration
`MissionManager`	Runs a sequence of navigation goals as one action.

Functions#

`dwell_seconds`	How long to dwell at one waypoint.
`goto_step`	One waypoint, as a goal on the planner’s action server.
`stop_step`	Call one component action that stops the robot before holding position.
`mission_routine_spec`	Turn a mission goal into a routine specification.
`reached_waypoints`	Which waypoints the cursor says were reached.
`ended_on_pause_timeout`	Whether the mission failed because a conditional pause expired.

API#

kompass.components.mission.dwell_seconds(pause_duration: List[float], index: int) → float#

How long to dwell at one waypoint.

One entry applies to every waypoint, so a caller with a uniform dwell does not have to repeat it.

Parameters:

pause_duration – The goal’s dwell list, empty, of length one, or one entry per waypoint
index – Which waypoint

Return type:

float

kompass.components.mission.goto_step(index: int, pose, *, planner_ref: str, algorithm_name: str, tolerance, timeout: float) → Dict[str, Any]#

One waypoint, as a goal on the planner’s action server.

Named goto_<index>, because a routine requires distinct step names and the name is what identifies the step in the cursor.

kompass.components.mission.stop_step(index: Union[int, str], ref: str, retries: int = 0) → Dict[str, Any]#

Call one component action that stops the robot before holding position.

Named after the component and where it runs, so the stops stay distinct

Parameters:

index – The waypoint the stop is at, or what else it runs for
retries – Extra attempts at stopping. A stop reports failure while the robot is still rolling, or before its location has arrived, and a step that runs out of attempts ends the mission

kompass.components.mission.mission_routine_spec(goal, *, name: str, planner_ref: str, stop_refs: List[str], waypoint_timeout: float, start_pose=None, stop_retries: int = 0) → Dict[str, Any]#

Turn a mission goal into a routine specification.

Parameters:

goal – A MultiGoalPlanPath.Goal
name – Unique routine name, which is how it is controlled and where its cursor is published
planner_ref – The planner’s action server, as component/action
stop_refs – Component actions that stop the robot, as component/action, called in order before holding position at a waypoint
waypoint_timeout – Seconds allowed for one waypoint
start_pose – Where the robot was when the mission began. Read only to refuse a return-to-start mission with nowhere to return to: the journey itself never drives back, the mission does that after it
stop_retries – Extra attempts at each stop before it ends the mission

Raises:

ValueError – If the goal describes no mission, or describes one that cannot be carried out as asked

Return type:

Dict[str, Any]

kompass.components.mission.reached_waypoints(cursor: Dict[str, Any], total: int) → List[bool]#

Which waypoints the cursor says were reached.

Read from the step names rather than counted, because a mission’s steps are not one per waypoint: dwells and conditional pauses sit between them.

kompass.components.mission.ended_on_pause_timeout(cursor: Dict[str, Any]) → bool#

Whether the mission failed because a conditional pause expired.

Read from the step the routine failed on rather than from its message. A pause step’s work always succeeds and its success condition is what holds it open, so the only way it fails is its timeout running out under an ending policy. An aborted routine is not a timeout, whichever step it was stopped at.

class kompass.components.mission.MissionManagerConfig#

Bases: kompass.config.ComponentConfig

Mission manager configuration

Parameters:

waypoint_timeout – Seconds allowed for one waypoint before the mission gives up on it
cursor_poll_rate – How often the routine’s cursor is read while a mission runs, in Hz. Only affects how promptly feedback is published
retries – Extra attempts at whatever a mission does that is worth trying again before giving up on it, and giving up on any of them ends the mission: stopping the robot before it holds position at a waypoint, which reports failure while the robot is still rolling or before its location has arrived, and reading the routine’s cursor, which fails while nothing answers for the routine
end_mission_timeout – Seconds a deactivation waits for the ongoing mission to end before taking its action server down. Ending it cancels the planner goal in flight, which the planner notices once a loop. Also what one call to the Monitor’s runtime API is given a share of, so that a Monitor that stopped answering is noticed rather than waited for and the mission still ends within this
ui_waypoints_topic – Where the UI publishes a waypoint picked on the map, for the mission’s card to collect into a journey
routine_name – Name of the routine carrying out a mission on the Monitor, the same for every mission as they run one at a time. What a UI is told to follow, with launcher.enable_ui(routines=[...])
planner_action – The planner’s main action server, as <planner component name>/<action name> (e.g. planner/navigate_to_goal). Filled in from the planner when one is given to the MissionManager
controller_name – Name of the Controller component, stopped before holding position at a waypoint. Filled in from the controller when given
drive_manager_name – Name of the DriveManager component, stopped before holding position at a waypoint. Filled in from the drive manager when given

class kompass.components.mission.MissionManager(component_name: str, planner: Optional[kompass.components.component.Component] = None, controller: Optional[kompass.components.controller.Controller] = None, drive_manager: Optional[kompass.components.drive_manager.DriveManager] = None, config: Optional[kompass.components.mission.MissionManagerConfig] = None, config_file: Optional[str] = None, inputs: Optional[Dict[kompass.components.defaults.TopicsKeys, Any]] = None, outputs: Optional[Dict[kompass.components.defaults.TopicsKeys, Any]] = None, **kwargs)#

Bases: kompass.components.component.Component

Runs a sequence of navigation goals as one action.

from kompass.components import MissionManager, MissionManagerConfig

planner = Planner(component_name="planner")
planner.run_type = "ActionServer"
controller = Controller(component_name="controller")
driver = DriveManager(component_name="drive_manager")
mission = MissionManager(
    component_name="mission",
    planner=planner,
    controller=controller,
    drive_manager=driver,
)

Sends MultiGoalPlanPath goals to mission/run_mission. Each mission is identified by the time it was received, and the ongoing one is published on mission_status. The mission is carried out by a Sugarcoat routine on the Monitor, named routine_name for every mission as they run one at a time. Its progress is also visible on routine/<routine_name>/state, it can be controlled by name through the Monitor’s runtime API, and it can be followed in the UI with launcher.enable_ui(routines=[mission.routine_name]).

custom_on_configure()#: Check the components once the config file has been applied

on_deactivate(state)#

End an ongoing mission before the action server is taken down.

Deactivating destroys the action server, and with it the goal handle: a client left waiting on a mission would never get a result, and the routine would carry on with nothing following it. So the mission ends itself first: its loop aborts the routine, which cancels the planner goal in flight, and reports the goal as aborted. The action server only sends that result once the mission callback has returned, which is what the wait is for

pause_mission(**_) → kompass.components.ros.ActionReturnType#

Pause the ongoing mission, stopping the robot until resume_mission.

On resuming, a waypoint that was being driven to is driven to again, a dwell starts over and a conditional pause waits for a new go-ahead

Return type:: ActionReturnType

resume_mission(**_) → kompass.components.ros.ActionReturnType#

Resume a paused mission where it was paused

Return type:: ActionReturnType

property ui_waypoints: ros_sugar.io.Topic#

The topic a waypoint picked on the UI’s map is published on.

Declared to the UI both ways: the map publishes a click on it, and the mission’s card reads it back to collect the journey

Return type:: Topic

property ui_inputs: List[Any]#

What the mission’s card needs, for launcher.enable_ui(inputs=...).

The action a mission is sent to, the service the card pauses and resumes through, and the topic it collects waypoints from

Return type:: List[Any]

property ui_outputs: List[ros_sugar.io.Topic]#

What the mission’s card follows, for launcher.enable_ui(outputs=...).

Its status, and the waypoints picked on the map

Return type:: List[Topic]

property routine_name: str#

Name of the routine carrying out a mission, the same for every one.

Missions run one at a time, so one name serves them all, and it is known before any mission starts, which is what lets a UI follow it

property stop_refs: List[str]#: Component actions that stop the robot, controller first so no new commands reach the drive manager

start_pose() → Optional[geometry_msgs.msg.Pose]#

Where the robot is now, as a geometry_msgs/Pose in the world frame.

Read when a mission starts, so the return-to-start policy has somewhere to return to. None when nothing has published a location yet, which mission_routine_spec turns into a refusal rather than a wrong pose.

current_pose() → Optional[geometry_msgs.msg.PoseStamped]#

Where the robot is now, as a geometry_msgs/PoseStamped, or None.

Stamped in the world frame, which is the frame the pose was brought into, whatever frame the location messages themselves are in

property monitor: ros_sugar.base_clients.ServiceClientHandler#

Client for the Monitor’s runtime API.

NOTE: created lazily, in a callback group of its own. The mission blocks in its action callback while calling this, so the response has to be free to arrive on another thread of the component’s executor, including while a lifecycle transition, which runs in the node’s default group, waits for the mission to end.

call_monitor(method: str, **kwargs) → Any#: Run one runtime API method, returning its response or None

cursor(routine_name: str) → Optional[Dict[str, Any]]#: Where the routine has got to, or None if it could not be read

custom_on_activate()#: Publish that there is no ongoing mission yet

main_action_callback(goal_handle)#

Carry out one mission, as a routine on the Monitor.

NOTE: The action server takes one goal at a time, a new mission is rejected until this callback returns

property robot: kompass.config.RobotConfig#

property robot_geometry_type: kompass_core.models.RobotGeometry.Type#

property robot_ctrl_limits: kompass_core.models.RobotCtrlLimits#

property run_type: kompass.config.ComponentRunType#

property inputs_keys: List[kompass.components.defaults.TopicsKeys]#

property outputs_keys: List[kompass.components.defaults.TopicsKeys]#

inputs(**kwargs)#

outputs(**kwargs)#

set_input(**kwargs) → bool#

set_output(**kwargs) → bool#

config_from_file(config_file: str)#

property odom_tf_listener: Optional[ros_sugar.tf.TFListener]#

transform_inputs_to(topic_key: kompass.components.defaults.TopicsKeys, goal_frame: str, static_tf: bool = False) → None#

input_tf_listener(topic_key: kompass.components.defaults.TopicsKeys, goal_frame: str, static_tf: bool = False, idx: int = 0) → Optional[ros_sugar.tf.TFListener]#

resolve_input_tf(topic_key: kompass.components.defaults.TopicsKeys, goal_frame: Optional[str] = None, idx: int = 0) → Tuple[bool, Optional[tf2_ros.TransformStamped]]#

wait_input_tf(topic_key: kompass.components.defaults.TopicsKeys, idx: int = 0, timeout: Optional[float] = None, static_tf: bool = True) → Optional[ros_sugar.tf.TFListener]#

wait_sensor_config(topic_key: kompass.components.defaults.TopicsKeys, idx: int = 0, timeout: Optional[float] = None) → Optional[kompass_cpp.types.SensorConfig]#

in_topic_name(key: Union[str, kompass.components.defaults.TopicsKeys]) → Union[str, List[str], None]#

out_topic_name(key: Union[str, kompass.components.defaults.TopicsKeys]) → Union[str, List[str], None]#

get_in_topic(key: Union[str, kompass.components.defaults.TopicsKeys]) → Union[kompass.components.ros.Topic, List[kompass.components.ros.Topic], None]#

get_out_topic(key: Union[str, kompass.components.defaults.TopicsKeys]) → Union[kompass.components.ros.Topic, List[kompass.components.ros.Topic], None]#

get_callback(key: Union[str, kompass.components.defaults.TopicsKeys], idx: int = 0) → Optional[kompass.callbacks.GenericCallback]#

get_publisher(key: Union[str, kompass.components.defaults.TopicsKeys], idx: int = 0) → ros_sugar.io.Publisher#

callbacks_inputs_check(inputs_to_check: Optional[List[str]] = None, inputs_to_exclude: Optional[List[str]] = None) → bool#

kompass.components.mission#

Module Contents#

Classes#

Functions#

API#

`kompass.components.mission`#