一、前言：为什么需要 ROS Bridge？

二、基础类详解

1.Callbacks类

2.TopicCallbacks(utils.Callbacks)类

3.ROSPeerChangeCallbacks(utils.Callbacks)类

4.ProjectAirSimTopicsManager类

5.ROSTopicsManager类

5.1 ROS1Node

5.1.1 嵌套的Publisher类

5.1.2 嵌套SensorHelper类

5.1.3 嵌套Subscriber类

5.1.4 ROS1Node

5.2 ROSTopicsManager

6.TopicsManagers类

7.AutoSubscriber类

8.BasicROSSubscriber

三、功能集成类详解

1.BasicBridgeToROS类

2.BasicBridgeFromROS(BasicROSSubscriber)

四、gps话题订阅示例

五、相关链接

一、前言：为什么需要 ROS Bridge？

在自动驾驶与机器人研究领域，Project AirSim 凭借其基于虚幻引擎（Unreal Engine）的高保真渲染和强大的物理引擎，成为了无人机及地面车辆仿真的首选平台。然而，AirSim 本身是一套独立的仿真系统，其原生的通信接口（基于 C++ 或 Python 的 RPC 调用）虽然功能强大，但在机器人工程化应用中存在一定的“生态孤岛”效应。ROS Bridge 的出现，正是为了打破这种隔阂，构建起高保真虚拟世界与标准化机器人开发范式之间的桥梁。

二、基础类详解

1.Callbacks类

__init__：初始化列表self.callbacks = []；

__bool__：判断是否有注册的回调函数，如果存在，则返回True；

__call__(*args, **kwargs)：接受任意参数，依次调用列表self.callbacks中所有的回调函数；

 __len__：返回列表中回调函数的数量；

add(回调函数)：向列表中添加回调函数，如果已经存在，则添加失败，返回False；

remove(回调函数)：删除对应的回调函数。

2.TopicCallbacks(utils.Callbacks)类

继承Callbacks类，并重写__call__函数。

__call__(topic_name, message)：依次调用回调函数列表self.callbacks中的所有回调函数，并传入topic_name,、message两个参数。

3.ROSPeerChangeCallbacks(utils.Callbacks)类

继承Callbacks类，并重写__call__函数。

__call__(topic_name, num_peers)：依次调用回调函数列表self.callbacks中的所有回调函数，并传入topic_name,、num_peers两个参数。其中，num_peers表示外部订阅者新的数量。

4.ProjectAirSimTopicsManager类

该类的作用地位等同于ProjectAirSim Client，使用ProjectAirSim Client的函数与ProjectAirSim Server(或UE)交互，即向ProjectAirSim Server(或UE)话题发布数据，并订阅来自ProjectAirSim Server(或UE)话题的数据。

功能优点：

  • 当至少有一个订阅者时，ProjectAirSim Server(或UE)话题才会被订阅；
  • 支持一个话题存在多个订阅者时，只会向ProjectAirSim Server订阅一次；当订阅获得ProjectAirSim Server发布的话题数据后，会依次调用订阅者的回调函数。

关键成员变量和成员函数：

  • sub_topic_callbacks：self.sub_topic_callbacks = ({})是字典的数据结构，key是topic_name，value是TopicCallbacks类型的实例。该成员变量的作用是将话题名称映射为回调函数。
  • add_subscriber(topic_name, topic_callback)：将回调函数添加到sub_topic_callbacks的键（topic_name）对应TopicCallbacks实例的self.callbacks=[]列表中。
    def add_subscriber(self, topic_name, topic_callback):
        if topic_name not in self.sub_topic_callbacks:
            callbacks = TopicCallbacks()
            self.sub_topic_callbacks[topic_name] = callbacks
        else:
            callbacks = self.sub_topic_callbacks[topic_name]

        callback_added = callbacks.add(topic_callback)

        # If this is the first subscriber, subscribe to the Project AirSim topic
        if callback_added and (len(callbacks) == 1):
            self._subscribe_to_topic(topic_name)
  • _subscribe_to_topic(topic_name)：当add_subscriber函数添加回调函数时，如果发现是topic_name话题的第一个订阅者，那么就会调用该函数。该函数通过projectairsim client的subscribe函数订阅topic_name话题（话题的发布者是ProjectAirSim Server/UE）。当接收到话题信息后，会自动触发subscribe是设置的回调函数，即_topic_update_cb。
    def _subscribe_to_topic(self, topic_name):
        if self.projectairsim_client is not None:
            self.logger.debug(f"Subscribing to AirSim topic {topic_name}")
            self.projectairsim_client.subscribe(topic_name, self._topic_update_cb)
  • _topic_update_cb(topic, message)：该函数通过self.sub_topic_callbacks[topic.path](topic, message)调用TopicCallbacks实例中所有的回调函数。
  • publish(topic_name, message)：通过projectairsim client的publish函数向对应话题发布消息。
    def publish(self, topic_name, message):
        if self.projectairsim_client is not None:
            self.projectairsim_client.publish(topic_name, message)
  • _unsubscribe_from_topic(topic_name)：与_subscribe_to_topic相反，通过projectairsimclient的unsubscribe取消订阅topic_name话题。
  • remove_subscriber(topic_name, topic_callback)：移除话题topic_name中的回调函数topic_callback（实际上就是删除一个订阅者），如果移除后不存在订阅者了，则调用_unsubscribe_from_topic取消话题的订阅。

