ROS应用层通信协议解析

参考:http://wiki.ros.org/ROS/Master_API

http://wiki.ros.org/ROS/Connection Header

说明

ROS本质上就是一个松耦合的通信框架,通信模式包括:远程调用(service-client)、订阅发布(topic)、持续通信(action)和全局参数(参数服务器),这四种模式基本已经能够涵盖百分之九十的应用场景了

本次针对订阅发布模式,探究一下ROS通信中的具体通信协议,读完本文后,你可以在不依赖ROS的情况下和ROS通信

本次通信采用从机订阅主机数据,通过wireshark抓包,得到具体xmlrpc协议数据内容,根据xmlrpc协议格式,找到对应代码

(因为时间有限,部分协议可能有跳过的地方)

1、registerPublisher

从机创建节点的第一步

这个方法用于注册一个发布者的caller

request报文body:

<?xml version="1.0"?> <methodCall>     <methodName>registerPublisher</methodName>     <params>         <param>             <value>/test_sub</value>         </param>         <param>             <value>/rosout</value>         </param>         <param>             <value>rosgraph_msgs/Log</value>         </param>         <param>             <value>http://192.168.1.150:40209</value>         </param>     </params> </methodCall> 

response报文body:

<?xml version='1.0'?> <methodResponse>     <params>         <param>             <value>                 <array>                     <data>                         <value>                             <int>1</int>                         </value>                         <value>                             <string>Registered [/test_sub] as publisher of [/rosout]</string>                         </value>                         <value>                             <array>                                 <data>                                     <value>                                         <string>http://sherlock:35861/</string>                                     </value>                                 </data>                             </array>                         </value>                     </data>                 </array>             </value>         </param>     </params> </methodResponse> 

先说结论:

ROS有一个日志相关的topic,名称是 /rosout,所有节点的日志信息都会通过这个 topic 发布出来

ROS还有一个日志相关的节点,名称是 /rosout,负责订阅 /rosout数据,然后使用名称为 /rosout_agg 的topic发布出来, /rosout_agg 的订阅者是rqt等调试工具

所以,结合上面的xml内容,我们可以大致推断,创建一个新的节点,那么这个节点必定会发布一个topic,就是/rosout,所以上面的XMLRPC协议内容,就是网rosmaster内注册一个publisher,用于发布/rosout

整体来说,就是调用接口

registerPublisher("/test_sub", "/rosout", "rosgraph_msgs/Log", "http://192.168.1.150:40209") 

返回值是:

1 Registered [/test_sub] as publisher of [/rosout] http://sherlock:35861/ 

1是固定数据

第二行是message

最后一个返回值表示订阅者的uri列表,这里因为只有一个订阅者,所有只有一个uri

再看代码:

函数声明如下:

registerPublisher(caller_id, topic, topic_type, caller_api) 	Register the caller as a publisher the topic. 	参数 		caller_id (str) 			ROS caller ID 		topic (str) 			Fully-qualified name of topic to register. 		topic_type (str) 			Datatype for topic. Must be a package-resource name, i.e. the .msg name. 		caller_api (str) 			API URI of publisher to register. 	返回值(int, str, [str]) 		(code, statusMessage, subscriberApis) 		List of current subscribers of topic in the form of XMLRPC URIs. 

找到 registerPublisher 接接口,位于ros_comm/rosmaster 包中,文件为:master_api.py(ROS主从机制利用python实现,拿掉python则无法实现主从)

