فصل سوم: ارتباطات عملی ARCHO

۳.۱از یک Node ساکت به یک تیم پرسروصدا

در فصل دوم، simple_node فقط یک جمله چاپ کرد و ساکت ماند. این برای یاد گرفتن ساختار خوب بود، اما یک ربات واقعی مثل ARCHO این‌طور کار نمی‌کند. باتری‌اش باید هر چند ثانیه وضعیتش را اعلام کند، کسی باید بتواند از موتور بخواهد Encoderها را صفر کند، و وقتی فرمان «برو به قفسه ۱۲» می‌آید، ربات باید واقعاً حرکت کند و گزارش پیشرفت بدهد. در این فصل، دقیقاً همان چهار ابزار ارتباطی فصل اول — Topic، Service، Action و Parameter — را برای اولین بار به کد واقعی تبدیل می‌کنیم.

۳.۲Publisher و Subscriber واقعی

اول ساده‌ترین حالت را می‌سازیم: یک Node که هر یک ثانیه وضعیت باتری ARCHO را منتشر می‌کند. با پیام استاندارد std_msgs/Float32 شروع می‌کنیم تا بعد در بخش بعد ببینیم چرا در دنیای واقعی این کافی نیست.

# archo_bringup/battery_monitor.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import Float32
import random


class BatteryMonitor(Node):
    def __init__(self):
        super().__init__('battery_monitor')
        self.publisher_ = self.create_publisher(Float32, 'battery_level', 10)
        self.timer = self.create_timer(1.0, self.publish_battery)
        self.level = 100.0

    def publish_battery(self):
        self.level = max(0.0, self.level - random.uniform(0.05, 0.2))
        msg = Float32()
        msg.data = self.level
        self.publisher_.publish(msg)
        self.get_logger().info(f'Battery: {self.level:.1f}%')


def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(BatteryMonitor())
    rclpy.shutdown()


if __name__ == '__main__':
    main()

سه چیز اینجا تازه است نسبت به فصل قبل:

حالا Subscriber را می‌سازیم — یک داشبورد ساده که به همین Topic گوش می‌دهد:

# archo_bringup/dashboard.py
import rclpy
from rclpy.node import Node
from std_msgs.msg import Float32


class Dashboard(Node):
    def __init__(self):
        super().__init__('dashboard')
        self.subscription = self.create_subscription(
            Float32, 'battery_level', self.battery_callback, 10)

    def battery_callback(self, msg):
        if msg.data < 20.0:
            self.get_logger().warning(f'Low battery: {msg.data:.1f}%')
        else:
            self.get_logger().info(f'Dashboard sees: {msg.data:.1f}%')


def main(args=None):
    rclpy.init(args=args)
    rclpy.spin(Dashboard())
    rclpy.shutdown()


if __name__ == '__main__':
    main()

۳.۳ساخت یک Message سفارشی

یک عدد خام مثل Float32 برای باتری کافی نیست. تیم ARCHO می‌خواهد هم‌زمان درصد باتری، ولتاژ، و اینکه آیا در حال شارژ است یا نه را بفرستد. اینجاست که یک Message سفارشی می‌سازیم — دقیقاً همان‌طور که در فصل اول با sensor_msgs/LaserScan آشنا شدیم، اما این‌بار قالب خودمان را طراحی می‌کنیم.

archo_ws/src/
├── archo_interfaces/
│   ├── msg/
│   │   └── BatteryStatus.msg
│   ├── srv/
│   │   └── ResetEncoder.srv
│   ├── action/
│   │   └── MoveToShelf.action
│   ├── CMakeLists.txt
│   └── package.xml
└── archo_bringup/

فایل BatteryStatus.msg ساختار پیام را به زبان ساده تعریف می‌کند:

# archo_interfaces/msg/BatteryStatus.msg
float32 percentage
float32 voltage
bool is_charging

این فایل فقط یک فرم است، نه کد اجرایی. وقتی این Package را Build کنی، ROS 2 به‌طور خودکار کلاس‌های پایتون و C++ متناظر با آن را می‌سازد. بعد از Build، در کد پایتون می‌توانی این‌طور وارد و استفاده‌اش کنی:

from archo_interfaces.msg import BatteryStatus

msg = BatteryStatus()
msg.percentage = 87.5
msg.voltage = 24.1
msg.is_charging = False
self.publisher_.publish(msg)

۳.۴Service در عمل: Reset Encoder

یادت هست در فصل اول گفتیم Service برای کارهایی است که فقط یک‌بار و بر اساس درخواست انجام می‌شوند؟ «صفر کردن Encoder چرخ‌ها» دقیقاً یکی از این کارهاست. اول قالب Service را تعریف می‌کنیم:

# archo_interfaces/srv/ResetEncoder.srv
---
bool success
string message

خط --- Request را از Response جدا می‌کند. اینجا Request خالی است (فقط فراخوانی لازم است، داده‌ای نمی‌فرستیم)، اما Response دو فیلد دارد: آیا موفق بود، و یک پیام توضیحی.

سمت Server

# archo_bringup/motor_controller.py (بخش Service)
import rclpy
from rclpy.node import Node
from archo_interfaces.srv import ResetEncoder


class MotorController(Node):
    def __init__(self):
        super().__init__('motor_controller')
        self.encoder_ticks = 15420
        self.srv = self.create_service(
            ResetEncoder, 'reset_encoder', self.handle_reset)

    def handle_reset(self, request, response):
        self.get_logger().info(f'Resetting encoder from {self.encoder_ticks} ticks')
        self.encoder_ticks = 0
        response.success = True
        response.message = 'Encoder reset to zero'
        return response

سمت Client (از خط فرمان، برای تست سریع)

و همین درخواست از داخل یک Node پایتون دیگر (مثلاً یک ابزار نگهداری):

client = self.create_client(ResetEncoder, 'reset_encoder')
while not client.wait_for_service(timeout_sec=1.0):
    self.get_logger().info('Waiting for reset_encoder service...')

request = ResetEncoder.Request()
future = client.call_async(request)

۳.۵Action در عمل: برو به قفسه

حالا به سراغ کاری می‌رویم که واقعاً طول می‌کشد: فرمان «برو به قفسه ۱۲». قالب Action سه بخش دارد — Goal، Feedback و Result — و همه در یک فایل با همین ترتیب نوشته می‌شوند:

# archo_interfaces/action/MoveToShelf.action
int32 shelf_number
---
bool success
string final_message
---
float32 distance_remaining
string status

سمت Action Server

# archo_bringup/motor_controller.py (بخش Action)
import time
from rclpy.action import ActionServer
from archo_interfaces.action import MoveToShelf


class MotorController(Node):
    def __init__(self):
        super().__init__('motor_controller')
        # ... کد Service قبلی هم اینجا می‌ماند ...
        self._action_server = ActionServer(
            self, MoveToShelf, 'move_to_shelf', self.execute_move)

    def execute_move(self, goal_handle):
        target = goal_handle.request.shelf_number
        self.get_logger().info(f'Moving to shelf {target}')
        distance = 18.0
        feedback = MoveToShelf.Feedback()

        while distance > 0:
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                result = MoveToShelf.Result()
                result.success = False
                result.final_message = 'Cancelled mid-route'
                return result

            distance -= 3.0
            feedback.distance_remaining = max(distance, 0.0)
            feedback.status = 'moving'
            goal_handle.publish_feedback(feedback)
            time.sleep(0.5)

        goal_handle.succeed()
        result = MoveToShelf.Result()
        result.success = True
        result.final_message = f'Arrived at shelf {target}'
        return result

و از خط فرمان، برای تست سریع بدون نوشتن هیچ کد Clientی:

۳.۶Parameter در عمل

حالا سرعت پیش‌فرض حرکت ARCHO را به‌جای نوشتن ثابت در کد، به یک Parameter تبدیل می‌کنیم:

class MotorController(Node):
    def __init__(self):
        super().__init__('motor_controller')
        self.declare_parameter('max_speed', 0.8)
        self.declare_parameter('wheel_radius', 0.05)

    def get_max_speed(self):
        return self.get_parameter('max_speed').get_parameter_value().double_value

