如何使用OpenCV和Python構(gòu)建人員計數(shù)器

這篇文章主要介紹了如何使用OpenCV和Python構(gòu)建人員計數(shù)器的相關(guān)知識，內(nèi)容詳細(xì)易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇如何使用OpenCV和Python構(gòu)建人員計數(shù)器文章都會有所收獲，下面我們一起來看看吧。

1.了解對象檢測與對象跟蹤

在繼續(xù)本教程的其余部分之前，您必須了解對象檢測和對象跟蹤之間的根本區(qū)別。

當(dāng)我們應(yīng)用對象檢測時，我們是在確定一個對象在圖像/幀中的位置。與目標(biāo)跟蹤算法相比，目標(biāo)檢測器通常在計算上更昂貴，因此也更慢。目標(biāo)檢測算法的例子包括Haar級聯(lián)、HOG +線性支持向量機(HOG + Linear SVM)和基于深度學(xué)習(xí)的目標(biāo)檢測器，如Faster R-CNN、YOLO和Single Shot檢測器(SSD)。

另一方面，對象跟蹤器將接受對象在圖像中位置的輸入 (x, y) 坐標(biāo)，并將：

1.為該特定對象分配唯一 ID

2.在對象圍繞視頻流移動時跟蹤對象，根據(jù)幀的各種屬性（梯度、光流等）預(yù)測下一幀中的新對象位置

對象跟蹤算法的示例包括 MedianFlow、MOSSE、GOTURN、核化相關(guān)濾波器和判別相關(guān)濾波器等。

2.結(jié)合對象檢測和對象跟蹤

高精度目標(biāo)跟蹤器將目標(biāo)檢測和目標(biāo)跟蹤的概念結(jié)合到一個算法中，通常分為兩個階段：

1.階段1 檢測：在檢測階段，我們正在運行計算成本更高的對象跟蹤器，以 (1) 檢測是否有新對象進(jìn)入我們的視野，以及 (2) 看看我們是否可以找到在跟蹤階段“丟失”的對象。對于每個檢測到的對象，我們使用新的邊界框坐標(biāo)創(chuàng)建或更新對象跟蹤器。由于我們的目標(biāo)檢測器的計算成本更高，我們每 N 幀只運行一次此階段。

2.階段2 跟蹤：當(dāng)我們不處于“檢測”階段時，我們處于“跟蹤”階段。對于我們檢測到的每個對象，我們創(chuàng)建一個對象跟蹤器來跟蹤對象在框架周圍的移動。我們的目標(biāo)跟蹤器應(yīng)該比目標(biāo)檢測器更快、更高效。我們將繼續(xù)跟蹤，直到我們到達(dá)第 N 幀，然后重新運行我們的目標(biāo)檢測器。然后重復(fù)整個過程。

這種混合方法的好處是我們可以應(yīng)用高度準(zhǔn)確的對象檢測方法，而無需太多的計算負(fù)擔(dān)。我們將實施這樣一個跟蹤系統(tǒng)來建立我們的人員計數(shù)器。

3.項目結(jié)構(gòu)

讓我們回顧一下今天博客文章的項目結(jié)構(gòu)。獲取代碼后，您可以使用 tree 命令檢查目錄結(jié)構(gòu)：

最重要的兩個目錄：

1.pyimagesearch/：該模塊包含質(zhì)心跟蹤算法。 “組合對象跟蹤算法”部分介紹了質(zhì)心跟蹤算法。

2.mobilenet_ssd/：包含 Caffe 深度學(xué)習(xí)模型文件。

今天項目的核心包含在 people_counter.py 腳本中——這是我們將花費大部分時間的地方。今天我們還將回顧 trackableobject.py 腳本。

4.結(jié)合對象跟蹤算法

為了實現(xiàn)我們的人員計數(shù)器，我們將同時使用 OpenCV 和 dlib。我們將 OpenCV 用于標(biāo)準(zhǔn)的計算機視覺/圖像處理功能，以及用于人數(shù)統(tǒng)計的深度學(xué)習(xí)對象檢測器。

然后我們將使用 dlib 來實現(xiàn)相關(guān)過濾器。我們也可以在這里使用 OpenCV；但是，對于這個項目，dlib 對象跟蹤實現(xiàn)更容易使用。

除了 dlib 的對象跟蹤實現(xiàn)，我們還將使用質(zhì)心跟蹤實現(xiàn)?；仡櫿麄€質(zhì)心跟蹤算法超出了這篇博文的范圍，但我在下面提供了一個簡短的概述。

在步驟#1，我們接受一組邊界框并計算它們對應(yīng)的質(zhì)心（即邊界框的中心）：

要使用 Python 通過質(zhì)心腳本構(gòu)建簡單的對象跟蹤，第一步是接受邊界框坐標(biāo)并使用它們來計算質(zhì)心。

邊界框本身可以由以下任一方式提供：

1.目標(biāo)檢測器（如 HOG + Linear SVM、Faster R-CNN、SSDs 等）

2.或?qū)ο蟾櫰鳎ɡ缦嚓P(guān)過濾器）

在上圖中，您可以看到我們在算法的初始迭代中有兩個對象要跟蹤。

在步驟#2中，我們計算任何新質(zhì)心（黃色）和現(xiàn)有質(zhì)心（紫色）之間的歐幾里得距離：

此圖像中存在三個對象。我們需要計算每對原始質(zhì)心（紫色）和新質(zhì)心（黃色）之間的歐幾里得距離。

質(zhì)心跟蹤算法假設(shè)它們之間具有最小歐幾里德距離的質(zhì)心對必須是相同的對象 ID。

