rclpy Python Client Library Integration

Learning Objectives

By the end of this chapter, you will be able to:

• Develop ROS 2 applications using the rclpy Python client library
• Create and manage rclpy nodes
• Implement data exchange using topics and services in Python
• Set up a ROS 2 workspace with Python packages

Introduction to rclpy

rclpy is the Python client library for ROS 2. It provides a Python API for creating ROS 2 nodes, publishers, subscribers, services, and other ROS 2 concepts. rclpy is built on top of rcl (the C client library) and provides a more Pythonic interface.

Setting Up a ROS 2 Workspace with Python Packages

Creating a Workspace

mkdir -p ~/ros2_workspace/src
cd ~/ros2_workspace

Creating a Python Package

cd src
ros2 pkg create --build-type ament_python my_robot_package

Package Structure

A typical Python package in ROS 2 has the following structure:

my_robot_package/
├── my_robot_package/
│   ├── __init__.py
│   └── my_node.py
├── setup.cfg
├── setup.py
├── package.xml
└── test/
    └── test_copyright.py
    └── test_flake8.py
    └── test_pep257.py

setup.py Configuration

from setuptools import find_packages, setup

package_name = 'my_robot_package'

setup(
    name=package_name,
    version='0.0.0',
    packages=find_packages(exclude=['test']),
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='Your Name',
    maintainer_email='your.email@example.com',
    description='TODO: Package description',
    license='TODO: License declaration',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'my_node = my_robot_package.my_node:main',
        ],
    },
)

Creating and Managing rclpy Nodes

Basic Node Structure

import rclpy
from rclpy.node import Node

class MyRobotNode(Node):
    def __init__(self):
        super().__init__('my_robot_node')
        self.get_logger().info('My Robot Node has been started')

def main(args=None):
    rclpy.init(args=args)
    node = MyRobotNode()

    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Node Lifecycle

The typical lifecycle of an rclpy node includes:

1. Initialization with rclpy.init()
2. Node creation inheriting from rclpy.node.Node
3. Setting up publishers, subscribers, services, etc.
4. Spinning the node with rclpy.spin() or rclpy.spin_once()
5. Cleanup with node.destroy_node() and rclpy.shutdown()

Implementing Data Exchange with Topics

Publisher Implementation

import rclpy
from rclpy.node import Node
from std_msgs.msg import String, Int32, Float32

class DataPublisher(Node):
    def __init__(self):
        super().__init__('data_publisher')

        # Create publishers for different message types
        self.string_publisher = self.create_publisher(String, 'string_topic', 10)
        self.int_publisher = self.create_publisher(Int32, 'int_topic', 10)
        self.float_publisher = self.create_publisher(Float32, 'float_topic', 10)

        # Create a timer to publish data periodically
        timer_period = 1.0  # seconds
        self.timer = self.create_timer(timer_period, self.timer_callback)

        self.counter = 0

    def timer_callback(self):
        # Publish string message
        string_msg = String()
        string_msg.data = f'Hello from Python: {self.counter}'
        self.string_publisher.publish(string_msg)

        # Publish integer message
        int_msg = Int32()
        int_msg.data = self.counter
        self.int_publisher.publish(int_msg)

        # Publish float message
        float_msg = Float32()
        float_msg.data = float(self.counter) * 1.5
        self.float_publisher.publish(float_msg)

        self.get_logger().info(f'Published: {string_msg.data}')
        self.counter += 1

def main(args=None):
    rclpy.init(args=args)
    publisher = DataPublisher()

    try:
        rclpy.spin(publisher)
    except KeyboardInterrupt:
        pass
    finally:
        publisher.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Subscriber Implementation

import rclpy
from rclpy.node import Node
from std_msgs.msg import String, Int32, Float32

class DataSubscriber(Node):
    def __init__(self):
        super().__init__('data_subscriber')

        # Create subscribers for different message types
        self.string_subscription = self.create_subscription(
            String,
            'string_topic',
            self.string_callback,
            10)

        self.int_subscription = self.create_subscription(
            Int32,
            'int_topic',
            self.int_callback,
            10)

        self.float_subscription = self.create_subscription(
            Float32,
            'float_topic',
            self.float_callback,
            10)

    def string_callback(self, msg):
        self.get_logger().info(f'Received string: {msg.data}')

    def int_callback(self, msg):
        self.get_logger().info(f'Received integer: {msg.data}')

    def float_callback(self, msg):
        self.get_logger().info(f'Received float: {msg.data}')