5.ROSTopicsManager类

5.1 ROS1Node

这一部分涉及ROSNode，具体而言是ROS1Node。

5.1.1 嵌套的Publisher类

在Publisher类__init__函数实例化rospy.Publisher时，传入参数subscriber_listener=self。subscriber_listener值赋值为一个实例化对象（即这里的self），该实例化对象要包含两个特定的函数peer_subscribe（当有一个新加入的订阅者连接到这个发布者时触发）、peer_unsubscribe（当一个订阅者断开连接时触发）。

    class Publisher(node.ROSNode.Publisher):
        def __init__(
            self,
            topic: str,
            msg_type,
            subscriber_listener,
            latch: bool,
            queue_size: int,
        ):
            self.rospy_publisher = rospy.Publisher(
                name=topic,
                data_class=msg_type,
                latch=latch,
                queue_size=queue_size,
                subscriber_listener=self,
            )
            self.subscriber_listener_target = subscriber_listener

subscriber_listener_target被赋值为外部传入的目标对象，peer_subscribe和peer_unsubscribe函数实际调用的是subscriber_listener_target的peer_subscribe和peer_unsubscribe函数。

def peer_subscribe(self, topic_name, publish_topic, publish_subscribers):
    if self.subscriber_listener_target:
        self.subscriber_listener_target.peer_subscribe(topic_name)

def peer_unsubscribe(self, topic_name, num_peers):
    if self.subscriber_listener_target:
        self.subscriber_listener_target.peer_unsubscribe(topic_name, num_peers)

综上所述，每当有新的订阅者连接或断开时，都会调用传入函数__init__中参数subscriber_listener的peer_subscribe或者peer_unsubscribe函数。

5.1.2 嵌套SensorHelper类

保存定义相机标定参数

5.1.3 嵌套Subscriber类

__inti__时实例化rospy.Subscriber

5.1.4 ROS1Node

create_publisher：一般使用create_publisher创建发布者。上述表述修改为：每当有新的订阅者连接或断开时，都会调用传入函数create_publisher中参数subscriber_listener的peer_subscribe或者peer_unsubscribe函数。

create_subscriber：一般使用create_subscriber创建订阅者。

5.2 ROSTopicsManager

该类用于订阅和发布ROS的所有话题：当至少有一个订阅者时，ROS话题才会被订阅；当没有订阅者时，ROS话题话题会被取消订阅。支持一个ROS话题存在多个订阅者。

关键成员变量：

  • self.ros_node：上述ROS1类型的节点
  • self.ros_publishers：{“publisher”：Publisher，“callbacks”：ROSPeerChangeCallbacks，"num_peers"：0}
  • self.ros_subscribers：{“subscriber”：Subscriber，“callbacks”：TopicCallbacks}