在上面的示例圖像中，我們有兩個現(xiàn)有的質(zhì)心（紫色）和三個新的質(zhì)心（黃色），這意味著已經(jīng)檢測到一個新對象（因為與舊質(zhì)心相比，還有一個新質(zhì)心）。

然后箭頭表示計算所有紫色質(zhì)心和所有黃色質(zhì)心之間的歐幾里得距離。一旦我們有了歐幾里得距離，我們就會在步驟#3 中嘗試關(guān)聯(lián)對象 ID：

您可以看到我們的質(zhì)心跟蹤器已選擇關(guān)聯(lián)使它們各自的歐幾里得距離最小化的質(zhì)心。但是左下角的點呢？它沒有與任何東西相關(guān)聯(lián)——我們該怎么辦？ 要回答這個問題，我們需要執(zhí)行步驟#4，注冊新對象：

注冊意味著我們通過以下方式將新對象添加到我們的跟蹤對象列表中：

1.為其分配一個新的對象 ID

2.存儲新對象的邊界框坐標(biāo)的質(zhì)心

如果對象丟失或離開視野，我們可以簡單地取消注冊對象（步驟#5）。

5.創(chuàng)建可追蹤對象

為了跟蹤和計算視頻流中的對象，我們需要一種簡單的方法來存儲有關(guān)對象本身的信息，包括：

• 對象ID

• 以前的質(zhì)心（所以我們可以很容易地計算出物體移動的方向）

• 對象是否已被計數(shù)

為了實現(xiàn)所有這些目標(biāo)，我們可以定義一個 TrackableObject 實例——打開 trackableobject.py 文件并插入以下代碼：

class TrackableObject:
    def __init__(self, objectID, centroid):
        # store the object ID, then initialize a list of centroids
        # using the current centroid
        self.objectID = objectID
        self.centroids = [centroid]
        # initialize a boolean used to indicate if the object has
        # already been counted or not
        self.counted = False

TrackableObject 構(gòu)造函數(shù)接受 objectID + centroid 并存儲它們。 centroids 變量是一個列表，因為它將包含對象的質(zhì)心位置歷史記錄。 構(gòu)造函數(shù)還將 counted 初始化為 False ，表示該對象還沒有被計數(shù)。

6.使用 OpenCV + Python 實現(xiàn)我們的人員計數(shù)器

# import the necessary packages
from pyimagesearch.centroidtracker import CentroidTracker
from pyimagesearch.trackableobject import TrackableObject
from imutils.video import VideoStream
from imutils.video import FPS
import numpy as np
import argparse
import imutils
import time
import dlib
import cv2

我們首先導(dǎo)入必要的包：

• 從 pyimagesearch 模塊，我們導(dǎo)入自定義的 CentroidTracker 和 TrackableObject 類。

• imutils.video 中的 VideoStream 和 FPS 模塊將幫助我們使用網(wǎng)絡(luò)攝像頭并計算估計的每秒幀數(shù) (FPS) 吞吐率。

• 我們需要 imutils 的 OpenCV 便利功能。

• dlib 庫將用于其相關(guān)跟蹤器實現(xiàn)。

• OpenCV 將用于深度神經(jīng)網(wǎng)絡(luò)推理、打開視頻文件、寫入視頻文件以及在我們的屏幕上顯示輸出幀。

現(xiàn)在所有工具都觸手可及，讓我們解析命令行參數(shù)：

# construct the argument parse and parse the arguments
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("-i", "--input", type=str,
    help="path to optional 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.4,
    help="minimum probability to filter weak detections")
ap.add_argument("-s", "--skip-frames", type=int, default=30,
    help="# of skip frames between detections")
args = vars(ap.parse_args())

我們有六個命令行參數(shù)，它們允許我們在運行時從終端將信息傳遞給我們的人員計數(shù)器腳本：

• --prototxt ：Caffe 部署 prototxt 文件的路徑。

• --model ：Caffe 預(yù)訓(xùn)練 CNN 模型的路徑。

• --input : 可選的輸入視頻文件路徑。如果未指定路徑，將使用您的網(wǎng)絡(luò)攝像頭。

• --output ：可選的輸出視頻路徑。如果未指定路徑，則不會錄制視頻。

• --confidence ：默認(rèn)值為 0.4 ，這是有助于過濾掉弱檢測的最小概率閾值。

• --skip-frames ：在跟蹤對象上再次運行我們的 DNN 檢測器之前要跳過的幀數(shù)。請記住，對象檢測的計算成本很高，但它確實有助于我們的跟蹤器重新評估幀中的

對象。默認(rèn)情況下，我們在使用 OpenCV DNN 模塊和我們的 CNN 單次檢測器模型檢測對象之間跳過 30 幀。

現(xiàn)在我們的腳本可以在運行時動態(tài)處理命令行參數(shù)，讓我們準(zhǔn)備我們的 SSD：

# initialize the list of class labels MobileNet SSD was trained to detect
CLASSES = ["background", "aeroplane", "bicycle", "bird", "boat",
    "bottle", "bus", "car", "cat", "chair", "cow", "diningtable",
    "dog", "horse", "motorbike", "person", "pottedplant", "sheep",
    "sofa", "train", "tvmonitor"]
# load our serialized model from disk
print("[INFO] loading model...")
net = cv2.dnn.readNetFromCaffe(args["prototxt"], args["model"])

首先，我們將初始化 CLASSES——SSD 支持的類列表。我們只對“人”類感興趣，但您也可以計算其他移動對象。

我們加載用于檢測對象的預(yù)訓(xùn)練 MobileNet SSD（但同樣，我們只對檢測和跟蹤人感興趣，而不是任何其他類）。

