Get ros message type at runtime

Question

I'm hoping to have a config file loaded in via rosparam that includes the type for a generic callback to subscribe to. Right now this is what I have but can't figure out how to make it work with the subscriber.

import rospy
from std_msgs.msg import *

def generic_cb(data):
    print(data.data)

if __name__ == '__main__':
    rospy.init_node('generic_node')
    topic_name = rospy.get_param('topic_name')
    topic_type = rospy.get_param('topic_type')
    rospy.Subscriber(topic_name, topic_type, generic_cb)

    rospy.spin()

Answer 1

In Python you can use globals to lookup and return the message type from a string

import rospy
from std_msgs.msg import *

def generic_cb(data):
    print(data.data)

if __name__ == '__main__':
    rospy.init_node('generic_node')

    topic_name = rospy.get_param('topic_name')
    topic_type = rospy.get_param('topic_type')
    
    ros_msg_type = globals()[topic_type]

    rospy.Subscriber(topic_name, ros_msg_type, generic_cb)

    rospy.spin()

Answer 2

Looking into the rostopic API, which dynamically figures out the topic type, you can get the message type of a topic without knowing it beforehand:

import rostopic
import rospy

rospy.init_node('test_node')

TopicType, topic_str, _ = rostopic.get_topic_class('/some/topic')

def callback(msg):
    print(msg)

rospy.Subscriber(topic_str, TopicType, callback)

rospy.spin()