如何通过 ZeroMQ 套接字发送 OpenCV 视频片段？

Question

我有一个简单的网络摄像头，我使用 OpenCV 读出了它，我现在正尝试使用 ZeroMQ 将此视频片段发送到另一个 (Python) 程序。所以我有以下简单的脚本来读取网络摄像头并使用 ZeroMQ 套接字发送：

import cv2
import os
import zmq
import base64

context = zmq.Context()
footage_socket = context.socket(zmq.PUB)
footage_socket.connect('tcp://localhost:5555')

# init the camera
camera = cv2.VideoCapture(0)

while True:
    try:
        (grabbed, frame) = camera.read()            # grab the current frame
        frame = cv2.resize(frame, (640, 480))       # resize the frame
        footage_socket.send_string(base64.b64encode(frame))

        # Show the video in a window
        cv2.imshow("Frame", frame)                  # show the frame to our screen
        cv2.waitKey(1)                              # Display it at least one ms
        #                                           # before going to the next frame

    except KeyboardInterrupt:
        camera.release()
        cv2.destroyAllWindows()
        print "\n\nBye bye\n"
        break

这很有效，因为它可以显示视频并且不会出现任何错误。

我注释掉了显示图像的两行（cv2.imshow() 和cv2.waitKey(1)）。然后我并行启动了下面的脚本。第二个脚本应该接收视频片段并显示它。

import cv2
import zmq
import base64
import numpy as np

context = zmq.Context()
footage_socket = context.socket(zmq.SUB)
footage_socket.bind('tcp://*:5555')
footage_socket.setsockopt_string(zmq.SUBSCRIBE, unicode(''))

# camera = cv2.VideoCapture("output.avi")

while True:
    try:
        frame = footage_socket.recv_string()
        frame = np.fromstring(base64.b64decode(frame), dtype=np.uint8)
        cv2.imshow("Frame", frame)                  # show the frame to our screen
        cv2.waitKey(1)                              # Display it at least one ms
        #                                           # before going to the next frame
    except KeyboardInterrupt:
        cv2.destroyAllWindows()
        break

print "\n\nBye bye\n"

不幸的是，这冻结在cv2.waitKey(1)。

有人知道我在这里做错了什么吗？我需要以不同的方式解码素材吗？欢迎所有提示！

Answer 1

最后我通过中间步骤解决了这个问题。我首先将单个图像写入磁盘，然后再次读出这些图像。这使我意识到我需要将帧编码为图像（我选择了 jpg），并且使用魔术方法 cv2.imencode('.jpg', frame) 和 cv2.imdecode(npimg, 1) 我可以让它工作。我在下面粘贴了完整的工作代码。

第一个脚本读取网络摄像头并通过 zeromq 套接字发送素材：

import cv2
import zmq
import base64

context = zmq.Context()
footage_socket = context.socket(zmq.PUB)
footage_socket.connect('tcp://localhost:5555')

camera = cv2.VideoCapture(0)  # init the camera

while True:
    try:
        (grabbed, frame) = camera.read()  # grab the current frame
        frame = cv2.resize(frame, (640, 480))  # resize the frame
        encoded, buffer = cv2.imencode('.jpg', frame)
        footage_socket.send_string(base64.b64encode(buffer))

    except KeyboardInterrupt:
        camera.release()
        cv2.destroyAllWindows()
        print "\n\nBye bye\n"
        break

第二个脚本接收帧图像并显示它们：

import cv2
import zmq
import base64
import numpy as np

context = zmq.Context()
footage_socket = context.socket(zmq.SUB)
footage_socket.bind('tcp://*:5555')
footage_socket.setsockopt_string(zmq.SUBSCRIBE, unicode(''))

while True:
    try:
        frame = footage_socket.recv_string()
        img = base64.b64decode(frame)
        npimg = np.fromstring(img, dtype=np.uint8)
        source = cv2.imdecode(npimg, 1)
        cv2.imshow("image", source)
        cv2.waitKey(1)

    except KeyboardInterrupt:
        cv2.destroyAllWindows()
        print "\n\nBye bye\n"
        break

无论如何，祝你有个美好的一天！

Answer 2

步骤 0：风险清单

鉴于目标很明确，分布式应用程序基础架构的快速原型设计取决于几个风险点。

0) OpenCV cv2 模块大量使用基于 C 的底层组件，如果尝试使用 cv2.imshow() 设施和 cv2 (外部 FSA ) 窗口管理和框架显示服务。

1) ZeroMQ 框架可以为您提供更多帮助，而不仅仅是尝试强制将数据转换为 (string) 只是为了使用 .send_string() / .recv_string() - 在这里可能会变得更好，而不是移动已知图像( pxW * pxH )geometry * 在一些更智能的 BLOB 映射对象中的 RGB 深度（将在下面的架构展望中更多地讨论这方面）。

2) 给定第 0 点和第 1 点，ZeroMQ 基础架构（原型设计中的更多）应该对未处理的异常（这将使 zmq.Context() 实例及其关联的 .Socket() 子级挂起）变得健壮以防cv2.imshow() 崩溃，因为它经常在快速原型设计循环中这样做。

因此，在 try: except: finally: 处理程序和显式初始 .setsockopt( zmq.LINGER, 0 ) 内的代码的彻底和自律框架在套接字实例化后立即 + 最终 @ finally: 处理程序部分内的 987654335@ + context.term() 是必须的。