@apivalidate([], ( is_topic('topic'), valid_type_name('topic_type'), is_api('caller_api'))) def registerPublisher(self, caller_id, topic, topic_type, caller_api): 	"""     Register the caller as a publisher the topic.     @param caller_id: ROS caller id     @type  caller_id: str     @param topic: Fully-qualified name of topic to register.     @type  topic: str 	@param topic_type: Datatype for topic. Must be a     package-resource name, i.e. the .msg name. 	@type  topic_type: str     @param caller_api str: ROS caller XML-RPC API URI     @type  caller_api: str     @return: (code, statusMessage, subscriberApis).     List of current subscribers of topic in the form of XMLRPC URIs.     @rtype: (int, str, [str])     """     #NOTE: we need topic_type for getPublishedTopics. 	try:         self.ps_lock.acquire()         self.reg_manager.register_publisher(topic, caller_id, caller_api)         # don't let '*' type squash valid typing         if topic_type != rosgraph.names.ANYTYPE or not topic in self.topics_types:             self.topics_types[topic] = topic_type         pub_uris = self.publishers.get_apis(topic)         sub_uris = self.subscribers.get_apis(topic)         self._notify_topic_subscribers(topic, pub_uris, sub_uris)         mloginfo("+PUB [%s] %s %s",topic, caller_id, caller_api)         sub_uris = self.subscribers.get_apis(topic)                 finally:     	self.ps_lock.release() 	return 1, "Registered [%s] as publisher of [%s]"%(caller_id, topic), sub_uris 

registerPublisher 接口的注释:Register the caller as a publisher the topic,将调用者注册为一个topic发布者

可以对应xmlrpc中对应参数,加上猜测:

caller_id:调用者,可以认为是节点,/test_sub,从及创建的节点

topic:发布的topic name,/rosout

topic_type:发布的topic数据类型,rosgraph_msgs/Log

caller_api:调用者发布数据的API接口,http://192.168.1.150:40209

总上,我们大概有几点猜测:

  1. 接口在rosmaster中,接口是registerPublisher,表示,这是注册节点的
  2. 告诉master节点,我创建了一个节点,节点名是/test_sub
  3. 告诉master,这个节点需要发布topic,topic名是/rosout,数据类型是rosgraph_msgs/Log

registerPublisher 接口中有三个地方需要注意:

  1. register_publisher接口调用
  2. _notify_topic_subscribers接口调用,告知当前所有的subscriber,有新的publisher,他们需要再次到新的publisher中去订阅数据
  3. return 内容,最后会拼接成xmlrpc的报文,response 回去,这也就顺便解释了第二条xmlrpc报文(response)

register_publisher

先看 register_publisher,代码在rosmaster中的registrations.py文件中

def register_publisher(self, topic, caller_id, caller_api):     """ 	Register topic publisher     @return: None     """     self._register(self.publishers, topic, caller_id, caller_api) 

_register 接口,这个节点做了三件事

  1. 调用内部接口保存节点信息
  2. 如果这个节点之前已经存在,就表明它是在更新,则发布数据的接口改变,且之前已经有订阅,则此时所有订阅该接口的所有subscriber解除订阅
  3. 调用register接口保存
def _register(self, r, key, caller_id, caller_api, service_api=None):     # update node information     node_ref, changed = self._register_node_api(caller_id, caller_api)     node_ref.add(r.type, key)     # update pub/sub/service indicies     if changed:         self.publishers.unregister_all(caller_id)         self.subscribers.unregister_all(caller_id)         self.services.unregister_all(caller_id)         self.param_subscribers.unregister_all(caller_id) 	r.register(key, caller_id, caller_api, service_api) 

_register_node_api 接口,我们可以看到,它主要做两件事

  1. 更新master中节点信息(节点名、节点发布数据的接口)
  2. 检查这个节点是不是已经存在,如果是,则告诉调用者
def _register_node_api(self, caller_id, caller_api):     """     @param caller_id: caller_id of provider     @type  caller_id: str     @param caller_api: caller_api of provider     @type  caller_api: str     @return: (registration_information, changed_registration). changed_registration is true if      caller_api is differet than the one registered with caller_id     @rtype: (NodeRef, bool)     """     node_ref = self.nodes.get(caller_id, None)      bumped_api = None     if node_ref is not None:         if node_ref.api == caller_api:             return node_ref, False         else:             bumped_api = node_ref.api             self.thread_pool.queue_task(bumped_api, shutdown_node_task,                                         (bumped_api, caller_id, "new node registered with same name"))      node_ref = NodeRef(caller_id, caller_api)     self.nodes[caller_id] = node_ref     return (node_ref, bumped_api != None) 

