边缘设备的实时 AI 人员检测：实时模式下检测视频中的人员

• 下载数据 - 19.3 MB
• 下载模型 - 43.5 MB
• 下载结果 - 36.66 MB

这是七篇文章系列的最后一篇。到目前为止，我们已经有了一个用于人脸检测的DNN模型，以及在树莓派设备上运行该模型的Python代码。我们使用MobileNet DNN构建的人工智能算法在Pi 3B板上的平均帧率约为1.25 FPS。正如我们在上一篇文章中所示，这个速度足以在直播摄像机的视频流中检测到人的出现。但我们的算法在传统意义上并非为实时而设计。

视频监控系统中的直播摄像头通常以15-30 FPS的速度运行。我们的算法以1.25 FPS的速度处理帧，实在太慢了。是什么导致它变慢？让我们看看我们的代码来寻找线索。

这是我们在之前的一篇文章中开发的代码片段

fps.start()
obj_data = self.ssd.detect(frame)
persons = self.ssd.get_objects(frame, obj_data, class_num, min_confidence)
fps.stop()

正如你所见，我们只测量了人脸检测操作（在帧上运行SSD模型）的速度。所以这个操作需要0.8秒/帧才能执行。算法的其余部分足够快，可以以实时视频系统的速度运行。

考虑到这一点，我们能否重新设计我们的算法以适应处理实时视频流所需的高速？让我们试试。

在算法的当前版本中，我们顺序运行SSD模型处理（约0.8秒/帧）以及其余可以实时运行的代码。因此，较慢的部分决定了整体速度。将算法适应实时模式的最简单方法是将其分成几部分并在并行运行这些部分。“快速”部分将以其实时速度运行，“慢速”部分将以1.25 FPS运行。但由于这些部分将异步运行，我们可以处理“慢速”部分（SSD模型）以及只有那些值得处理的帧。

看看下面的代码。这是一个实现异步处理思想的新类

# Real-Time Video detector
import sys
from multiprocessing import Process
from multiprocessing import Queue
 
def detect_in_process(proto, model, ssd_proc, frame_queue, person_queue, class_num, min_confidence):
    ssd_net = CaffeModelLoader.load(proto, model)
    ssd = SSD(ssd_proc, ssd_net)
    while True:
        if not frame_queue.empty():
            frame = frame_queue.get()
         	obj_data = ssd.detect(frame)
            persons = ssd.get_objects(frame, obj_data, class_num, min_confidence)
            person_queue.put(persons)
 
class RealtimeVideoSSD:	
    def __init__(self, proto, model, ssd_proc):
        self.ssd_proc = ssd_proc
    	  self.proto = proto
    	  self.model = model
	
    def detect(self, video, class_num, min_confidence):
        detection_num = 0
    	  capture = cv2.VideoCapture(video)
    	  frame_queue = Queue(maxsize=1)
    	  person_queue = Queue(maxsize=1)
    	  detect_proc = Process(target=detect_in_process, args=(self.proto, self.model, self.ssd_proc, frame_queue, person_queue, class_num, min_confidence))
        detect_proc.daemon = True
        detect_proc.start()
    	
    	  frame_num = 0
    	  persons = None
    	  # Capture all frames
    	  while(True):	
    	      t1 = time.time()
 
    	      (ret, frame) = capture.read()
        	if frame is None:
                break
        	
        	if frame_queue.empty():
                print ("Put into frame queue ..."+str(frame_num))
                frame_queue.put(frame)
        	
            t2 = time.time()
        	dt = t2-t1
        	if dt<0.040:
                st = 0.040-dt
                print ("Sleep..."+str(st))
                time.sleep(st)
        	
        	if not person_queue.empty():
                persons = person_queue.get()
                print ("Get from person queue ..."+str(len(persons)))
        	
        	if (persons is not None) and (len(persons)>0):
                detection_num += len(persons)
                Utils.draw_objects(persons, "PERSON", (0, 0, 255), frame)
        	
        	# Display the resulting frame
            cv2.imshow('Person detection',frame)
            if cv2.waitKey(1) & 0xFF == ord('q'):
                break
 
            frame_num += 1
 
        	
        capture.release()
    	  cv2.destroyAllWindows()
    	
    	  return (frame_num, detection_num)

