Ubuntu ROS 실습(34)

zeusqoi·5일 전
ubuntu

개인공부 - ubuntu

목록 보기
53/57

이번 실습에서는 ROS 2의 Action 통신을 Python으로 구현하는 방법을 실습했다.
checker 노드가 Action Client로 목표 합계를 전달하고, calculator 노드가 Action Server로 연산을 수행하면서 중간 Feedback과 최종 Result를 전달하는 과정을 확인했다.

1. Action Client / Action Server

ROS 2의 Action은 Goal, Feedback, Result를 주고받는 방식으로 동작한다.

이번 실습에서는 다음과 같은 구조로 구성되어 있다.

  • checker → Action Client
  • calculator → Action Server
  • Action 이름 → arithmetic_checker
  • Action 타입 → ArithmeticChecker

checker가 목표 합계값을 Goal로 전달하면 calculator는 연산 결과를 누적하면서 중간 계산식을 Feedback으로 전달한다. 목표 합계에 도달하면 최종 계산식과 합계를 Result로 반환한다. 

checker
(Action Client)
      │
      │ Goal
      ▼
arithmetic_checker
      │
      ▼
calculator
(Action Server)
      │
      ├── Feedback
      │
      └── Result

2. Action Server 코드 확인

먼저 calculator.py에서 Action Server가 어떻게 생성되어 있는지 확인했다.

grep -n -A7 "ActionServer" ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/calculator/calculator.py

확인한 코드는 다음과 같다.

self.arithmetic_action_server = ActionServer(
    self,
    ArithmeticChecker,
    'arithmetic_checker',
    self.execute_checker,
    callback_group=self.callback_group)

여기서

  • ArithmeticChecker → Action 타입
  • arithmetic_checker → Action 이름
  • execute_checker → Goal을 전달받았을 때 실행되는 콜백 함수
  • callback_group → 콜백 그룹

으로 사용된다.

3. execute_checker() 확인

다음으로 Action Goal을 전달받은 후 실제로 어떤 작업을 수행하는지 확인했다.

sed -n '105,135p' ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/calculator/calculator.py

주요 코드는 다음과 같다.

def execute_checker(self, goal_handle):
    self.get_logger().info('Execute arithmetic_checker action!')
    feedback_msg = ArithmeticChecker.Feedback()
    feedback_msg.formula = []
    total_sum = 0.0
    goal_sum = goal_handle.request.goal_sum

    while total_sum < goal_sum:
        total_sum += self.argument_result
        feedback_msg.formula.append(self.argument_formula)
        self.get_logger().info(
            'Feedback: {0}'.format(feedback_msg.formula))
        goal_handle.publish_feedback(feedback_msg)
        time.sleep(1)

    goal_handle.succeed()

    result = ArithmeticChecker.Result()
    result.all_formula = feedback_msg.formula
    result.total_sum = total_sum

    return result

goal_sum을 목표값으로 받아 total_sum이 목표값에 도달할 때까지 연산 결과를 누적한다.

이 과정에서 계산식을 feedback_msg.formula에 저장하고 publish_feedback()을 이용하여 Action Client에 Feedback을 전달한다.

목표값을 넘으면 goal_handle.succeed()를 실행하고 전체 계산식과 최종 합계를 Result에 저장하여 반환한다.

4. Action Server 실행 코드 확인

calculator 노드의 실행 코드도 확인했다.

sed -n '1,80p' ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/calculator/main.py

calculator는 MultiThreadedExecutor를 이용하여 실행된다.

calculator = Calculator()
executor = MultiThreadedExecutor(num_threads=4)
executor.add_node(calculator)
executor.spin()

또한 종료할 때 Action Server를 별도로 소멸시키도록 구성되어 있다.

calculator.arithmetic_action_server.destroy()
calculator.destroy_node()

calculator는 기존의 Topic Subscriber와 Service Server에 더해 Action Server 역할까지 수행하는 복합 기능의 노드이다.

5. Action Client 코드 확인

다음으로 checker.py의 Action Client 코드를 확인했다.

sed -n '1,140p' ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/checker/checker.py

Checker 클래스는 Node를 상속하고 checker라는 이름으로 생성된다.

class Checker(Node):

    def __init__(self):
        super().__init__('checker')
        self.arithmetic_action_client = ActionClient(
            self,
            ArithmeticChecker,
            'arithmetic_checker')

Action Server와 동일하게 ArithmeticChecker와 arithmetic_checker를 사용한다.

6. Action Goal 전송

send_goal_total_sum() 함수에서는 Action Server에 Goal을 전달한다.

goal_msg = ArithmeticChecker.Goal()
goal_msg.goal_sum = (float)(goal_sum)

self.send_goal_future = self.arithmetic_action_client.send_goal_async(
    goal_msg,
    feedback_callback=self.get_arithmetic_action_feedback)

self.send_goal_future.add_done_callback(
    self.get_arithmetic_action_goal)

먼저 ArithmeticChecker.Goal()을 이용하여 Goal을 생성하고 goal_sum에 목표 합계값을 저장한다.

그 후 send_goal_async()를 이용하여 Goal을 전달하면서 Feedback을 받을 콜백 함수도 지정한다.

7. Action Feedback 확인

Action Server에서 Feedback이 전달되면 다음 함수가 실행된다.

def get_arithmetic_action_feedback(self, feedback_msg):
    action_feedback = feedback_msg.feedback.formula
    self.get_logger().info(
        'Action feedback: {0}'.format(action_feedback))

