Subscriber Member Function

Dynamic Drone State Vector Subscriber Module

Note

Features: - Automatic discovery of new drone topics - Thread-safe queue integration for state vector data - Regular expression-based topic matching - Multi-threaded ROS 2 executor

class subscriber_member_function.DynamicDroneSubscriber(*args: Any, **kwargs: Any)

Bases: Node

Dynamic subscriber for drone state vector topics.

Parameters:

• receive_sv_queue – Thread-safe queue for state vector data

• sv_logger – Configured logger instance

__init__(receive_sv_queue, sv_logger)

Initialize dynamic subscriber node.

scan_for_drones()

Discover and subscribe to new drone topics.

Note

Scans all ROS 2 topics every SCAN_INTERVAL seconds Matches using predefined TOPIC_PATTERN

extract_drone_id(topic)

Extract numeric drone ID from topic name.

Returns:

Drone ID if valid format, None otherwise

create_drone_subscription(topic, drone_id)

Create subscription for a new drone topic.

create_callback(drone_id)

Generate state vector callback with drone context.

Parameters:

drone_id – Numeric drone identifier

Returns:

Configured callback function

subscriber_member_function.main(args=None, receive_sv_queue=None, sv_logger: Logger = None)

Initialize and run dynamic drone subscriber.

Parameters:

• args – ROS 2 command line arguments

• receive_sv_queue – Thread-safe queue for state vectors

• sv_logger – Configured logger instance

Returns:

Initialized node instance or None on failure

Raises:

ROSError – If ROS 2 initialization fails

Example

from queue import Queue
import logging

sv_queue = Queue()
logger = logging.getLogger('SVSubscriber')
node = main(receive_sv_queue=sv_queue, sv_logger=logger)