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()
微信扫一扫 
支付宝扫一扫 