我正在读取 OSX 上的网络摄像头，使用这个简单的脚本可以正常工作:

import cv2
camera = cv2.VideoCapture(0)

while True:
    try:
        (grabbed, frame) = camera.read()  # grab the current frame
        frame = cv2.resize(frame, (640, 480))  # resize the frame
        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:
        # cleanup the camera and close any open windows
        camera.release()
        cv2.destroyAllWindows()
        print "\n\nBye bye\n"
        break

我现在想在一个单独的进程中阅读视频，我有一个更长的脚本，它在 Linux 上的一个单独的进程中正确读出视频:

import numpy as np
import time
import ctypes
import argparse

from multiprocessing import Array, Value, Process
import cv2


class VideoCapture:
    """
    Class that handles video capture from device or video file
    """
    def __init__(self, device=0, delay=0.):
        """
        :param device: device index or video filename
        :param delay: delay between frame captures in seconds(floating point is allowed)
        """
        self._cap = cv2.VideoCapture(device)
        self._delay = delay

    def _proper_frame(self, delay=None):
        """
        :param delay: delay between frames capture(in seconds)
        :param finished: synchronized wrapper for int(see multiprocessing.Value)
        :return: frame
        """
        snapshot = None
        correct_img = False
        fail_counter = -1
        while not correct_img:
            # Capture the frame
            correct_img, snapshot = self._cap.read()
            fail_counter += 1
            # Raise exception if there's no output from the device
            if fail_counter > 10:
                raise Exception("Capture: exceeded number of tries to capture the frame.")
            # Delay before we get a new frame
            time.sleep(delay)
        return snapshot

    def get_size(self):
        """
        :return: size of the captured image
        """
        return (int(self._cap.get(int(cv2.CAP_PROP_FRAME_HEIGHT))),
                int(self._cap.get(int(cv2.CAP_PROP_FRAME_WIDTH))), 3)

    def get_stream_function(self):
        """
        Returns stream_function object function
        """

        def stream_function(image, finished):
            """
            Function keeps capturing frames until finished = 1
            :param image: shared numpy array for multiprocessing(see multiprocessing.Array)
            :param finished: synchronized wrapper for int(see multiprocessing.Value)
            :return: nothing
            """
            # Incorrect input array
            if image.shape != self.get_size():
                raise Exception("Capture: improper size of the input image")
            print("Capture: start streaming")
            # Capture frame until we get finished flag set to True
            while not finished.value:
                image[:, :, :] = self._proper_frame(self._delay)
            # Release the device
            self.release()

        return stream_function

    def release(self):
        self._cap.release()


