Chapter 3: Practical Communication with ARCHO

3.1From a Quiet Node to a Chatty Team

In Chapter 2, simple_node just printed a single sentence and stayed quiet. That was fine for learning the basic structure, but a real robot like ARCHO doesn't work that way. Its battery needs to announce its status every few seconds, someone needs to be able to ask the motor to zero the encoders, and when the command "go to shelf 12" arrives, the robot actually needs to move and report progress. In this chapter, we turn exactly the same four communication tools from Chapter 1 — Topic, Service, Action, and Parameter — into real code for the first time.

3.2A Real Publisher and Subscriber

First we'll build the simplest case: a node that publishes ARCHO's battery status every second. We'll start with the standard std_msgs/Float32 message so that in the next section we can see why this isn't enough in the real world.

# 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()

Three things here are new compared to the previous chapter:

Now let's build the subscriber — a simple dashboard that listens on the same 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()

3.3Building a Custom Message

A raw number like Float32 isn't enough for the battery. The ARCHO team wants to send the battery percentage, voltage, and whether it's charging, all at once. This is where we build a custom message — exactly as we became familiar with sensor_msgs/LaserScan in Chapter 1, but this time we design our own format.

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

The BatteryStatus.msg file defines the message structure in plain form:

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

This file is just a form, not executable code. When you build this package, ROS 2 automatically generates the corresponding Python and C++ classes. After building, you can import and use it in Python code like this:

from archo_interfaces.msg import BatteryStatus

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

3.4Service in Practice: Reset Encoder

Remember in Chapter 1 we said a service is for tasks that happen just once and on request? "Zeroing the wheel encoders" is exactly one of those tasks. First we define the service format:

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

The --- line separates the request from the response. Here the request is empty (only the call itself is needed, we don't send any data), but the response has two fields: whether it succeeded, and an explanatory message.

Server Side

# archo_bringup/motor_controller.py (Service section)
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 Side (from the command line, for quick testing)

And the same request from within another Python node (for example, a maintenance tool):

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)

3.5Action in Practice: Go to Shelf

Now let's tackle something that actually takes time: the command "go to shelf 12." The action format has three parts — Goal, Feedback, and Result — and they're all written in one file, in that order:

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

Action Server Side

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


class MotorController(Node):
    def __init__(self):
        super().__init__('motor_controller')
        # ... the previous Service code also stays here ...
        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

And from the command line, for quick testing without writing any client code:

3.6Parameters in Practice

Now let's turn ARCHO's default movement speed into a parameter instead of writing it as a constant in the code:

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

And the corresponding YAML file, loaded at launch:

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

3.7A Launch File That Starts Everything

Now let's connect the three nodes, the parameter file, and everything else with a single launch file:

# 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

With one command, all three nodes above come up: battery_monitor starts publishing, dashboard starts listening, and motor_controller prepares both the reset_encoder service and the move_to_shelf action, with parameters loaded from the YAML file.

3.8Chapter 3 Summary

From Chapter 1 to here, we've covered a complete journey: you saw the abstract concepts of Node, Topic, Service, Action, and Parameter; in Chapter 2 you built your first quiet node; and in this chapter, for the first time, you wrote three real ARCHO nodes — battery_monitor, dashboard, and motor_controller — that actually talk to each other: one publishes, one listens, one responds to quick requests, and one runs a long-running operation with live progress reporting.

Connection to the Main Project

ARCHO Project now has a complete, real communication infrastructure. The robot still has no physical body or wheels — that's exactly what we build in the next chapter.

What the Next Chapter Adds

In Chapter 4, we give ARCHO a real body for the first time: a chassis, two drive wheels, a caster wheel, and mounting points for a LiDAR and an IMU — all using ROS 2's robot description language, URDF, and its smarter version, Xacro.

Chapter 3 Glossary

create_publisher / create_subscription

Functions in the Node class for creating a publisher and subscriber on a given topic.

Timer

A mechanism in rclpy that automatically runs a function at a fixed time interval.

Interfaces Package

A package that holds only custom message, service, and action definitions, without executable logic.

ActionServer / call_async

rclpy tools for implementing the server side of an action and for calling a service asynchronously.

declare_parameter

A function that registers a parameter with a default value on a node.

DeclareLaunchArgument

A launch tool for accepting command-line input arguments and using them conditionally.

Chapter 3 Common Mistakes — Summary

• Forgetting to rebuild after changing a .msg/.srv/.action file.
• Making a synchronous service call that locks up the node, instead of using call_async.
• Not checking is_cancel_requested inside an action server, which means the operation can never actually be cancelled.
• Hardcoding configuration values like speed or thresholds instead of defining them as parameters.