正如你所见，检测算法被提取到一个名为detect_in_process的单独函数中。该函数有两个特殊参数：frame_queue和person_queue。这些队列旨在支持帧捕获进程和人脸检测进程之间的交互。

让我们来看看实时人脸检测的主类RealtimeVideoSSD。其构造函数接收三个参数：SSD结构的原型、SSD模型和帧处理器。请注意，前两个参数是模型数据的路径，而不是加载的模型本身。模型在执行它的进程内部加载。这使我们能够避免在进程之间传输大量数据；我们只传输轻量级字符串数据。

该类的detect方法接收视频文件的路径、人脸类的编号和置信度阈值。首先，它创建视频流的捕获器以及帧和人脸检测的队列。然后，它使用指定的参数初始化Process类的一个实例。其中一个参数是detect_in_process函数，它将与进程一起运行。检测进程在后台运行（daemon = True）。

接下来，我们循环遍历所有接收到的帧。当我们获得一个新帧时，我们查看帧队列，如果它为空，我们就将帧放入队列。后台进程查看队列并开始处理帧；完成后，进程将结果放入人脸队列。

帧处理循环查看人脸队列，如果有检测到人脸，就将其绘制在当前帧上。所以，如果一帧中检测到人，该人就会显示在检测结果的帧中，而不是原始帧中。

帧处理循环中还有一个技巧。请注意，我们测量了capture.read方法的耗时。然后我们将帧处理切换到睡眠模式，以迫使其减慢速度。在这种情况下，我们将速度降低到大约25 FPS（0.040秒=1/25 FPS）——我们测试视频的实际帧率。

现在我们可以使用以下Python代码以实时模式测试人脸检测

# testing SSD in real-time mode
if __name__ == '__main__':
    proto_file = r"/home/pi/Desktop/PI_RPD/mobilenet.prototxt"
    model_file = r"/home/pi/Desktop/PI_RPD/mobilenet.caffemodel"
    proc_frame_size = 300
    person_class = 15
	
    # frame processor for MobileNet
    frame_proc = FrameProcessor(proc_frame_size, 1.0/127.5, 127.5)
	
    video_file = r"/home/pi/Desktop/PI_RPD/video/persons_4.mp4"
    video_ssd = RealtimeVideoSSD(proto_file, model_file, frame_proc)
    (frames, detections) = video_ssd.detect(video_file, person_class, 0.5)
 
    print ("Frames count: "+str(frames))
    print ("Detection count: "+str(detections))

这是在Pi 3B设备上测试运行的屏幕视频

你可以看到我们的视频以FPS=25的实时模式运行。人脸检测存在延迟——检测框与人脸位置不匹配，如下图所示

这种延迟是我们检测算法明显的缺点：我们在人脸出现后0.8秒才显示检测结果。随着人的移动，检测结果显示的是过去的位置。尽管如此，该算法是实时的：它确实以视频本身的帧率25 FPS处理视频。该算法执行其预期任务：它在视频流中检测人脸。考虑到一个人在摄像头视野中移动几秒钟，发现这个人脸出现的概率非常高。

在本系列文章中，我们使用了现代人工智能算法在计算资源有限的边缘设备上进行人脸检测。我们考虑了三种用于对象检测的DNN方法，并选择了最佳模型——适合在树莓派设备上进行图像处理。我们使用预训练的DNN模型开发了用于人脸检测的Python代码，并展示了如何在树莓派设备上运行该代码。最后，我们调整了代码以适应实时处理，并在视频上进行了测试，证明了它可以在直播视频流中检测到人脸。