【征集/转载】角蜂鸟多线程简易教程 - 从10FPS到真·实时

feixiang20

[征集]你玩AI，我送幸运，为AI套件专区添砖加瓦--
角蜂鸟多线程简易教程 - 从10FPS到真·实时
终于！经历了各种坎坷我们从加工厂迎来了新的一批角蜂鸟，暂时告别了缺货危机。你们肯定想不到在此之前我们这几个开发者为了争夺角蜂鸟的使用权是如何头破血流……
于是面对着一大箱（看上面题图！）的角蜂鸟，自然要琢磨着搞点新事情：比如说一个PC机插20只角蜂鸟会是什么体验… 野心归野心，也要先始于足下。所以这一篇文章我们来讲一下在Python下多线程的实现。（什么，你说talk is cheap show me the code？直接下拉到最后拷代码吧！）
首先，为什么要多线程？
答案很简单：单线程是串联，多线程是并联。
比如我有一个摄像头30FPS，但我的检测算法只能跑9FPS，于是单线程就是： 取图 - 检测 - 取结果 - 取图 - 检测 - 取结果 … 所以无论如何两次取图之间都要做一次检测，花费时间全部叠加。 但多线程的话就可以：
线程一：取图 - 取图 - 取图 - 取图 - 取图 - 取图
↓ ↓
线程二：检测 …… 取结果 - 检测 …… 取结果
然后合并两个线程返回的结果即可。这样显示图像的帧率仍然是30FPS，只是隔几帧才显示一次检测结果。
那么难点在哪？
排除掉实现方面的难点（用Python实现多线程语法方面非常简单），将不同线程的结果进行合并归纳是比起单线程额外的主要麻烦点。比如说上面范例的黑体字，在取得第一帧的检测结果时，线程一已经取得第三张图了，要是直接合并结果显示必然会错位。比如说Google的AIY， 直接将好几帧前的检测结果拍在当前帧上，设想如果用这个检测结果抓取人脸去做识别，大部分人都得变无机物。真的不是因为是竞品才恶意攻击…（手动狗头）




那么让我们扛起键盘，开始写代码吧。

*注意：需要使用额外的网络摄像头。用角蜂鸟自带的也可以实现但稍微麻烦一些。

先把必须品import出来，注意两个关键的库：队列库queue和线程库threading。

import numpy as np, cv2, sys, queue, threading
sys.path.append('../../api')
import hsapi as hs
然后来创建一个网络摄像头（30FPS）的队列和线程：

def get_image(cap, image_queue):
  while 1:
    _, img = cap.read()
    image_queue.put(img, True)

image_queue = queue.Queue(5)
cap = cv2.VideoCapture(0)
video_thread = threading.Thread(target=get_image, args=(cap, image_queue))
video_thread.start()
在这段代码我们创造了一个用来放图像的队列 image_queue ,和一个要被单独扔在一个线程里让它自己跑的函数 get_image。 这样运行的时候摄像头会自动往image_queue里灌图像，满5张图的话会进入等待状态。注意那个put里的True，用来等待，如果是False将会直接报异常，也非常关键，后面会用到。

然后我们在后面写一个while循环来取图测试一下：

while 1:
  cv2.imshow('', image_queue.get(True))
  cv2.waitKey(1)
应该可以正确的显示摄像头图像。

现在我们来创建第二个线程，用于角蜂鸟的人脸检测，加在while 1的前面：

def get_result(net, image_queue, result_queue):
  while 1:
    result = net.run(image_queue.get(True))
    while 1:
      try:
        result_queue.put(result.copy(), False)
        result[0] = image_queue.get(False)
      except:
        break

net = hs.HS('FaceDetector', zoom=True, verbose=0, threshSSD=0.5)
result_queue = queue.Queue(5)
hs_thread = threading.Thread(target=get_result, args=(net, image_queue, result_queue))
hs_thread.start()
这个线程把图像队列连接了过来， 然后输出结果到另一个结果队列result_queue。

在这里需要额外解释一下第二个while循环的作用。因为通常检测1次会对应N帧，为了显示结果方便，我们将N-1帧都赋予第1帧的结果，所以通过while把堆在队列里的帧全部拿出来赋值然后放入结果队列。

其实就这么简单，接下来在外面获取队列并且显示结果就好了。

while(1):
  img = net.plotSSD(result_queue.get(True))
  cv2.imshow('', img)
  cv2.waitKey(30)