关键成员函数：

  • add_subscriber(topic_name, ros_message_type, topic_callback)：订阅消息，触发回调函数。
    def add_subscriber(self, topic_name: str, ros_message_type: type, topic_callback):
        if topic_name in self.ros_subscribers:
            topic_callbacks = self.ros_subscribers[topic_name]["callbacks"]
        else:
            self.logger.info(f'Subscribing to ROS topic "{topic_name}"')
            topic_callbacks = TopicCallbacks()
            self.ros_subscribers[topic_name] = {
                "subscriber": self.ros_node.create_subscriber(
                    msg_type=ros_message_type,
                    topic=topic_name,
                    callback=self._topic_cb_decorator(
                        self._topic_update_cb, topic_name
                    ),
                ),
                "callbacks": topic_callbacks,
            }

        topic_callbacks.add(topic_callback)
  • add_publisher(...，peer_change_callback，...)：传入的peer_change_callback会被封装在“callbacks”的值中，即ROSPeerChangeCallbacks类的self.callbacks列表中。回调函数列表在每次有人订阅或取消时都会被触发。
   def add_publisher(
        self,
        topic_name: str,
        ros_message_type: type,
        peer_change_callback=None,
        is_latching: bool = True,
        ros_queue_size: int = 1,
    ):
        if topic_name in self.ros_publishers:
            ros_peer_change_callbacks = self.ros_publishers[topic_name]["callbacks"]
        else:
            self.logger.info(f'Advertising ROS topic "{topic_name}"')
            ros_peer_change_callbacks = ROSPeerChangeCallbacks()
            self.ros_publishers[topic_name] = {
                "publisher": self.ros_node.create_publisher(
                    topic=topic_name,
                    msg_type=ros_message_type,
                    queue_size=ros_queue_size,
                    latch=is_latching,
                    subscriber_listener=self,
                ),
                "callbacks": ros_peer_change_callbacks,
                "num_peers": 0,
            }

        ros_peer_change_callbacks.add(peer_change_callback)
  • peer_subscribe(topic_name)：在订阅时，ROS只会通知有订阅事件发生，但是不会传会订阅者的数量。当有新的订阅者订阅了对应的话题，便会触发该类的peer_subscribe函数，函数内部num_peers手动加1，随后依次调用回调函数列表中的所有函数。
    def peer_subscribe(self, topic_name):
        if topic_name in self.ros_publishers:
            self.logger.debug(f"Peer subscribed to {topic_name}")
            num_peers = self.ros_publishers[topic_name]["num_peers"] + 1
            self.ros_publishers[topic_name]["num_peers"] = num_peers
            self.ros_publishers[topic_name]["callbacks"](topic_name, num_peers)
  • peer_unsubscribe(topic_name, num_peers)：在取消订阅时，ROS会直接传回剩余订阅者的数量，即变量num_peers。因此，函数直接更新订阅者数量，并依次调用回调函数列表中的所有函数。
    def peer_unsubscribe(self, topic_name, num_peers):
        if topic_name in self.ros_publishers:
            self.logger.debug(f"Peer unsubscribed from {topic_name}")
            self.ros_publishers[topic_name]["num_peers"] = num_peers
            self.ros_publishers[topic_name]["callbacks"](topic_name, num_peers)

6.TopicsManagers类

将ProjectAirSimTopicsManager和ROSTopicsManager组合在一起

7.AutoSubscriber类

如果有ROS节点想要获取ProjectAirSim的信息时，自动订阅对应的面向ProjectAirSim Server(或UE)主题，并在获取信息后自动执行回调函数。

AutoSubscriber(projectairsim_topic_name, projectairsim_topic_callback, airsim_topics_manager)

peer_change_cb：调用者会将该方法作为 peer_change_callback 参数传递给 ROSTopicManager.add_publisher()，存放在回调函数列表中。因此，当发布者对应订阅者的数量发生变化时，就会触发执行该函数。该函数的功能是：当没有ROS订阅者消息时，取消对ProjectAirSim Server(UE)的订阅；当有订阅者时，向ProjectAirSim Server(UE)订阅。

    def peer_change_cb(self, ros_topic_name: str, num_peers: int):
        if num_peers == 0:
            if self.is_subscribed:
                self.unsubscribe()
        else:
            if not self.is_subscribed:
                self.subscribe()

    def subscribe(self):
        self.projectairsim_topics_manager.add_subscriber(
            self.projectairsim_topic_name, self.projectairsim_topic_callback
        )
        self.is_subscribed = True

    def unsubscribe(self):
        self.is_subscribed = False
        self.projectairsim_topics_manager.remove_subscriber(
            self.projectairsim_topic_name, self.projectairsim_topic_callback
        )

8.BasicROSSubscriber

订阅话题数据，并执行实例化时传入的回调函数message_callback

订阅ros_topic_name话题，并设置回调函数

        topics_managers.ros_topics_manager.add_subscriber(
            ros_topic_name, ros_message_type, self._ros_topic_update_cb
        )

执行回调函数_ros_topic_update_cb

    def _ros_topic_update_cb(self, ros_topic_name, ros_message_data):
        if self.ros_topic_name:
            self.message_callback(ros_topic_name, ros_message_data)

三、功能集成类详解

1.BasicBridgeToROS类

        self._auto_subscriber = AutoSubscriber(
            projectairsim_topic_name,
            self._projectairsim_topic_update_cb,
            topics_managers.projectairsim_topics_manager,
        )

        # Advertise the ROS topic
        topics_managers.ros_topics_manager.add_publisher(
            topic_name=ros_topic_name,
            ros_message_type=ros_message_type,
            is_latching=ros_topic_is_latching,
            peer_change_callback=self._auto_subscriber.peer_change_cb,
        )