def main(args=None):
    rclpy.init(args=args)
    subscriber = DataSubscriber()

    try:
        rclpy.spin(subscriber)
    except KeyboardInterrupt:
        pass
    finally:
        subscriber.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Implementing Data Exchange with Services

Service Server

from example_interfaces.srv import AddTwoInts
import rclpy
from rclpy.node import Node

class AddTwoIntsService(Node):
    def __init__(self):
        super().__init__('add_two_ints_service')
        self.srv = self.create_service(
            AddTwoInts,
            'add_two_ints',
            self.add_two_ints_callback
        )

    def add_two_ints_callback(self, request, response):
        response.sum = request.a + request.b
        self.get_logger().info(f'Request: {request.a} + {request.b} = {response.sum}')
        return response

def main(args=None):
    rclpy.init(args=args)
    service = AddTwoIntsService()

    try:
        rclpy.spin(service)
    except KeyboardInterrupt:
        pass
    finally:
        service.destroy_node()
        rclpy.shutdown()

if __name__ == '__main__':
    main()

Service Client

from example_interfaces.srv import AddTwoInts
import rclpy
from rclpy.node import Node

class AddTwoIntsClient(Node):
    def __init__(self):
        super().__init__('add_two_ints_client')
        self.cli = self.create_client(AddTwoInts, 'add_two_ints')
        while not self.cli.wait_for_service(timeout_sec=1.0):
            self.get_logger().info('Service not available, waiting again...')
        self.req = AddTwoInts.Request()

    def send_request(self, a, b):
        self.req.a = a
        self.req.b = b
        self.future = self.cli.call_async(self.req)
        rclpy.spin_until_future_complete(self, self.future)
        return self.future.result()

def main(args=None):
    rclpy.init(args=args)
    client = AddTwoIntsClient()

    response = client.send_request(2, 3)
    if response:
        client.get_logger().info(f'Result: {response.sum}')
    else:
        client.get_logger().info('Service call failed')

    client.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Advanced rclpy Features

Parameters

import rclpy
from rclpy.node import Node

class ParameterNode(Node):
    def __init__(self):
        super().__init__('parameter_node')

        # Declare parameters with default values
        self.declare_parameter('robot_name', 'my_robot')
        self.declare_parameter('max_velocity', 1.0)
        self.declare_parameter('sensors_enabled', True)

        # Get parameter values
        self.robot_name = self.get_parameter('robot_name').value
        self.max_velocity = self.get_parameter('max_velocity').value
        self.sensors_enabled = self.get_parameter('sensors_enabled').value

        self.get_logger().info(
            f'Robot: {self.robot_name}, Max Vel: {self.max_velocity}, Sensors: {self.sensors_enabled}'
        )

def main(args=None):
    rclpy.init(args=args)
    node = ParameterNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Timers

import rclpy
from rclpy.node import Node

class TimerNode(Node):
    def __init__(self):
        super().__init__('timer_node')

        # Create timers with different periods
        self.timer1 = self.create_timer(0.5, self.timer1_callback)  # 0.5 second period
        self.timer2 = self.create_timer(1.0, self.timer2_callback)  # 1.0 second period

        self.counter1 = 0
        self.counter2 = 0

    def timer1_callback(self):
        self.get_logger().info(f'Timer 1: {self.counter1}')
        self.counter1 += 1

    def timer2_callback(self):
        self.get_logger().info(f'Timer 2: {self.counter2}')
        self.counter2 += 1

def main(args=None):
    rclpy.init(args=args)
    node = TimerNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()

if __name__ == '__main__':
    main()

Best Practices

1. Error Handling: Always include proper error handling in your nodes
2. Resource Management: Properly clean up resources with destroy_node()
3. Logging: Use the node's logger for consistent logging
4. Parameter Validation: Validate parameters before using them
5. Threading: Be aware of threading implications when using rclpy

Exercises

1. Create a Python package with multiple nodes that communicate via topics
2. Implement a service that performs a complex calculation based on multiple parameters
3. Create a node that uses parameters to configure its behavior
4. Build a simple robot controller using publishers and subscribers

Summary

The rclpy client library provides a comprehensive Python interface for ROS 2 development. Understanding how to create nodes, publishers, subscribers, and services is essential for developing robotic applications in Python. The library provides all the necessary tools to build complex, distributed robotic systems.