Creating a Custom Navigation Component#

Every execution unit in Kompass is a Component, which maps to a ROS 2 Lifecycle Node. This guide explains how to create a new component by subclassing kompass.components.component.Component.

Component Base Class#

Component (defined in kompass/kompass/components/component.py) extends BaseComponent from ros-sugar. It adds Kompass-specific features:

Keyed I/O via TopicsKeys enum – each input/output has a semantic key, not just a topic name.
Allowed topics enforcement via AllowedTopics – restricts which message types a component accepts.
Default topic configurations – sensible defaults for rapid prototyping.
TF listener helpers – odom_tf_listener, scan_tf_listener, depth_tf_listener, pc_tf_listener.

Constructor Signature#

def __init__(
    self,
    component_name: str,
    config: Optional[ComponentConfig] = None,
    config_file: Optional[str] = None,
    inputs: Optional[Dict[TopicsKeys, Union[Topic, List[Topic], None]]] = None,
    outputs: Optional[Dict[TopicsKeys, Union[Topic, List[Topic], None]]] = None,
    fallbacks: Optional[ComponentFallbacks] = None,
    allowed_inputs: Optional[Dict[str, AllowedTopics]] = None,
    allowed_outputs: Optional[Dict[str, AllowedTopics]] = None,
    allowed_run_types: Optional[List[ComponentRunType]] = None,
    callback_group=None,
    **kwargs,
):

TopicsKeys Enum#

TopicsKeys (in kompass.components.defaults) provides standardized key names for all topics in the Kompass navigation stack:

Key	Value	Purpose
`GOAL_POINT`	`"goal_point"`	Navigation goal
`GLOBAL_PLAN`	`"plan"`	Global path from planner
`GLOBAL_MAP`	`"map"`	Global occupancy grid
`ROBOT_LOCATION`	`"location"`	Robot odometry/pose
`SPATIAL_SENSOR`	`"sensor_data"`	LiDAR, depth, point cloud
`VISION_DETECTIONS`	`"vision_detections"`	Object detection results
`DEPTH_CAM_INFO`	`"depth_camera_info"`	Depth camera intrinsics
`LOCAL_PLAN`	`"local_plan"`	Local path segment
`PATH_SAMPLES`	`"path_samples"`	Sampled path candidates
`INTERMEDIATE_CMD`	`"command"`	Single velocity command
`INTERMEDIATE_CMD_LIST`	`"multi_command"`	Velocity command array
`LOCAL_MAP`	`"local_map"`	Local occupancy grid
`INTERPOLATED_PATH`	`"interpolation"`	Interpolated path
`TRACKED_POINT`	`"tracked_point"`	Currently tracked point on path
`FINAL_COMMAND`	`"robot_command"`	Command sent to robot driver
`EMERGENCY`	`"emergency_stop"`	Emergency stop signal
`REACHED_END`	`"reached_end"`	Goal reached flag

Using TopicsKeys ensures consistent naming across components and enables automatic wiring in recipes.

AllowedTopics#

Each component defines which message types are valid for each key using AllowedTopics (from ros_sugar.io):

from ros_sugar.io import AllowedTopics

my_allowed_inputs = {
    TopicsKeys.ROBOT_LOCATION: AllowedTopics(
        types=["Odometry", "PoseStamped", "Pose"]
    ),
    TopicsKeys.GOAL_POINT: AllowedTopics(
        types=["PointStamped", "PoseStamped"]
    ),
}

AllowedTopics parameters:

Parameter	Default	Description
`types`	(required)	List of allowed message type names
`number_required`	`1`	Minimum number of topics required for this key
`number_optional`	`0`	Additional optional topics allowed (`-1` for unlimited)

For example, a component that requires 1 sensor input but can optionally accept up to 3:

TopicsKeys.SPATIAL_SENSOR: AllowedTopics(
    types=["LaserScan", "PointCloud2"],
    number_required=1,
    number_optional=2,
)

Default Topic Configurations#

Existing components define their defaults in kompass.components.defaults. For example, the Planner:

