Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
A comprehensive guide to building robot applications with ROS2 (Robot Operating System 2) using
Python. This skill covers the full stack: node lifecycle, topic publish/subscribe, service and
action patterns, tf2 coordinate transforms, URDF loading, and autonomous navigation with Nav2.
When to Use This Skill
Use this skill when you need to:
Build ROS2 nodes in Python (rclpy) for robot control, sensing, or data processing
Implement publish/subscribe communication between robot components
Create service servers/clients for synchronous request-response interactions
Implement action servers/clients for long-running tasks with feedback (e.g., move to goal)
Transform coordinates between reference frames using tf2
Load and parse URDF robot description files
Send navigation goals to Nav2 and monitor their execution
Integrate sensor data (LiDAR, camera, IMU) with ROS2 message types
Debug and introspect a running ROS2 system
Do NOT use this skill for:
Non-ROS robotics frameworks (e.g., raw serial, pure OpenCV pipelines without ROS)
ROS1 (rospy) — the APIs differ significantly; see a dedicated ROS1 skill
Simulation setup (Gazebo/Isaac Sim configuration) — those have their own workflows
Background & Key Concepts
ROS2 Architecture Overview
ROS2 is built on DDS (Data Distribution Service) middleware, providing a distributed,
real-time communication backbone. The key abstractions are:
Concept
Description
Node
A process that participates in the ROS2 graph; the fundamental unit of computation
The tf2 library maintains a directed tree of coordinate frames. Each edge is a TransformStamped
message. Common frames: map -> odom -> base_link -> base_laser, etc.
Nav2 Simple Commander
nav2_simple_commander provides a Python API to interact with the Nav2 stack without writing
low-level action clients. It handles lifecycle management of Nav2 servers internally.
# Add to ~/.bashrc for persistence
source /opt/ros/humble/setup.bash
# Python packages used alongside ROS2
pip install pandas>=2.0 numpy>=1.24 matplotlib>=3.7
3. Create a ROS2 Package
# Create a colcon workspace
mkdir -p ~/ros2_ws/src && cd ~/ros2_ws/src
# Create a Python package
ros2 pkg create --build-type ament_python my_robot_pkg \
--dependencies rclpy std_msgs geometry_msgs sensor_msgs nav_msgs tf2_ros
cd ~/ros2_ws
colcon build --symlink-install
source install/setup.bash
4. Verify Installation
# Check ROS2 is sourced
ros2 --version
# List available message types
ros2 interface list | grep geometry_msgs
# Run a demo node
ros2 run demo_nodes_py talker &
ros2 run demo_nodes_py listener
Core Workflow
Step 1: Create and Spin a Basic Node
The minimal skeleton for any ROS2 Python node:
#!/usr/bin/env python3
"""minimal_node.py — bare-minimum rclpy node."""
import rclpy
from rclpy.node import Node
class MinimalNode(Node):
"""A node that logs a greeting on startup."""
def __init__(self) -> None:
super().__init__("minimal_node")
self.get_logger().info("MinimalNode has started!")
# Create a wall timer: callback fires every 1.0 second
self.timer = self.create_timer(1.0, self.timer_callback)
self._count = 0
def timer_callback(self) -> None:
self._count += 1
self.get_logger().info(f"Tick #{self._count}")
def main(args=None) -> None:
rclpy.init(args=args)
node = MinimalNode()
try:
rclpy.spin(node)
except KeyboardInterrupt:
pass
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == "__main__":
main()
#!/usr/bin/env python3
"""service_demo.py — service server + client for a simple math operation."""
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts
# ──────────────────────────────────────────────────────────────────────────────
# Service SERVER
# ──────────────────────────────────────────────────────────────────────────────
class AddTwoIntsServer(Node):
"""Provides an AddTwoInts service on /add_two_ints."""
def __init__(self) -> None:
super().__init__("add_two_ints_server")
self.srv = self.create_service(
AddTwoInts,
"/add_two_ints",
self.handle_request,
)
self.get_logger().info("AddTwoInts service is ready.")
def handle_request(
self,
request: AddTwoInts.Request,
response: AddTwoInts.Response,
) -> AddTwoInts.Response:
response.sum = request.a + request.b
self.get_logger().info(
f"Request: {request.a} + {request.b} = {response.sum}"
)
return response
# ──────────────────────────────────────────────────────────────────────────────
# Service CLIENT
# ──────────────────────────────────────────────────────────────────────────────
class AddTwoIntsClient(Node):
"""Calls the AddTwoInts service synchronously."""
def __init__(self) -> None:
super().__init__("add_two_ints_client")
self.client = self.create_client(AddTwoInts, "/add_two_ints")
# Wait for the server to come online (timeout 5 s)
if not self.client.wait_for_service(timeout_sec=5.0):
self.get_logger().error("Service /add_two_ints not available!")
def call(self, a: int, b: int) -> int:
"""Send a blocking service call and return the sum."""
request = AddTwoInts.Request()
request.a = a
request.b = b
future = self.client.call_async(request)
rclpy.spin_until_future_complete(self, future)
if future.result() is not None:
return future.result().sum
else:
raise RuntimeError("Service call failed")
def main(args=None) -> None:
rclpy.init(args=args)
server = AddTwoIntsServer()
client = AddTwoIntsClient()
# In a real application, server and client run in separate processes.
# Here we use a MultiThreadedExecutor to run both in one process.
from rclpy.executors import MultiThreadedExecutor
executor = MultiThreadedExecutor()
executor.add_node(server)
executor.add_node(client)
import threading
spin_thread = threading.Thread(target=executor.spin, daemon=True)
spin_thread.start()
result = client.call(3, 7)
client.get_logger().info(f"3 + 7 = {result}")
executor.shutdown()
rclpy.shutdown()
if __name__ == "__main__":
main()
Step 4: Action Server and Client (Long-Running Tasks)
#!/usr/bin/env python3
"""action_demo.py — Fibonacci action server with feedback streaming."""
import time
import rclpy
from rclpy.node import Node
from rclpy.action import ActionServer, ActionClient, GoalResponse, CancelResponse
from example_interfaces.action import Fibonacci
class FibonacciActionServer(Node):
"""Computes Fibonacci sequence up to requested order, streaming partial results."""
def __init__(self) -> None:
super().__init__("fibonacci_action_server")
self._action_server = ActionServer(
self,
Fibonacci,
"fibonacci",
execute_callback=self.execute_callback,
goal_callback=self.goal_callback,
cancel_callback=self.cancel_callback,
)
def goal_callback(self, goal_request) -> GoalResponse:
self.get_logger().info(f"Received goal: order={goal_request.order}")
if goal_request.order <= 0:
return GoalResponse.REJECT
return GoalResponse.ACCEPT
def cancel_callback(self, goal_handle) -> CancelResponse:
self.get_logger().info("Cancellation requested.")
return CancelResponse.ACCEPT
def execute_callback(self, goal_handle) -> Fibonacci.Result:
self.get_logger().info("Executing Fibonacci goal...")
feedback_msg = Fibonacci.Feedback()
sequence = [0, 1]
for i in range(1, goal_handle.request.order):
if goal_handle.is_cancel_requested:
goal_handle.canceled()
self.get_logger().info("Goal was cancelled.")
return Fibonacci.Result()
sequence.append(sequence[-1] + sequence[-2])
feedback_msg.partial_sequence = sequence
goal_handle.publish_feedback(feedback_msg)
time.sleep(0.1) # simulate computation time
goal_handle.succeed()
result = Fibonacci.Result()
result.sequence = sequence
self.get_logger().info(f"Result: {sequence}")
return result
class FibonacciActionClient(Node):
"""Sends a Fibonacci goal and prints feedback as it arrives."""
def __init__(self) -> None:
super().__init__("fibonacci_action_client")
self._client = ActionClient(self, Fibonacci, "fibonacci")
def send_goal(self, order: int) -> None:
self._client.wait_for_server()
goal = Fibonacci.Goal()
goal.order = order
self.get_logger().info(f"Sending goal: order={order}")
send_goal_future = self._client.send_goal_async(
goal,
feedback_callback=self.feedback_callback,
)
send_goal_future.add_done_callback(self.goal_response_callback)
def goal_response_callback(self, future) -> None:
goal_handle = future.result()
if not goal_handle.accepted:
self.get_logger().error("Goal rejected.")
return
self.get_logger().info("Goal accepted, waiting for result...")
result_future = goal_handle.get_result_async()
result_future.add_done_callback(self.result_callback)
def feedback_callback(self, feedback_msg) -> None:
partial = feedback_msg.feedback.partial_sequence
self.get_logger().info(f"Feedback: {partial}")
def result_callback(self, future) -> None:
result = future.result().result
self.get_logger().info(f"Final sequence: {result.sequence}")
rclpy.shutdown()
Step 5: tf2 Coordinate Transforms
#!/usr/bin/env python3
"""tf2_demo.py — broadcast and lookup coordinate transforms."""
import rclpy
from rclpy.node import Node
from tf2_ros import TransformBroadcaster, Buffer, TransformListener
from geometry_msgs.msg import TransformStamped
import math
class TF2Demo(Node):
"""
Broadcasts a rotating transform from 'world' to 'robot_base',
then looks up the transform and logs it.
"""
def __init__(self) -> None:
super().__init__("tf2_demo")
# Broadcaster sends our custom transforms
self.broadcaster = TransformBroadcaster(self)
# Buffer + Listener receive and cache all transforms in the system
self.tf_buffer = Buffer()
self.tf_listener = TransformListener(self.tf_buffer, self)
self._angle = 0.0
self.timer = self.create_timer(0.05, self.update) # 20 Hz
def update(self) -> None:
now = self.get_clock().now().to_msg()
self._angle += 0.02 # radians per tick
# ------------------------------------------------------------------
# Broadcast world -> robot_base
# ------------------------------------------------------------------
t = TransformStamped()
t.header.stamp = now
t.header.frame_id = "world"
t.child_frame_id = "robot_base"
t.transform.translation.x = math.cos(self._angle) * 1.0
t.transform.translation.y = math.sin(self._angle) * 1.0
t.transform.translation.z = 0.0
# Quaternion for rotation about Z by self._angle
t.transform.rotation.z = math.sin(self._angle / 2.0)
t.transform.rotation.w = math.cos(self._angle / 2.0)
self.broadcaster.sendTransform(t)
# ------------------------------------------------------------------
# Broadcast robot_base -> sensor_frame (static offset)
# ------------------------------------------------------------------
t2 = TransformStamped()
t2.header.stamp = now
t2.header.frame_id = "robot_base"
t2.child_frame_id = "sensor_frame"
t2.transform.translation.x = 0.3 # 30 cm in front of base
t2.transform.translation.z = 0.15 # 15 cm above base
t2.transform.rotation.w = 1.0 # no rotation
self.broadcaster.sendTransform(t2)
# ------------------------------------------------------------------
# Lookup: where is sensor_frame in world coordinates?
# ------------------------------------------------------------------
try:
tf = self.tf_buffer.lookup_transform(
"world",
"sensor_frame",
rclpy.time.Time(), # latest available
)
tx = tf.transform.translation.x
ty = tf.transform.translation.y
tz = tf.transform.translation.z
self.get_logger().debug(
f"sensor_frame in world: ({tx:.3f}, {ty:.3f}, {tz:.3f})"
)
except Exception as e:
self.get_logger().warning(f"TF lookup failed: {e}")
def main(args=None) -> None:
rclpy.init(args=args)
node = TF2Demo()
try:
rclpy.spin(node)
except KeyboardInterrupt:
pass
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == "__main__":
main()
Advanced Usage
Nav2 Simple Commander — Send Navigation Goals
#!/usr/bin/env python3
"""nav2_navigate.py — send waypoints to Nav2 and monitor execution."""
import rclpy
from nav2_simple_commander.robot_navigator import BasicNavigator, TaskResult
from geometry_msgs.msg import PoseStamped
from builtin_interfaces.msg import Duration
def make_pose(navigator: BasicNavigator, x: float, y: float, yaw: float = 0.0) -> PoseStamped:
"""Helper: build a PoseStamped in the 'map' frame."""
import math
pose = PoseStamped()
pose.header.frame_id = "map"
pose.header.stamp = navigator.get_clock().now().to_msg()
pose.pose.position.x = x
pose.pose.position.y = y
pose.pose.orientation.z = math.sin(yaw / 2.0)
pose.pose.orientation.w = math.cos(yaw / 2.0)
return pose
def main() -> None:
rclpy.init()
navigator = BasicNavigator()
# Set initial pose (must match the robot's actual starting position)
initial_pose = make_pose(navigator, x=0.0, y=0.0, yaw=0.0)
navigator.setInitialPose(initial_pose)
# Wait for Nav2 to fully activate
navigator.waitUntilNav2Active()
# ── Navigate to a single goal ─────────────────────────────────────────
goal = make_pose(navigator, x=3.0, y=1.5, yaw=1.57)
navigator.goToPose(goal)
while not navigator.isTaskComplete():
feedback = navigator.getFeedback()
if feedback:
remaining = Duration.from_msg(feedback.estimated_time_remaining)
print(f"ETA: {remaining.sec:.1f}s remaining")
result = navigator.getResult()
if result == TaskResult.SUCCEEDED:
print("Navigation succeeded!")
elif result == TaskResult.CANCELED:
print("Navigation was canceled.")
elif result == TaskResult.FAILED:
print("Navigation failed — check costmap or planner logs.")
# ── Follow a sequence of waypoints ───────────────────────────────────
waypoints = [
make_pose(navigator, 1.0, 0.0),
make_pose(navigator, 2.0, 1.0),
make_pose(navigator, 3.0, 0.0),
make_pose(navigator, 0.0, 0.0), # return to origin
]
navigator.followWaypoints(waypoints)
while not navigator.isTaskComplete():
feedback = navigator.getFeedback()
if feedback:
idx = feedback.current_waypoint
print(f"Visiting waypoint {idx + 1}/{len(waypoints)}")
print("Waypoint following complete.")
navigator.lifecycleShutdown()
rclpy.shutdown()
if __name__ == "__main__":
main()
Cause: ROS2 environment is not sourced, or another ROS2 instance is running on the same domain.
# Check that ROS2 is sourced
echo $ROS_DISTRO # should print 'humble', 'iron', etc.
source /opt/ros/humble/setup.bash
# Isolate from other ROS2 systems on the network
export ROS_DOMAIN_ID=42 # pick any 0–101; default is 0
Issue: Subscriber never receives messages
Cause: QoS mismatch between publisher and subscriber, or wrong topic name.
# List active topics
ros2 topic list
# Inspect QoS profile of a topic
ros2 topic info /cmd_vel --verbose
# Echo messages to verify data is flowing
ros2 topic echo /cmd_vel
# Use a compatible QoS profile (reliable, transient_local for latched topics)
from rclpy.qos import QoSProfile, ReliabilityPolicy, DurabilityPolicy
qos = QoSProfile(
reliability=ReliabilityPolicy.RELIABLE,
durability=DurabilityPolicy.TRANSIENT_LOCAL,
depth=1,
)
self.sub = self.create_subscription(msg_type, topic, callback, qos)
Issue: tf2 lookup raises LookupException
Cause: The requested transform has not been broadcast yet, or the parent/child frame names are wrong.
from rclpy.duration import Duration
try:
tf = self.tf_buffer.lookup_transform(
"world", "robot_base",
rclpy.time.Time(),
timeout=Duration(seconds=1.0), # wait up to 1 s
)
except Exception as e:
self.get_logger().error(f"TF lookup error: {e}")
# Visualize the tf tree
ros2 run tf2_tools view_frames
evince frames.pdf
Issue: Nav2 action server not available
# Check Nav2 nodes are running
ros2 node list | grep nav
# Restart Nav2 lifecycle
ros2 lifecycle set /bt_navigator configure
ros2 lifecycle set /bt_navigator activate
Issue: High CPU usage in spin loop
# Use MultiThreadedExecutor for I/O-bound nodes
from rclpy.executors import MultiThreadedExecutor
executor = MultiThreadedExecutor(num_threads=4)
executor.add_node(node_a)
executor.add_node(node_b)
executor.spin()
# Check the fused frame appears in the tf tree
ros2 run tf2_tools view_frames
# Confirm base_link_fused appears under odom
# Monitor the fused yaw in real time
ros2 run tf2_ros tf2_echo odom base_link_fused
