서비스의 개념 및 기본 구조

ROS2에서 서비스(Service)는 요청(request)과 응답(response) 방식으로 동작하는 통신 개념이다. 이는 클라이언트-서버 모델을 기반으로 하며, 클라이언트가 요청을 보내고 서버가 해당 요청에 대한 응답을 보내는 형태로 이루어진다. 서비스는 일회성 통신에 적합하며, 특정한 작업을 완료하는 데 자주 사용된다.

다음은 ROS2에서 서비스 서버와 클라이언트를 설정하고, 이를 활용한 간단한 예제 코드이다.

서버 구현 예제

#include "rclcpp/rclcpp.hpp"
#include "example_interfaces/srv/add_two_ints.hpp"

class AddTwoIntsServer : public rclcpp::Node
{
public:
  AddTwoIntsServer() : Node("add_two_ints_server")
  {
    service_ = this->create_service<example_interfaces::srv::AddTwoInts>(
        "add_two_ints", std::bind(&AddTwoIntsServer::handle_add_two_ints, this, _1, _2));
  }

private:
  void handle_add_two_ints(const std::shared_ptr<example_interfaces::srv::AddTwoInts::Request> request,
                           std::shared_ptr<example_interfaces::srv::AddTwoInts::Response> response)
  {
    response->sum = request->a + request->b;
    RCLCPP_INFO(this->get_logger(), "Incoming request\na: %ld b: %ld", request->a, request->b);
  }

  rclcpp::Service<example_interfaces::srv::AddTwoInts>::SharedPtr service_;
};

int main(int argc, char **argv)
{
  rclcpp::init(argc, argv);
  rclcpp::spin(std::make_shared<AddTwoIntsServer>());
  rclcpp::shutdown();
  return 0;
}

위 코드는 두 개의 정수를 더하는 간단한 서비스 서버이다. 클라이언트가 두 개의 정수를 요청으로 보내면 서버가 해당 요청을 처리하고 결과를 반환한다.

액션의 개념 및 기본 구조

액션(Action)은 서비스와 달리 장시간 걸리는 비동기 작업을 처리하는 데 적합한 구조이다. 액션 서버는 클라이언트로부터 목표(goal)를 수신한 후, 이를 처리하면서 중간 결과(feedback)를 제공하고, 최종적으로 완료되었을 때 결과(result)를 반환한다. 액션은 장시간 작업을 잘 지원하며, 작업 도중 피드백을 받을 수 있는 구조적 장점이 있다.

액션의 구조는 크게 세 부분으로 나뉜다:

• Goal: 작업의 목표를 설정하는 부분
• Feedback: 작업이 진행 중일 때 클라이언트에 보내는 중간 결과
• Result: 작업이 완료되었을 때 반환되는 최종 결과

액션 서버와 클라이언트의 상호작용

액션은 다양한 실용적인 예시에서 활용된다. 예를 들어, 로봇이 지정된 목표 지점으로 이동할 때 액션 서버는 로봇이 목표 지점에 도달할 때까지 주기적으로 피드백을 클라이언트로 전송할 수 있다.

다음은 이동 관련 액션 서버 예시이다.

#include "rclcpp/rclcpp.hpp"
#include "example_interfaces/action/move_to_goal.hpp"
#include "rclcpp_action/rclcpp_action.hpp"

class MoveToGoalActionServer : public rclcpp::Node
{
public:
  using MoveToGoal = example_interfaces::action::MoveToGoal;
  using GoalHandleMoveToGoal = rclcpp_action::ServerGoalHandle<MoveToGoal>;

  MoveToGoalActionServer() : Node("move_to_goal_action_server")
  {
    action_server_ = rclcpp_action::create_server<MoveToGoal>(
        this,
        "move_to_goal",
        std::bind(&MoveToGoalActionServer::handle_goal, this, _1, _2),
        std::bind(&MoveToGoalActionServer::handle_cancel, this, _1),
        std::bind(&MoveToGoalActionServer::handle_accepted, this, _1));
  }

private:
  rclcpp_action::GoalResponse handle_goal(const rclcpp_action::GoalUUID &uuid,
                                          std::shared_ptr<const MoveToGoal::Goal> goal)
  {
    RCLCPP_INFO(this->get_logger(), "Received goal request with distance %f", goal->distance);
    return rclcpp_action::GoalResponse::ACCEPT_AND_EXECUTE;
  }