def main():
    # Add program arguments
    parser = argparse.ArgumentParser(description='Captures the video from the webcamera and \nwrites it into the output file with predefined fps.', formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    parser.add_argument('-output', dest="output",  default="output.avi", help='name of the output video file')
    parser.add_argument('-log', dest="log",  default="frames.log", help='name of the log file')
    parser.add_argument('-fps', dest="fps",  default=25., help='frames per second value')

    # Read the arguments if any
    result = parser.parse_args()
    fps = float(result.fps)
    output = result.output
    log = result.log

    # Initialize VideoCapture object and auxilary objects
    cap = VideoCapture()
    shape = cap.get_size()
    stream = cap.get_stream_function()

    # Define shared variables(which are synchronised so race condition is excluded)
    shared_array_base = Array(ctypes.c_uint8, shape[0] * shape[1] * shape[2])
    frame = np.ctypeslib.as_array(shared_array_base.get_obj())
    frame = frame.reshape(shape[0], shape[1], shape[2])
    finished = Value('i', 0)

    # Start processes which run in parallel
    video_process = Process(target=stream, args=(frame, finished))
    video_process.start()  # Launch capture process

    # Sleep for some time to allow videocapture start working first
    time.sleep(2)

    # Termination function
    def terminate():
        print("Main: termination")
        finished.value = True
        # Wait for all processes to finish
        time.sleep(1)
        # Terminate working processes
        video_process.terminate()

    # The capturing works until keyboard interrupt is pressed.
    while True:
        try:
            # Display the resulting frame
            cv2.imshow('frame', frame)
            cv2.waitKey(1)  # Display it at least one ms before going to the next frame
            time.sleep(0.1)

        except KeyboardInterrupt:
            cv2.destroyAllWindows()
            terminate()
            break

if __name__ == '__main__':
    main()

这在 Linux 上运行良好，但在 OSX 上我遇到了麻烦，因为它似乎无法执行 .read()在创建cv2.VideoCapture(device)对象(存储在 var self._cap 中)。

经过一番搜索，我找到了 this SO answer , 建议使用 Billiard ，pythons 多处理的替代品，据说有一些非常有用的改进。所以在文件的顶部，我只是在我之前的多处理导入之后添加了导入(有效地覆盖了 multiprocessing.Process ):

from billiard import Process, forking_enable

就在 video_process 的实例化之前我使用的变量 forking_enable如下:

forking_enable(0)  # Supposedly this is all I need for billiard to do it's magic
video_process = Process(target=stream, args=(frame, finished))

所以在这个版本 ( here on pastebin ) 中，我再次运行该文件，这给了我这个错误:

pickle.PicklingError: Can't pickle : it's not found as main.stream_function

搜索该错误后我找到了 an SO question with a long list of answers其中一个给了我使用 dill serialization lib 的建议解决这个问题。但是，该库应该与 Pathos multiprocessing fork 一起使用.所以我只是尝试更改我的多处理导入行

from multiprocessing import Array, Value, Process

到

from pathos.multiprocessing import Array, Value, Process

但是 Array 都没有, Value和 Process似乎存在于 pathos.multiprocessing包。

从这一点来看，我完全迷路了。我正在搜索我几乎没有足够知识的东西，我什至不知道我需要向哪个方向搜索或调试。

那么，有没有比我聪明的人可以帮助我在单独的过程中捕捉视频？欢迎所有提示!

最佳答案

您的第一个问题是您无法在 forked 进程中访问网络摄像头。当外部库与 fork 一起使用时会出现几个问题，因为 fork 操作不会清除父进程打开的所有文件描述符，从而导致奇怪的行为。该库通常对 Linux 上的此类问题更为稳健，但在两个进程之间共享诸如 cv2.VideoCapture 之类的 IO 对象并不是一个好主意。

当您使用 billard.forking_enabled 并将其设置为 False 时，您要求库不要使用 fork 来生成新进程，但是 spawn 或 forkserver 方法，它们关闭所有文件描述符时更干净，但启动速度也较慢，这在您的情况下应该不是问题。如果您使用的是 python3.4+，您可以使用 multiprocessing.set_start_method('forkserver') 来执行此操作。

当您使用这些方法之一时，目标函数和参数需要序列化才能传递给子进程。默认情况下，序列化是使用 pickle 完成的，它有多个流程，如您提到的那样，无法序列化本地定义的对象以及 cv2.VideoCapture。但是您可以简化您的程序，使您的 Process 的所有参数都可以 picklelisable。这是解决您的问题的尝试:

import numpy as np
import time
import ctypes

from multiprocessing import set_start_method
from multiprocessing import Process, Array, Value
import cv2


class VideoCapture:
    """
    Class that handles video capture from device or video file
    """
    def __init__(self, device=0, delay=0.):
        """
        :param device: device index or video filename
        :param delay: delay between frame captures in seconds(float allowed)
        """
        self._delay = delay
        self._device = device
        self._cap = cv2.VideoCapture(device)
        assert self._cap.isOpened()

    def __getstate__(self):
        self._cap.release()
        return (self._delay, self._device)

    def __setstate__(self, state):
        self._delay, self._device = state
        self._cap = cv2.VideoCapture(self._device)
        assert self._cap.grab(), "The child could not grab the video capture"

    def _proper_frame(self, delay=None):
        """
        :param delay: delay between frames capture(in seconds)
        :param finished: synchronized wrapper for int
        :return: frame
        """
        snapshot = None
        correct_img = False
        fail_counter = -1
        while not correct_img:
            # Capture the frame
            correct_img, snapshot = self._cap.read()
            fail_counter += 1
            # Raise exception if there's no output from the device
            if fail_counter > 10:
                raise Exception("Capture: exceeded number of tries to capture "
                                "the frame.")
            # Delay before we get a new frame
            time.sleep(delay)
        return snapshot

    def get_size(self):
        """
        :return: size of the captured image
        """
        return (int(self._cap.get(int(cv2.CAP_PROP_FRAME_HEIGHT))),
                int(self._cap.get(int(cv2.CAP_PROP_FRAME_WIDTH))), 3)

    def release(self):
        self._cap.release()


def stream(capturer, image, finished):
    """
    Function keeps capturing frames until finished = 1
    :param image: shared numpy array for multiprocessing
    :param finished: synchronized wrapper for int
    :return: nothing
    """
    shape = capturer.get_size()

    # Define shared variables
    frame = np.ctypeslib.as_array(image.get_obj())
    frame = frame.reshape(shape[0], shape[1], shape[2])

    # Incorrect input array
    if frame.shape != capturer.get_size():
        raise Exception("Capture: improper size of the input image")
    print("Capture: start streaming")
    # Capture frame until we get finished flag set to True
    while not finished.value:
        frame[:, :, :] = capturer._proper_frame(capturer._delay)

    # Release the device
    capturer.release()


def main():

    # Initialize VideoCapture object and auxilary objects
    cap = VideoCapture()
    shape = cap.get_size()

    # Define shared variables
    shared_array_base = Array(ctypes.c_uint8, shape[0] * shape[1] * shape[2])
    frame = np.ctypeslib.as_array(shared_array_base.get_obj())
    frame = frame.reshape(shape[0], shape[1], shape[2])
    finished = Value('i', 0)

    # Start processes which run in parallel
    video_process = Process(target=stream,
                            args=(cap, shared_array_base, finished))
    video_process.start()  # Launch capture process

    # Sleep for some time to allow videocapture start working first
    time.sleep(2)

    # Termination function
    def terminate():
        print("Main: termination")
        finished.value = True
        # Wait for all processes to finish
        time.sleep(1)
        # Terminate working processes
        video_process.join()

    # The capturing works until keyboard interrupt is pressed.
    while True:
        try:
            # Display the resulting frame
            cv2.imshow('frame', frame)
            # Display it at least one ms before going to the next frame
            time.sleep(0.1)
            cv2.waitKey(1)

        except KeyboardInterrupt:
            cv2.destroyAllWindows()
            terminate()
            break


if __name__ == '__main__':
    set_start_method("spawn")
    main()

我目前无法在 mac 上测试它，所以它可能无法开箱即用，但不应该有 multiprocessing 相关错误。一些注意事项:

• 我在新子对象中实例化 cv2.VideoCapture 对象并抓取摄像头，因为只有一个进程应该从摄像头读取数据。
• 也许您使用 fork 的第一个程序中的问题只是由于共享 cv2.VideoCapture 并在 stream 函数中重新创建它会解决您的问题。
• 您不能将 numpy 包装器传递给 child ，因为它不会共享 mp.Array 缓冲区(这真的很奇怪，我花了一段时间才弄清楚)。您需要显式传递 Array 并重新创建包装器。

• 也许您使用 fork 的第一个程序中的问题只是由于共享 cv2.VideoCapture 并在 stream 中重新创建它> 功能将解决您的问题。

• 我假设您在 python3.4+ 中运行代码，所以我没有使用 billard，而是使用了 forking_enabled(False) 而不是 set_start_method 应该有点相似。

让我知道这是否有效!

关于python - 如何在 OSX 上的单独进程中读取网络摄像头？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43572744/