我們可以初始化我們的視頻流：

# if a video path was not supplied, grab a reference to the webcam
if not args.get("input", False):
    print("[INFO] starting video stream...")
    vs = VideoStream(src=0).start()
    time.sleep(2.0)
# otherwise, grab a reference to the video file
else:
    print("[INFO] opening video file...")
    vs = cv2.VideoCapture(args["input"])

首先，我們處理使用網(wǎng)絡(luò)攝像頭視頻流的情況。否則，我們將從視頻文件中捕獲幀。在開始循環(huán)幀之前，我們還有一些初始化要執(zhí)行：

# initialize the video writer (we'll instantiate later if need be)
writer = None
# initialize the frame dimensions (we'll set them as soon as we read
# the first frame from the video)
W = None
H = None
# instantiate our centroid tracker, then initialize a list to store
# each of our dlib correlation trackers, followed by a dictionary to
# map each unique object ID to a TrackableObject
ct = CentroidTracker(maxDisappeared=40, maxDistance=50)
trackers = []
trackableObjects = {}
# initialize the total number of frames processed thus far, along
# with the total number of objects that have moved either up or down
totalFrames = 0
totalDown = 0
totalUp = 0
# start the frames per second throughput estimator
fps = FPS().start()

其余的初始化包括：

• writer：我們的視頻寫入器。如果我們正在寫入視頻，我們稍后會實例化這個對象。

• W 和 H：我們的幀尺寸。我們需要將這些插入到 cv2.VideoWriter 中。

• ct：我們的 CentroidTracker。

• trackers ：存儲 dlib 相關(guān)跟蹤器的列表。

• trackableObjects ：將 objectID 映射到 TrackableObject 的字典。

• totalFrames ：處理的幀總數(shù)。

• totalDown 和 totalUp ：向下或向上移動的對象/人的總數(shù)。

• fps ：我們用于基準(zhǔn)測試的每秒幀數(shù)估計器。

現(xiàn)在我們所有的初始化都處理好了，讓我們循環(huán)傳入的幀：

# loop over frames from the video stream
while True:
    # grab the next frame and handle if we are reading from either
    # VideoCapture or VideoStream
    frame = vs.read()
    frame = frame[1] if args.get("input", False) else frame
    # if we are viewing a video and we did not grab a frame then we
    # have reached the end of the video
    if args["input"] is not None and frame is None:
        break
    # resize the frame to have a maximum width of 500 pixels (the
    # less data we have, the faster we can process it), then convert
    # the frame from BGR to RGB for dlib
    frame = imutils.resize(frame, width=500)
    rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    # if the frame dimensions are empty, set them
    if W is None or H is None:
        (H, W) = frame.shape[:2]
    # if we are supposed to be writing a video to disk, initialize
    # the writer
    if args["output"] is not None and writer is None:
        fourcc = cv2.VideoWriter_fourcc(*"MJPG")
        writer = cv2.VideoWriter(args["output"], fourcc, 30,
            (W, H), True)

我們開始循環(huán)。在循環(huán)的頂部，我們抓取下一幀。如果我們已經(jīng)到達(dá)視頻的結(jié)尾，我們將跳出循環(huán)。

幀進(jìn)行預(yù)處理。這包括調(diào)整大小和交換顏色通道，因為 dlib 需要 rgb 圖像。我們?yōu)橐曨l編寫器獲取幀的尺寸。 如果通過命令行參數(shù)提供了輸出路徑，我們將從那里實例化視頻編寫器。

現(xiàn)在讓我們使用 SSD檢測人：

    # initialize the current status along with our list of bounding
    # box rectangles returned by either (1) our object detector or
    # (2) the correlation trackers
    status = "Waiting"
    rects = []
    # check to see if we should run a more computationally expensive
    # object detection method to aid our tracker
    if totalFrames % args["skip_frames"] == 0:
        # set the status and initialize our new set of object trackers
        status = "Detecting"
        trackers = []
        # convert the frame to a blob and pass the blob through the
        # network and obtain the detections
        blob = cv2.dnn.blobFromImage(frame, 0.007843, (W, H), 127.5)
        net.setInput(blob)
        detections = net.forward()

我們將狀態(tài)初始化為Waiting。可能的狀態(tài)包括：

• Waiting：在這種狀態(tài)下，我們正在等待檢測和跟蹤人員。

• Detecting：我們正在使用 MobileNet SSD 檢測人員。

• Tracking：人們在幀中被跟蹤，我們正在計算 totalUp 和 totalDown 。

我們的 rects 列表將通過檢測或跟蹤來填充。我們繼續(xù)初始化rects 。

重要的是要了解深度學(xué)習(xí)對象檢測器的計算成本非常高，尤其是當(dāng)您在 CPU 上運行它們時。

為了避免在每一幀上運行我們的目標(biāo)檢測器，并加快我們的跟蹤管道，我們將跳過 N 幀（由命令行參數(shù)設(shè)置 --skip-frames ，其中 30 是默認(rèn)值）。只有每 N 幀，我們才會使用 SSD 進(jìn)行對象檢測。否則，我們將只是跟蹤中間的移動對象。

使用模運算符，我們確保每 N 幀執(zhí)行一次 if 語句中的代碼。 進(jìn)入if語句后，我們會將狀態(tài)更新為Detecting。 然后我們初始化新的跟蹤器列表。

