python - 如何在Python中将实时数据读入循环与更多处理密集型操作分开？

Question

因此，我有一个OpenCV网络摄像头供稿，希望尽快读取它的帧。由于使用了Python GIL，脚本在帧中读取的最快速度似乎如下:

#Parent or maybe client(?) script

#initilize the video capture object
cam  = cv2.VideoCapture(0)

while True:
    ret, frame = cam.read()

    # Some code to pass this numpy frame array to another python script 
    # (which has a queue) that is not under time constraint and also
    # is doing some more computationally intensive post-processing...

    if exit_condition is True:
        break

我想发生的事情是将这些帧(Numpy数组)添加到第二个Python脚本(或可能是多处理实例)中的某种处理队列中，然后将执行一些不在像cam.read()循环这样的时间限制是...

因此，基本思路如下所示:

实时(或尽快获取)数据收集(摄像头读取)脚本
---->
分析脚本(将进行后处理，写入结果并生成matplotlib图，该图会落后于数据收集)

我已经做过一些研究，看起来像:管道，套接字，pyzmq和python多处理都可以实现我想要的东西。问题是我没有上述任何经验。

所以我的问题是，哪种方法最能实现我想要的目标，并且有人可以提供一个简短的示例，甚至是一些想法/想法来指出正确的方向吗？

非常感谢!

编辑:非常感谢史蒂夫让我开始正确的道路。这是我所想到的一个工作要点...代码可以正常工作，但是如果添加更多的后处理步骤，则队列大小可能会比主进程可以通过的速度更快。限制帧速率的建议很可能会成为我最终使用的策略。

import time
import cv2
import multiprocessing as mp

def control_expt(connection_obj, q_obj, expt_dur):

    def elapsed_time(start_time):
        return time.clock()-start_time

    #Wait for the signal from the parent process to begin grabbing frames
    while True:
        msg = connection_obj.recv()     
        if msg == 'Start!':
            break    

    #initilize the video capture object
    cam  = cv2.VideoCapture(cv2.CAP_DSHOW + 0)  

    #start the clock!!
    expt_start_time = time.clock() 

    while True:
        ret, frame = cam.read()          
        q_obj.put_nowait((elapsed_time(expt_start_time), frame))

        if elapsed_time(expt_start_time) >= expt_dur:
            q_obj.put_nowait((elapsed_time(expt_start_time),'stop'))
            connection_obj.close()
            q_obj.close()
            cam.release()
            break

class test_class(object):
    def __init__(self, expt_dur):

        self.parent_conn, self.child_conn = mp.Pipe()
        self.q  = mp.Queue()
        self.control_expt_process = mp.Process(target=control_expt, args=(self.child_conn, self.q, expt_dur))
        self.control_expt_process.start()

    def frame_processor(self):
        self.parent_conn.send('Start!')

        prev_time_stamp = 0

        while True:
           time_stamp, frame = self.q.get()                        
           #print (time_stamp, stim_bool)          
           fps = 1/(time_stamp-prev_time_stamp)
           prev_time_stamp = time_stamp      

            #Do post processing of frame here but need to be careful that q.qsize doesn't end up growing too quickly...
            print (int(self.q.qsize()), fps)

            if frame == 'stop':
                print 'destroy all frames!'
                cv2.destroyAllWindows()
                break               
            else:
                cv2.imshow('test', frame)        
                cv2.waitKey(30)

        self.control_expt_process.terminate()

if __name__ == '__main__':  
    x = test_class(expt_dur = 60)
    x.frame_processor()

Answer 1

多处理文档是一个很好的起点。 https://docs.python.org/2/library/multiprocessing.html#exchanging-objects-between-processes
即使您现在可能不了解，我也建议您阅读。

在您提到的其他技术上使用管道可以使您保持性能并使代码保持简单。

以下是一些我尚未测试过的代码，它们应该为您提供一个良好的起点。

from multiprocessing import Process, Pipe

def read_frames(connection_obj):
  #initilize the video capture object
  cam  = cv2.VideoCapture(0)
  while True:
    ret, frame = cam.read()
    connection_obj.send(frame) # is this what you want to send?

    if exit_condition is True:
        connection_obj.close()
        break

def launch_read_frames(connection_obj):
    """
    Starts the read_frames function in a separate process.
    param connection_obj: pipe to send frames into.
    Returns a pipe object
    """
    read_frames_process = Process(target=read_frames, args=(connection_obj,)) # this trailing comma is intentional
    read_frames_process.start()
    read_frames_process.join()

    return parent_conn

def act_on_frames(connection_obj):
    while True:
        frame = connection_obj.recv()
        # Do some stuff with each frame here

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    launch_read_frames(child_conn)

    # You could also call this function as a separate process, though in
    # this instance I see no performance benefit.
    act_on_frames(parent_conn)