_notify_topic_subscribers

_notify_topic_subscribers 代码,根据注释说明,接口的作用就是通知所有的subscriber,有新的publisher

def _notify_topic_subscribers(self, topic, pub_uris, sub_uris):     """     Notify subscribers with new publisher list     @param topic: name of topic     @type  topic: str     @param pub_uris: list of URIs of publishers.     @type  pub_uris: [str]     """     self._notify(self.subscribers, publisher_update_task, topic, pub_uris, sub_uris)  

_notify 代码,将更新的通知任务(publisher_update_task)放进事件队列中,等待执行:

def _notify(self, registrations, task, key, value, node_apis):     """     Generic implementation of callback notification     @param registrations: Registrations     @type  registrations: L{Registrations}     @param task: task to queue     @type  task: fn     @param key: registration key     @type  key: str     @param value: value to pass to task     @type  value: Any     """     # cache thread_pool for thread safety     thread_pool = self.thread_pool     if not thread_pool:         return          try:                     for node_api in node_apis:             # use the api as a marker so that we limit one thread per subscriber             thread_pool.queue_task(node_api, task, (node_api, key, value))     except KeyError:         _logger.warn('subscriber data stale (key [%s], listener [%s]): node API unknown'%(key, s)) 

publisher_update_task 代码,传入的三个参数分别是:新节点的接口、topic名称、订阅者的接口:

def publisher_update_task(api, topic, pub_uris):     """     Contact api.publisherUpdate with specified parameters     @param api: XML-RPC URI of node to contact     @type  api: str     @param topic: Topic name to send to node     @type  topic: str     @param pub_uris: list of publisher APIs to send to node     @type  pub_uris: [str]     """     msg = "publisherUpdate[%s] -> %s %s" % (topic, api, pub_uris)     mloginfo(msg)     start_sec = time.time()     try:         #TODO: check return value for errors so we can unsubscribe if stale         ret = xmlrpcapi(api).publisherUpdate('/master', topic, pub_uris)         msg_suffix = "result=%s" % ret     except Exception as ex:         msg_suffix = "exception=%s" % ex         raise     finally:         delta_sec = time.time() - start_sec         mloginfo("%s: sec=%0.2f, %s", msg, delta_sec, msg_suffix) 

publisherUpdate 接口在 rospy 模块的 masterslave.py 文件中,猜测是使用XMLRPC协议,通知所有的订阅者节点,发布者更新了

@apivalidate(-1, (is_topic('topic'), is_publishers_list('publishers'))) def publisherUpdate(self, caller_id, topic, publishers):     """     Callback from master of current publisher list for specified topic.     @param caller_id: ROS caller id     @type  caller_id: str     @param topic str: topic name     @type  topic: str     @param publishers: list of current publishers for topic in the form of XMLRPC URIs     @type  publishers: [str]     @return: [code, status, ignore]     @rtype: [int, str, int]     """     if self.reg_man:         for uri in publishers:             self.reg_man.publisher_update(topic, publishers)     return 1, "", 0 

2、hasParam

request报文body

<?xml version='1.0'?> <methodResponse>     <params>         <param>             <value>                 <array>                     <data>                         <value>                             <int>1</int>                         </value>                         <value>                             <string>/use_sim_time</string>                         </value>                         <value>                             <boolean>0</boolean>                         </value>                     </data>                 </array>             </value>         </param>     </params> </methodResponse> 

response报文body

<?xml version="1.0"?> <methodCall>     <methodName>requestTopic</methodName>     <params>         <param>             <value>/rosout</value>         </param>         <param>             <value>/rosout</value>         </param>         <param>             <value>                 <array>                     <data>                         <value>                             <array>                                 <data>                                     <value>TCPROS</value>                                 </data>                             </array>                         </value>                     </data>                 </array>             </value>         </param>     </params> </methodCall> 

