Add node defined in python to launch description
I want to create a simple publisher to be used in an integration test based on launch_testing
. All the examples I can find show how to launch a node defined in C++ in some package. But how I can add an instance of a subclass of rclpy.node.Node
to a launch description? Here is what this could look like, following the minimal pub/sub tutorial in python:
from launch import LaunchDescription
import launch_ros
import launch_testing
import rclpy
from std_msgs.msg import String
class MinimalPublisher(Node):
def __init__(self):
super().__init__('minimal_publisher')
self._publisher = self.create_publisher(String, 'tracked_objects', 10)
timer_period = 1 # seconds
self.timer = self.create_timer(timer_period, self.timer_callback)
self.i = 0
def timer_callback(self):
msg = String()
msg.data = f'Hello World: {self.i}'
self._publisher.publish(msg)
self.get_logger().info(f'Publishing: "{msg.data}"')
self.i += 1
@pytest.mark.launch_test
def generate_test_description():
publisher_node = MinimalPublisher()
####
# how to add `publisher_node` to launch description? The following does not work because the rclpy.node.Node type is different from launch_ros.actions.Node
#####
return LaunchDescription([publisher_node, launch_testing.actions.ReadyToTest()])
Workaround
I put the node definition into its own executable. This means it's more cumbersome to pass arguments to its constructor and one can't call any methods directly from a test. But for the simple example above, it's not a real limitation:
#!/bin/env python3
import rclpy
from rclpy.node import Node
class MinimalPublisher(Node):
# implement
def main(args=None):
rclpy.init(args=args)
minimal_publisher = MinimalPublisher()
rclpy.spin(minimal_publisher)
# Destroy the node explicitly
# (optional - otherwise it will be done automatically
# when the garbage collector destroys the node object)
minimal_publisher.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
And lauch it with ExecuteProcess
@pytest.mark.launch_test
def generate_test_description():
publisher_node = launch.actions.ExecuteProcess(
cmd=['python3', 'publisher_node.py'],
cwd=os.path.join(get_package_share_directory('some_pkg'), 'test/'))