---

第 1 步：通过发送纯 int-s 的 SEQ 来验证 ZeroMQ 部分

最好的第一级问题隔离是设置流，只是提供一个不受控制的整数SEQ，即从PUB一侧广播的.send( )。

...                                                   # FOR STREAMING, ALWAYS PREFER DESIGNS USING A NONBLOCKING MODE
SEQ += 1
footage_socket.send( SEQ, zmq.NOBLOCK )               # PUB.send( SEQ ) -> *SUB*
...

除非您的接收方证明它对 int-s 流的强大处理，否则继续进行是没有意义的。

...
aMsgIN = footage_socket.recv( zmq.NOBLOCK )           # FOR STREAMING, ALWAYS PREFER DESIGNS USING A NONBLOCKING MODE
print "{0:}".format( aMsgIN if len( aMsgIN ) > 0 else "." ),
# sleep(...)                                          # backthrottle the loop a bit
...

---

第 2 步：将 .imshow() 与 .recv() 数据和事件循环解耦

如果您的数据泵按需要工作，则远程显示器开始成为下一个目标。

ZeroMQ 要么提供完整的消息（一个 BLOB），要么什么都不提供。这是事实。其次，ZeroMQ 不保证任何人交付时不会出错。这些是事实，您的设计必须接受。

采集是更简单的部分，只需获取数据（也许一些色彩空间转换可能会在这里集中进行，但除此之外，任务是（对于低于 4K/低于 30fps 的图像处理）通常在这一侧没有错误。

如上所述，

frame 是一个 numpy.ndarray 实例。发送硬编码、二进制映射的 BLOB 时将获得最佳性能，无需任何“智能”转换，显然，frame 只是一大堆比特（尽管零可能有一些更高级的美味佳肴-ZeroMQ 的复制机制，但这些在发送方的这一阶段不会产生直接的好处）。

#                               struct.unpack() / .pack() seems "just"-enough for fast & smart custom Payload protocol designs
DecodeWireMSG_DATA( aMSG_DATA = struct.unpack( "!" + ( aPayloadHEADER + ( ( pxW * pxH ) * RGB_DEPTH * CELL_SIZE ) ) * "I", aMsgIN ) )

较难的部分在接收方。如果尝试使用隐藏在.imshow() 中的cv2 内置事件循环引擎，则此循环将与您通过.recv() 读取从PUB 端发布的更新流的外部逻辑发生冲突。

作为一种合理的折衷方案，可以忽略所有“延迟”frame-s，它没有使进程与PUB-side 采集/广播节奏同步，而只显示最新的.. . 在 ZeroMQ 传输基础设施上使用 zmq.CONFLATE 选项（如果重建事件流的目的只是视觉感知，则延迟的“旧”图片已经失去意义，相反，如果目的是以 1:1 记录完整的采集，zmq.CONFLATE 将丢弃应该处理的 frame 实例，因此应该为这种 1:1 的文档目的添加一些其他架构，最好与数据流/处理流的“可视化”分支）。

完成此操作后，.recv()（可能是Poller() + .recv() 循环的组合）将为SUB 端提供适当的数据泵，该数据泵独立于cv2 工具和这是.imshow()（隐藏）-FSA 事件循环。

---

架构和性能提示：

• 对于更高 fps + FullHD / 2K / 4K 项目，使用 zmq.Stopwatch() 实例的{ .start(), .stop() } 方法系统地分析cv2 处理的累积处理延迟/延迟。

• 拥有硬数据，您可能会及时发现，当出现额外需求以解决一些更难的实时限制时，请考虑：

• 原则上避免陷入任何无法控制的 Python 垃圾收集黑洞的风险 -- 始终控制 { gc.disable() | gc.enable();gc.collect() } 围绕您的关键路径部分并在其中启动显式 gc.collect()你的设计知道这是可行的。

• 避免新的内存分配延迟——可能预分配所有必要的numpy 数组，然后只强制numpy 用于那些数据修改的就地模式，从而避免任何进一步的 ad-hoc 内存管理相关的等待状态

• 通过单独、多流、独立更新整个大/（彩色）深图像的部分（条纹）来设计流量以提高抗错性（记住零保修 - 获得完整的“胖”-消息或 None )

• 调整 .Context() 的性能，使用 zmq.AFFINITY 将不同类别的 I/O 流量优先级映射到隔离的 zmq.Context( N ) I /O 线程。

• 微调 zmq.SNDBUF + zmq.SNDHWM 在 PUB 端，如果预期有多个订阅者并且 zmq.CONFLATE 未使用。

• 最后但同样重要的是，可以利用 numba.jit() LLVM 预编译加速可重用代码的关键路径功能（通常是繁重的 numpy 处理），额外的微秒被削去对你的视频处理管道带来最有益的影响，同时仍然保持纯 python 的舒适度（当然，仍然有一些cv2 警告）。

---

在原型设计阶段使用cv2 的更多提示和技巧：

可能喜欢this 以进行基于cv2 的图像处理。

可能喜欢this 为cv2 方法的简单GUI 交互式参数调整。

可能喜欢this 用于cv2 处理管道分析，zmq.Stopwatch() 详细信息到[usec]

Answer 3

帧对象包含服务器状态的内存状态，当它被发送到客户端时它会冻结，因为它接收到的帧是浅拷贝。 尝试查找如何为框架制作深层副本。