先说结论:

调用 hasParam 接口,检查参数服务器是否有参数 /test_sub/use_sim_time

返回值是 [1, /use_sim_time, 0],表示没有

再看代码:

hasParam

有了上面的分析经验,我们可以很轻松地的出结论,这是在调用 hasParam 接口

我们可以很轻松地找到这个方法的代码,在rosmaster模块的master_api.py文件中

@apivalidate(False, (non_empty_str('key'),)) def hasParam(self, caller_id, key):     """     Check if parameter is stored on server.      @param caller_id str: ROS caller id     @type  caller_id: str     @param key: parameter to check     @type  key: str     @return: [code, statusMessage, hasParam]     @rtype: [int, str, bool]     """     key = resolve_name(key, caller_id)     if self.param_server.has_param(key):         return 1, key, True     else:         return 1, key, False  

根据协议

  1. caller_id 传参是 /test_sub
  2. key 传参是 /use_sim_time

根据注释和代码,我们可以确认,这个就口就是在检查,参数服务器是否有参数 /test_sub/use_sim_time

resolve_name 接口接收两个参数,根据调用:

  1. name是/use_sim_time
  2. namespace_是/test_sub

所以,才能确认上面的全局参数 /test_sub/use_sim_time

hasParam 接口的返回值有三个

  1. code,整型,这里无论有没有,都返回1,可以忽略
  2. key,这里就是 /use_sim_time
  3. hasParam,表示是否有这个参数,True/False

根据 response 报文,这里应该返回 [1, /use_sim_time, 0],表示没有这个参数

use_sim_time

想要理解为什么要调用这个接口,就要理解 use_sim_time 参数的作用

use_sim_time是一个重要的参数,它默认值为false,可以配合Rosbag使用,是一个很重要的离线调试工具

我们都知道,ROS 中的时间有两种:

  1. ROS::Time()
  2. ROS::WallTime()

ROS::Time()和ROS::WallTime()

表示ROS网络中的时间,如果当时在非仿真环境里运行,那它就是当前的时间。但是假设去回放当时的情况,那就需要把当时的时间录下来

以控制为例,很多的数据处理需要知道当时某一个时刻发生了什么。Wall Time可以理解为墙上时间,墙上挂着的时间没有人改变的了,永远在往前走;ROS Time可以被人为修改,你可以暂停它,可以加速,可以减速,但是Wall Time不可以。

在开启一个Node之前,当把use_sim_time设置为true时,这个节点会从clock Topic获得时间。所以操作这个clock的发布者,可以实现一个让Node中得到ROS Time暂停、加速、减速的效果。同时下面这些方面都是跟Node透明的,所以非常适合离线的调试方式。当把ROSbag记下来以后重新play出来时,加两个横杠,--clock,它就会发布出这个消息

3、registerSubscriber

先看报文:

request报文body

<?xml version="1.0"?> <methodCall>     <methodName>registerSubscriber</methodName>     <params>         <param>             <value>/test_sub</value>         </param>         <param>             <value>/ros_message</value>         </param>         <param>             <value>my_package/MessageDefine</value>         </param>         <param>             <value>http://192.168.1.150:43597</value>         </param>     </params> </methodCall> 

response报文body

<?xml version='1.0'?> <methodResponse>     <params>         <param>             <value>                 <array>                     <data>                         <value>                             <int>1</int>                         </value>                         <value>                             <string>Subscribed to [/ros_message]</string>                         </value>                         <value>                             <array>                                 <data>                                     <value>                                         <string>http://sherlock:41689/</string>                                     </value>                                 </data>                             </array>                         </value>                     </data>                 </array>             </value>         </param>     </params> </methodResponse> 

registerSubscriber

registerSubscriber 代码在rosmaster包中的master_api.py文件中,如下:

