[Robocup@Home 2026] 2026.01.23 (manipulation task, 서비스 노드와 클라이언트 노드 구현)

Notion 경로: RoboCup Main Page → 리뷰 대응 → 진행 현황 → manipulation task

 

 RB-Y1의 manual mode를 사용하여 특정 동작 수행 자세를 만들고 데이터 (해당 자세에서의 관절 값) 취득하기

 

** 주의할 점 **

• 직접 교시 버튼을 양쪽 동시에 누르면 토르소 움직임을 제어하게 되므로 사용하지 말 것

 

Step 1) RB-Y1 bringup 실행

ros2 launch rby1_bringup bringup.launch.py robot_ip:=192.168.30.1:50051

 

Step 2) RB-Y1 조인트 별 값 확인

ros2 topic echo /joint_states --once

 

Step 3) 만들어둔 자세의 조인트를 아래 함수 형식으로 처리

def _create_bag_picking_pose(self):
""" 가방 집는 자세 (Mode 4)"""
    joint_data = [0.0] * 24
    joint_data[2:8] = [0.0, 0.7853943664153001, -1.5707963267948966, 0.7853981633974483, 0.0, 0.0]
    joint_data[8:15] = [0.0, -0.08726604071281112, 0.0, -2.094388774089615, 0.0, 1.2217313286075795, 0.0]
    joint_data[15:22] = [-1.2063391983147205, 0.2946799875421206, -0.07251856204907886, -0.3033902645901643, -1.1480043769898258, -0.5605050750375041, -0.07613146650276437]
    return joint_data

 

 

현재 태스크 목록

• 양 팔로 빨래통 내려놓고, 빨래통에서 수건 꺼내서 책상 위에 두기
• 잘 펴져있는 수건을 반으로 접기
• 빈 그릇에 시리얼 붓기
• 수세미로 책상 위 얼룩 닦기
• 음료 뚜껑 열기
• 문 열기

 

Task 1) 양 팔로 빨래통 내려놓고, 빨래통에서 수건 꺼내서 책상 위에 두기

• 1. READY 자세에서 시작
• 2. 양 팔 그리퍼 닫기 (양 팔로 빨래통을 잡고 있다고 가정)
• 3. 양 팔을 아래로 내려서 빨래통 내려놓기 (모바일로 RB-Y1을 회전했다고 가정)
• 4. 오른쪽 팔을 빨래통 속에 있는 수건으로 뻗기
• 5. 오른쪽 팔의 그리퍼 닫기 (수건을 잡았다고 가정)
• 6. 오른쪽 팔을 책상 위로 뻗기 (모바일로 RB-Y1이 책상 앞에 있다고 가정 + 수건을 잡고 있다고 가정)
• 7. 오른쪽 팔의 그리퍼 열기 
• 8. READY 자세로 돌아오기

 

Task 2) 잘 펴져있는 수건을 반으로 접기

• 1. READY 자세에서 시작
• 2. 양 팔로 수건

 

ROS 2 서비스 (Service) 개념

 토픽이 데이터를 지속적으로 방송하는 것이라면, 서비스(Service)는 요청(Request)과 응답(Response)의 구조를 가진다.

• Server (Service Node): 요청을 기다렸다가, 요청이 오면 작업을 수행하고 결과를 보낸다.
• Client (Client Node): 요청을 보내고, 결과가 올 때까지 기다린다.

 

1단계: 패키지 생성

 먼저 터미널을 열고 작업 공간 (workspace)의 src 폴더로 이동하여 새로운 패키지 (my_service_pkg)를 만든다. 

newhaneul@newhaneul-950QDB:~/ros2_ws$ ros2 pkg create --build-type ament_python my_service_pkg --dependencies rclpy example_interfaces
going to create a new package
package name: my_service_pkg
destination directory: /home/newhaneul/ros2_ws
package format: 3
version: 0.0.0
description: TODO: Package description
maintainer: ['newhaneul <newhaneul@todo.todo>']
licenses: ['TODO: License declaration']
build type: ament_python
dependencies: ['rclpy', 'example_interfaces']
creating folder ./my_service_pkg
creating ./my_service_pkg/package.xml
creating source folder
creating folder ./my_service_pkg/my_service_pkg
creating ./my_service_pkg/setup.py
creating ./my_service_pkg/setup.cfg
creating folder ./my_service_pkg/resource
creating ./my_service_pkg/resource/my_service_pkg
creating ./my_service_pkg/my_service_pkg/__init__.py
creating folder ./my_service_pkg/test
creating ./my_service_pkg/test/test_copyright.py
creating ./my_service_pkg/test/test_flake8.py
creating ./my_service_pkg/test/test_pep257.py

[WARNING]: Unknown license 'TODO: License declaration'.  This has been set in the package.xml, but no LICENSE file has been created.
It is recommended to use one of the ament license identitifers:
Apache-2.0
BSL-1.0
BSD-2.0
BSD-2-Clause
BSD-3-Clause
GPL-3.0-only
LGPL-3.0-only
MIT
MIT-0

 

