python - 如何在 ZMQ 中使用序列化发送图像和数据字符串？

Question

我的目标是将图像和数据字符串从 RPi(服务器)发送到客户端。我使用 send_json(data)，其中数据是字典 {'img': img_ls, 'telemetry':'0.01, 320, -10'}。 img_ls 是转换为列表的图像。问题是我得到 len( img_ls ) = 57556，而原始图像的大小为:320 x 240 = 76800。我不明白为什么会出现差异。这是代码:

服务器端

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://0.0.0.0:5557")


def outputs():
    stream = io.BytesIO()
    while True:
        yield stream
        stream.seek(0)
        sensors = '0.01, 320, -10'
        img_ls = np.fromstring(stream.getvalue(), dtype=np.uint8).tolist()
        data = {'telemetry': sensors, 'img': img_ls}
        socket.send_json(data)
        stream.seek(0)
        stream.truncate()

with picamera.PiCamera() as camera:
        camera.resolution = (320, 240)
        camera.framerate = 80
        time.sleep(2)
        camera.capture_sequence(outputs(), 'jpeg', use_video_port=True)

客户端

ip_server = "192.168.42.1"
context = zmq.Context()
zmq_socket = context.socket(zmq.SUB)
zmq_socket.setsockopt(zmq.SUBSCRIBE, b'')
zmq_socket.setsockopt(zmq.CONFLATE, 1)
zmq_socket.connect("tcp://{}:5557".format(ip_server))

try:
    img_nbr = 1
    while True:
        start = time.time()
        frames = zmq_socket.recv_json()
        img_ls = frames['img']
        telemetry = frames['telemetry']
        #convert img list to array
        img_arr = np.asarray(img_ls)
        #reshape gives error because 320*240 != len(img_ls)
        image = np.reshape(img_ls, (320, 240))
        #save image file locally 
        image = Image.fromarray(image)
        #timestamp in ms
        timestamp = int(time.time() * 1000 )
        image.save('img_'+str(timestamp)+'.jpg')
        print('Frame number: ', str(img_nbr))
        img_nbr += 1
finally:
    pass

最后说明:这是我尝试将图像和传感器数据从 RPi 同步流式传输到客户端。恐怕数组和列表转换(在 RPi 端完成)可能会减慢流式传输。如果有更好的方法(仍然)使用 zmq 来做到这一点，请告诉我。

Answer 1

图像处理非常耗费 CPU。所以，性能第一:

ZeroMQ 应允许人们享受零拷贝操作方式，因此要防止任何不利的操作，破坏它。

我只使用了一个通用的 OpenCV 相机，而不是 RPi/PiCamera，我总是更喜欢在受控事件循环下的采集端拍摄单个相机帧(而不是序列)。

相机获取已知的固定几何图片(在 OpenCV 中是 numpy.ndarray 3D-structure [X,Y,[B,G,R]])，因此速度最快且最多直接序列化是在发送方使用 struct.pack( CONST_FRAME_STRUCT_MASK, aFrame ) ，在接收方使用 struct.unpack( CONST_FRAME_STRUCT_MASK, aMessage ) ).

是的，struct.pack() 是迄今为止最快的方法，即使文档提供了其他方法(灵 active 需要额外成本，这是不合理的):

import numpy

def send_array( socket, A, flags = 0, copy = True, track = False ):
    """send a numpy array with metadata"""
    md = dict( dtype = str( A.dtype ),
               shape =      A.shape,
               )
    pass;  socket.send_json( md, flags | zmq.SNDMORE )
    return socket.send(      A,  flags, copy = copy, track = track )

def recv_array( socket, flags = 0, copy = True, track = False ):
    """recv a numpy array"""
    md = socket.recv_json( flags = flags )
    msg = socket.recv(     flags = flags, copy = copy, track = track )
    buf = buffer( msg )
    pass;  A = numpy.frombuffer( buf, dtype = md['dtype'] )
    return A.reshape(                         md['shape'] )

任何颜色转换和类似的源端转换都可能消耗 +150 ~ 180 [ms]，因此请尽量避免任何和所有不必要的颜色空间或 reshape 或类似的非核心转换，因为这些会不利地增加累积管道延迟包络。

使用 struct.pack() 还可以避免任何类型的大小不匹配，因此您加载到二进制负载着陆台上的内容正是您在接收器端接收到的内容).

如果确实希望在消息核心数据周围也有 JSON 相关的开销，那么可以设置一个双套接字范例，两者都具有 ZMQ_CONFLATE == 1，其中第一个移动 struct-payloads 和第二个 JSON-decorated 遥测。

如果 RPi 允许，zmq.Context( nIOthreads ) 可能会进一步增加两侧的数据泵吞吐量，nIOthreads >= 2，以及额外的 JSON_socket.setsockopt( ZMQ_AFFINITY, 1 ); VIDEO_socket.setsockopt( ZMQ_AFFINITY, 0 ) 映射可以分离/分配工作负载，以在不同的、单独的 IOthread 上运行。