`kompass.components.component`#

Module Contents#

Classes#

Component

Component is the main execution entity in Kompass, every Component is equivalent to a ROS2 Lifecycle Node.

API#

class kompass.components.component.Component(component_name: str, config: Optional[kompass.config.ComponentConfig] = None, config_file: Optional[str] = None, inputs: Optional[Dict[kompass.components.defaults.TopicsKeys, Union[kompass.components.ros.Topic, List[kompass.components.ros.Topic], None]]] = None, outputs: Optional[Dict[kompass.components.defaults.TopicsKeys, Union[kompass.components.ros.Topic, List[kompass.components.ros.Topic], None]]] = None, fallbacks: Optional[ros_sugar.core.ComponentFallbacks] = None, allowed_inputs: Optional[Dict[str, ros_sugar.io.AllowedTopics]] = None, allowed_outputs: Optional[Dict[str, ros_sugar.io.AllowedTopics]] = None, allowed_run_types: Optional[List[kompass.config.ComponentRunType]] = None, callback_group=None, **kwargs)#

Bases: ros_sugar.core.BaseComponent

Component is the main execution entity in Kompass, every Component is equivalent to a ROS2 Lifecycle Node.

A Component requires a set of input and/or output topic(s). All the functionalities implemented in ROS2 nodes can be implemented in the Component. Inputs/Outputs are attrs classes with the attribute name representing a unique input/output key name and the value equal to the Topic.

Example:

    from kompass.topic import Topic, create_topics_config
    from kompass.components.component import Component

    NewTopicsClass = create_topics_config(
                        "ClassName",
                        input_1=Topic(name="/plan", msg_type="Path"),
                        input_2=Topic(name="/location", msg_type="Odometry"),
                    )
    inputs = NewTopicsClass()
    my_component = Component(component_name='my_node', inputs=inputs)

Components in Kompass can be defined to accept only restricted types of inputs/outputs to help lock the functionality of a specific Component implementation.

Example:

    from kompass.topic import AllowedTopic, RestrictedTopicsConfig

    class AllowedInputs(RestrictedTopicsConfig):
        PLAN = AllowedTopic(key="input_1", types=["Path"])
        LOCATION = AllowedTopic(key="input_2", types=["Odometry", "PoseStamped", "Pose"])

    my_component = Component(component_name='my_node', allowed_inputs=AllowedInputs)

custom_on_configure()#: Component custom configuration method to set the core debug level

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) → 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

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

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.component#

Module Contents#

Classes#

API#

`kompass.components.component`#