  rclcpp_action::CancelResponse handle_cancel(const std::shared_ptr<GoalHandleMoveToGoal> goal_handle)
  {
    RCLCPP_INFO(this->get_logger(), "Received request to cancel goal");
    return rclcpp_action::CancelResponse::ACCEPT;
  }

  void handle_accepted(const std::shared_ptr<GoalHandleMoveToGoal> goal_handle)
  {
    using namespace std::placeholders;
    std::thread{std::bind(&MoveToGoalActionServer::execute, this, _1), goal_handle}.detach();
  }

  void execute(const std::shared_ptr<GoalHandleMoveToGoal> goal_handle)
  {
    const auto goal = goal_handle->get_goal();
    auto feedback = std::make_shared<MoveToGoal::Feedback>();
    auto result = std::make_shared<MoveToGoal::Result>();

    for (int i = 0; i <= 100; ++i)
    {
      if (goal_handle->is_canceling())
      {
        result->success = false;
        goal_handle->canceled(result);
        RCLCPP_INFO(this->get_logger(), "Goal canceled");
        return;
      }
      feedback->progress = i;
      goal_handle->publish_feedback(feedback);
      RCLCPP_INFO(this->get_logger(), "Progress: %d%%", i);
      std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }

    result->success = true;
    goal_handle->succeed(result);
    RCLCPP_INFO(this->get_logger(), "Goal succeeded");
  }

  rclcpp_action::Server<MoveToGoal>::SharedPtr action_server_;
};

int main(int argc, char **argv)
{
  rclcpp::init(argc, argv);
  rclcpp::spin(std::make_shared<MoveToGoalActionServer>());
  rclcpp::shutdown();
  return 0;
}

이 코드에서는 액션 서버가 목표 지점까지 이동하는 작업을 수행하며, 진행 중에 피드백을 클라이언트로 전송한다.

서비스와 액션의 활용 차이점

서비스와 액션은 서로 다른 통신 요구에 적합하게 설계되었다. 서비스는 짧은 작업에 적합하며, 비동기적 요구는 액션을 통해 해결할 수 있다. 예를 들어, 로봇이 특정 위치에 이동하거나, 이미지를 처리하는 등의 장기 작업에서는 액션이 더 유리한다.

서비스와 액션의 결합

다음은 서비스와 액션을 결합하여 더 복잡한 로봇 시스템을 구현하는 방법에 대한 예이다. 예를 들어, 특정 작업(예: 로봇 팔 움직임)의 성공 여부를 서비스로 처리하고, 장기 작업(예: 로봇의 이동)은 액션을 통해 처리할 수 있다.

서비스 활용 사례

로봇 제어에서의 서비스 호출

로봇 제어 시스템에서 서비스를 이용하여 특정한 작업을 요청하거나 명령을 전달하는 경우가 빈번한다. 예를 들어, 로봇 팔이 특정 위치로 이동해야 하는 상황을 생각해 봅시다. 이때 사용자는 서비스 서버로 목표 위치를 전달하고, 서버는 이 요청을 받아들여 로봇 팔이 해당 위치로 이동할 수 있도록 명령을 실행하게 된다.

서비스 요청 예제

로봇 팔 제어에서, 목표 좌표를 전달하고 이동 명령을 실행하는 서비스의 인터페이스는 다음과 같이 구성될 수 있다.

• 입력: 목표 위치 $\mathbf{p} = [x, y, z]$
• 출력: 성공 여부와 완료 시간 $t$

사용자는 서비스 요청을 통해 로봇 팔이 특정 좌표로 이동할 것을 요구할 수 있다. 이때, 목표 좌표 $\mathbf{p}$는 3차원 벡터로 표현되며, 서비스 호출 시 이를 매개변수로 전달한다.

class MoveRobotArmService:
    def __init__(self):
        self.service = create_service(MoveRobot, 'move_robot_arm', self.handle_move_robot_arm)

    def handle_move_robot_arm(self, request, response):
        target_position = request.position  # 목표 좌표 [x, y, z]
        # 로봇 팔 이동 로직 수행
        response.success = move_to_position(target_position)
        response.completion_time = get_current_time()
        return response

