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

Getter of robot config

Every Kompass component reasons about the robot’s body – its size for collision checking, its velocity envelope for control – so a missing robot config is an error rather than something to default around.

Returns:: Robot configuration
Return type:: RobotConfig
Raises:: ValueError – If no robot configuration has been set

property robot_geometry_type: kompass_core.models.RobotGeometry.Type#

Getter of the robot geometry as the type the algorithms take

The description carries the geometry as a plain name, since Sugarcoat does no geometry math. The algorithms take the type the C++ library defines, so convert on the way in.

Returns:: Robot geometry type
Return type:: RobotGeometry.Type

property robot_ctrl_limits: kompass_core.models.RobotCtrlLimits#

Getter of the robot velocity envelope as the type the algorithms take

Counterpart of robot_geometry_type for the control limits: the description declares them, the C++ library enforces them, and this is the one place the two are matched up.

Returns:: Robot control limits
Return type:: RobotCtrlLimits

property run_type: kompass.config.ComponentRunType#

Component run type: Timed, ActionServer or Event

Returns:: Timed, ActionServer or Server
Return type:: str

property inputs_keys: List[kompass.components.defaults.TopicsKeys]#: Getter of component inputs key names that should be used in the inputs dictionary

property outputs_keys: List[kompass.components.defaults.TopicsKeys]#: Getter of component outputs key names that should be used in the inputs dictionary

inputs(**kwargs)#: Set component input streams (topics) : kwargs[topic_keyword]=Topic()

outputs(**kwargs)#: Set component output streams (topics)

set_input(**kwargs) → bool#

Set value of an input(s) topic

Returns:: If input(s) successfully updated
Return type:: bool

set_output(**kwargs) → bool#

Set value of an output(s) topic

Returns:: If output is successfully updates
Return type:: bool

config_from_file(config_file: str)#

Configure component from file

Parameters:: config_file (str) – Path to file (yaml, json, toml)

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

Gets a transform listener from the robot location frame to the world.

The localization frame is not configured: it is read from the header.frame_id of the location messages, so this is None until the first one arrives. When the robot is already localized in the world frame the lookup resolves to the identity transform, so “no odometry” needs no special case.

Returns:: TF listener from the location frame to the world frame
Return type:: Optional[TFListener]

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

Ask for every input under a key to be delivered in a given frame.

Handles keys bound to several topics (such as multiple proximity sensors), each of which carries its own source frame in its messages.

Parameters:

topic_key (TopicsKeys) – Key of the component input topic(s)
goal_frame (str) – Frame the data should be expressed in
static_tf (bool) – Whether the sensors are rigidly mounted

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

Gets a transform listener from an input’s own frame to a given frame.

The source frame is not configured anywhere: it is read from the header.frame_id of the data itself. This is the counterpart of transform_input_to, for algorithms that need the transform (its translation and rotation) rather than transformed data.

Parameters:

topic_key (TopicsKeys) – Key of the component input topic
goal_frame (str) – Frame to transform into, usually one of config.frames
static_tf (bool) – Whether the sensor is rigidly mounted
idx (int) – Index of the input, for keys bound to several topics

Returns:

TF listener, or None until the first message on that input arrives. Data already in the goal frame yields a listener that resolves to the identity transform.

Return type:

Optional[TFListener]

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

Resolve, at call time, the transform taking an input into a given frame.

Unlike the transform cached on a callback (refreshed only on message arrival), this looks the TF up live, so a TF that came up late is still picked up. Pass the returned transform to the callback’s get_output to deliver the data in the goal frame.

Parameters:

topic_key (TopicsKeys) – Key of the component input topic
goal_frame (Optional[str]) – Frame to deliver the data in, defaults to the configured world frame
idx (int) – Index of the input, for keys bound to several topics

Returns:

Whether the input can be delivered in the goal frame, and the transform to ask for it with (None when none is needed)

Return type:

Tuple[bool, Optional[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]#

Block until an input has delivered its first message and the transform from the input’s own frame to the robot base is available. The first message is names the input’s frame.

Parameters:

topic_key (TopicsKeys) – Key of the component input topic
idx (int) – Index of the input, for keys bound to several topics
static_tf (bool) – Whether the input’s frame is rigidly mounted

Returns:

The listener holding the transform, or None when the timeout passed

Return type:

Optional[TFListener]

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

Describe a spatial sensor input (LaserScan or PointCloud2) as the core’s SensorConfig. Its mount pose from the transform of its own frame to the robot base and, for a point cloud, the encoding of its x/y/z fields read from the first message.

Parameters:

topic_key (TopicsKeys) – Key of the sensor input topic
idx (int) – Index of the input, for keys bound to several topics
timeout (Optional[float]) – Seconds to wait before giving up, None waits indefinitely

Returns:

The sensor description, or None when the timeout passed or the first message could not be decoded

Return type:

Optional[SensorConfig]

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

Get the topic(s) name(s) corresponding to an input key name

Parameters:: key (str) – Input key name
Returns:: Topic(s) name(s)
Return type:: Union[str, List[str], None]

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

Get the topic(s) name(s) corresponding to an output key name

Parameters:: key (str) – Output key name
Returns:: Topic(s) name(s)
Return type:: 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 the topic(s) corresponding to an input key name

Parameters:: key (str) – Input key name
Returns:: Topic(s)
Return type:: Union[Topic, List[Topic], None]

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

Get the topic(s) corresponding to an output key name

Parameters:: key (str) – Output key name
Returns:: Topic(s)
Return type:: Union[Topic, List[Topic], None]

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

Get callback with given input key name

Parameters:

key (str) – Input key name
idx (int, optional) – Index of the input of multiple inputs correspond to the same key, defaults to 0

Raises:

KeyError – If key is not found in component inputs

Returns:

Callback object

Return type:

GenericCallback

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

Get publisher with given output key name

Parameters:

key (str) – Output topic key name
idx (int, optional) – Index of the output of multiple inputs correspond to the same key, defaults to 0

Raises:

KeyError – If key is not found in component outputs

Returns:

Publisher object

Return type:

Publisher

callbacks_inputs_check(inputs_to_check: Optional[List[str]] = None, inputs_to_exclude: Optional[List[str]] = None) → bool#: Check that all node inputs are provided before executing callback

kompass.components.mission#

Module Contents#

Classes#

Functions#

API#

`kompass.components.mission`#