planner_default_inputs = {
    TopicsKeys.GLOBAL_MAP: Topic(name="/map", msg_type="OccupancyGrid", ...),
    TopicsKeys.GOAL_POINT: Topic(name="/goal", msg_type="PointStamped"),
    TopicsKeys.ROBOT_LOCATION: Topic(name="/odom", msg_type="Odometry"),
}

When users provide custom topics, they are merged with your defaults using update_topics():

from kompass.components.ros import update_topics

in_topics = (
    update_topics(my_default_inputs, **inputs)
    if inputs
    else my_default_inputs
)

This lets users override individual topics without having to redefine the entire dictionary.

Lifecycle and Key Methods#

Components follow the ROS 2 Lifecycle pattern. The methods you override to add your logic:

`init_variables()`#

Called during node initialization (before configure). Use this to set up algorithm instances, internal state, and any variables that depend on self.config:

def init_variables(self):
    self.goal = None
    self.robot_state = None
    self._my_algorithm = MyAlgorithm(
        robot_type=self.config.robot.model_type,
    )

`custom_on_configure()`#

Called when the lifecycle node transitions to Inactive. Subscriptions and publishers are created by the base class. Always call super() first:

def custom_on_configure(self):
    super().custom_on_configure()
    # Additional configuration here

`custom_on_activate()`#

Called when the lifecycle node transitions to Active. The main execution loop starts after this.

`_execution_step()`#

The main loop body, called at config.loop_rate Hz (for TIMED run type). This is where your component reads inputs, runs its algorithm, and publishes outputs:

def _execution_step(self):
    self._update_state()  # Read latest values from all callbacks
    if self.robot_state and self.goal:
        result = self._my_algorithm.compute(self.robot_state, self.goal)
        self.get_publisher(TopicsKeys.GOAL_POINT).publish(result)

Run Types#

The ComponentRunType enum controls how the component executes:

Run Type	Description
`TIMED`	Calls `_execution_step()` at a fixed rate
`EVENT`	Executes in response to an event trigger
`SERVER`	Runs as a ROS 2 service server
`ACTION_SERVER`	Runs as a ROS 2 action server (supports feedback/cancel)

Set allowed_run_types in your constructor to restrict which modes are valid for your component.

Accessing Inputs and Outputs#

Reading Inputs: `get_callback()`#

callback = self.get_callback(TopicsKeys.ROBOT_LOCATION)

Returns a GenericCallback object (or None if the input is not configured). Key API:

Method / Property	Description
`.got_msg`	`True` if at least one message has been received
`.get_output()`	Returns the latest processed message data
`.get_output(clear_last=True)`	Returns data and clears the internal buffer
`.clear_last_msg()`	Clears the last received message
`.frame_id`	Frame ID from the message header (if available)

For keys with multiple topics (e.g. multiple sensors), use the idx parameter:

sensor_0 = self.get_callback(TopicsKeys.SPATIAL_SENSOR, idx=0)
sensor_1 = self.get_callback(TopicsKeys.SPATIAL_SENSOR, idx=1)

Publishing Outputs: `get_publisher()`#

publisher = self.get_publisher(TopicsKeys.GOAL_POINT)
publisher.publish(my_msg)

Returns a Publisher object. Call .publish() with the message data. The publisher handles ROS message conversion automatically.

Checking All Inputs#

Use callbacks_inputs_check() to block until all required inputs are available:

def _execution_step(self):
    if not self.callbacks_inputs_check():
        return
    # All inputs are ready

Skeleton Example: WaypointManager Component#

Below is a complete skeleton for a hypothetical WaypointManager component that accepts a set of waypoints and publishes the next goal point.

from typing import Dict, List, Optional, Union
from attrs import define, field

from kompass.components.component import Component
from kompass.components.ros import Topic, update_topics
from kompass.components.defaults import TopicsKeys
from kompass.config import ComponentConfig, ComponentRunType
from ros_sugar.io import AllowedTopics


# --- Configuration ---
@define
class WaypointManagerConfig(ComponentConfig):
    """Configuration for the WaypointManager component."""
    waypoint_reach_threshold: float = field(default=0.5)
    loop_waypoints: bool = field(default=False)


