ROS(2) : ROS2에서 Python으로 토픽, 서비스 다루기

ROS2

rclpy(ROS Client Library for Python)의 줄임말, ROS를 파이썬 유저들이 사용할 수 있도록 해주는 라이브러리이다.

Topic Publisher

import rclpy as rp
from geometry_msgs.msg import Twist

rp.init()
test_node = rp.create_node('pub_test') #pub_test라는 노드의 이름을 코드 상에서는 test_node로 지정

현재 상황 : /turtle1/cmd_vel 이라는 토픽을 발행하려고 하며, cmd_vel이라는 토픽의 데이터 타입이 geometry_msgs의 Twist 데이터 타입이다. 따라서 import로 Twist 데이터 타입을 불러온다.

rp.create_node('pub_test') : pub_test라는 이름의 노드를 생성한다.

#데이터 타입 : Twist, 토픽 이름 : /turtle1/cmd_vel
pub = test_node.create_publisher(Twist, '/turtle1/cmd_vel', 10)

'pub_test' 라는 노드가 'Twist'라는 데이터 타입을 가진 '/turtle1/cmd_vel'이라는 토픽을 발행하도록 한다.

msg = Twist()
msg.linear.x = 2.0
pub.publish(msg)

실제로 토픽을 발행한다.

cnt = 0

def timer_callback():
    global cnt
    cnt += 1   

    print(cnt)
    pub.publish(msg)
    
    if cnt>30 :
        raise Exception("publisher stop")

# 일정 주기로 토픽을 발행 한다
timer_period = 0.1
test_node.create_timer(timer_period, timer_callback)
rp.spin(test_node)

callback 함수를 설정하고 일정 주기로 토픽을 발행한다.
rp.spin(test_node) : test_node의 callback 함수의 종료 조건이 실행될 때까지 작업을 실행한다.

test_node.destroy_node()

해당 노드를 삭제한다.

Topic Subscriber

import rclpy as rp
from turtlesim.msg import Pose

rp.init()
test_node = rp.create_node('sub_test') #sub_test라는 노드의 이름을 코드 상에서는 test_node로 지정

현재 상황 : /turtle1/pose 라는 토픽을 구독하려고 하며, pose 라는 토픽의 데이터 타입이 geometry_msgs의 Twist 데이터 타입이 /turtlesim/pose 이다. 따라서 import로 Pose 데이터 타입을 불러온다.

#callback 함수가 받는 data = turtlesim/msg/Pose

count = 0

def callback(data):
    global count
    count += 1
    print("--->")
    print("/turtle1/pose : ", data, "count : ", count)
    print("X : ", data.x)
    print("Y : ", data.y)
    print("Theta : ", data.theta)
    if count > 50:
        count = 0
        raise Exception("Stop Subscription")

#데이터 타입 : Pose, 토픽 이름 : /turtle1/pose
test_node.create_subscription(Pose, '/turtle1/pose', callback, 10)

callback 함수가 받는 data는 turtlesim/msg/Pose이다. test_node.create_subscription()을 통해 test_node가 'Pose'라는 데이터 타입을 가진 '/turtle1/pose' 토픽을 구독하도록 한다.

rp.spin(test_node)

test_node의 callback 함수의 종료 조건이 실행될 때까지 작업을 실행한다.

Service Client

현재 상황 : /turtle1/teleport_absolute 이라는 서비스를 사용하려고 하며, teleport_absolute라는 서비스의 타입이 turtlesim/srv/TeleportAbsolute이다. 따라서 import로 TeleportAbsolute를 불러온다.

import rclpy as rp
from turtlesim.srv import TeleportAbsolute

rp.init()
test_node = rp.create_node('client_test') #client_test라는 노드를 생성하고 이를 test_node라는 이름으로 지정한다

test_node라는 변수는 create_node가 반환한 node라는 class를 받게 된다.

service_name = '/turtle1/teleport_absolute'

#TeleportAbsolute라는 서비스를 요청할 클라이언트 'test_node'를 생성하고, 이를 client라는 변수에 대입
client = test_node.create_client(TeleportAbsolute, service_name)

변수 'client'는 Client라는 클래스를 반환 받는다.

client.call_async(req)
rp.spin_once(test_node)

해당하는 서비스를 불러온다.

참고 자료

• ROS2 혼자 공부하는 로봇 SW
• 오로카 https://cafe.naver.com/openrt?iframe_url=/MyCafeIntro.nhn%3Fclubid=25572101