接下來，我們將通過對象檢測進(jìn)行推理。我們首先從圖像中創(chuàng)建一個 blob，然后將該 blob 通過網(wǎng)絡(luò)傳遞以獲得檢測。 現(xiàn)在我們將遍歷每個檢測，希望找到屬于person類的對象：

        # loop over the detections
        for i in np.arange(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 requiring a minimum
            # confidence
            if confidence > args["confidence"]:
                # extract the index of the class label from the
                # detections list
                idx = int(detections[0, 0, i, 1])
                # if the class label is not a person, ignore it
                if CLASSES[idx] != "person":
                    continue

循環(huán)檢測，我們繼續(xù)獲取置信度并過濾掉那些不屬于人類的結(jié)果。

現(xiàn)在我們可以為每個人計算一個邊界框并開始相關(guān)性跟蹤：

                # 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])
                (startX, startY, endX, endY) = box.astype("int")
                # construct a dlib rectangle object from the bounding
                # box coordinates and then start the dlib correlation
                # tracker
                tracker = dlib.correlation_tracker()
                rect = dlib.rectangle(startX, startY, endX, endY)
                tracker.start_track(rgb, rect)
                # add the tracker to our list of trackers so we can
                # utilize it during skip frames
                trackers.append(tracker)

計算我們的box。 然后實例化我們的 dlib 相關(guān)跟蹤器，然后將對象的邊界框坐標(biāo)傳遞給 dlib.rectangle，將結(jié)果存儲為 rect。 隨后，我們開始跟蹤，并將跟蹤器附加到跟蹤器列表中。 這是我們每 N 個跳幀執(zhí)行的所有操作的封裝！ 讓我們處理在 else 塊中進(jìn)行跟蹤的典型操作：

    # otherwise, we should utilize our object *trackers* rather than
    # object *detectors* to obtain a higher frame processing throughput
    else:
        # loop over the trackers
        for tracker in trackers:
            # set the status of our system to be 'tracking' rather
            # than 'waiting' or 'detecting'
            status = "Tracking"
            # update the tracker and grab the updated position
            tracker.update(rgb)
            pos = tracker.get_position()
            # unpack the position object
            startX = int(pos.left())
            startY = int(pos.top())
            endX = int(pos.right())
            endY = int(pos.bottom())
            # add the bounding box coordinates to the rectangles list
            rects.append((startX, startY, endX, endY))

大多數(shù)時候，并沒有發(fā)生在跳幀倍數(shù)上。在此期間，我們將利用跟蹤器來跟蹤對象，而不是應(yīng)用檢測。 我們開始遍歷可用跟蹤器。 我們繼續(xù)將狀態(tài)更新為Tracking并獲取對象位置。 我們提取位置坐標(biāo)，然后在我們的 rects 列表中填充信息。 現(xiàn)在讓我們畫一條水平可視化線（人們必須穿過它才能被跟蹤）并使用質(zhì)心跟蹤器來更新我們的對象質(zhì)心：

    # draw a horizontal line in the center of the frame -- once an
    # object crosses this line we will determine whether they were
    # moving 'up' or 'down'
    cv2.line(frame, (0, H // 2), (W, H // 2), (0, 255, 255), 2)
    # use the centroid tracker to associate the (1) old object
    # centroids with (2) the newly computed object centroids
    objects = ct.update(rects)

我們畫一條水平線，我們將用它來可視化人們“越過”——一旦人們越過這條線，我們將增加各自的計數(shù)器 然后，我們利用 CentroidTracker 實例化來接受 rects 列表，無論它們是通過對象檢測還是對象跟蹤生成的。我們的質(zhì)心跟蹤器會將對象 ID 與對象位置相關(guān)聯(lián)。 在下一個代碼塊中，我們將回顧一個人在幀中向上或向下移動的邏輯：

    # loop over the tracked objects
    for (objectID, centroid) in objects.items():
        # check to see if a trackable object exists for the current
        # object ID
        to = trackableObjects.get(objectID, None)
        # if there is no existing trackable object, create one
        if to is None:
            to = TrackableObject(objectID, centroid)
        # otherwise, there is a trackable object so we can utilize it
        # to determine direction
        else:
            # the difference between the y-coordinate of the *current*
            # centroid and the mean of *previous* centroids will tell
            # us in which direction the object is moving (negative for
            # 'up' and positive for 'down')
            y = [c[1] for c in to.centroids]
            direction = centroid[1] - np.mean(y)
            to.centroids.append(centroid)
            # check to see if the object has been counted or not
            if not to.counted:
                # if the direction is negative (indicating the object
                # is moving up) AND the centroid is above the center
                # line, count the object
                if direction < 0 and centroid[1] < H // 2:
                    totalUp += 1
                    to.counted = True
                # if the direction is positive (indicating the object
                # is moving down) AND the centroid is below the
                # center line, count the object
                elif direction > 0 and centroid[1] > H // 2:
                    totalDown += 1
                    to.counted = True
        # store the trackable object in our dictionary
        trackableObjects[objectID] = to

我們首先遍歷更新后的對象id的邊界框坐標(biāo)。我們嘗試為當(dāng)前的objectID獲取TrackableObject。如果objectID的TrackableObject不存在，我們就創(chuàng)建一個。否則，已經(jīng)存在一個 TrackableObject ，所以我們需要弄清楚對象（人）是向上還是向下移動。

為此，我們獲取給定對象之前所有質(zhì)心位置的y坐標(biāo)值。然后，通過取當(dāng)前質(zhì)心位置與之前所有質(zhì)心位置的平均值之間的差來計算方向。

我們?nèi)【档脑蚴菫榱舜_保我們的方向跟蹤更穩(wěn)定。如果我們只存儲這個人之前的質(zhì)心位置，我們就有可能出現(xiàn)錯誤的方向計數(shù)。記住，目標(biāo)檢測和目標(biāo)跟蹤算法不是“魔術(shù)”——有時它們會預(yù)測出可能稍微偏離你預(yù)期的邊界盒;因此，通過取均值，我們可以讓我們的人計算得更準(zhǔn)確。

如果TrackableObject還沒有被計數(shù)，我們需要確定它是否已經(jīng)準(zhǔn)備好被計數(shù)，通過:

1.檢查direction是否為負(fù)（表示對象向上移動）并且質(zhì)心在中心線上方。在這種情況下，我們增加 totalUp。

2.或者檢查direction是否為正(表示物體正在向下移動)且質(zhì)心在中心線以下。如果這是真的，我們增加totalDown。

最后，我們將TrackableObject存儲在trackableObjects字典中，這樣我們就可以在捕獲下一幀時獲取并更新它。

接下來的三個代碼塊處理:

• 顯示(繪圖并向幀寫入文本)

• 將幀寫入磁盤上的視頻文件(如果存在--output命令行參數(shù))

• 捕獲按鍵

• 清理

首先，我們將在框架上繪制一些信息以進(jìn)行可視化：

        # draw both the ID of the object and the centroid of the
        # object on the output frame
        text = "ID {}".format(objectID)
        cv2.putText(frame, text, (centroid[0] - 10, centroid[1] - 10),
            cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
        cv2.circle(frame, (centroid[0], centroid[1]), 4, (0, 255, 0), -1)
    # construct a tuple of information we will be displaying on the
    # frame
    info = [
        ("Up", totalUp),
        ("Down", totalDown),
        ("Status", status),
    ]
    # loop over the info tuples and draw them on our frame
    for (i, (k, v)) in enumerate(info):
        text = "{}: {}".format(k, v)
        cv2.putText(frame, text, (10, H - ((i * 20) + 20)),
            cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)

在這里，我們在幀上覆蓋以下數(shù)據(jù)：

• ObjectID ：每個對象的ID。

• centroid ：對象的中心將由一個點表示，該點是通過填充一個圓圈而創(chuàng)建的。

• info : 包括 totalUp 、 totalDown 和 status

然后我們將把幀寫入視頻文件(如果需要的話)并處理按鍵:

    # check to see if we should write the frame to disk
    if writer is not None:
        writer.write(frame)
    # show the output frame
    cv2.imshow("Frame", frame)
    key = cv2.waitKey(1) & 0xFF
    # if the `q` key was pressed, break from the loop
    if key == ord("q"):
        break
    # increment the total number of frames processed thus far and
    # then update the FPS counter
    totalFrames += 1
    fps.update()

在這個代碼塊中我們:

• 如果需要，將幀寫入輸出視頻文件

• 顯示幀并處理按鍵。如果q被按下，我們將跳出幀處理循環(huán)。

• 更新我們的fps計數(shù)器

現(xiàn)在是時候清理了:

# stop the timer and display FPS information
fps.stop()
print("[INFO] elapsed time: {:.2f}".format(fps.elapsed()))
print("[INFO] approx. FPS: {:.2f}".format(fps.fps()))
# check to see if we need to release the video writer pointer
if writer is not None:
    writer.release()
# if we are not using a video file, stop the camera video stream
if not args.get("input", False):
    vs.stop()
# otherwise, release the video file pointer
else:
    vs.release()
# close any open windows
cv2.destroyAllWindows()

為了完成腳本，我們向終端顯示 FPS 信息，釋放所有指針，并關(guān)閉所有打開的窗口。

7.完整代碼

people_counter.py

from pyimagesearch.centroidtracker import CentroidTracker
from pyimagesearch.trackableobject import TrackableObject
from imutils.video import VideoStream
from imutils.video import FPS
import numpy as np
import argparse
import imutils
import time
import dlib
import cv2


# 構(gòu)造參數(shù)解析并解析參數(shù)
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("-i", "--input", type=str,
	help="path to optional 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.4,
	help="minimum probability to filter weak detections")
ap.add_argument("-s", "--skip-frames", type=int, default=30,
	help="# of skip frames between detections")
args = vars(ap.parse_args())

# 初始化類標(biāo)簽列表
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...")
net = cv2.dnn.readNetFromCaffe(args["prototxt"], args["model"])

# 如果未提供視頻路徑，請獲取網(wǎng)絡(luò)攝像頭的引用
if not args.get("input", False):
	print("[INFO] starting video stream...")
	vs = VideoStream(src=0).start()
	time.sleep(2.0)
# 否則，獲取對視頻文件的引用
else:
	print("[INFO] opening video file...")
	vs = cv2.VideoCapture(args["input"])

# 初始化視頻寫入器(如果需要，我們稍后將進(jìn)行實例化)
writer = None
# 初始化幀尺寸（我們將在從視頻中讀取第一幀后立即設(shè)置它們）
W = None
H = None
# 實例化我們的質(zhì)心跟蹤器，然后初始化一個列表來存儲每個dlib相關(guān)跟蹤器，
# 然后是一個字典來將每個唯一的對象ID映射到TrackableObject
ct = CentroidTracker(maxDisappeared=40, maxDistance=50)
trackers = []
trackableObjects = {}
# 初始化到目前為止處理的幀總數(shù)，以及向上或向下移動的對象總數(shù)
totalFrames = 0
totalDown = 0
totalUp = 0
# 啟動FPS評估器
fps = FPS().start()

# 循環(huán)視頻流中的幀
while True:
	# 如果我們正在從 VideoCapture 或 VideoStream 讀取數(shù)據(jù)，則抓取下一幀并處理
	frame = vs.read()
	frame = frame[1] if args.get("input", False) else frame
	# 如果我們正在觀看視頻并且我們沒有抓取幀，那么我們已經(jīng)到了視頻的結(jié)尾
	if args["input"] is not None and frame is None:
		break
	# 調(diào)整幀的最大寬度為 500 像素（我們擁有的數(shù)據(jù)越少，我們處理它的速度就越快），
	# 然后將幀從 BGR 轉(zhuǎn)換為 RGB 用于 dlib
	frame = imutils.resize(frame, width=500)
	rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
	# 如果幀尺寸為空，則設(shè)置它們
	if W is None or H is None:
		(H, W) = frame.shape[:2]
	# 如果我們應(yīng)該將視頻寫入磁盤，請初始化寫入器
	if args["output"] is not None and writer is None:
		fourcc = cv2.VideoWriter_fourcc(*"MJPG")
		writer = cv2.VideoWriter(args["output"], fourcc, 30,
			(W, H), True)


	# 初始化當(dāng)前狀態(tài)以及由（1）我們的對象檢測器或（2）相關(guān)跟蹤器返回的邊界框矩形列表
	status = "Waiting"
	rects = []
	# 檢查我們是否應(yīng)該運行計算量更大的目標(biāo)檢測方法來幫助我們的跟蹤器
	if totalFrames % args["skip_frames"] == 0:
		# 設(shè)置狀態(tài)并初始化我們的新對象跟蹤器集
		status = "Detecting"
		trackers = []
		# 將幀轉(zhuǎn)換為 blob 并通過網(wǎng)絡(luò)傳遞 blob 并獲得檢測結(jié)果
		blob = cv2.dnn.blobFromImage(frame, 0.007843, (W, H), 127.5)
		net.setInput(blob)
		detections = net.forward()

		# 循環(huán)檢測結(jié)果
		for i in np.arange(0, detections.shape[2]):
			# 提取與預(yù)測相關(guān)的置信度(即概率)
			confidence = detections[0, 0, i, 2]
			# 通過要求最小置信度過濾掉弱檢測
			if confidence > args["confidence"]:
				# 從檢測列表中提取類標(biāo)簽的索引
				idx = int(detections[0, 0, i, 1])
				# 如果類標(biāo)簽不是人，則忽略它
				if CLASSES[idx] != "person":
					continue
				# 計算對象邊界框的 (x, y) 坐標(biāo)
				box = detections[0, 0, i, 3:7] * np.array([W, H, W, H])
				(startX, startY, endX, endY) = box.astype("int")
				# 利用邊界框坐標(biāo)構(gòu)造一個 dlib 矩形對象，然后啟動 dlib 相關(guān)跟蹤器
				tracker = dlib.correlation_tracker()
				rect = dlib.rectangle(startX, startY, endX, endY)
				tracker.start_track(rgb, rect)
				# 將跟蹤器添加到我們的跟蹤器列表中，以便我們可以在跳幀期間使用它
				trackers.append(tracker)

	# 否則，我們應(yīng)該利用目標(biāo)跟蹤器而不是目標(biāo)檢測器來獲得更高的FPS
	else:
		# 遍歷跟蹤器
		for tracker in trackers:
			# 將系統(tǒng)的狀態(tài)設(shè)置為“跟蹤”而不是“等待”或“檢測”
			status = "Tracking"
			# 更新跟蹤器并獲取更新的位置
			tracker.update(rgb)
			pos = tracker.get_position()
			# 解包位置對象
			startX = int(pos.left())
			startY = int(pos.top())
			endX = int(pos.right())
			endY = int(pos.bottom())
			# 將邊界框坐標(biāo)添加到矩形列表
			rects.append((startX, startY, endX, endY))

	# 在幀中心畫一條水平線——一旦一個物體穿過這條線，我們將確定他們是在“向上”還是“向下”移動。
	cv2.line(frame, (0, H // 2), (W, H // 2), (0, 255, 255), 2)
	# 使用質(zhì)心跟蹤器將 (1) 舊對象質(zhì)心與 (2) 新計算的對象質(zhì)心相關(guān)聯(lián)
	objects = ct.update(rects)

	# 循環(huán)遍歷被跟蹤的對象
	for (objectID, centroid) in objects.items():
		# 檢查當(dāng)前對象 ID 是否存在可跟蹤對象
		to = trackableObjects.get(objectID, None)
		# 如果沒有現(xiàn)有的可跟蹤對象，則創(chuàng)建一個
		if to is None:
			to = TrackableObject(objectID, centroid)
		# 否則，有一個可追蹤的物體，所以我們可以利用它來確定方向
		else:
			# *當(dāng)前*質(zhì)心的 y 坐標(biāo)與 *previous* 質(zhì)心的平均值之間的差異
			# 將告訴我們物體在哪個方向移動（“向上”為負(fù)，“向下”為正）
			y = [c[1] for c in to.centroids]
			direction = centroid[1] - np.mean(y)
			to.centroids.append(centroid)
			# 檢查對象是否已被計數(shù)
			if not to.counted:
				# 如果方向為負(fù)(表示物體向上移動)且質(zhì)心在中線以上，則計算物體
				if direction < 0 and centroid[1] < H // 2:
					totalUp += 1
					to.counted = True
				# 如果方向為正（表示物體正在向下移動）并且質(zhì)心低于中心線，則計算物體
				elif direction > 0 and centroid[1] > H // 2:
					totalDown += 1
					to.counted = True
		# 將可跟蹤對象存儲在我們的字典中
		trackableObjects[objectID] = to

		# 在輸出幀上繪制對象的 ID 和對象的質(zhì)心
		text = "ID {}".format(objectID)
		cv2.putText(frame, text, (centroid[0] - 10, centroid[1] - 10),
			cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2)
		cv2.circle(frame, (centroid[0], centroid[1]), 4, (0, 255, 0), -1)
	# 構(gòu)建我們將在幀上顯示的信息元組
	info = [
		("Up", totalUp),
		("Down", totalDown),
		("Status", status),
	]
	# 遍歷信息元組并將它們繪制在我們的幀上
	for (i, (k, v)) in enumerate(info):
		text = "{}: {}".format(k, v)
		cv2.putText(frame, text, (10, H - ((i * 20) + 20)),
			cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)



	# 檢查我們是否應(yīng)該將幀寫入磁盤
	if writer is not None:
		writer.write(frame)
	# 顯示輸出幀
	cv2.imshow("Frame", frame)
	key = cv2.waitKey(1) & 0xFF
	# 如果' q '鍵被按下，中斷循環(huán)
	if key == ord("q"):
		break
	# 增加到目前為止處理的幀總數(shù)，然后更新 FPS 計數(shù)器
	totalFrames += 1
	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()
# 如果我們不使用視頻文件，請停止攝像頭視頻流
if not args.get("input", False):
	vs.stop()
# 否則，釋放視頻文件指針
else:
	vs.release()
# 關(guān)閉所有打開的窗口
cv2.destroyAllWindows()

centroidtracker.py

（1）質(zhì)心跟蹤器是最可靠的跟蹤器之一。

（2）為了簡單起見，質(zhì)心跟蹤器計算包圍框的質(zhì)心。

（3）也就是說，邊界框是圖像中對象的(x, y)坐標(biāo)。

（4）一旦我們的SSD獲得了坐標(biāo)，跟蹤器就會計算包圍框的質(zhì)心(中心)。換句話說，就是物體的中心。

（5）然后為每一個被檢測到的特定對象分配一個唯一的ID，用于跟蹤幀序列。

from scipy.spatial import distance as dist
from collections import OrderedDict
import numpy as np

class CentroidTracker:
	def __init__(self, maxDisappeared=50, maxDistance=50):
		# 初始化下一個唯一的對象ID，并使用兩個有序字典來跟蹤給定對象ID到其質(zhì)心的映射，
		# 以及它被標(biāo)記為“消失”的連續(xù)幀數(shù)
		self.nextObjectID = 0
		self.objects = OrderedDict()
		self.disappeared = OrderedDict()

		# 存儲一個給定對象允許被標(biāo)記為“消失”的最大連續(xù)幀數(shù)，直到我們需要從跟蹤中注銷該對象
		self.maxDisappeared = maxDisappeared

		# 存儲質(zhì)心之間的最大距離以關(guān)聯(lián)對象——如果距離大于這個最大距離，我們開始將對象標(biāo)記為“消失”
		self.maxDistance = maxDistance

	def register(self, centroid):
		# 注冊對象時，我們使用下一個可用的對象 ID 來存儲質(zhì)心
		self.objects[self.nextObjectID] = centroid
		self.disappeared[self.nextObjectID] = 0
		self.nextObjectID += 1

	def deregister(self, objectID):
		# 要注銷對象 ID，我們從各自的字典中刪除對象 ID
		del self.objects[objectID]
		del self.disappeared[objectID]

	def update(self, rects):
		# 檢查輸入邊界框矩形列表是否為空
		if len(rects) == 0:
			# 循環(huán)遍歷任何現(xiàn)有的跟蹤對象并將它們標(biāo)記為消失
			for objectID in list(self.disappeared.keys()):
				self.disappeared[objectID] += 1

				# 如果我們已經(jīng)達(dá)到給定對象被標(biāo)記為消失的最大連續(xù)幀數(shù)，則取消注冊它
				if self.disappeared[objectID] > self.maxDisappeared:
					self.deregister(objectID)

			# 早點返回，因為沒有要更新的質(zhì)心或跟蹤信息
			return self.objects

		# 初始化當(dāng)前幀的輸入質(zhì)心數(shù)組
		inputCentroids = np.zeros((len(rects), 2), dtype="int")

		# 循環(huán)邊界框矩形
		for (i, (startX, startY, endX, endY)) in enumerate(rects):
			# 使用邊界框坐標(biāo)推導(dǎo)出質(zhì)心
			cX = int((startX + endX) / 2.0)
			cY = int((startY + endY) / 2.0)
			inputCentroids[i] = (cX, cY)

		# 如果我們當(dāng)前沒有跟蹤任何對象，則獲取輸入質(zhì)心并注冊它們中的每一個
		if len(self.objects) == 0:
			for i in range(0, len(inputCentroids)):
				self.register(inputCentroids[i])

		# 否則，我們目前正在跟蹤對象，因此我們需要嘗試將輸入質(zhì)心與現(xiàn)有對象質(zhì)心匹配
		else:
			# 獲取一組對象 ID 和相應(yīng)的質(zhì)心
			objectIDs = list(self.objects.keys())
			objectCentroids = list(self.objects.values())

			# 分別計算每對對象質(zhì)心和輸入質(zhì)心之間的距離——我們的目標(biāo)是將輸入質(zhì)心與現(xiàn)有對象質(zhì)心匹配
			D = dist.cdist(np.array(objectCentroids), inputCentroids)

			# 為了執(zhí)行這種匹配，我們必須 (1) 找到每一行中的最小值，
			# 然后 (2) 根據(jù)它們的最小值對行索引進(jìn)行排序，以便具有最小值的行位于索引列表的 *front*
			rows = D.min(axis=1).argsort()

			# 接下來，我們對列執(zhí)行類似的處理，方法是在每個列中找到最小的值，
			# 然后使用之前計算的行索引列表進(jìn)行排序
			cols = D.argmin(axis=1)[rows]

			# 為了確定我們是否需要更新、注冊或取消注冊一個對象，我們需要跟蹤我們已經(jīng)檢查過的行和列索引
			usedRows = set()
			usedCols = set()

			# 循環(huán)遍歷（行，列）索引元組的組合
			for (row, col) in zip(rows, cols):
				# 如果我們之前已經(jīng)檢查過行值或列值，請忽略它
				if row in usedRows or col in usedCols:
					continue

				# 如果質(zhì)心之間的距離大于最大距離，則不要將兩個質(zhì)心關(guān)聯(lián)到同一個對象
				if D[row, col] > self.maxDistance:
					continue

				# 否則，獲取當(dāng)前行的對象 ID，設(shè)置其新質(zhì)心，并重置消失的計數(shù)器
				objectID = objectIDs[row]
				self.objects[objectID] = inputCentroids[col]
				self.disappeared[objectID] = 0

				# 表明我們已經(jīng)分別檢查了每個行和列索引
				usedRows.add(row)
				usedCols.add(col)

			# 計算我們尚未檢查的行和列索引
			unusedRows = set(range(0, D.shape[0])).difference(usedRows)
			unusedCols = set(range(0, D.shape[1])).difference(usedCols)

			# 如果對象質(zhì)心的數(shù)量等于或大于輸入質(zhì)心的數(shù)量，
			# 我們需要檢查并查看其中一些對象是否可能已經(jīng)消失
			if D.shape[0] >= D.shape[1]:
				# 循環(huán)未使用的行索引
				for row in unusedRows:
					# 獲取相應(yīng)行索引的對象 ID 并增加消失的計數(shù)器
					objectID = objectIDs[row]
					self.disappeared[objectID] += 1

					# 檢查對象的連續(xù)幀數(shù)是否被標(biāo)記為“消失”，以注銷該對象
					if self.disappeared[objectID] > self.maxDisappeared:
						self.deregister(objectID)

			# 否則，如果輸入質(zhì)心的數(shù)量大于現(xiàn)有對象質(zhì)心的數(shù)量，我們需要將每個新的輸入質(zhì)心注冊為可跟蹤對象
			else:
				for col in unusedCols:
					self.register(inputCentroids[col])

		# 返回可跟蹤對象的集合
		return self.objects

trackableobject.py

class TrackableObject:
	def __init__(self, objectID, centroid):
		# 存儲對象 ID，然后使用當(dāng)前質(zhì)心初始化質(zhì)心列表
		self.objectID = objectID
		self.centroids = [centroid]

		# 初始化一個布爾值，用于指示對象是否已被計數(shù)
		self.counted = False

8.運行結(jié)果

打開終端，執(zhí)行以下命令:

python people_counter.py --prototxt mobilenet_ssd/MobileNetSSD_deploy.prototxt \
    --model mobilenet_ssd/MobileNetSSD_deploy.caffemodel \
    --input videos/example_01.mp4 --output output/output_01.avi

我們的人員計數(shù)正在計算以下人數(shù):

• 正進(jìn)入百貨商店(下)

• 離開的人數(shù)(上)

在第一個視頻的最后，你會看到有7個人進(jìn)入，3個人離開。

此外，檢查終端輸出，你會發(fā)現(xiàn)我們的人計數(shù)器能夠?qū)崟r運行，達(dá)到34幀每秒。盡管我們正在使用深度學(xué)習(xí)對象檢測器來更準(zhǔn)確地檢測人。

我們的 34 FPS 幀率是通過我們的兩個階段過程實現(xiàn)的： 每 30 幀檢測一次人 然后在其間的所有幀中應(yīng)用更快、更有效的對象跟蹤算法。

9.改進(jìn)我們的人員計數(shù)器應(yīng)用程序

為了構(gòu)建我們的 OpenCV 人員計數(shù)器，我們使用了 dlib 的相關(guān)性跟蹤器。此方法易于使用，并且只需要很少的代碼。

然而，我們的實現(xiàn)有點低效——為了跟蹤多個對象，我們需要創(chuàng)建關(guān)聯(lián)跟蹤器對象的多個實例。然后當(dāng)我們需要在后續(xù)幀中計算對象的位置時，我們需要遍歷所有 N 個對象跟蹤器并獲取更新的位置。

所有這些計算都將發(fā)生在我們腳本的主執(zhí)行線程中，從而降低了我們的 FPS 速率。

因此，提高性能的一種簡單方法是使用dlib的多對象跟蹤器，以使我們的 FPS 速率提高 45%！ 注意：OpenCV 也實現(xiàn)了多對象跟蹤，但不是多進(jìn)程（至少在撰寫本文時）。 OpenCV 的多對象方法當(dāng)然更容易使用，但如果沒有多處理能力，在這種情況下它并沒有多大幫助。

最后，為了獲得更高的跟蹤精度（但在沒有快速 GPU 的情況下會犧牲速度），您可以研究基于深度學(xué)習(xí)的對象跟蹤器，例如 Deep SORT。

關(guān)于“如何使用OpenCV和Python構(gòu)建人員計數(shù)器”這篇文章的內(nèi)容就介紹到這里，感謝各位的閱讀！相信大家對“如何使用OpenCV和Python構(gòu)建人員計數(shù)器”知識都有一定的了解，大家如果還想學(xué)習(xí)更多知識，歡迎關(guān)注億速云行業(yè)資訊頻道。