目前可依靠模塊化方式實(shí)現(xiàn)圖像處理管道，檢測(cè)一堆圖像文件中的人臉，并將其與漂亮的結(jié)構(gòu)化JSON摘要文件一起保存在單獨(dú)的文件夾中。

讓我們對(duì)視頻流也可以進(jìn)行同樣的操作。

首先，我們需要捕獲視頻流。該管線任務(wù)將從視頻文件或網(wǎng)絡(luò)攝像頭（逐幀）生成一系列圖像。接下來，我們將檢測(cè)每個(gè)幀上的臉部并將其保存。接下來的三個(gè)塊是可選的，它們的目標(biāo)是創(chuàng)建帶有注釋的輸出視頻，例如在檢測(cè)到的人臉周圍的框。我們可以顯示帶注釋的視頻并將其保存。最后一個(gè)任務(wù)將收集有關(guān)檢測(cè)到的面部的信息，并保存帶有面部的框坐標(biāo)和置信度的JSON摘要文件。

如果尚未設(shè)置jagin / image-processing-pipeline存儲(chǔ)庫以查看源代碼并運(yùn)行一些示例，則可以立即執(zhí)行以下操作：

$ git clone git://github.com/jagin/image-processing-pipeline.git
$ cd image-processing-pipeline
$ git checkout 7df1963247caa01b503980fe152138b88df6c526
$ conda env create -f environment.yml
$ conda activate pipeline

如果已經(jīng)克隆了存儲(chǔ)庫并設(shè)置了環(huán)境，請(qǐng)使用以下命令對(duì)其進(jìn)行更新：

$ git pull
$ git checkout 7df1963247caa01b503980fe152138b88df6c526
$ conda env update -f environment.yml

拍攝影片

使用OpenCV捕獲視頻非常簡(jiǎn)單。我們需要?jiǎng)?chuàng)建一個(gè)VideoCapture對(duì)象，其中參數(shù)是設(shè)備索引（指定哪個(gè)攝像機(jī)的數(shù)字）或視頻文件的名稱。然后，我們可以逐幀捕獲視頻流。

我們可以使用以下CaptureVideo擴(kuò)展類來實(shí)現(xiàn)捕獲視頻任務(wù)Pipeline：

import cv2
from pipeline.pipeline import Pipeline


class CaptureVideo(Pipeline):
    def __init__(self, src=0):
        self.cap = cv2.VideoCapture(src)
        if not self.cap.isOpened():
            raise IOError(f"Cannot open video {src}")


        self.fps = int(self.cap.get(cv2.CAP_PROP_FPS))
        self.frame_count = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))


        super(CaptureVideo, self).__init__()


    def generator(self):
        image_idx = 0
        while self.has_next():
            ret, image = self.cap.read()
            if not ret:
                # no frames has been grabbed
                break


            data = {
                "image_id": f"{image_idx:05d}",
                "image": image,
            }


            if self.filter(data):
                image_idx += 1
                yield self.map(data)


    def cleanup(self):
        # Closes video file or capturing device
        self.cap.release()

使用__init__我們創(chuàng)建VideoCapture對(duì)象（第6行）并提取視頻流的屬性，例如每秒幀數(shù)和幀數(shù)。我們將需要它們顯示進(jìn)度條并正確保存視頻。圖像幀將在具有字典結(jié)構(gòu)的generator函數(shù)（第30行）中產(chǎn)生：

data = {
    "image_id": f"{image_idx:05d}",
    "image": image,
}

當(dāng)然，數(shù)據(jù)中也包括圖像的序列號(hào)和幀的二進(jìn)制數(shù)據(jù)。

檢測(cè)人臉

我們準(zhǔn)備檢測(cè)面部。這次，我們將使用OpenCV的深度神經(jīng)網(wǎng)絡(luò)模塊，而不是我在上一個(gè)故事中所承諾的Haar級(jí)聯(lián)。我們將要使用的模型更加準(zhǔn)確，并且還為我們提供了置信度得分。

從版本3.3開始，OpenCV支持許多深度學(xué)習(xí)框架，例如Caffe，TensorFlow和PyTorch，從而使我們能夠加載模型，預(yù)處理輸入圖像并進(jìn)行推理以獲得輸出分類。

有一位優(yōu)秀的博客文章中阿德里安·羅斯布魯克（Adrian Rosebrock）解釋如何使用OpenCV和深度學(xué)習(xí)實(shí)現(xiàn)人臉檢測(cè)。我們將在FaceDetector類中使用部分代碼：

import cv2
import numpy as np


class FaceDetector:
    def __init__(self, prototxt, model, confidence=0.5):
        self.confidence = confidence


        self.net = cv2.dnn.readNetFromCaffe(prototxt, model)


    def detect(self, images):
        # convert images into blob
        blob = self.preprocess(images)


        # pass the blob through the network and obtain the detections and predictions
        self.net.setInput(blob)
        detections = self.net.forward()
        # Prepare storage for faces for every image in the batch
        faces = dict(zip(range(len(images)), [[] for _ in range(len(images))]))


        # loop over the detections
        for i in range(0, detections.shape[2]):
            # extract the confidence (i.e., probability) associated with the prediction
            confidence = detections[0, 0, i, 2]


            # filter out weak detections by ensuring the `confidence` is
            # greater than the minimum confidence
            if confidence < self.confidence:
                continue


            # grab the image index
            image_idx = int(detections[0, 0, i, 0])
            # grab the image dimensions
            (h, w) = images[image_idx].shape[:2]
            # compute the (x, y)-coordinates of the bounding box for the object
            box = detections[0, 0, i, 3:7] * np.array([w, h, w, h])


            # Add result
            faces[image_idx].append((box, confidence))


        return faces


    def preprocess(self, images):
        return cv2.dnn.blobFromImages(images, 1.0, (300, 300), (104.0, 177.0, 123.0))