여기서 중요한 점은 서비스가 비동기적으로 동작할 수 있다는 점이다. 즉, 클라이언트는 서비스 요청을 보낸 후 즉시 결과를 기다리는 것이 아니라, 서비스 서버가 응답을 완료할 때까지 다른 작업을 수행할 수 있다.

센서 데이터 요청 서비스

ROS2 시스템에서는 여러 종류의 센서를 통해 데이터를 수집할 수 있다. 센서 데이터는 일반적으로 토픽을 통해 실시간으로 전달되지만, 특정 시점의 데이터를 요청하고자 할 때 서비스가 유용하게 사용될 수 있다.

센서 서비스 요청 예제

특정 센서 데이터, 예를 들어 로봇의 주변 온도를 조회하는 서비스를 고려해 보자. 서비스 요청 시, 현재 온도 $T(t)$를 제공하는 서버는 다음과 같은 방식으로 구성된다.

• 입력: 요청 시간 $t$
• 출력: 해당 시점의 온도 $T(t)$

class TemperatureService:
    def __init__(self):
        self.service = create_service(GetTemperature, 'get_temperature', self.handle_get_temperature)

    def handle_get_temperature(self, request, response):
        current_temperature = read_temperature_sensor()
        response.temperature = current_temperature
        return response

이와 같은 방식으로, 서비스 호출을 통해 특정 시점의 센서 데이터를 손쉽게 요청할 수 있으며, 이러한 접근 방식은 특히 실시간 데이터가 필요하지 않은 경우에 유용하다.

액션 활용 사례

로봇 이동 경로 추종에서의 액션 사용

액션 서버는 긴 시간이 소요되는 작업을 처리할 때 매우 유용하다. 예를 들어, 로봇이 지정된 경로를 따라 이동하는 작업을 생각해 봅시다. 경로를 추종하는 동안 주기적으로 현재 상태를 피드백으로 받아보아야 하며, 경로 이동이 완료되었을 때 완료 신호를 받아야 한다.

경로 추종 액션 예제

로봇의 경로 추종을 위한 액션 서버의 인터페이스는 다음과 같이 구성될 수 있다.

• 목표: 경로 $\mathbf{P} = [\mathbf{p}_1, \mathbf{p}_2, \dots, \mathbf{p}_n]$
• 피드백: 현재 위치 $\mathbf{p}(t)$
• 결과: 경로 완료 여부

경로 추종 중 로봇의 위치 $\mathbf{p}(t)$는 피드백으로 클라이언트에게 주기적으로 전달된다.

class FollowPathActionServer:
    def __init__(self):
        self.action_server = create_action_server(FollowPath, 'follow_path', self.execute_path)

    def execute_path(self, goal_handle):
        path = goal_handle.request.path  # 경로 [p1, p2, ..., pn]
        for waypoint in path:
            # 경로 추종 로직 수행
            move_to_waypoint(waypoint)
            feedback = FollowPath.Feedback()
            feedback.current_position = get_robot_position()
            goal_handle.publish_feedback(feedback)

        goal_handle.succeed()
        result = FollowPath.Result()
        result.success = True
        return result

액션 서버는 목표 경로를 따라 로봇이 이동하는 동안 지속적으로 현재 위치를 피드백으로 제공하며, 경로를 모두 따라간 후 성공 여부를 클라이언트에 알린다.

드론의 착륙 동작에서 액션 사용

드론이나 무인항공기(UAV)의 착륙 과정은 즉시 완료되는 작업이 아니기 때문에, 착륙 동작을 관리하는 액션 서버를 사용하는 것이 효과적이다. 드론이 착륙할 위치에 도달하고 실제로 착륙을 완료하는 과정 동안 실시간 상태 피드백이 필요하다.

착륙 동작 액션 예제

착륙을 위한 액션 서버의 인터페이스는 다음과 같이 구성될 수 있다.

• 목표: 착륙 좌표 $\mathbf{p}_l = [x_l, y_l, z_l]$
• 피드백: 현재 고도 $z(t)$
• 결과: 착륙 완료 여부

드론이 착륙 과정에서 고도 $z(t)$를 주기적으로 피드백으로 제공하며, 안전하게 착륙했을 때 착륙 완료 신호를 전송한다.

