前言
前段時間由於業務需要，做了一個接入攝像頭的視頻流然后做人臉識別的項目！！我們選用的是Caffe(卷積神經網絡框架)+SSD算法+OpenCV！！開始我們的處理是邏輯是一幀一幀的處理，但是測試的時候發現圖像處理的速度很慢，而且准確度也不高！后來我們就換了一套方案，利用多線程來實現，然而效果也並不是很理想！所以我們才想探索一下能不能實現多張圖像一起處理！今天就記錄一下我在實現過程中的一些問題和一些經驗！
首先來簡單的說明一下所選的工具！
caffe的全稱是Convolutional Architecture for Fast Feature Embedding（譯為：快速特征嵌入的卷積體系結構），核心語言是C++。caffe的基本工作流程是設計建立在神經網絡的一個簡單假設，所有的計算都是層的形式表示的，網絡層所做的事情就是輸入數據，然后輸出計算結果。比如卷積就是輸入一幅圖像，然后和這一層的參數（filter）做卷積，最終輸出卷積結果。每層需要兩種函數計算，一種是forward，從輸入計算到輸出；另一種是backward，從上層給的gradient來計算相對於輸入層的gradient。這兩個函數實現之后，我們就可以把許多層連接成一個網絡，這個網絡輸入數據（圖像，語音或其他原始數據），然后計算需要的輸出（比如識別的標簽）。在訓練的時候，可以根據已有的標簽計算loss和gradient，然后用gradient來更新網絡中的參數。
目標檢測近年來已經取得了很重要的進展，主流的算法主要分為兩個類型：（1）two-stage方法，如R-CNN系算法，其主要思路是先通過啟發式方法（selective search）或者CNN網絡（RPN)產生一系列稀疏的候選框，然后對這些候選框進行分類與回歸，two-stage方法的優勢是准確度高；（2）one-stage方法，如Yolo和SSD，其主要思路是均勻地在圖片的不同位置進行密集抽樣，抽樣時可以采用不同尺度和長寬比，然后利用CNN提取特征后直接進行分類與回歸，整個過程只需要一步，所以其優勢是速度快，但是均勻的密集采樣的一個重要缺點是訓練比較困難，這主要是因為正樣本與負樣本（背景）極其不均衡（參見Focal Loss），導致模型准確度稍低。不同算法的性能如圖1所示，可以看到兩類方法在准確度和速度上的差異。
OpenCV是一個基於BSD許可（開源）發行的跨平台計算機視覺和機器學習軟件庫，可以運行在Linux、Windows、Android和Mac OS操作系統上。 [1] 它輕量級而且高效——由一系列 C 函數和少量 C++ 類構成，同時提供了Python、Ruby、MATLAB等語言的接口，實現了圖像處理和計算機視覺方面的很多通用算法
首先來介紹一下blobFromImages函數的參數
image：這個就是我們將要輸入神經網絡進行處理或者分類的圖片。
mean：需要將圖片整體減去的平均值，如果我們需要對RGB圖片的三個通道分別減去不同的值，那么可以使用3組平均值，如果只使用一組，那么就默認對三個通道減去一樣的值。減去平均值（mean）：為了消除同一場景下不同光照的圖片，對我們最終的分類或者神經網絡的影響，我們常常對圖片的R、G、B通道的像素求一個平均值，然后將每個像素值減去我們的平均值，這樣就可以得到像素之間的相對值，就可以排除光照的影響。
scalefactor：當我們將圖片減去平均值之后，還可以對剩下的像素值進行一定的尺度縮放，它的默認值是1，如果希望減去平均像素之后的值，全部縮小一半，那么可以將scalefactor設為1/2。
size：這個參數是我們神經網絡在訓練的時候要求輸入的圖片尺寸。
swapRB：OpenCV中認為我們的圖片通道順序是BGR，但是我平均值假設的順序是RGB，所以如果需要交換R和G，那么就要使swapRB=true
實現代碼
from imutils.video import VideoStream from imutils.video import FPS import numpy as np import argparse import imutils import time import cv2 import threading import queue from flask import Flask, render_template, Response
#獲取輸入參數 ap = argparse.ArgumentParser() ap.add_argument("-p", "--prototxt", required=True, help="path to Caffe 'deploy' prototxt file") ap.add_argument("-m", "--model", required=True, help="path to Caffe pre-trained model") ap.add_argument("-v", "--video", required=True, help="path to Caffe video file") ap.add_argument("-c", "--confidence", type=float, default=0.2, help="minimum probability to filter weak detections") ap.add_argument("-t", "--threading", type=int, default=1, help="threading number") args = vars(ap.parse_args())
#SSD算法能夠識別的物體列表 CLASSES = ["background", "aeroplane", "bicycle", "bird", "boat", "bottle", "bus", "car", "cat", "chair", "cow", "diningtable", "dog", "horse", "motorbike", "person", "pottedplant", "sheep", "sofa", "train", "tvmonitor"] #為識別出來的物體畫框着色 COLORS = np.random.uniform(0, 255, size=(len(CLASSES), 3))
print("[INFO] starting video stream...") # 通過opencv獲取實時視頻流 class VideoCamera(object): def __init__(self):
#從視頻流抽幀放在列表中 self.Frame = []
#處理圖片的列表 self.Images = []
#每次處理圖像數 self.Nu = 10 # 控制圖片的結果賽選 self.NuDNN =50 self.done_Images = queue.Queue(maxsize=500) self.FrameQueue = queue.Queue(maxsize=self.Nu) self.ImageQueue = queue.Queue(maxsize=500) self.dataQueue = queue.Queue(maxsize=500) self.status = False self.isstop = False self.video = cv2.VideoCapture(args["video"]) def __del__(self): self.video.release() def start(self): # 把程序放進子線程，daemon=True 表示該線程會隨着主線程關閉而關閉。 print('ipcam started!')
#視頻流的圖像線程 threading.Thread(target=self.queryframe, daemon=True, args=()).start() print('dnn_frame started')
#圖像處理線程 threading.Thread(target=self.dnn_frame, daemon=True, args=()).start() print('get_frame started')
#圖像返回視頻流線程 threading.Thread(target=self.get_frame, daemon=True, args=()).start() def stop(self): # 記得要設計停止無限循環的開關。 self.isstop = True print('ipcam stopped!') #從視頻流讀取幀的線程 def queryframe(self): while (not self.isstop):
#讀取視頻流的幀並放在列表中 self.status, self.Frame = self.video.read() self.Frame = imutils.resize(self.Frame, width=1080) if self.status: self.video.grab() if self.FrameQueue.full(): self.FrameQueue.get() self.FrameQueue.put(self.Frame) else: self.FrameQueue.put(self.Frame) self.video.release() #圖像處理線程 def dnn_frame(self): time.sleep(1) print("[INFO] loading model...") net = cv2.dnn.Net_readFromModelOptimizer(args["prototxt"], args["model"]) # 設置運算設備 測試選用的CPU net.setPreferableTarget(cv2.dnn.DNN_TARGET_CPU) while (not self.isstop): start_time = time.time() Images=[] while (len(Images)) < self.Nu: image2 = self.FrameQueue.get() Images.append(image2)
#將圖像列表封裝成一個blob
blob = cv2.dnn.blobFromImages(Images, 0.007843, (384,672), 127.5,True)
#將blob送進模型進行識別 net.setInput(blob)
#拿到識別后的結果 detections = net.forward()
#這里新建一個列表是為了保證 識別結果和待識別圖像進行--對應 data=[] data.append(Images) data.append(detections) if self.dataQueue.full(): self.dataQueue.get() self.dataQueue.put(data) else: self.dataQueue.put(data) #將識別結果在原圖像上畫框的線程 def get_frame(self): time.sleep(4) n=0 while (not self.isstop): dataList=self.dataQueue.get() detections = dataList[1]#結果 image_list = dataList[0]#圖像 # 因為opencv讀取的圖片並非jpeg格式，因此要用motion JPEG模式需要先將圖片轉碼成jpg格式圖片 (h, w) = image_list[0].shape[:2]
#所有預測結果種取 self.NuDNN*self.Nu種 for i in range(0, self.NuDNN*self.Nu):
#拿到預測結果的置信度 confidence = detections[0, 0, i, 2]
#只有大於我輸入的置信度的結果才保留 if confidence > args["confidence"]:
#預測結果的下標 idx = int(detections[0, 0, i, 1])
#原圖片的id imageIndex = int(detections[0, 0, i, 0])
#在原圖上畫框 box = detections[0, 0, i, 3:7] * np.array([w, h, w, h]) (startX, startY, endX, endY) = box.astype("int") label = "{}: {:.2f}%".format(CLASSES[idx], confidence * 100) cv2.rectangle(image_list[imageIndex], (startX, startY), (endX, endY), COLORS[idx], 2) y = startY - 15 if startY - 15 > 15 else startY + 15 cv2.putText(image_list[imageIndex], label, (startX, y), cv2.FONT_HERSHEY_SIMPLEX, 0.5, COLORS[idx], 2) for x in range(0,self.Nu): ret, jpeg = cv2.imencode('.jpg', image_list[x]) if self.done_Images.full(): self.done_Images.get() self.done_Images.put(jpeg.tobytes()) else: self.done_Images.put(jpeg.tobytes()) app = Flask(__name__) @app.route('/') # 主頁 def index(): # jinja2模板，具體格式保存在index.html文件中 return render_template('index.html') def gen(camera): while True: if not camera.done_Images.empty(): frame = camera.done_Images.get() time.sleep(0.2) # 使用generator函數輸出視頻流， 每次請求輸出的content類型是image/jpeg yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + frame + b'\r\n\r\n') @app.route('/video_feed') # 這個地址返回視頻流響應 def video_feed(): global ipcam return Response(gen(ipcam), mimetype='multipart/x-mixed-replace; boundary=frame') if __name__ == '__main__': # 連接攝影機 ipcam = VideoCamera() # 啟動子線程 print("init") ipcam.start() time.sleep(4) app.run(host='0.0.0.0', debug=False, port=5000)
這里需要特別說明一下圖像識別后的結果detections，網上對blobFromImage函數講解的很詳細，但是對於輸出的結果幾乎沒有介紹！小編也是摸索了好久才把結果和圖片一一對應起來！
detections其實是一個四維數組，形式為（0, 0, n, s），如果直接循環取出里面的圖片進行計算識別，是不行的！這樣只會降低形式為detections的維度，先解釋一下單張圖片與多張圖片的blob的區別：
單張圖片blob：(1,3,224,224)
多張圖片blob：(10,3,224,224)
結果元組的格式如下(num_images, num_channels, width, height)
然后一張圖片默認情況下會預測100個結果，那么我們一次處理10張圖片就有1000個結果，並且會根據置信度(預測結果的可能性的大小)排序！
n表示第幾種結果(最大為100*圖片數)，s表示這個結果的多個屬性值，最大為6
0 表示image_id 圖片的id
1 表示label 預測結果的classes下標值
2 表示conf 置信度即預測結果的可能性
3-6表示這個圖片的的最大最小長寬
總結
至此，python的多線程進行人臉識別的功能就開發完成！！！但是，這僅僅是實現了功能而已！效果並不一定是最優的！后續可能還需要調整視頻顯示的幀數，一次處理多少張圖片比較合理，預測結果取多少種合適。。。。。個人認為最重要的是選取合適的模型和算法！我這里選取的SSD和Caffe並不一定是最優的！！！！
原創文章，轉載請標注出處！！！！！！