# --- Allowed I/O ---
waypoint_manager_allowed_inputs: Dict[TopicsKeys, AllowedTopics] = {
    TopicsKeys.ROBOT_LOCATION: AllowedTopics(
        types=["Odometry", "PoseStamped", "Pose"]
    ),
    TopicsKeys.GLOBAL_PLAN: AllowedTopics(types=["Path"]),
}

waypoint_manager_allowed_outputs: Dict[TopicsKeys, AllowedTopics] = {
    TopicsKeys.GOAL_POINT: AllowedTopics(
        types=["PointStamped", "PoseStamped"]
    ),
}

# --- Default Topics ---
waypoint_manager_default_inputs: Dict[TopicsKeys, Topic] = {
    TopicsKeys.ROBOT_LOCATION: Topic(name="/odom", msg_type="Odometry"),
    TopicsKeys.GLOBAL_PLAN: Topic(name="/waypoints", msg_type="Path"),
}

waypoint_manager_default_outputs: Dict[TopicsKeys, Topic] = {
    TopicsKeys.GOAL_POINT: Topic(name="/goal", msg_type="PointStamped"),
}


# --- Component ---
class WaypointManager(Component):
    """Manages a list of waypoints and publishes the next goal."""

    def __init__(
        self,
        component_name: str = "waypoint_manager",
        config: Optional[WaypointManagerConfig] = None,
        config_file: Optional[str] = None,
        inputs: Optional[Dict] = None,
        outputs: Optional[Dict] = None,
        **kwargs,
    ):
        config = config or WaypointManagerConfig()

        # Merge user-provided topics with defaults
        in_topics = (
            update_topics(waypoint_manager_default_inputs, **inputs)
            if inputs
            else waypoint_manager_default_inputs
        )
        out_topics = (
            update_topics(waypoint_manager_default_outputs, **outputs)
            if outputs
            else waypoint_manager_default_outputs
        )

        super().__init__(
            component_name=component_name,
            config=config,
            config_file=config_file,
            inputs=in_topics,
            outputs=out_topics,
            allowed_inputs=waypoint_manager_allowed_inputs,
            allowed_outputs=waypoint_manager_allowed_outputs,
            allowed_run_types=[ComponentRunType.TIMED],
            **kwargs,
        )

    def init_variables(self):
        """Called at node init -- set up internal state."""
        self._current_waypoint_idx = 0
        self._waypoints = []

    def custom_on_configure(self):
        """Called on lifecycle configure transition."""
        super().custom_on_configure()
        self._current_waypoint_idx = 0

    def _execution_step(self):
        """Main timed callback -- called at config.loop_rate Hz."""
        # Read current robot location
        location_cb = self.get_callback(TopicsKeys.ROBOT_LOCATION)
        if location_cb is None or not location_cb.got_msg:
            return

        robot_pose = location_cb.get_output()

        # Get the waypoints path
        path_cb = self.get_callback(TopicsKeys.GLOBAL_PLAN)
        if path_cb is None or not path_cb.got_msg:
            return

        waypoints = path_cb.get_output()

        # Compute and publish the next waypoint
        # ... your logic here ...
        publisher = self.get_publisher(TopicsKeys.GOAL_POINT)
        publisher.publish(next_goal)

Key Patterns to Follow#

Always define allowed_inputs and allowed_outputs to enforce type safety.
Provide default topics so the component works out of the box.
Merge with update_topics() so users can override individual topics without redefining everything.
Use TopicsKeys for all I/O keys to maintain stack-wide consistency.
Override init_variables() to set up algorithm instances and state that depends on self.config.
Override custom_on_configure() and call super() first.
Implement _execution_step() as your main loop body.
Use get_callback(key) and get_publisher(key) to access I/O by semantic key rather than topic name.
Set allowed_run_types to restrict how the component can be executed.

Next Steps#

See the Advanced Guide: Health Status, Fallbacks & Events for making your component resilient with automatic failure recovery.