*注意：因为队列中每次都一下获取N帧，所以破坏了原本连贯的间隔。如果直接显示视频会很不顺畅，所以需要cv2.waitKey(30)来手动设置一个间隔。 现在就获得了一个看似30FPS的实时检测器。并且检测框和图像同步！

全部代码：

import numpy as np, cv2, sys, queue, threading
sys.path.append('../../api')
import hsapi as hs


def get_image(cap, image_queue):
  while 1:
    _, img = cap.read()
    image_queue.put(img, True)

image_queue = queue.Queue(5)
cap = cv2.VideoCapture(0)
video_thread = threading.Thread(target=get_image, args=(cap, image_queue))
video_thread.start()

def get_result(net, image_queue, result_queue):
  while 1:
    result = net.run(image_queue.get(True))
    while 1:
      try:
        result_queue.put(result.copy(), False)
        img = image_queue.get(False)
        result[0] = img
      except:
        break

net = hs.HS('FaceDetector', zoom=True, verbose=0, threshSSD=0.5)
result_queue = queue.Queue(5)
hs_thread = threading.Thread(target=get_result, args=(net, image_queue, result_queue))
hs_thread.start()

while(1):
  img = net.plotSSD(result_queue.get(True))
  cv2.imshow('', img)
  cv2.waitKey(30)


等等， 先别右上角X。下面再提供一个完整版的多线程API，计划下次更新加入SDK，我先偷偷放出来… 小心BUG！

hsproc.py， 放入api文件夹即可：

#! /usr/bin/env python3

# Copyright(c) 2018 Senscape Corporation.
# License: Apache 2.0

import numpy as np, cv2, sys, queue, threading, time
import hsapi as hs

class HSProc:
        def __init__(self, modelName, useWebcam, **kwargs):
                self.modelName = modelName               
                self.useWebcam = useWebcam
                self.videoIndex = 0
                self.webcamFlip = True
                self.delay = 0.025 #ms
                self.autoStart = True
                self.image_queue = queue.Queue(5)
                self.res_queue = queue.Queue(10)
               
                for k,v in kwargs.items():
                        exec('self.'+k+'=v')
               
                self.net = hs.HS(modelName, **kwargs)
                if self.autoStart:
                        if self.useWebcam:
                                self.video_capture = cv2.VideoCapture(self.videoIndex)
                                self._video_thread = threading.Thread(target=self._get_image, args=())
                                self._video_thread.start()
                        self.start()

        def start(self):
                self._worker_thread = threading.Thread(target=self._do_work, args=())
                self._worker_thread.start()
       
        def stop(self):
                self.net.quit()
               
        def _do_work(self):
                while 1:
                        if self.useWebcam:
                                img = self.image_queue.get(True)
                        else: img = None
                        result = self.net.run(img)
                        result.append(True)
                        while 1:
                                try:
                                        self.res_queue.put(result.copy(), False)
                                        img = self.image_queue.get(False)
                                        result[0] = img
                                        result[2] = False
                                except:
                                        break

        def _get_image(self):
                while 1:
                        try:
                                _, img = self.video_capture.read()
                                if self.webcamFlip:
                                        img = cv2.flip(img, 1)
                                time.sleep(self.delay)
                                self.net.msg_debug('Img Q: %d' % self.image_queue.qsize())
                                self.net.msg_debug('Ret Q: %d' % self.res_queue.qsize())
                                self.image_queue.put(img, False)
                        except:
                                self.net.msg_debug('No image!')
使用范例：

多线程人脸检测，顺便带输出人脸，证明检测结果和视频同步。

MT_FaceDetection.py，放入examples/python

#! /usr/bin/env python3

# Copyright(c) 2018 Senscape Corporation.
# License: Apache 2.0

import cv2, sys
sys.path.append('../../api/')
import hsproc

useWebcam = True
sr = hsproc.HSProc('FaceDetector', useWebcam, zoom=True, verbose=0, threshSSD=0.5)

try:
        while(1):
                ret = sr.res_queue.get()
               
                # Get face images for further processing
                crops, info = sr.net.cropObjects(ret)
                if len(crops) > 0 and ret[2] is True:
                        # Display first face info/image as example
                        print(info)
                        try:
                                cv2.imshow('A Face', crops[0])
                        except:
                                pass

                img = sr.net.plotSSD(ret)
                cv2.imshow("Face Detector", img)
                cv2.waitKey(25)
finally:
        sr.stop()
大概多线程教程就到这里…