ROS Resources: Documentation | Support | Discussion Forum | Index | Service Status | ros @ Robotics Stack Exchange
Ask Your Question

Rospy service with numpy array message

asked 2021-02-17 09:28:43 -0500

7Z0nE gravatar image

During implementation of a service that responds with a large array (> 100000 values) I noticed that it takes a long time to create a numpy array from the ros message tuple. Rospy already has a solution for that considering basic messaged: Unfortunately I could not find any documentation whether this can be applied to services the same way. Further research leaves me with the conclusion that this is currently not supported. Can somebody confirm this assumption? Also, what would be the best solution to build something similar myself? Designing my own message type with my own serializer or implement my own service api on top of basic ROS publish/subscribe patterns.

Thanks for any help in advance!

Kind regards Fred

edit retag flag offensive close merge delete

1 Answer

Sort by ยป oldest newest most voted

answered 2023-08-07 07:33:55 -0500

kingsimba0511 gravatar image

updated 2023-08-07 07:37:31 -0500

I have the same performance issue and dig a bit... ,created a numpy_srv(), similar with numpy_msg()

But it will only improve performance of byte[], not uint8[]. By using this, to transfer 100MBytes bytes[]:

  • numpy_srv() with bytes[] uses: 1.2s
  • uint8[] uses: 2.5s

I think uint8[] is already very fast for most uses.

import sys

import numpy as np
import rospy
from ax_msgs.srv import GetMapTest, GetMapTestRequest, GetMapTestResponse
from rospy.numpy_msg import numpy_msg

SIZE = 1024 * 1024 * 100

_numpy_srv_types = {}

def numpy_srv(srv_type):
    if srv_type in _numpy_srv_types:
        return _numpy_srv_types[srv_type]

    classdict = {
        "_type": srv_type._type,
        "_md5sum": srv_type._md5sum,
        "_request_class": numpy_msg(srv_type._request_class),
        "_response_class": numpy_msg(srv_type._response_class),

    # create the numpy message type
    srv_type_name = "Numpy_%s" % srv_type._type.replace("/", "__")
    numpy_type = type(srv_type_name, (srv_type,), classdict)
    _numpy_srv_types[srv_type] = numpy_type
    return numpy_type

GetMapTestNumpy = numpy_srv(GetMapTest)  # numpy-lize service

if sys.argv[1] == "server":
    # server

    # create response content
    map_bytes = np.random.randint(0, 128, SIZE, dtype=np.uint8)

    def get_map_bytes_service(req: GetMapTestRequest):
        res = numpy_msg(GetMapTestResponse)()  # Must use numpy_msg() to create the response

        res.map_bytes = map_bytes  # map_bytes is `byte[]`
        return res

    rospy.Service("/get_map_bytes", GetMapTestNumpy, get_map_bytes_service)
    # client
    func = rospy.ServiceProxy("/get_map_bytes", GetMapTestNumpy)
    res: GetMapTestResponse = func(GetMapTestRequest())
    print(f"len(res.map_bytes) = {len(res.map_bytes)}")
edit flag offensive delete link more

Question Tools


Asked: 2021-02-17 09:28:43 -0500

Seen: 255 times

Last updated: Aug 07 '23