Python - 高性能视频流缓冲区

Question

如何构建可以执行 numpy 数组操作的高性能视频流缓冲区？

这是我目前的实现 - 我只是将前一个数组向前移动 1 帧并将最后一个元素分配给当前帧。

import numpy as np
import cv2
import time
cap = cv2.VideoCapture(0)
status, frame = cap.read()
buffer = np.empty([100, frame.shape[0], frame.shape[1], frame.shape[2]])
i=0
total = 100
while i < total:
    if not i: 
        start = time.time()
    status, frame = cap.read()
    t = time.time()
    if i < total/2:
        buffer[i] = frame
    else:
        buffer[:-1] = buffer[1:]
        buffer[-1] = frame
    if i == total/2: 
        middle = t
    i += 1
    # Calculations on the buffer ommitted for brevity but include mean, std, etc.

stop = time.time()
print((middle-start)/(total/2))
print((stop-middle)/(total/2))

与简单地将帧的值分配给数组的元素相比，移动数组需要大约 350 倍的时间。我知道这是因为我正在移动数组中的所有指针，这是不必要且昂贵的。保持框架井井有条很好，但不是必需的。

Answer 1

对此进行细微改进的一种非常简单的方法是使用 Python 列表进行实际的移位/追加，然后将缓冲区重新实例化为新的 NumPy 数组，如下所示：

import numpy as np
import cv2
import itertools
import time
cap = cv2.VideoCapture(0)
status, frame = cap.read()
buffer = np.empty([100, frame.shape[0], frame.shape[1], frame.shape[2]])
i=0
total = 100
middle = 0
while i < total:
    if not i: 
        start = time.time()
    status, frame = cap.read()
    t = time.time()
    if i < total/2:
        buffer[i] = frame
    else:
        list_buffer = [item for item in buffer[1:]]
        list_buffer.append(frame)
        buffer = np.asanyarray(list_buffer)
    if i == total/2: 
        middle = t
    i += 1
    # Calculations on the buffer ommitted for brevity but include mean, std, etc.

stop = time.time()
print((middle-start)/(total/2))
print((stop-middle)/(total/2))

在我的机器上，第二次总耗时从 1.7 秒减少到大约 1.36 秒。不是很大的改进，但也不是微不足道的（约 20% 的加速）。

但是，如果我们改为在整个循环中使用 list_buffer 来跟踪缓冲区的内容，并简单地在 that 上进行切片和追加：

import numpy as np
import cv2
import itertools
import time
cap = cv2.VideoCapture(0)
status, frame = cap.read()
buffer = np.empty([100, frame.shape[0], frame.shape[1], frame.shape[2]])
i=0
total = 100
middle = 0
list_buffer = []
while i < total:
    if not i: 
        start = time.time()
    status, frame = cap.read()
    t = time.time()
    if i < total/2:
        buffer[i] = frame
        list_buffer.append(frame)
    else:
        list_buffer = list_buffer[1:]
        list_buffer.append(frame)
        buffer = np.asanyarray(list_buffer)
    if i == total/2: 
        middle = t
    i += 1
    # Calculations on the buffer ommitted for brevity but include mean, std, etc.

stop = time.time()
print((middle-start)/(total/2))
print((stop-middle)/(total/2))

突然我们的输出看起来像这样：

>>> 0.08505516052246094
>>> 0.08459827899932862

希望有帮助！

Answer 2

将列表重新格式化为 numpy 数组的成本非常低。缓冲区中有 100 帧时，双端队列/链表实际上并没有效率更高。

import numpy as np
import cv2
import time
import collections

cap = cv2.VideoCapture(0)
i=0
buff_len = 100

# buffer = []                           #Standard list
# buffer = collections.deque()          #linked list 
status, frame = cap.read()              #numpy array - replaces the first frame once it reaches the last frame
buffer = np.empty([buff_len, frame.shape[0], frame.shape[1], frame.shape[2]])

times_through = 3
start = time.time()

while i < times_through*buff_len:
    t = time.time()
    status, frame = cap.read()
    # buffer.append(frame)              #list and linked list 
    buffer[i%(buff_len)] = frame        #numpy array 
    # if i >= buff_len:                 #list and linked list 
        # buffer.pop(0)                 #list  
        # buffer.popleft()              #linked list 
    if i == buff_len:
        full = t
    i += 1
    print(i, np.mean(buffer, dtype=np.int), int((time.time()-t)*100)/100.)
stop = time.time()
print((full-start)/(buff_len))
print((stop-full)/(buff_len*(times_through-1)))
print(len(buffer))

以秒/帧为单位的结果：

# list
# 0.19624330043792726
# 0.3691681241989136

# linked list
# 0.19301403045654297
# 0.3468885350227356

# numpy Array 
# 0.316677029132843
# 0.30973124504089355