و فایل YAML متناظر که در Launch بارگذاری می‌شود:

# config/motor_params.yaml
motor_controller:
  ros__parameters:
    max_speed: 0.8
    wheel_radius: 0.05

۳.۷یک Launch File که همه را روشن می‌کند

حالا سه Node، فایل Parameter و همه‌چیز را با یک فایل Launch به هم وصل می‌کنیم:

# launch/archo_comms.launch.py
import os
from ament_index_python.packages import get_package_share_directory
from launch import LaunchDescription
from launch_ros.actions import Node


def generate_launch_description():
    params_file = os.path.join(
        get_package_share_directory('archo_bringup'),
        'config', 'motor_params.yaml')

    return LaunchDescription([
        Node(
            package='archo_bringup',
            executable='battery_monitor',
        ),
        Node(
            package='archo_bringup',
            executable='dashboard',
        ),
        Node(
            package='archo_bringup',
            executable='motor_controller',
            parameters=[params_file],
        ),
    ])

ros2 launch archo_bringup archo_comms.launch.py

با یک دستور، هر سه Node بالا می‌آیند: battery_monitor شروع به انتشار می‌کند، dashboard گوش می‌دهد، و motor_controller هم Service reset_encoder و هم Action move_to_shelf را با Parameterهای بارگذاری‌شده از YAML آماده می‌کند.

۳.۸جمع‌بندی فصل سوم

از فصل اول تا اینجا یک مسیر کامل طی شد: مفاهیم انتزاعی Node، Topic، Service، Action و Parameter را دیدی؛ در فصل دوم اولین Node ساکت را ساختی؛ و در همین فصل، برای اولین بار سه Node واقعی ARCHO — battery_monitor، dashboard و motor_controller — را نوشتی که واقعاً با هم حرف می‌زنند: یکی منتشر می‌کند، یکی گوش می‌دهد، یکی به درخواست سریع پاسخ می‌دهد و یکی عملیات طولانی را با گزارش زنده اجرا می‌کند.

ارتباط با پروژه اصلی

پروژه ARCHO حالا یک زیرساخت ارتباطی کامل و واقعی دارد. هنوز ربات هیچ بدنه فیزیکی یا چرخی ندارد — این دقیقاً همان چیزی است که در فصل بعد می‌سازیم.

فصل بعد چه چیزی اضافه می‌کند

در فصل چهارم، برای اولین بار به ARCHO یک بدنه واقعی می‌دهیم: شاسی، دو چرخ محرک، یک چرخ هرزگرد، و محل نصب LiDAR و IMU — همه با زبان توصیف ربات ROS 2 به نام URDF، و نسخه هوشمندتر آن، Xacro.

واژه‌نامه فصل سوم

create_publisher / create_subscription

توابعی در کلاس Node برای ساخت Publisher و Subscriber روی یک Topic مشخص.

Timer

مکانیزمی در rclpy که یک تابع را با فاصله زمانی ثابت به‌طور خودکار اجرا می‌کند.

Interfaces Package

Packageای که فقط تعریف Message، Service و Action سفارشی را نگه می‌دارد، بدون منطق اجرایی.

ActionServer / call_async

ابزارهای rclpy برای پیاده‌سازی سمت سرور یک Action و فراخوانی ناهمزمان یک Service.

declare_parameter

تابعی که یک Parameter را با مقدار پیش‌فرض در Node ثبت می‌کند.

DeclareLaunchArgument

ابزار Launch برای گرفتن آرگومان‌های ورودی از خط فرمان و استفاده شرطی از آن‌ها.

خطاهای رایج فصل سوم — جمع‌بندی

• فراموش‌کردن Build مجدد بعد از تغییر فایل .msg/.srv/.action.
• فراخوانی همزمان (Sync) Service که Node را قفل می‌کند به‌جای استفاده از call_async.
• ننوشتن بررسی is_cancel_requested داخل یک Action Server، که باعث می‌شود عملیات هیچ‌وقت واقعاً قابل لغو نباشد.
• هاردکد کردن مقادیر تنظیمی مثل سرعت یا آستانه به‌جای تعریف آن‌ها به‌عنوان Parameter.