当有ros节点想要获取ProjectAirSim数据时，处理ProjectAirSim话题数据，并将其发送到ROS话题中，流程如下：

  1. 该实例订阅ProjectAirSim Server(或UE)的话题，并设置回调函数_projectairsim_topic_update_cb（该函数在ProjectAirSim Client接收到ProjectAirSim Server的信息后触发）

  2. 该实例创建一个发布者用于监听是否有ROS节点订阅消息，并设置订阅者变化的回调函数peer_change_cb（见5.1.4 ROS1Node，每当有新的订阅者连接或断开时，都会调用传入函数create_publisher中参数subscriber_listener的peer_subscribe或者peer_unsubscribe函数。此时，create_publisher传入的self为topics_managers.ros_topics_manager，对应函数见5.2 ROSTopicsManager）

  3. 现在有一个ros节点想要获取话题数据，订阅第2步发布者发布的话题

  4. 检测到有人订阅，程序执行ros_topics_manager的peer_subscribe，进而回调函数peer_change_cb，完成对ProjectAirSim Server(或UE)的订阅

  5. ProjectAirSim Server(或UE)会发布对应数据到该实例的AutoSubscriber中，触发回调函数_projectairsim_topic_update_cb。

  6. 回调函数先调用ros_topics_manager.publish函数，使用message_callback函数处理，进一步调用ROS1Node的publish方法，实现处理后信息的发布。

    def _projectairsim_topic_update_cb(
        self, projectairsim_topic, projectairsim_message_data
    ):
        if self.ros_topic_name:
            ros_message = self.message_callback(
                projectairsim_topic.path, projectairsim_message_data
            )
            if ros_message is not None:
                self.topics_managers.ros_topics_manager.publish(
                    self.ros_topic_name, ros_message
                )

2.BasicBridgeFromROS(BasicROSSubscriber)

重写BasicROSSubscriber类的_ros_topic_update_cb函数，将经过回调函数message_callbak处理的信息通过topics_managers.projectairsim_topics_manager.publish发布至ProjectAirSim中

    def _projectairsim_topic_update_cb(
        self, projectairsim_topic, projectairsim_message_data
    ):
        if self.ros_topic_name:
            ros_message = self.message_callback(
                projectairsim_topic.path, projectairsim_message_data
            )
            if ros_message is not None:
                self.topics_managers.ros_topics_manager.publish(
                    self.ros_topic_name, ros_message
                )

四、gps话题订阅示例

首先，ProjectAirSimROSBridge初始化时，系统只创建了一个订阅/ProjectAirSim/node /{self.ros_node.name}/load_scene话题的订阅者。当有ROS节点向该话题发布信息时，会触发_load_scene_message_cb函数。

该回调函数会调用update_topics函数，update_topics函数会搜索所有projectairsim client的topic，查看对应的topic是否需要被bridge处理。如果需要处理，那么就会根据topic_entries中定义的参数实例化topic_handler_type。

以gps为例，ProjectAirSimROSBridge类初始化时保存以下信息：

self.topic_entries = [
    self.TopicEntry(
        self.TopicEntry.MatchType.ENDS_WITH,
        "/gps",
        rossensmsg.NavSatFix,
        topic_handler_type=BasicBridgeToROS,
        message_callback=self.msg_converter.convert_gps_to_ros,
        ros_topic_is_latching=False,
    ),
]

Topic_entries的初始化参数如下所示：

    class TopicEntry:
        def __init__(
            self,
            match_type: MatchType,
            name_pattern: str,
            ros_message_type,
            topic_handler_type,
            **kwargs,
        ):

上述参数说明，话题匹配方式为字符串的结尾匹配，话题名称模式为/gps，即如果一个订阅话题是以/gps结尾，那么这个话题相关操作就会被这个TopicEntry捕获。ros的消息类型为rossensmsg.NavSatFix，topic_handler_type为BasicBridgeToROS。其他的信息均被保存在TopicEntry的self.topic_handler_params中（message_callback会在发布到ROS话题之前处理）。

因此，信息在发布到ROS话题中之前，调用convert_gps_to_ros，将ProjectAirSim格式的gps信息转化为ROS的消息格式。

至此，完成信息从ProjectAirSim Server到ProjectAirSim Client再到ROS的全流程。

五、相关链接

AirSim：https://github.com/microsoft/AirSim.git

Project AirSim：https://github.com/iamaisim/ProjectAirSim.git

本文仅为个人学习与理解笔记，水平有限，如有错误或理解偏差，欢迎评论区指正，轻喷！

如果大家有更好的实现思路或改进建议，也非常欢迎联系我一起讨论！

希望这篇文章能帮到正在研究 Project AirSim 的同学！

# 无人机 # 算法 # python # 开发语言 # ue5
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区
cover

最高赢取价值万元大奖，仓颉生态创新开发挑战赛开启！

avatar AtomGit开源社区