kompass.components.drive_manager

Module Contents

Classes

DriveManagerConfig	DriveManager component configuration parameters
DriveManager	DriveManager component used for direct communication with the robot.

API

class kompass.components.drive_manager.DriveManagerConfig

Bases: kompass.config.ComponentConfig

DriveManager component configuration parameters

Name	Type, Default	Description
closed_loop	bool, True	Publish commands in closed loop by checking the robot velocity from the odometry topic
closed_loop_span	int, 3	Max number of commands to send in a closed loop execution
smooth_commands	bool, False	Filter (smooth) incoming velocity commands to limit the acceleration
cmd_tolerance	float, 0.05	Tolerance value when checking for reaching the command in closed loop
critical_zone_angle	float, 45deg	Angle range for the emergency stop critical zone (deg)
critical_zone_distance	float, 0.05	Distance for the emergency stop critical zone (meters)
slowdown_zone_distance	float, 0.2	Distance for the slowdown zone (meters)
disable_safety_stop	bool, False	Set to True to disable the safety stop functionality
use_without_scan_sensor	bool, False	Set to True to allow running the drive manager with no spatial sensor at all. Range sensors (e.g. front and back ultrasounds) count as safety sensors and do not need this flag
use_gpu	bool, True	Use GPU implementation for the critical zone checking if available, otherwise use CPU implementation
sensor_data_timeout	float, 0.2	Maximum age (seconds) of a safety sensor’s last message before it is considered stale
stale_sensor_policy	str, "stop"	What to do when a safety sensor goes stale: "stop" triggers an emergency stop until data returns, "skip" runs the check on the remaining sensors only

class kompass.components.drive_manager.DriveManager(component_name: str, config_file: Optional[str] = None, config: Optional[kompass.components.drive_manager.DriveManagerConfig] = None, inputs: Optional[Dict[str, kompass.components.ros.Topic]] = None, outputs: Optional[Dict[str, kompass.components.ros.Topic]] = None, **kwargs)

Bases: kompass.components.component.Component

DriveManager component used for direct communication with the robot.

Inputs:

Key Name	Allowed Types	Number	Default
intermediate_cmd	geometry_msgs.msg.Twist	1	Topic(name="/control", msg_type="Twist")
intermediate_cmd_list	kompass_interfaces.msg.TwistArray	1	Topic(name="/control_list", msg_type="TwistArray")
spatial_sensor	sensor_msgs.msg.LaserScan, sensor_msgs.msg.PointCloud2, sensor_msgs.msg.Range	1 to 10	Topic(name="/scan", msg_type="LaserScan")
robot_location	nav_msgs.msg.Odometry, geometry_msgs.msg.PoseStamped, geometry_msgs.msg.Pose	1	Topic(name="/odom", msg_type="Odometry")

Outputs:

Key Name	Allowed Types	Number	Default
final_command	geometry_msgs.msg.Twist, geometry_msgs.msg.TwistStamped	1	Topic(name="/cmd_vel", msg_type="Twist")
emergency	std_msgs.msg.Bool	1	Topic(name="/emergency_stop", msg_type="Bool")

init_variables()

Overwrites the init variables method called at Node init

execute_cmd_closed_loop(output: geometry_msgs.msg.Twist, max_time: float)

Execute a control command in closed loop

Parameters:

• cmd (Twist) – Velocity Twist message

• max_time (float) – Maximum time for the closed loop execution (s)

move_forward(max_distance: float, **_) → bool

Moves the robot forward if the forward direction is clear of obstacles

Parameters:

max_distance (float) – Maximum distance (m)

Returns:

If the movement action is performed

Return type:

bool

move_backward(max_distance: float, **_) → bool

Moves the robot backwards if the backward direction is clear of obstacles

Parameters:

max_distance (float) – Maximum distance (m)

Returns:

If the movement action is performed

Return type:

bool

rotate_in_place(max_rotation: float, safety_margin: Optional[float] = None, **_) → bool

Rotates the robot in place if a safety margin around the robot is clear

Parameters:

safety_margin (Optional[float], optional) – Margin clear of obstacles to perform rotation, if None defaults to 5% of the robot_radius

Returns:

If the movement action is performed

Return type:

bool

move_to_unblock(max_distance_forward: Optional[float] = None, max_distance_backwards: Optional[float] = None, max_rotation: float = np.pi / 2, rotation_safety_margin: Optional[float] = None, **_) → bool

Moves the robot forward/backward or rotate in place to get out of blocking spots

Parameters:

• max_distance_forward (Optional[float], optional) – Maximum distance to move forward (meters), if None defaults to 2 * robot_radius

• max_distance_backwards (Optional[float], optional) – Maximum distance to move backwards (meters), if None defaults to 2 * robot_radius

• max_rotation (float, optional) – Maximum rotation angle (radians), defaults to np.pi/4

• rotation_safety_margin (Optional[float], optional) – Safety margin to perform rotation in place (meters), if None defaults to 5% of robot_radius

Returns:

If one of the movement actions is performed

Return type:

bool

custom_on_configure()

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