@apivalidate([], ( is_topic('topic'), valid_type_name('topic_type'), is_api('caller_api'))) def registerSubscriber(self, caller_id, topic, topic_type, caller_api):     """     Subscribe the caller to the specified topic. In addition to receiving     a list of current publishers, the subscriber will also receive notifications     of new publishers via the publisherUpdate API.             @param caller_id: ROS caller id     @type  caller_id: str     @param topic str: Fully-qualified name of topic to subscribe to.      @param topic_type: Datatype for topic. Must be a package-resource name, i.e. the .msg name.     @type  topic_type: str     @param caller_api: XML-RPC URI of caller node for new publisher notifications     @type  caller_api: str     @return: (code, message, publishers). Publishers is a list of XMLRPC API URIs        for nodes currently publishing the specified topic.     @rtype: (int, str, [str])     """     #NOTE: subscribers do not get to set topic type     try:         self.ps_lock.acquire()         self.reg_manager.register_subscriber(topic, caller_id, caller_api)          # ROS 1.1: subscriber can now set type if it is not already set         #  - don't let '*' type squash valid typing         if not topic in self.topics_types and topic_type != rosgraph.names.ANYTYPE:             self.topics_types[topic] = topic_type          mloginfo("+SUB [%s] %s %s",topic, caller_id, caller_api)         pub_uris = self.publishers.get_apis(topic)     finally:         self.ps_lock.release()     return 1, "Subscribed to [%s]"%topic, pub_uris 

根据协议往来,我们可以看到调用过程

registerSubscriber("/test_sub", "/ros_message", "my_package/MessageDefine", "http://192.168.1.150:43597") 

入参有4个:

  1. 订阅节点名:/test_sub
  2. 需要订阅的topic 名称:/ros_message
  3. topic的数据类型:my_package/MessageDefine
  4. 订阅节点自己的uri,即发布者通知时的发送目标

返回值有3个:

  1. code,这里固定是1

  2. message,这里是 Subscribed to [/ros_message]

  3. publisher 的订阅URI 列表,因为这里只有一个publisher,所以只有一个 http://sherlock:41689/,首先,这可能是主机里面某个几点的uri,需要从机去订阅

代码说明:

和第一条,注册publisher相反,这里是注册subscriber

有几个关键代码

register_subscriber 代码,位于rosmaster包中的registerations.py文件中:

def register_subscriber(self, topic, caller_id, caller_api):     """     Register topic subscriber     @return: None     """     self._register(self.subscribers, topic, caller_id, caller_api) 

_register 代码,调用 _register_node_api 接口更新节点信息,如果之前有该节点的注册信息,则先删除:

def _register(self, r, key, caller_id, caller_api, service_api=None):     # update node information     node_ref, changed = self._register_node_api(caller_id, caller_api)     node_ref.add(r.type, key)     # update pub/sub/service indicies     if changed:         self.publishers.unregister_all(caller_id)         self.subscribers.unregister_all(caller_id)         self.services.unregister_all(caller_id)         self.param_subscribers.unregister_all(caller_id)     r.register(key, caller_id, caller_api, service_api) 

_register_node_api 代码:

def _register_node_api(self, caller_id, caller_api):     """     @param caller_id: caller_id of provider     @type  caller_id: str     @param caller_api: caller_api of provider     @type  caller_api: str     @return: (registration_information, changed_registration). changed_registration is true if      caller_api is differet than the one registered with caller_id     @rtype: (NodeRef, bool)     """     node_ref = self.nodes.get(caller_id, None)      bumped_api = None     if node_ref is not None:         if node_ref.api == caller_api:             return node_ref, False         else:             bumped_api = node_ref.api             self.thread_pool.queue_task(bumped_api, shutdown_node_task,                                         (bumped_api, caller_id, "new node registered with same name"))      node_ref = NodeRef(caller_id, caller_api)     self.nodes[caller_id] = node_ref     return (node_ref, bumped_api != None) 

shutdown_node_task 代码,如果订阅节点退出了,则需要通知:

def shutdown_node_task(api, caller_id, reason):     """     Method to shutdown another ROS node. Generally invoked within a     separate thread as this is used to cleanup hung nodes.          @param api: XML-RPC API of node to shutdown     @type  api: str     @param caller_id: name of node being shutdown     @type  caller_id: str     @param reason: human-readable reason why node is being shutdown     @type  reason: str     """     try:         xmlrpcapi(api).shutdown('/master', "[{}] Reason: {}".format(caller_id, reason))     except:         pass #expected in many common cases     remove_server_proxy(api) 

4、requestTopic

request报文body,如下,我们可以看到,xmlrpc发送的host是sherlock:41689,是上一步骤,收到的publisher的uri,如果有多个publisher,要request多次

POST /RPC2 HTTP/1.1 Host: sherlock:41689 User-Agent: Go-http-client/1.1 Content-Length: 307 Content-Type: text/xml Accept-Encoding: gzip  <?xml version="1.0"?> <methodCall>     <methodName>requestTopic</methodName>     <params>         <param>             <value>/test_sub</value>         </param>         <param>             <value>/ros_message</value>         </param>         <param>             <value>                 <array>                     <data>                         <value>                             <array>                                 <data>                                     <value>TCPROS</value>                                 </data>                             </array>                         </value>                     </data>                 </array>             </value>         </param>     </params> </methodCall> 

response报文

<?xml version="1.0"?> <methodResponse>     <params>         <param>             <value>                 <array>                     <data>                         <value>                             <i4>1</i4>                         </value>                         <value></value>                         <value>                             <array>                                 <data>                                     <value>TCPROS</value>                                     <value>sherlock</value>                                     <value>                                         <i4>33173</i4>                                     </value>                                 </data>                             </array>                         </value>                     </data>                 </array>             </value>         </param>     </params> </methodResponse> 

根据协议,调用过程如下:

requestTopic("/test_sub", "/ros_message", ["TCPROS"]) 

告诉publisher,subscriber准备好了,可以发数据了

requestTopic

_remap_table['requestTopic'] = [0] # remap topic  @apivalidate([], (is_topic('topic'), non_empty('protocols'))) def requestTopic(self, caller_id, topic, protocols):     """     Publisher node API method called by a subscriber node.      Request that source allocate a channel for communication. Subscriber provides     a list of desired protocols for communication. Publisher returns the     selected protocol along with any additional params required for     establishing connection. For example, for a TCP/IP-based connection,     the source node may return a port number of TCP/IP server.      @param caller_id str: ROS caller id         @type  caller_id: str     @param topic: topic name     @type  topic: str     @param protocols: list of desired      protocols for communication in order of preference. Each      protocol is a list of the form [ProtocolName,      ProtocolParam1, ProtocolParam2...N]     @type  protocols: [[str, XmlRpcLegalValue*]]     @return: [code, msg, protocolParams]. protocolParams may be an     empty list if there are no compatible protocols.     @rtype: [int, str, [str, XmlRpcLegalValue*]]     """     if not get_topic_manager().has_publication(topic):         return -1, "Not a publisher of [%s]"%topic, []     for protocol in protocols: #simple for now: select first implementation          protocol_id = protocol[0]         for h in self.protocol_handlers:             if h.supports(protocol_id):                 _logger.debug("requestTopic[%s]: choosing protocol %s", topic, protocol_id)                 return h.init_publisher(topic, protocol)     return 0, "no supported protocol implementations", [] 

init_publisher 代码

def init_publisher(self, resolved_name, protocol):     """     Initialize this node to receive an inbound TCP connection,     i.e. startup a TCP server if one is not already running.          @param resolved_name: topic name     @type  resolved__name: str          @param protocol: negotiated protocol       parameters. protocol[0] must be the string 'TCPROS'     @type  protocol: [str, value*]     @return: (code, msg, [TCPROS, addr, port])     @rtype: (int, str, list)     """     if protocol[0] != TCPROS:         return 0, "Internal error: protocol does not match TCPROS: %s"%protocol, []     start_tcpros_server()     addr, port = get_tcpros_server_address()     return 1, "ready on %s:%s"%(addr, port), [TCPROS, addr, port] 