Action Server가 계산식을 Feedback으로 전달하면 checker 터미널에서 해당 계산식을 확인할 수 있다.

8. Action Result 확인

최종 결과는 get_arithmetic_action_result()에서 처리한다.

def get_arithmetic_action_result(self, future):
    action_status = future.result().status
    action_result = future.result().result

    if action_status == GoalStatus.STATUS_SUCCEEDED:
        self.get_logger().info('Action succeeded!')
        self.get_logger().info(
            'Action result(all formula): {0}'.format(
                action_result.all_formula))
        self.get_logger().info(
            'Action result(total sum): {0}'.format(
                action_result.total_sum))

Action 상태가 STATUS_SUCCEEDED이면 전체 계산식과 최종 합계를 출력한다.

9. Action Client 실행 코드 확인

마지막으로 checker/main.py를 확인했다.

sed -n '1,120p' ~/robot_ws/src/ros2-seminar-examples/topic_service_action_rclpy_example/topic_service_action_rclpy_example/checker/main.py

실행 인자에서 목표 합계값을 설정할 수 있도록 되어 있었다.

parser.add_argument(
    '-g',
    '--goal_total_sum',
    type=int,
    default=50,
    help='Target goal value of total sum')

따라서 별도의 값을 입력하지 않으면 목표값은 50으로 설정된다.

-g 옵션을 이용하면 원하는 목표값을 지정할 수 있다.

ros2 run topic_service_action_rclpy_example checker -g 100

PPT에서도 기본 목표값을 50으로 설정하고 -g 100과 같이 실행 인자를 이용하여 목표값을 변경할 수 있도록 설명하고 있다.

10. Calculator 실행

이제 실제 Action 통신을 확인하기 위해 calculator를 실행했다.

source ~/robot_ws/install/setup.zsh
ros2 run topic_service_action_rclpy_example calculator

실행 후에는 별도의 오류나 출력 없이 Action Server가 요청을 기다리는 상태가 되었다.

11. Checker 실행

새로운 터미널에서 checker를 실행했다.

source ~/robot_ws/install/setup.zsh
ros2 run topic_service_action_rclpy_example checker

처음에는 다음과 같이 Action Goal이 정상적으로 받아들여진 것을 확인했다.

[checker]: Action goal accepted.

이후 Action Server에서 Feedback을 계속 전달했다.

[checker]: Action feedback: [...]
[checker]: Action feedback: [...]
[checker]: Action feedback: [...]

12. Argument 실행

Action에서 사용할 계산값을 만들기 위해 새로운 터미널에서 argument도 실행했다.

source ~/robot_ws/install/setup.zsh
ros2 run topic_service_action_rclpy_example argument

argument 노드에서 a, b 값이 계속 생성되었고 calculator에서는 다음과 같이 값을 받아오는 것을 확인했다.

[calculator]: Subscribed argument a: 4.0
[calculator]: Subscribed argument b: 4.0

13. Operator 실행

다음으로 operator를 실행하여 계산 결과를 생성했다.

source ~/robot_ws/install/setup.zsh
ros2 run topic_service_action_rclpy_example operator

실행 결과 다음과 같이 Service를 통한 계산 결과가 출력되는 것을 확인했다.

[operator]: Result: 5.0
[operator]: Result: 1.0
[operator]: Result: 0.0
[operator]: Result: 17.0
[operator]: Result: 8.0
[operator]: Result: 25.0

이렇게 생성된 연산 결과를 calculator의 Action Server가 누적하여 목표 합계와 비교한다.

14. Action 최종 결과 확인

checker 터미널에서 최종 결과를 확인했다.

[checker]: Action succeeded!

그리고 전체 계산식과 최종 합계가 출력되었다.

[checker]: Action result(all formula): [...]
[checker]: Action result(total sum): 62.0

이번 실습에서는 기본 목표값인 50을 설정했으며, 연산 결과를 누적한 최종 합계가 62.0이 되어 Action이 성공적으로 종료되는 것을 확인했다.

PPT에서 설명한 것처럼 목표 합계에 도달할 때까지 Feedback을 전달하고, 목표를 넘으면 최종 Result를 반환하는 과정을 실제로 확인할 수 있었다.

마무리

이번 실습에서는 ROS 2의 Action Client와 Action Server를 Python으로 구현하고 실제 통신 과정을 확인했다.

checker
(Action Client)
      │
      │ Goal
      ▼
arithmetic_checker
      │
      ▼
calculator
(Action Server)
      │
      ├── Feedback
      │
      └── Result
      │
      ▼
checker

checker가 목표 합계값을 전달하면 calculator가 연산 결과를 계속 누적하고, 중간 계산식을 Feedback으로 전달했다.

목표값에 도달한 이후에는 Action succeeded!가 출력되고 최종 결과로 62.0을 확인했다.

Topic은 지속적인 데이터 전달, Service는 요청과 응답 방식으로 통신하는 반면, Action은 목표를 전달하고 작업이 진행되는 동안 Feedback을 받으며 최종 Result를 받는 방식으로 동작한다.

이번 실습을 통해 Topic → Service → Action으로 이어지는 ROS 2의 주요 통신 방식을 Python으로 직접 실행해 볼 수 있었다.

profile
zeusqoi
이전 포스트

Ubuntu ROS 실습(33)

다음 포스트

Ubuntu ROS 실습(35)

0개의 댓글