Writing a ROS 2 Python Publisher
A ROS 2 publisher node in Python is built on rclpy — the Python client library. The pattern consists of: inheriting from rclpy.node.Node, creating a publisher with self.create_publisher(msg_type, topic_name, qos_depth), and scheduling periodic publication with self.create_timer(period_sec, callback).
The most common robot command topic is /cmd_vel accepting geometry_msgs/msg/Twist messages, which encode the desired linear and angular velocity. Publishing to /cmd_vel drives any ROS-compatible mobile robot (TurtleBot, Husky, simulated Turtlesim).
Node lifecycle in rclpy:
- Call
rclpy.init()— initialise the ROS 2 Python bindings - Instantiate your Node subclass
- Call
rclpy.spin(node)— hand control to the executor, which calls timer/subscription callbacks - Call
rclpy.shutdown()in a finally block for clean teardown
The spin() call blocks the main thread and processes callbacks until the node is shut down (Ctrl-C or rclpy.shutdown()). For concurrent callbacks, use rclpy.spin(node, executor=MultiThreadedExecutor()).
Subscribers, Services, and Package Structure
Writing a subscriber uses self.create_subscription(msg_type, topic_name, callback, qos_depth). The callback receives the message object and runs in the executor thread whenever a new message arrives on the topic.
self.sub = self.create_subscription(
LaserScan, '/scan', self.scan_callback, 10)
def scan_callback(self, msg: LaserScan):
min_range = min(r for r in msg.ranges if r > 0.01)
self.get_logger().info(f'Closest obstacle: {min_range:.2f} m')
Writing a service server:
from std_srvs.srv import SetBool
self.srv = self.create_service(SetBool, 'enable_drive', self.enable_cb)
def enable_cb(self, request, response):
self.enabled = request.data
response.success = True
response.message = f'Drive enabled: {self.enabled}'
return response
ROS 2 Python package structure:
my_pkg/
package.xml # dependencies, maintainer, license
setup.py # entry_points console_scripts
setup.cfg # install prefix config
resource/my_pkg # ament resource index marker (empty file)
my_pkg/
__init__.py
drive_node.py
subscriber_node.py
For C++ packages, replace setup.py with CMakeLists.txt using ament_cmake. Source files go in src/, headers in include/my_pkg/. The rclcpp API mirrors rclpy with the same concepts.
ROS 2 Package Creation Cheatsheet
| Task | Command | Notes |
|---|---|---|
| Create workspace | mkdir -p ~/ros2_ws/src && cd ~/ros2_ws | src/ holds all package clones |
| Create Python pkg | ros2 pkg create --build-type ament_python my_pkg --dependencies rclpy std_msgs | Generates setup.py, package.xml, resource/ |
| Create C++ pkg | ros2 pkg create --build-type ament_cmake my_cpp_pkg --dependencies rclcpp std_msgs | Generates CMakeLists.txt, package.xml |
| Build workspace | cd ~/ros2_ws && colcon build --symlink-install | --symlink-install avoids rebuilding on Python file changes |
| Source workspace | source ~/ros2_ws/install/setup.bash | Add to ~/.bashrc for automatic sourcing |
| Run a node | ros2 run my_pkg drive_node | Requires sourced workspace and built package |
| List topics | ros2 topic list | Shows all currently active topics |
| Echo topic | ros2 topic echo /cmd_vel | Prints messages in YAML format to terminal |
| Record bag | ros2 bag record -o my_bag -a | Records all topics to my_bag/ directory |
| Play bag | ros2 bag play my_bag | Republishes recorded messages on original topics |
Custom Message and Service Types
When standard message types are insufficient, create custom interfaces in a dedicated *_interfaces package. This separation is a best practice: other packages can depend on your interfaces without depending on your node implementations.
Directory structure for an interfaces package:
my_robot_interfaces/
package.xml # build_type: ament_cmake, depend: rosidl_default_generators
CMakeLists.txt # rosidl_generate_interfaces() call
msg/
RobotStatus.msg # custom message
srv/
SetSpeed.srv # custom service
action/
MoveToGoal.action # custom action
Example .msg file (msg/RobotStatus.msg):
std_msgs/Header header
float32 battery_voltage
bool motors_enabled
string error_message
Example .srv file (srv/SetSpeed.srv):
float32 linear_speed
float32 angular_speed
---
bool success
string message
After building, use interfaces as: from my_robot_interfaces.msg import RobotStatus. Inspect with: ros2 interface show my_robot_interfaces/msg/RobotStatus.
6 ROS 2 Programming Patterns and Anti-Patterns
- 01
DO: use create_timer for periodic publishing — NEVER spin in a callback or use time.sleep() inside a callback; blocking a callback blocks the entire executor thread and causes all other callbacks in that node to stall
- 02
DO: keep callbacks short and non-blocking — offload heavy computation (ML inference, path planning) to a separate thread or a dedicated node; use concurrent.futures.ThreadPoolExecutor if computation must be inside the node
- 03
AVOID: global mutable state shared between callbacks without locks — in MultiThreadedExecutor, callbacks may execute concurrently; protect shared variables with threading.Lock() or use rclpy reentrant callback groups carefully
- 04
AVOID: publishing large messages at high frequency — a 6 MP image at 60 Hz is ~1.3 GB/s; use compressed image topics, reduce resolution, use intra-process communication, or publish only on demand to prevent bandwidth saturation
- 05
DO: always check publisher has subscribers before expensive computation — use self.pub.get_subscription_count() > 0 to gate expensive image processing; this avoids wasted CPU cycles when no consumer is active
- 06
AVOID: hardcoding topic and service names — always make them parameters or use remappings in launch files; hardcoded names prevent node reuse across different robot platforms and cause silent communication failures when names differ by even one character