publisher 检查,是否支持指定协议,如果不支持,则返回1,否则返回0

返回值的第二个参数有三个值,分别是协议类型、ip地址 和 端口

5、unregisterSubscriber

request报文body

<?xml version="1.0"?> <methodCall>     <methodName>unregisterSubscriber</methodName>     <params>         <param>             <value>/test_sub</value>         </param>         <param>             <value>/ros_message</value>         </param>         <param>             <value>http://192.168.1.150:43597</value>         </param>     </params> </methodCall> 

response报文body

<?xml version='1.0'?> <methodResponse>     <params>         <param>             <value>                 <array>                     <data>                         <value>                             <int>1</int>                         </value>                         <value>                             <string>Unregistered [/test_sub] as provider of [/ros_message]</string>                         </value>                         <value>                             <int>1</int>                         </value>                     </data>                 </array>             </value>         </param>     </params> </methodResponse> 

unregisterSubscriber

@apivalidate(0, (is_topic('topic'), is_api('caller_api'))) def unregisterSubscriber(self, caller_id, topic, caller_api):     """     Unregister the caller as a subscriber of the topic.     @param caller_id: ROS caller id     @type  caller_id: str     @param topic: Fully-qualified name of topic to unregister.     @type  topic: str     @param caller_api: API URI of service to unregister. Unregistration will only occur if current        registration matches.         @type  caller_api: str     @return: (code, statusMessage, numUnsubscribed).        If numUnsubscribed is zero it means that the caller was not registered as a subscriber.       The call still succeeds as the intended final state is reached.     @rtype: (int, str, int)     """     try:         self.ps_lock.acquire()         retval = self.reg_manager.unregister_subscriber(topic, caller_id, caller_api)         mloginfo("-SUB [%s] %s %s",topic, caller_id, caller_api)         return retval     finally:         self.ps_lock.release() 

取消订阅,固定返回1

6、TCP数据私有协议

首先保证主从机的数据类型一致,包括字段的顺序,实际ROS框架内是通过md5检测,保证数据类型一致的

数据传输前提:

  1. 数据类型一致
  2. 字段名一致
  3. 字段顺序一致

数据传输模式:小端hex

数据包结构

数据域长度 数据域
4 byte n byte

数据域长度固定4byte,长度不包括自身

数据域

数据域根据字段类型解析,ros 通信的内置数据类型有:

原始类型 字节数
bool 1
int8 1
uint8 1
int16 2
uint16 2
int32 4
uint32 4
int64 8
uint64 8
float32 4
float64 8
string n(n > 4)
time 8
duration 8
数组 n(n > 4)

其中,除 string、time、duration数组 类型外的其余类型,直接根据字节数读取即可

string

字符串类型,也可认为是字符数组(则可以和数组类型复用),因为是不定长度,所以需要知道字符串的长度,ROS中使用uint32类型表示长度/数组元素数量,即4byte

所以,如果出现字符串类型,则数据域为:

字符串长度 字符
4 byte n byte

数组

数组类型,因为是不定长度,所以需要知道数组的元素数量,和string同理,ROS 中使用uint32类型表示数组的元素数量,再结合数组元素的类型,即可得到总长度

所以,出现数组类型,则数据域为:

数组元素数量 数组数据
4 byte n byte

如果数组类型是 int32,则数组数据占 4 * n byte,其余类型以此类推

time

ROS 中把 time 单独提取作为基本数据类型,对应 ROS 中的 ros::Time 类,因为我们可以认为是嵌套类型

ros::Time 有两个字段:

  1. sec: uint32
  2. nsec: uint32

所以,time 类型在数据域占8byte,如果出现 time 类型,则数据域为:

sec nsec
4 byte 4 byte

duration

duration 类型和 time 相同,在 ROS 中对应 ros::Duration 类,可以认为是嵌套类型

ros::Duration 有两个字段:

  1. sec: uint32
  2. nsec: uint32

