Writing a simple service and client(Python)
Background
node는 service를 이용해 communicate할 때, data를 request하는 node는 client node, respond하는 node를 service node라고 한다.
이는 .srv 파일로 정의된다.
Tasks
Create a package
ros2_ws/src directory를 만들어 새로운 package를 만들어보자.
ros2 pkg create --build-type ament_python --license Apache-2.0 py_srvcli --dependencies rclpy example_interfaces
필요한 file, folder를 만들었음을 terminal로 확인할 수 있을 것이다.
--dependencies는 자동으로 필요한 dependency를 package.xml에 추가해준다.
example_interfaces는 request, response구조에 필요한 .srv file을 포함하는 package이다.
int64 a
int64 b
---
int64 sum처음 2줄은 request에 필요한 parameter, 아래는 response이다.
Update package.xml
--dependencies 옵션을 사용하였기 때문에 package.xml에 dependency를 추가할 필요가 없다.
하지만 description, maintainer, license는 추가해야 한다!
<description>Python client server tutorial</description>
<maintainer email="you@email.com">Your Name</maintainer>
<license>Apache License 2.0</license>
Update setup.py
setup.py도 동일하게 추가해주자.
maintainer='Your Name',
maintainer_email='you@email.com',
description='Python client server tutorial',
license='Apache License 2.0',
Write the service node
ros2_ws/src/py_srvcli/py_srvcli directory에 service_member_function.py 파일을 추가하고 아래 코드를 넣어주자.
from example_interfaces.srv import AddTwoInts
import rclpy
from rclpy.node import Node
class MinimalService(Node):
def __init__(self):
super().__init__('minimal_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('Incoming request\na: %d b: %d' % (request.a, request.b))
return response
def main():
rclpy.init()
minimal_service = MinimalService()
rclpy.spin(minimal_service)
rclpy.shutdown()
if __name__ == '__main__':
main()
Examine the code
필요한 library를 import한다.
import sys
from example_interfaces.srv import AddTwoInts
import rclpy
from rclpy.node import Node
MinimalService class constructor가 node(minimal_service)를 초기화하고 service를 만들어 type, name, callback을 정의한다.
def __init__(self):
super().__init__('minimal_service')
self.srv = self.create_service(AddTwoInts, 'add_two_ints', self.add_two_ints_callback)
service callback은 request data를 받고, 이를 합쳐 response로 return한다.
def add_two_ints_callback(self, request, response):
response.sum = request.a + request.b
self.get_logger().info('Incoming request\na: %d b: %d' % (request.a, request.b))
return response
최종적으로 main class는 ROS2 Python client library를 초기화하고, MinimalService class를 객체화(instantiate)해서 service node를 생성하여 callback을 다루기 위해 node를 spin한다.
Add an entry point
ros2 run이 node를 실행하기 위해 setup.py에 entry point를 추가해준다.
entry_points={
'console_scripts': [
'service = py_srvcli.service_member_function:main',
],
},
Write the client node
# ros2_ws/src/py_srvcli/py_srvcli/client_member_function.py
import sys
from example_interfaces.srv import AddTwoInts
import rclpy
from rclpy.node import Node
class MinimalClientAsync(Node):
def __init__(self):
super().__init__('minimal_client_async')
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
return self.cli.call_async(self.req)
def main():
rclpy.init()
minimal_client = MinimalClientAsync()
future = minimal_client.send_request(int(sys.argv[1]), int(sys.argv[2]))
rclpy.spin_until_future_complete(minimal_client, future)
response = future.result()
minimal_client.get_logger().info(
'Result of add_two_ints: for %d + %d = %d' %
(int(sys.argv[1]), int(sys.argv[2]), response.sum))
minimal_client.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
Examine the code
필요한 library들을 import한다.
import sys
from example_interfaces.srv import AddTwoInts
import rclpy
from rclpy.node import Node
MinimalClientAsync class contructor는 minimal_client_async 이름의 node를 초기화한다.
constructor는 service node와 동일한 type, 이름의 client를 생성한다.
type, 이름은 client, service가 communicate하기 위해 매칭되어야 한다.
while loop는 service가 이용가능한지 1초마다 확인한다.
마지막으로 AddTwoInts request를 생성한다.
def __init__(self):
super().__init__('minimal_client_async')
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()
send_request는 request를 보내고 response 또는 fail을 받을 때까지 spin한다. (spin_until_futer_complete)
def send_request(self, a, b):
self.req.a = a
self.req.b = b
return self.cli.call_async(self.req)
마지막으로 main을 정의한다.
MinimalClientAsync를 생성하고, request를 보내고 result를 기록한다.
def main():
rclpy.init()
minimal_client = MinimalClientAsync()
future = minimal_client.send_request(int(sys.argv[1]), int(sys.argv[2]))
rclpy.spin_until_future_complete(minimal_client, future)
response = future.result()
minimal_client.get_logger().info(
'Result of add_two_ints: for %d + %d = %d' %
(int(sys.argv[1]), int(sys.argv[2]), response.sum))
minimal_client.destroy_node()
rclpy.shutdown()
Add an entry point
setup.py에서 entry_points에 service와 같이 client node를 실행하기 위해 추가한다.
entry_points={
'console_scripts': [
'service = py_srvcli.service_member_function:main',
'client = py_srvcli.client_member_function:main',
],
},
Build and run
dependency 확인을 위해 rosdep을 실행한다.
# ros2_ws
rosdep install -i --from-path src --rosdistro humble -y
package를 build하고, sourcing 후 service node를 실행해보자.
colcon build --packages-select py_srvclisource install/setup.bash
service는 client의 request를 기다리고 있다.
ros2 run py_srvcli service
request를 보내면 2 + 3의 response를 출력한다.
ros2 run py_srvcli client 2 3
ROS2_Humble Documentation : https://docs.ros.org/en/humble/Tutorials/Beginner-Client-Libraries.html