我对Python中的多进程编程还不熟悉。我想从每个小时长的视频文件的每一帧中提取特征。处理每一帧需要大约30毫秒。因为每一帧都是独立处理的,所以我认为使用多进程是一个好主意。
我想将特征提取的结果存储在一个自定义类中。
我阅读了一些示例代码,并按照这里的建议使用了多进程和队列。但结果令人失望,现在每帧处理时间约为1000毫秒。我猜测我产生了大量的开销。
有更有效的方法可以并行处理帧并收集结果吗?
为了说明问题,我准备了一个虚拟示例。
我想将特征提取的结果存储在一个自定义类中。
我阅读了一些示例代码,并按照这里的建议使用了多进程和队列。但结果令人失望,现在每帧处理时间约为1000毫秒。我猜测我产生了大量的开销。
有更有效的方法可以并行处理帧并收集结果吗?
为了说明问题,我准备了一个虚拟示例。
import multiprocessing as mp
from multiprocessing import Process, Queue
import numpy as np
import cv2
def main():
#path='path\to\some\video.avi'
coordinates=np.random.random((1000,2))
#video = cv2.VideoCapture(path)
listOf_FuncAndArgLists=[]
for i in range(50):
#video.set(cv2.CAP_PROP_POS_FRAMES,i)
#img_frame_original = video.read()[1]
#img_frame_original=cv2.cvtColor(img_frame_original, cv2.COLOR_BGR2GRAY)
img_frame_dummy=np.random.random((300,300)) #using dummy image for this example
frame_coordinates=coordinates[i,:]
listOf_FuncAndArgLists.append([parallel_function,frame_coordinates,i,img_frame_dummy])
queues=[Queue() for fff in listOf_FuncAndArgLists] #create a queue object for each function
jobs = [Process(target=storeOutputFFF,args=[funcArgs[0],funcArgs[1:],queues[iii]]) for iii,funcArgs in enumerate(listOf_FuncAndArgLists)]
for job in jobs: job.start() # Launch them all
for job in jobs: job.join() # Wait for them all to finish
# And now, collect all the outputs:
return([queue.get() for queue in queues])
def storeOutputFFF(fff,theArgs,que): #add a argument to function for assigning a queue
print 'MULTIPROCESSING: Launching %s in parallel '%fff.func_name
que.put(fff(*theArgs)) #we're putting return value into queue
def parallel_function(frame_coordinates,i,img_frame_original):
#do some image processing that takes about 20-30 ms
dummyResult=np.argmax(img_frame_original)
return(resultClass(dummyResult,i))
class resultClass(object):
def __init__(self,maxIntensity,i):
self.maxIntensity=maxIntensity
self.i=i
if __name__ == '__main__':
mp.freeze_support()
a=main()
[x.maxIntensity for x in a]
multiprocessing.Pool.map
或...map_async
方法。当我必须直接使用Process
和Queue
时,我会制作自己的类似于Pool
的类,并通过向工作进程发送特定的“终止”命令来处理清理。然后,在清理期间,Process.join()
实际上是最后完成的事情,因为它们将已退出(或很快将退出)。 - Wuggy