1. cv2.dnn.readNetFromCaffe(prototxt, model) 用于进行SSD网络的caffe框架的加载
参数说明:prototxt表示caffe网络的结构文本，model表示已经训练好的参数结果
2.t=delib.correlation_tracker() 使用delib生成单目标的追踪器
3.delib.rectangle(int(box[0]), int(box[1]), int(box[2]), int(box[3])) 用于生成追踪器所需要的矩形框[(startX, startY), (endX, endY)]
4.t.start_track(rgb, rect) # 初始化生成器的开始状态
5.cv2.Writer(name, fourcc, (frame.shape[1], frame.shape[0]), True)进行图片写入到视频里面
参数说明: name表示视频的名字，fourcc表示视频格式，frame.shape[1] 表示视频的长和宽,
6.cv2.dnn.blobFromImage(frame, 0.007843, (w, h), 127.5) 对图像进行归一化操作(-1, 1)，
参数说明：frame表示输入图片，0.007843表示需要乘的数，即1/127.5，(w, h)表示图像大小，127.5表示需要减去的数
7. net.SetInput(rgb) 表示将图片输入到caffe网络中
参数说明: rgb表示已经经过归一化的图片
8. net.forward() 输出前向传播的预测结果
9. oq = multiprocessing.Queue() 生成用于多进行传输过程中的线程
10.p = multiprocessing.Process(target=start_track, args=(bb, label, rgb, iq, oq)) # 用于对函数创建进程
参数说明：target表示需要转换为进程的函数，args表示传入到进程里函数的参数
SSD是一种目标检测的算法，其使用多个卷积层进行预测，原理在后续的博客中进行补充
对于目标追踪的视频，我们先使用SSD找出图片中人物的位置，然后使用dlib中的跟踪器对物体进行跟踪
由于每一个人物框对应一个跟踪器，因此我们可以对每一个跟踪器起一个进程，使用输入和输出线程，用于构造多进程
使用的数据，需要一个训练好的SSD权重参数，还需要caffe关于SSD的prototxt文件
代码说明：
下面的代码可以近似认为是由两部分构成
第一部分：使用SSD网络进行预测，获得box的位置
第二部分：使用dlib构造tracker跟踪器，带入box构造带有矩形框的追踪器，然后使用dlib的追踪器对图像每一帧的位置进行追踪
代码：
第一步：构造进程函数，使用iq.get 和oq.put进行追踪器的位置更新
第二步：构造输入的参数, 使用cv2.dnn.readNetFromCaffe()构造SSD网络模型
第三步：使用cv2.Videocapture视频读入，fps=FPS().start() 用于计算FPS
第四步：进入循环，使用.read()读取图片
第五步：使用cv2.resize()对图片大小进行放缩变化，使用cv2.cvtColor(img, cv2.COLOR_BGR2RGB) #将读入的BGR转换为RGB，用于模型的预测
第六步：如果需要进行输出，使用cv2.VideoWriter实例化视频存储器
第七步：如果还没有使用SSD获得矩形框，使用cv2.dnn.blobFromImage对图像进行归一化操作
第八步：使用net.setInput将图片传入，使用net.forward获得前向传播输出的结果
第九步：如果置信度大于给定的置信度，获得SSD的标签，以及前向传播的位置信息
第十步：使用multiprocessing.Queue构造线程iq和oq，将线程添加到列表中，使用multiprocessing.process构造多进程，用于分别建立单个跟踪器
第十一步：如果已经生成了通道，使用iq.put(rgb)传入图像，使用oq.get()获得追踪器更新的位置
第十二步：进行画图操作，如果存在writer就进行写入
第十三步：更新fps.update
第十四步：统计运行的时间和FPS，并对vs进行释放内存
import cv2
import numpy as np
import argparse
import dlib
import multiprocessing
from utils import FPS # 第一步：构造追踪器并进行结果的更新
def start_tracker(box, label, rgb, inputQueue, outputQueue): # 构造追踪器
t = dlib.correlation_tracker()
# rect为SSD获得的矩形框的位置
rect = dlib.rectangle(int(box[0]), int(box[1]), int(box[2]), int(box[3]))
# 设置追踪器的初始位置
t.start_track(rgb, rect)
# 获得下一帧图片
while True:
# 传入的图片
rgb = inputQueue.get()
if rgb is not None:
# 更新追踪器
t.update(rgb)
# 获得追踪器的当前位置
pos = t.get_position() startX = int(pos.left())
startY = int(pos.top())
endX = int(pos.right())
endY = int(pos.bottom()) # 把结果输出放入到output里面, 返回标签和位置
outputQueue.put((label, (startX, startY, endX, endY))) # 第二步：设置参数，并使用cv2.dnn.readFrameCaffe构造SSD的网络模型
ap = argparse.ArgumentParser()
ap.add_argument('-p', '--prototxt', default='mobilenet_ssd/MobileNetSSD_deploy.prototxt',
help='path to caffe "deploy" prototxt file')
ap.add_argument('-m', '--model', default='mobilenet_ssd/MobileNetSSD_deploy.caffemodel',
help='path to Caffe pre-trained model')
ap.add_argument('-v', '--video', default='race.mp4',
help='path to input video file')
ap.add_argument('-o', '--output', type=str,
help='path to optional output video file')
ap.add_argument('-c', '--confidence', type=float, default=0.2,
help='minimu probability to filter weak detections') args = vars(ap.parse_args()) # 用于存放输入线程和输出线程
inputQueues = []
outputQueues = []
# 21种分类的结果
CLASSES = ["background", "aeroplane", "bicycle", "bird", "boat",
"bottle", "bus", "car", "cat", "chair", "cow", "diningtable",
"dog", "horse", "motorbike", "person", "pottedplant", "sheep",
"sofa", "train", "tvmonitor"] print('[INFO] loading model...')
# 构造SSD网络模型
net =cv2.dnn.readNetFromCaffe(args['prototxt'], args['model']) print('[INFO] starting video stream...')
# 第三步：使用cv2.VideoCapture读取视频
vs = cv2.VideoCapture(args['video'])
writer = None fps = FPS().start() if __name__ == '__main__':
# 第四步：进入循环，使用.read() 读取图片
while True:
ret, frame = vs.read() if frame is None:
break # 第五步：进行图像的维度变化, 并且将BGR转换为RGB格式
h, w = frame.shape[:2]
width = 600
r = width / float(w)
dim = (width, int(r*h))
frame = cv2.resize(frame, dim, cv2.INTER_AREA)
rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # 第六步：进行视频的保存
if args['output'] is not None and writer is None:
fourcc = cv2.VideoWriter_fourcc(*'MJPG')
writer = cv2.VideoWriter(args['output'], fourcc,
(frame.shape[1], frame.shape[0]), True) # 如果输入进程的维度为0，进入循环首先检测位置
if len(inputQueues) == 0:
# 第七步：使用cv2.dnn.blobFromImage()对图片进行归一化操作
(h, w) = frame.shape[:2]
blob = cv2.dnn.blobFromImage(frame, 0.007843, (w, h), 127.5)
# 第八步：使用net.setInput输入图片，net.forward()获得前向传播的结果
net.setInput(blob)
detections = net.forward()
# 第九步：对结果进行循环，如果置信度大于阈值，则获得其标签和box位置信息
for i in np.arange(0, detections.shape[2]):
confidence = detections[0, 0, i, 2]
if confidence > args['confidence']:
idx = int(detections[0, 0, i, 1])
label = CLASSES[idx] if label != 'person':
continue box = detections[0, 0, i, 3:7] * np.array([w, h, w, h])
(startX, startY, endX, endY) = box.astype('int')
bb = (startX, startY, endX, endY) # 第十步：创建输入q和输出q，创建process线程,使用process.start()启动线程
iq = multiprocessing.Queue()
oq = multiprocessing.Queue()
inputQueues.append(iq)
outputQueues.append(oq) # 在多个核上运行, 创建多核
p = multiprocessing.Process(
target=start_tracker,
args=(bb, label, rgb, iq, oq)
)
p.daemon = True
p.start() cv2.rectangle(frame, (startX, startY), (endX, endY),
(0, 255, 0), 2)
cv2.putText(frame, label, (startX, startY-15),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 0), 2) else:
# 第十一步：如果生成了进程，循环输入线程，传入rgb图片，获得输出线程的label和更新位置的输出
for iq in inputQueues:
iq.put(rgb) for oq in outputQueues:
(label, (startX, startY, endX, endY)) = oq.get()
# 在frame图像上绘制矩形框和text
cv2.rectangle(frame, (startX, startY), (endX, endY),
(0, 255, 0), 2)
cv2.putText(frame, label, (startX, startY - 15),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (0, 255, 0), 2)
# 第十二步：进行绘图，并进行视频的写入操作
if writer is not None:
writer.write(frame) cv2.imshow('Frame', frame)
key = cv2.waitKey(1) & 0xFF
if key == 27:
break
# 第十三步 fps更新
fps.update()
# 第十四步：统计运行时间和FPS，释放内存
fps.stop()
print('[INFO] elapsed time {:.2f}'.format(fps.elapsed()))
print('[INFO] approx. FPS:{:.2f}'.format(fps.fps())) if writer is not None:
writer.release() cv2.destroyAllWindows()
vs.release()
效果展示：
FPS副代码
import cv2
import numpy as np
import datetime class FPS: def __init__(self): self._start = None
self._end = None
self._numFrames = 0 def start(self):
# start the timer
self._start = datetime.datetime.now()
return self def stop(self):
# stop the timer
self._end = datetime.datetime.now() def update(self):
self._numFrames += 1 def elapsed(self):
return (self._end - self._start).total_seconds() def fps(self):
return self._numFrames / self.elapsed()
