Python3并行处理opencv视频帧

Question

我有一个视频文件，我需要逐帧处理，然后需要在帧中显示结果。目前我正在按顺序进行处理并一一显示帧。

现在我想并行处理帧而不是顺序处理。一旦处理了 X 个帧，则 cv2.imshow 必须出现，并且必须以正确的顺序显示已处理的帧。

目前我的顺序代码如下所示

import cv2
import requests


def process_frame(bgr_image, jpg_as_text):
    try:
        # Post to api for processing and get the results
        # result = requests.post("example.com", data={"jpg": jpg_as_text})

        # Add results to bgr_image
        # cv2.putText()
    except Exception as e:
        print(e)
        pass
    # Show the frame
    cv2.imshow("frame", bgr_image)


video = cv2.VideoCapture("video.mp4")
i = 0


while video.isOpened():
    ret, bgr_image = video.read()
    if ret == True:
        img_height, img_width, _ = bgr_image.shape
        jpg_as_text = cv2.imencode(".jpg", bgr_image)[1].tostring()
        process_frame(bgr_image, jpg_as_text)
        print(i)
        i += 1
    else:
        break
    if cv2.waitKey(1) & 0xFF == ord("q"):
        break

video.release()
cv2.destroyAllWindows()

现在我应该重构什么来进行并行处理并在处理 X 帧后预览帧。

Answer 1

OpenCV 在其 github 存储库中有一个多线程视频处理示例。

https://github.com/opencv/opencv/blob/master/samples/python/video_threaded.py

从 multiprocessing.pool 导入 ThreadPool 并为每个 cpu 内核启动一个新线程。 OpenCV 有一个函数叫做 getNumberOfCPUs()

例子：

from __future__ import print_function

import numpy as np
import cv2 as cv

from multiprocessing.pool import ThreadPool
from collections import deque

from common import clock, draw_str, StatValue
import video


class DummyTask:
    def __init__(self, data):
    self.data = data
def ready(self):
    return True
def get(self):
    return self.data

if __name__ == '__main__':
    import sys

    print(__doc__)

    try:
        fn = sys.argv[1]
    except:
        fn = 0
    cap = video.create_capture(fn)


    def process_frame(frame, t0):
    # some intensive computation...
        frame = cv.medianBlur(frame, 19)
        frame = cv.medianBlur(frame, 19)
        return frame, t0

    threadn = cv.getNumberOfCPUs()
    pool = ThreadPool(processes = threadn)
    pending = deque()

    threaded_mode = True

    latency = StatValue()
    frame_interval = StatValue()
    last_frame_time = clock()
    while True:
        while len(pending) > 0 and pending[0].ready():
            res, t0 = pending.popleft().get()
            latency.update(clock() - t0)
            draw_str(res, (20, 20), "threaded      :  " +             str(threaded_mode))
            draw_str(res, (20, 40), "latency        :  %.1f ms" %     (latency.value*1000))
            draw_str(res, (20, 60), "frame interval :  %.1f ms" % (frame_interval.value*1000))
            cv.imshow('threaded video', res)
        if len(pending) < threadn:
            ret, frame = cap.read()
            t = clock()
            frame_interval.update(t - last_frame_time)
            last_frame_time = t
            if threaded_mode:
                task = pool.apply_async(process_frame, (frame.copy(), t))
            else:
                task = DummyTask(process_frame(frame, t))
            pending.append(task)
        ch = cv.waitKey(1)
        if ch == ord(' '):
            threaded_mode = not threaded_mode
        if ch == 27:
            break
    cv.destroyAllWindows()

您应该能够使用该示例代码并将您的图像处理放在 process_frame 函数中。

在循环中添加一个计数器并在 count == X 时调用 cv2.imshow