2단계: 서비스 노드 (Server) 작성

'my_service_pkg/my_service_pkg' 폴더 안에 service_member_function.py 파일을 생성하고 다음과 같이 작성한다.

newhaneul@newhaneul-950QDB:~/ros2_ws/my_service_pkg/my_service_pkg$ code service_member_function.py

# service_member_function.py

from example_interfaces.srv import AddTwoInts # Import the service definition
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(f'Incoming request \n a: {request.a} b: {request.b}')

        # 응답 반환
        return response
    
def main(args=None):
    rclpy.init(args=args)

    minimal_service = MinimalService()

    rclpy.spin(minimal_service) # 노드가 종료될 때까지 대기

    rclpy.shutdown()

if __name__ == '__main__':
    main()

 

Q1. super().__init__('minimal_service')이 무엇이지?

• 상속(Inheritance)의 핵심이다.
• MinimalService는 Node라는 부모 클래스를 상속받았다.
• super()는 부모 클래스(Node)를 가리킵니다.
• 즉, "부모님(Node)이 가진 초기화 기능(__init__)을 먼저 실행해서, ROS 통신 기능, 파라미터 관리 기능, 로깅 기능 등을 내 로봇에 장착해 줘!"라는 뜻이다.
• 이 줄이 없으면 이 클래스는 껍데기만 노드일 뿐, ROS 2 기능을 전혀 수행하지 못합니다. 여기서 넣어준 'minimal_service'가 노드의 이름이 된다.

Q2. if __name__ == '__main__':을 하는 이유?

• 직접 실행 vs 모듈로 불러오기 구분이다.
• 파이썬 파일은 두 가지 방법으로 쓰인다.
  1. 터미널에서 python3 file.py로 직접 실행.
  2. 다른 파일에서 import file로 불러와서 사용.
• __name__이라는 변수는 직접 실행하면 '__main__'이 되고, 임포트되면 파일 이름이 된다.
• 따라서 이 코드가 있으면 "이 파일을 직접 실행했을 때만 main() 함수를 켜라."라는 뜻이다.
• ROS 2에서 중요한 이유는 setup.py에서 entry_points를 설정할 때, ROS 2는 파일을 임포트해서 main 함수만 쏙 빼서 실행한다. 이 구문이 없으면 임포트하는 순간 의도치 않게 코드가 실행되어 에러가 날 수 있다.

Q3 AddTwoInts라는 클래스를 확인하는 방법?

•  이 클래스는 코드를 실행하기 전(빌드 타임)에 .srv 파일로부터 자동으로 생성되는 파이썬 클래스이다. 내부 구조가 궁금하다면 터미널 명령어로 아래처럼 확인할 수 있다. (--- 위쪽이 Request 구조, 아래쪽이 Response 구조이다.)

newhaneul@newhaneul-950QDB:~/ros2_ws/my_service_pkg/my_service_pkg$ ros2 interface show example_interfaces/srv/AddTwoInts
int64 a
int64 b
---
int64 sum

 

 

 

3단계: 클라이언트 노드 (Client) 작성

 같은 폴더에 client_member_function.py 파일을 생성하고 다음 코드를 작성한다.

# 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

        # 비동기 (Async)로 요청 보내기
        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)
    minimal_client = MinimalClientAsync()

    # 실행 시 입력받은 인자 (숫자 2개)를 전송
    response = minimal_client.send_request(int(sys.argv[1]), int(sys.argv[2]))

    minimal_client.get_logger().info(f'Result of add_two_ints: {minimal_client.req.a} + {minimal_client.req.b} = {response.sum}')

    minimal_client.destroy_node()
    rclpy.shutdown()

if __name__ == "__main__":
    main()

 

self.cli = self.create_client(AddTwoInts, 'add_two_ints')

• server의 이름이 똑같아야 연결이 정상적으로 될 수 있다.

 

self.future = self.cli.call_async(self.req)

• server에게 요청서를 보낸다. 이때 call_async는 "일단 보내놓고, 답장은 나중에 받을게"라는 뜻이다. 리턴값으로 future라는 미래의 결과표를 받는다.

 

rclpy.spin_until_future_complete(self, self.future)

• 아까 받은 미래의 결과표에 실제 답장이 도착해서 완료될 때까지, 로봇(노드)을 계속 가동(spin)시킨다. 답장이 오면 이 줄이 끝나고 다음 줄로 넘어간다.

 

Q1. 비동기(Async)와 동기(Sync)의 차이?

• 동기 (Synchronous): "진동벨 없는 카페"
  • 손님이 주문합니다.
  • 점원은 커피를 만들기 시작합니다.
  • 손님은 커피가 나올 때까지 계산대 앞에서 꼼짝도 못 하고 서 있어야 합니다. (Blocking)
  • 점원도 커피를 다 줄 때까지 다음 손님 주문을 못 받습니다.
  • 장점: 순서가 확실함. 단점: 비효율적임.