class LandDroneActionServer:
    def __init__(self):
        self.action_server = create_action_server(LandDrone, 'land_drone', self.execute_landing)

    def execute_landing(self, goal_handle):
        landing_position = goal_handle.request.position  # 착륙 좌표 [x_l, y_l, z_l]
        current_altitude = get_current_altitude()

        while current_altitude > landing_position[2]:
            # 착륙 로직 수행
            decrease_altitude()
            feedback = LandDrone.Feedback()
            feedback.current_altitude = get_current_altitude()
            goal_handle.publish_feedback(feedback)
            current_altitude = get_current_altitude()

        goal_handle.succeed()
        result = LandDrone.Result()
        result.success = True
        return result

이 예제에서 액션 서버는 착륙 중에 드론의 현재 고도를 클라이언트에게 피드백으로 제공하며, 최종적으로 드론이 착륙 완료 상태에 도달했을 때 성공 신호를 보낸다.

로봇의 비전 기반 객체 인식에서의 액션 사용

로봇이 카메라를 통해 객체를 인식하는 작업은 상당히 시간이 걸리며, 지속적인 피드백이 필요한 경우가 많다. 이때 액션 서버를 사용하면, 객체를 찾는 동안 피드백을 받을 수 있고, 객체 인식이 완료되면 결과를 받을 수 있다.

객체 인식 액션 예제

로봇이 특정 객체를 인식하는 작업을 위한 액션 서버는 다음과 같이 설계될 수 있다.

• 목표: 인식할 객체의 ID $\mathbf{id}$
• 피드백: 현재 인식 중인 객체의 상태
• 결과: 인식 성공 여부 및 객체 위치 $\mathbf{p}_o = [x_o, y_o, z_o]$

class ObjectRecognitionActionServer:
    def __init__(self):
        self.action_server = create_action_server(ObjectRecognition, 'recognize_object', self.execute_recognition)

    def execute_recognition(self, goal_handle):
        object_id = goal_handle.request.object_id  # 인식할 객체 ID
        while not object_recognized(object_id):
            # 객체 인식 로직 수행
            feedback = ObjectRecognition.Feedback()
            feedback.recognition_progress = get_recognition_progress()
            goal_handle.publish_feedback(feedback)

        goal_handle.succeed()
        result = ObjectRecognition.Result()
        result.object_position = get_object_position()
        return result

액션 서버는 객체 인식이 진행되는 동안 인식 상태를 피드백으로 제공하며, 객체 인식이 완료되면 객체의 위치와 함께 성공 여부를 클라이언트에 전달한다.

로봇 암의 조립 작업에서 액션 사용

로봇 암이 여러 부품을 조립하는 작업에서는 각 단계별로 부품을 정확한 위치에 놓아야 하는 경우가 많다. 이러한 작업은 장시간에 걸쳐 수행되며, 각 부품을 놓을 때마다 피드백을 통해 현재 상태를 확인할 수 있다.

조립 작업 액션 예제

조립 작업의 액션 서버는 다음과 같이 구성될 수 있다.

• 목표: 조립할 부품들의 위치 리스트 $\mathbf{P}_a = [\mathbf{p}_1, \mathbf{p}_2, \dots, \mathbf{p}_n]$
• 피드백: 현재 조립 단계 $i$
• 결과: 조립 완료 여부

class AssemblyActionServer:
    def __init__(self):
        self.action_server = create_action_server(Assembly, 'assembly_robot_arm', self.execute_assembly)

    def execute_assembly(self, goal_handle):
        assembly_positions = goal_handle.request.assembly_positions  # 조립할 위치 리스트
        for i, position in enumerate(assembly_positions):
            move_to_position(position)
            feedback = Assembly.Feedback()
            feedback.current_step = i
            goal_handle.publish_feedback(feedback)

        goal_handle.succeed()
        result = Assembly.Result()
        result.success = True
        return result

이 액션 서버는 로봇이 각 부품을 지정된 위치에 놓는 동안 현재 조립 단계에 대한 피드백을 제공하며, 모든 부품이 조립되었을 때 성공 여부를 클라이언트에 알린다.

서비스와 액션의 실제 사례

사례 1: 로봇의 이동과 목표 지점 도달

로봇의 네비게이션에서 자주 사용되는 예는 특정 목표 지점으로 로봇을 이동시키는 작업이다. 이러한 작업에서는 액션 서버가 로봇이 이동할 목표 좌표를 목표(goal)로 받아, 이동하는 동안 피드백을 클라이언트에 제공하며, 도착 시 결과를 반환한다. 이와 동시에 서비스는 더 작은 단위의 작업, 예를 들어 특정 장애물을 우회하기 위한 순간적인 회전이나 직진 같은 작업을 처리하는 데 사용될 수 있다.

