Utility function that gets a single message from a ROS topic

get_single_msg.py

import threading

import rospy


def get_single_msg(topic, type_, timeout=None):
    """ Utility function that gets a single message from a ROS topic.
    
    Args:
        topic: ROS topic (string).
        type_: topic message type.
        timeout: max waiting time (None or float in seconds).
    Returns:
        instance of first received message or None on timeout.
    """
    msgs = []
    received = threading.Event()

    def cb(msg):
        msgs.append(msg)
        received.set()

    sub = rospy.Subscriber(topic, type_, cb, queue_size=1)
    no_timeout = received.wait(timeout)
    sub.unregister()
    if no_timeout:
        return msgs[0]