• 비동기 (Asynchronous): "진동벨 있는 카페"
  • 손님이 주문합니다. (call_async)
  • 점원은 진동벨(Future)을 줍니다. "나중에 울리면 오세요."
  • 손님은 자리에 가서 핸드폰을 하거나 친구랑 떠듭니다. (Non-blocking, 다른 일 처리 가능)
  • 커피가 다 되면 진동벨이 울립니다. (Future complete)
  • 손님은 그때 커피를 받으러 갑니다.
  • 장점: 기다리는 시간 동안 다른 일을 할 수 있음 (로봇이 센서를 계속 읽거나 움직일 수 있음).

 ROS 2에서 비동기를 쓰는 이유: 로봇에게 "저기까지 가서 물건 가져와"라고 시켰을 때(서비스 요청), 로봇이 가는 데 1분이 걸린다고 칩시다. 동기 방식이면 내 프로그램은 1분 동안 아무것도 못 하고 멈춰버립니다 (Freeze). 비동기 방식을 쓰면, 명령만 보내놓고 로봇은 장애물을 피하거나 배터리를 체크하는 등 다른 일을 계속할 수 있습니다.

 

Q2. spin_until_future_complete가 뭐지?

이 함수는 "반쪽짜리 비동기"를 처리해 주는 도구입니다.

• call_async로 진동벨(Future)은 받았는데, 이 간단한 예제 프로그램에서는 기다리는 동안 딱히 할 다른 일이 없다.
• 그래서 spin_until_future_complete (노드, future)는 "이 진동벨이 울릴 때까지만 (until complete) ROS를 가동 (spin)하면서 여기서 딱 기다려."라고 명령하는 것이다.
• 즉, 비동기 방식으로 요청을 보냈지만, 결과를 받아야만 다음 코드 (결과 출력)를 실행할 수 있으므로, 결과가 올 때까지 강제로 기다리게 만드는 역할을 합니다.

Q3. result()하면 뭐가 출력되는 거지?

 result()는 Response 객체 전체를 반환합니다. AddTwoInts.srv 파일의 구조는 아래와 같다.

int64 a
int64 b
---
int64 sum

 

 여기서 --- 아래쪽이 Response이다. 따라서 future.result()를 하면 sum이라는 변수를 담고 있는 객체가 나온다. 그래서 코드에서 response.sum이라고 해야 비로소 숫자 (합계)를 꺼낼 수 있다. 만약 .srv 파일에 message라는 변수도 있었다면 response.message로 꺼낼 수 있다.

 

Q4. self.cli와 self.srv의 메시지 타입은 모두 같게 설정해야 하는가?

• 비유: 한국인(Client)이 한국어(AddTwoInts)로 말을 걸었는데, 미국인(Server)이 영어(SetBool)로 대답하려고 하면 대화가 안 되는 것 처럼..
• 원리: ROS 2 통신은 데이터를 0과 1(직렬화)로 바꿔서 보냅니다. 받는 쪽에서 이걸 다시 복구하려면 "첫 64비트는 숫자 A고, 다음 64비트는 숫자 B야"라는 설계도(타입)가 양쪽 다 똑같이 있어야 한다.
• 타입이 다르면 아예 연결조차 되지 않거나 에러가 발생하게 된다.

 

4단계: setup.py 설정

 패키지 루트의 setup.py 파일을 열어 entry_points 부분을 수정해야 ROS 2가 실행 가능한 노드를 찾을 수 있다.

entry_points={
        'console_scripts': [
            'service = my_service_pkg.service_member_function:main',
            'client = my_service_pkg.client_member_function:main',
        ],
    },

 

5단계: 빌드 및 실행

 작업 공간의 루트 (~/ros2_ws)로 돌아가서 빌드한다.

cd ~/ros2_ws
colcon build --packages-select my_service_pkg
source install/setup.bash

newhaneul@newhaneul-950QDB:~/ros2_ws$ colcon build --symlink-install
Starting >>> my_service_pkg
Starting >>> py_pubsub
Finished <<< my_service_pkg [1.02s]                                         
Finished <<< py_pubsub [1.02s]

Summary: 2 packages finished [1.27s]
newhaneul@newhaneul-950QDB:~/ros2_ws$ source install/setup.bash

 

터미널 1 (서비스 실행):

newhaneul@newhaneul-950QDB:~/ros2_ws$ ros2 run my_service_pkg service
[INFO] [1769174789.622063416] [minimal_service]: Incoming request 
 a: 2 b: 3
[INFO] [1769174795.541533386] [minimal_service]: Incoming request 
 a: 4 b: 5

 

터미널 2 (클라이언트 실행):

newhaneul@newhaneul-950QDB:~/ros2_ws$ ros2 run my_service_pkg client 2 3
[INFO] [1769174789.628932867] [minimal_client_async]: Result of add_two_ints: 2 + 3 = 5
newhaneul@newhaneul-950QDB:~/ros2_ws$ ros2 run my_service_pkg client 4 5
[INFO] [1769174795.548268699] [minimal_client_async]: Result of add_two_ints: 4 + 5 = 9