액션은 클라이언트에서 비동기적으로 실행되며, 도중에 취소 요청을 보내거나 진행 중인 피드백을 받으면서도 자유롭게 다른 작업을 처리할 수 있다.

사례 2: 로봇 팔의 제어

로봇 팔을 제어할 때, 서비스와 액션은 매우 유용하다. 간단한 작업, 예를 들어 로봇 팔의 조인트를 특정 각도로 회전시키는 작업은 서비스로 쉽게 처리할 수 있다. 그러나 로봇 팔이 복잡한 동작을 수행하거나 연속적인 움직임을 필요로 할 때, 예를 들어 특정 위치에서 물체를 집고 다른 위치로 이동하는 작업에는 액션을 사용하여 장기적인 동작을 비동기적으로 처리하고, 중간에 피드백을 통해 동작 상태를 파악할 수 있다.

이 예에서, 서비스는 주로 개별적이고 짧은 동작을 실행하는 데 적합하며, 액션은 장기적이고 연속적인 작업에 더 적합한다. 서비스와 액션을 동시에 활용함으로써, 로봇 팔의 정교한 동작과 유연한 제어를 실현할 수 있다.

사례 3: 산업 자동화 시스템에서의 활용

산업 자동화 시스템에서는 액션과 서비스가 다양한 방식으로 결합되어 사용된다. 예를 들어, 공장에서 컨베이어 벨트를 제어하는 시스템에서는 다음과 같은 시나리오를 생각할 수 있다:

• 컨베이어 벨트를 특정 위치로 이동시키는 작업은 액션으로 처리된다. 이때 이동 중에 발생하는 상태 정보(예: 현재 위치, 목표 위치에 대한 진행률)는 클라이언트로 피드백되어 시스템이 동작 상태를 실시간으로 모니터링할 수 있다.
• 컨베이어 벨트가 멈추거나, 특정한 간격으로 동작하는 작은 작업들은 서비스를 통해 처리된다. 이러한 서비스는 단일 요청-응답 모델로 설계되어, 즉시 처리되고 응답을 반환한다.

이러한 복잡한 시스템에서는 서비스와 액션을 적절히 결합하여 시스템의 유연성과 반응성을 높이는 것이 중요하다. 다음은 이를 구현하기 위한 통합된 시스템 예이다:

위의 다이어그램은 클라이언트가 서비스 서버와 액션 서버에 각각 요청을 보내는 과정을 보여준다. 서비스는 즉각적으로 응답을 반환하고, 액션은 목표에 도달할 때까지 주기적으로 피드백을 보낸다.

사례 4: 자율 주행 차량의 작업 분할

자율 주행 차량에서도 서비스와 액션의 활용이 매우 중요하다. 예를 들어, 차량이 특정 목적지까지 주행하는 동안에는 액션을 사용하여 이동 상태를 지속적으로 모니터링하고 피드백을 받을 수 있다. 또한, 차량이 주행 중 만나는 짧은 작업들(예: 차선 변경, 장애물 회피 등)은 서비스로 처리된다. 이 과정에서 액션과 서비스는 상호 보완적인 역할을 한다.

액션 피드백과 결과 처리

액션을 통해 피드백을 제공하는 과정에서 클라이언트는 중간에 작업을 취소할 수 있다. 예를 들어, 자율 주행 차량이 주행 중에 목적지 변경 요청을 받았을 경우, 액션을 취소하고 새로운 목표를 설정할 수 있다. 반면, 서비스는 이미 진행 중인 작업을 취소하거나 변경할 수 없으므로, 작업이 완료된 후 새로운 작업을 요청해야 한다.

서비스와 액션은 각각 짧은 일회성 작업과 장시간에 걸친 비동기 작업을 효과적으로 처리하는 데 최적화되어 있다. 이를 통해 다양한 로봇 및 자율 시스템에서 유연한 통신을 구현할 수 있다. 실제 로봇 제어나 복잡한 자동화 시스템에서는 이 두 가지 방법을 적절히 결합하여 사용할 수 있으며, 이러한 통합은 시스템의 성능과 반응성을 높이는 데 중요한 역할을 한다.