所以,duration 类型在数据域中占8byte,如果出现 duration 类型,则数据域为:

sec nsec
4 byte 4 byte

嵌套类型

嵌套类型可以认为是数据域的组合,如果发现字段类型不是内置数据类型,则可认为是嵌套类型,嵌套类型按照类型的字段,递归处理即可

协议分析示例

示例1:

.msg 文件为:

int8 shutdown_time string text 

主机发出数据为:

shutdown_time = 123 text = abc 

从机收到数据为:

08 00 00 00 7b 03 00 00 00 61 62 63 

分析如下:

  1. 包头4 byte表示数据与长度

    08 00 00 00,表示数据域长度为8,即后续数据总长度为8

  2. 字段1为shutdown_time,类型是int8,1byte

    7b转10进制,为123

  3. 字段2为text,类型是字符串 (4+n)byte

    4byte 长度:03 00 00 00,表示字符串长度为3,后面3byte 为字符串内容:61 62 63,ASCII转换为:abc)

示例2:

.msg 文件为:

Header header int8 shutdown_time int32 shutdown_time2 string text float32 num string text2 int8[] data int16[] data2 

Header的数据类型为:

uint32 seq time stamp string frame_id 

主机发出数据为:

//header一般由ROS系统自己处理,这里写出来是为了方便观察 header.seq = 29; header.time.sec = 0; header.time.nsec = 0; header.frame_id = ""; shutdown_time = 123; shutdown_time2 = 987654; text2 = "lmn"; text = "abc"; num = 23.4; data = [1, 2, 4, 89]; data2 = [11, 22, 908] 

从机收到的数据为:

39 00 00 00 1d 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 7b 06 12 0f 00 03 00 00 00 61 62 63 33 33 bb 41 03 00 00 00 6c 6d 6e 04 00 00 00 01 02 04 59 03 00 00 00 0b 00 16 00 8c 03 

分析如下:

  1. 包头4byte表示数据域长度

    0x39 00 00 00,10进制57,表明后续数据域长度57byte

  2. 字段1,Header 类型,可以认为是嵌套类型,Header字段如下:

    1. 字段1,seq,uint32,4byte,数据为,0x1d 00 00 00,十进制29;
    2. 字段2,time类型,可以认为是嵌套类型,字段如下:
      1. 字段1,sec,uint32,4byte,数据为:0x00 00 00 00,十进制0;
      2. 字段2,nsec,uint32,4byte,数据为:0x00 00 00 00,十进制0;
    3. 字段3,frame_id,字符串类型,4byte 表示长度,00 00 00 00,表示长度为0,字符串为空
  3. 字段2,shutdown_time,int8,1byte,数据为:0x7b,十进制123;

  4. 字段3,shutdown_time2,int32,4byte,数据为:0x06 12 0f 00,十进制:987654;

  5. 字段4,text,字符串:

    1. 4byte 长度,数据为:0x03 00 00 00 ,表示字符产长度为3;
    2. 字符串内容,数据为:0x61 62 63 ,ASCII对应:abc;
  6. 字段5,num,flota32,4byte,数据为:33 33 bb 41,十进制:23.4;

  7. 字段6:text2,字符串:

    1. 4byte长度,数据为:0x03 00 00 00,表示字符串长度为3;
    2. 字符产内容,数据为:0x6c 6d 6e,ASCII对应lmn;
  8. 字段7,data,int8数组:

    1. 4byte表示数组元素数量,数据为:0x04 00 00 00,表示有4个int8元素:
    2. 数组内容:[0x01, 0x02, 0x04, 0x59,],表示:[1,2,4,89];
  9. 字段8,data2,int16数组:

    1. 4byte表示长度,数据为:0x03 00 00 00,表示有3个int16数据;
    2. 数组内容:[0x0b00, 0x1600, 0x8c03],表示:[11, 22, 908]

7、小结

宗上,整体的从机订阅时序图如下:

ROS应用层通信协议解析

发表评论

相关文章

当前内容话题