Commands Action Client

This module implements a ROS 2 Action Client for drone command management, featuring: - Multi-drone command handling - Asynchronous goal management - Feedback and result processing - Thread-safe queue integration - Concurrent logging

Classes

CommandsActionClient: ROS 2 Node managing action client communication with drones

Functions

create_ros2_action_client: Factory function for initializing ROS 2 components
main: Main execution entry point
create_logger: Configure concurrent-safe logging

Usage

Start action client:

from ros2_action_client import create_ros2_action_client

ros2_queue = queue.Queue()
logger = create_logger('ROS2', 'ros2.log')
client = create_ros2_action_client(ros2_queue, logger)
client.send_goal(1, ["takeoff", "scan_area"])

class commands_action_client.CommandsActionClient(*args: Any, **kwargs: Any)[source]

Bases: Node

ROS 2 Action Client node for drone command management.

Manages bidirectional communication with drone action servers including: - Goal submission tracking - Feedback processing - Result handling - Connection management

Parameters:

ros2_action_queue (queue.Queue) – Thread-safe queue for action status updates
ros2_logger (logging.Logger) – Configured logger instance for ROS operations

Variables:

_action_clients (dict[int, ActionClient]) – Active drone action clients (drone_id → ActionClient)

__init__(ros2_action_queue, ros2_logger)[source]: Initialize action client node with communication infrastructure.

send_goal(drone_id, commands_list)[source]

Submit command sequence to target drone action server.

Parameters:

drone_id (int) – Unique drone identifier
commands_list (list[str]) – Ordered command sequence for execution

Raises:

RuntimeError – If action server unavailable after timeout

Note

Creates new ActionClient for unrecognized drone IDs Maintains 2-second server connection timeout

goal_response_callback(future, drone_id)[source]

Handle action server goal acceptance/rejection.

Parameters:

future (rclpy.task.Future) – Pending goal submission future
drone_id (int) – Target drone identifier

get_result_callback(future, drone_id)[source]

Process final command execution results.

Parameters:

future (rclpy.task.Future) – Completed result future
drone_id (int) – Originating drone ID

feedback_callback(feedback_msg, drone_id)[source]

Handle intermediate command execution updates.

Parameters:

feedback_msg (Commands.FeedbackMessage) – ROS 2 feedback message
drone_id (int) – Source drone identifier

commands_action_client.create_ros2_action_client(ros2_action_queue, ros2_logger)[source]

Initialize and configure ROS 2 action client infrastructure.

Parameters:

ros2_action_queue (queue.Queue) – Shared queue for action status updates
ros2_logger (logging.Logger) – Configured logger instance

Returns:

Initialized action client node

Return type:

CommandsActionClient

Note

Starts independent executor thread for ROS 2 node Sets up MultiThreadedExecutor for concurrent callbacks

commands_action_client.main(args=None)[source]

Main execution entry point for standalone operation.

Parameters:: args (list | None) – ROS 2 initialization arguments

Example

Basic usage:

$ python3 ros2_action_client.py
[INFO][ROS2] - Sending commands to drone 1
[INFO][ROS2] - Sending commands to drone 2

commands_action_client.create_logger(logger_name, log_file)[source]

Configure concurrent-safe logging with rotation.

Parameters:

logger_name (str) – Namespace for logger instance
log_file (str) – Path to primary log file

Returns:

Configured logger instance

Return type:

logging.Logger

Configuration: - 1MB log rotation with 5 backups - Concurrent write safety - Unified console/file formatting