무작정 공부하기 - 혼자 공부하는 로봇 SW 따라하기 (4)

썸네일용

Ch4. Python으로 ROS2 토픽 다루기

이번 글에서는 위 책에서 Ch4에서의 Python으로 ROS2 토픽을 다루는 법에 대해서 작성하고자 한다.

VScode를 이용해서 컴퓨터의 메모리를 바로 이용하여 할 수도 있지만, 책에서 나오는 방법처럼 jupyter notebook을 이용하여 진행해보고자 한다.

 

1. 설치 및 준비

터미널에서 다음 명령어를 이용해서 pip를 설치하고, 업데이트를 진행한다.

sudo apt install python3-pip
pip3 install --upgrade pip

글을 쓰는 시점의 최신 버전으로 업데이트가 되는 것을 볼 수 있고, 그 이후 다음 명령어를 이용하여 pip를 이용하여 jupyter, ipywidegets, pyyaml, bqplot을 설치하고자 한다.

pip3 install jupyter ipywidgets pyyaml bqplot

설치가 끝나면 재부팅후 다음 명령어를 이용하여 웹브라우저를 실행시킨다.

jupyter notebook

혹시라도 에러가 난다면 다음 명령어를 실행한 뒤 재부팅을 한 후 다시 시도하면 된다.

sudo apt install jupyter-core

 

2. Python으로 토픽 구독하기

먼저 home에 python 폴더를 만들고 거기서 humble을 실행한 뒤 jupyter notebook을 실행한다.

이후 python 폴더에서 turtlesim 패키지의 turtlesim_node를 실행한다.

이후 Jupyter에서 Subscriber Test라는 ipynb 파일을 만든 뒤 코딩을 시작한다.

가장 먼저 모듈을 import 하기 위해서 다음과 같은 명령어를 입력한다.

import rclpy as rp
from turtlesim.msg import Pose

가장 먼저 alias를  이용해 rclpy 모듈을 불러왔으며, turtlesim_node의 토픽 중 /turtle1/pose라는 토픽을 구독하기 위해서 turtlesim/msg/Pose 라는 데이터 타입을 이용함을 알고, 이를 이용해 다음과 같은 명령어를 이용해 Pose라는 클래스를 이용한다.

 

이후 rclpy를 초기화 하기 위해서 다음 코드를 실행 후, rcply의 create_node를 이용해 sub_test라는 이름의 노드를 만든다.

rp.init()
test_node = rp.create_node('sub_test')

그 작업까지 한다면 ros2 node list를 통해서 노드를 확인하면 /sub_test와 /turtlesim 이렇게 2개가 정상적으로 잡히는 것을 볼 수 있다.

※ 지금 작성중인 코드 ※

 : 간단히 토픽을 구독하는 python 코드

 > 어떤 노드가 토픽을 발행하면 python으로 노드를 구독하는 subscriber가 동작함.

 >> callback 함수

 

def callback(data):
    print("--->")
    print("/turtle1/pose : ", data)
    print("X : ", data.x)
    print("Y : ", data.y)
    print("Theta : ", data.theta)

원래는 log에  기록을 남기면 좋지만, 간단한 공부이므로 print로 진행하였다.

 

이후 토픽 subscriber를 만들기 위해서 다음 명령을 입력한다.

test_node.create_subscription(Pose, '/turtle1/pose', callback, 10)
#test_node.create_subscription(<data_type>, '<topic_name>', <callback>, <QoS History>)

이때 호출은 바로 밑 줄과 같은 방식으로 진행하면 된다.

여기서 QoS History는 간단하게 저장할 샘플의 수라고 생각하면 된다.

rqt_graph를 실행한 모습은 다음과 같다.

 

이후 다음의 명령을 이용해 토픽을 구독할 수 있다.

rp.spin_once(test_node)	# 한 번만 구독
rp.spin(test_node)		# 무한히 구독

 

 - 제한된 횟수만큼 토픽을 받는 함수 만들기

import rclpy as rp
from turtlesim.msg import Pose

rp.init()
test_node = rp.create_node('sub_test')

cnt = 0
def callback(data):
    global cnt
    cnt += 1
    print(">", cnt, " -> X : ", data.x, " -> Y : ", data.y, " -> Theta : ", data.theta)
    if cnt > 3:
        raise Exception("Subscription Stop")
        
test_node.create_subscription(Pose, '/turtle1/pose', callback, 10)

rp.spin(test_node)

이를 통해 제대로 구독을 실행함을 알 수 있다.

 

 

3. Python으로 토픽 발행하기

Publisher Test.ipynb 파일을 만든 뒤 다음과 같은 코드를 작성한다.

import rclpy as rp
from geometry_msgs.msg import Twist

rp.init()
test_node = rp.create_node('pub_test')

msg = Twist()
print(msg)

msg.linear.x = 2.0
print(msg)

pub = test_node.create_publisher(Twist, '/turtle1/cmd_vel', 10)
pub.publish(msg)

msg.linear.x = 2.0
msg.angular.z = 2.0
pub.publish(msg)

이후 일정 주기로 토픽을 발행하고자 하는데, 이럴 때는 일정 시간을 유지해야 하므로 timer 기능을 활용해야 함.

이는 create_timer 를 이용하며 됨.

cnt = 0

def timer_callback():
    global cnt
    
    cnt += 1
    
    print(cnt)
    pub.publish(msg)
    
    if cnt > 3:
        raise Exception("Publisher Stop")
        
timer_period = 0.1
timer = test_node.create_timer(timer_period, timer_callback)
rp.spin(test_node)

이후 확실히 해당 노드를 종료하기 위해서 다음 코드를 실행하면 된다.

test_node.destroy_node()

 

다음에는 Ch5. Python으로 서비스 클라이언트 다루기 에 대해서 써보고자 한다.

Subscriber Test.ipynb

0.01MB

Publisher Test.ipynb

0.01MB

혹시 몰라 파일을 남긴다.(추후 깃헙에 다시 정리할 예정이다.)