0 簡(jiǎn)介

?? 優(yōu)質(zhì)競(jìng)賽項(xiàng)目系列，今天要分享的是

基于深度學(xué)習(xí)的口罩佩戴檢測(cè)【全網(wǎng)最詳細(xì)】 - opencv 卷積神經(jīng)網(wǎng)絡(luò) 機(jī)器視覺 深度學(xué)習(xí)

該項(xiàng)目較為新穎，適合作為競(jìng)賽課題方向，學(xué)長(zhǎng)非常推薦！

?? 更多資料, 項(xiàng)目分享：

https://gitee.com/dancheng-senior/postgraduate文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-642096.html

1 課題背景

從2019年末開始，新型冠狀病毒肺炎（COVID-19）在我國(guó)全面爆發(fā)并迅速傳播，同時(shí)國(guó)家衛(wèi)生健康委員會(huì)也積極響應(yīng)密切關(guān)注全國(guó)疫情的動(dòng)態(tài)變化并且發(fā)布了相關(guān)的預(yù)防指南，強(qiáng)調(diào)個(gè)人出行需要做好安全措施，在公共場(chǎng)合必須嚴(yán)格按照要求佩戴口罩。自從新型冠狀病毒蔓延以來(lái)，各行各業(yè)都受到了巨大的沖擊，嚴(yán)重影響到人們正常生產(chǎn)和生活。

新型冠狀病毒具有很強(qiáng)的傳播和生存能力，只要條件合適能存活五天之久，并且可以通過(guò)唾液，飛沫等多種方式進(jìn)行傳播，為有效的減少病毒的傳播佩戴口罩是一個(gè)很好的辦法。盡管這一時(shí)期國(guó)外的形勢(shì)不容樂觀，但是在全國(guó)上下齊心努力之下我國(guó)的防疫取得了階段性成功，各行業(yè)都在積極復(fù)蘇，管理也隨之變化進(jìn)入到常態(tài)化階段。在這一階段復(fù)工復(fù)產(chǎn)也是大勢(shì)所趨，口罩出行也成為了一種常態(tài)。正確佩戴口罩能夠有效減少飛沫傳染的風(fēng)險(xiǎn)，特別是在公共場(chǎng)所，這種舉措尤為重要。但是，仍然還需要提高公眾對(duì)主動(dòng)佩戴口罩的觀念，在常態(tài)化管理下人們的防范意識(shí)越來(lái)越薄弱，口罩隨意佩戴或者不佩戴的情況屢見不鮮。

因此，在這期間，有意識(shí)地戴口罩不僅僅是每個(gè)公民的公共道德還是自我修養(yǎng)的表現(xiàn)。這不但需要個(gè)人積極配合，而且還需要某些監(jiān)管以及有效的治理方法。

?? 2 口罩佩戴算法實(shí)現(xiàn)

2.1 YOLO 模型概覽

YOLO 的縮寫是 You only look once。YOLO 模型可以直接根據(jù)圖片輸出包含對(duì)象的區(qū)域與區(qū)域?qū)?yīng)的分類，一步到位，不像 RCNN
系列的模型需要先計(jì)算包含對(duì)象的區(qū)域，再根據(jù)區(qū)域判斷對(duì)應(yīng)的分類，YOLO 模型的速度比 RCNN 系列的模型要快很多。

YOLO 模型的結(jié)構(gòu)如下：

是不是覺得有點(diǎn)熟悉？看上去就像 Faster-RCNN 的區(qū)域生成網(wǎng)絡(luò) (RPN) 啊。的確，YOLO
模型原理上就是尋找區(qū)域的同時(shí)判斷區(qū)域包含的對(duì)象分類，YOLO 模型與區(qū)域生成網(wǎng)絡(luò)有以下的不同：

• YOLO 模型會(huì)輸出各個(gè)區(qū)域是否包含對(duì)象中心，而不是包含對(duì)象的一部分
• YOLO 模型會(huì)同時(shí)輸出對(duì)象分類
• YOLO 模型輸出的區(qū)域偏移會(huì)根據(jù)對(duì)象中心點(diǎn)計(jì)算，具體算法在下面說(shuō)明

YOLO 模型與 Faster-RCNN
的區(qū)域生成網(wǎng)絡(luò)最大的不同是會(huì)判斷各個(gè)區(qū)域是否包含對(duì)象中心，如下圖中狗臉覆蓋了四個(gè)區(qū)域，但只有左下角的區(qū)域包含了狗臉的中心，YOLO
模型應(yīng)該只判斷這個(gè)區(qū)域包含對(duì)象。

當(dāng)然，如果對(duì)象中心非常接近區(qū)域的邊界，那么判斷起來(lái)將會(huì)很困難，YOLO 模型在訓(xùn)練的時(shí)候會(huì)忽略對(duì)象重疊率高于一定水平的區(qū)域，具體可以參考后面給出的代碼。

YOLO 模型會(huì)針對(duì)各個(gè)區(qū)域輸出以下的結(jié)果，這里假設(shè)有三個(gè)分類：

• 是否包含對(duì)象中心 (是為 1, 否為 0)
• 區(qū)域偏移 x
• 區(qū)域偏移 y
• 區(qū)域偏移 w
• 區(qū)域偏移 h
• 分類 1 的可能性 (0 ~ 1)
• 分類 2 的可能性 (0 ~ 1)
• 分類 3 的可能性 (0 ~ 1)

輸出結(jié)果的維度是 批次大小, 區(qū)域數(shù)量, 5 + 分類數(shù)量。

區(qū)域偏移用于調(diào)整輸出的區(qū)域范圍，例如上圖中狗臉的中心點(diǎn)大約在區(qū)域的右上角，如果把區(qū)域左上角看作 (0, 0)，右下角看作 (1, 1)，那么狗臉中心點(diǎn)應(yīng)該在
(0.95, 0.1) 的位置，而狗臉大小相對(duì)于區(qū)域長(zhǎng)寬大概是 (1.3, 1.5) 倍，生成訓(xùn)練數(shù)據(jù)的時(shí)候會(huì)根據(jù)這 4
個(gè)值計(jì)算區(qū)域偏移，具體計(jì)算代碼在下面給出。

看到這里你可能會(huì)想，YOLO 模型看起來(lái)很簡(jiǎn)單啊，我可以丟掉操蛋的 Faster-RCNN 模型了??。

不，沒那么簡(jiǎn)單，以上介紹的只是 YOLOv1 模型，YOLOv1 模型的精度非常低，后面為了改進(jìn)識(shí)別精度還發(fā)展出 YOLOv2, YOLOv3,
YOLOv4, YOLOv5 模型??，學(xué)長(zhǎng)下面會(huì)給出 YOLOv3 模型的實(shí)現(xiàn)。Y

OLOv4 和 YOLOv5 模型主要改進(jìn)了提取特征用的 CNN 模型 (也稱骨干網(wǎng)絡(luò) Backbone Network)，原始的 YOLO 模型使用了
C 語(yǔ)言編寫的 Darknet 作為骨干網(wǎng)絡(luò)，而這篇使用 Resnet 作為骨干網(wǎng)絡(luò)，所以今天學(xué)長(zhǎng)只介紹到 YOLOv3。

2.2 YOLOv3

YOLOv3 引入了多尺度檢測(cè)機(jī)制 (Multi-Scale Detection)，這個(gè)機(jī)制可以說(shuō)是 YOLO 模型的精華，引入這個(gè)機(jī)制之前 YOLO
模型的精度很不理想，而引入之后 YOLO 模型達(dá)到了接近 Faster-RCNN 的精度，并且速度還是比 Faster-RCNN 要快。

多尺度檢測(cè)機(jī)制簡(jiǎn)單的來(lái)說(shuō)就是按不同的尺度劃分區(qū)域，然后再檢測(cè)這些不同大小的區(qū)域是否包含對(duì)象，檢測(cè)的時(shí)候大區(qū)域的特征會(huì)混合到小區(qū)域中，使得小區(qū)域判斷時(shí)擁有一定程度的上下文信息。

2.3 YOLO 口罩佩戴檢測(cè)實(shí)現(xiàn)

接下來(lái)學(xué)長(zhǎng)帶大家用 YOLO 模型把沒帶口罩的家伙抓出來(lái)吧。

數(shù)據(jù)集

學(xué)長(zhǎng)的這個(gè)數(shù)據(jù)集包含了 8535 張圖片 (部分圖片沒有使用)，其中各個(gè)分類的數(shù)量如下：

• 戴口罩的區(qū)域 (with_mask): 3232 個(gè)
• 不戴口罩的區(qū)域 (without_mask): 717 個(gè)
• 帶了口罩但姿勢(shì)不正確的區(qū)域 (mask_weared_incorrect): 123 個(gè)

因?yàn)閹Я丝谡值藙?shì)不正確的樣本數(shù)量很少，所以都?xì)w到戴口罩里面去??。

2.4 實(shí)現(xiàn)代碼

使用這個(gè)數(shù)據(jù)集訓(xùn)練，并且訓(xùn)練成功以后使用模型識(shí)別圖片或視頻的完整代碼如下：

?

import os
import sys
import torch
import gzip
import itertools
import random
import numpy
import math
import pandas
import json
from PIL import Image
from PIL import ImageDraw
from PIL import ImageFont
from torch import nn
from matplotlib import pyplot
from collections import defaultdict
from collections import deque
import xml.etree.cElementTree as ET

# 縮放圖片的大小
IMAGE_SIZE = (256, 192)
# 訓(xùn)練使用的數(shù)據(jù)集路徑
DATASET_1_IMAGE_DIR = "./archive/images"
DATASET_1_ANNOTATION_DIR = "./archive/annotations"
DATASET_2_IMAGE_DIR = "./784145_1347673_bundle_archive/train/image_data"
DATASET_2_BOX_CSV_PATH = "./784145_1347673_bundle_archive/train/bbox_train.csv"
# 分類列表
# YOLO 原則上不需要 other 分類，但實(shí)測(cè)中添加這個(gè)分類有助于提升標(biāo)簽分類的精確度
CLASSES = [ "other", "with_mask", "without_mask" ]
CLASSES_MAPPING = { c: index for index, c in enumerate(CLASSES) }
# 判斷是否存在對(duì)象使用的區(qū)域重疊率的閾值 (另外要求對(duì)象中心在區(qū)域內(nèi))
IOU_POSITIVE_THRESHOLD = 0.30
IOU_NEGATIVE_THRESHOLD = 0.30

# 用于啟用 GPU 支持
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

class BasicBlock(nn.Module):
    """ResNet 使用的基礎(chǔ)塊"""
    expansion = 1 # 定義這個(gè)塊的實(shí)際出通道是 channels_out 的幾倍，這里的實(shí)現(xiàn)固定是一倍
    def __init__(self, channels_in, channels_out, stride):
        super().__init__()
        # 生成 3x3 的卷積層
        # 處理間隔 stride = 1 時(shí)，輸出的長(zhǎng)寬會(huì)等于輸入的長(zhǎng)寬，例如 (32-3+2)//1+1 == 32
        # 處理間隔 stride = 2 時(shí)，輸出的長(zhǎng)寬會(huì)等于輸入的長(zhǎng)寬的一半，例如 (32-3+2)//2+1 == 16
        # 此外 resnet 的 3x3 卷積層不使用偏移值 bias
        self.conv1 = nn.Sequential(
            nn.Conv2d(channels_in, channels_out, kernel_size=3, stride=stride, padding=1, bias=False),
            nn.BatchNorm2d(channels_out))
        # 再定義一個(gè)讓輸出和輸入維度相同的 3x3 卷積層
        self.conv2 = nn.Sequential(
            nn.Conv2d(channels_out, channels_out, kernel_size=3, stride=1, padding=1, bias=False),
            nn.BatchNorm2d(channels_out))
        # 讓原始輸入和輸出相加的時(shí)候，需要維度一致，如果維度不一致則需要整合
        self.identity = nn.Sequential()
        if stride != 1 or channels_in != channels_out * self.expansion:
            self.identity = nn.Sequential(
                nn.Conv2d(channels_in, channels_out * self.expansion, kernel_size=1, stride=stride, bias=False),
                nn.BatchNorm2d(channels_out * self.expansion))

    def forward(self, x):
        # x => conv1 => relu => conv2 => + => relu
        # |                              ^
        # |==============================|
        tmp = self.conv1(x)
        tmp = nn.functional.relu(tmp, inplace=True)
        tmp = self.conv2(tmp)
        tmp += self.identity(x)
        y = nn.functional.relu(tmp, inplace=True)
        return y

class MyModel(nn.Module):
    """YOLO (基于 ResNet 的變種)"""
    Anchors = None # 錨點(diǎn)列表，包含 錨點(diǎn)數(shù)量 * 形狀數(shù)量 的范圍
    AnchorSpans = (16, 32, 64) # 尺度列表，值為錨點(diǎn)之間的距離
    AnchorAspects = ((1, 1), (1.5, 1.5)) # 錨點(diǎn)對(duì)應(yīng)區(qū)域的長(zhǎng)寬比例列表
    AnchorOutputs = 1 + 4 + len(CLASSES) # 每個(gè)錨點(diǎn)范圍對(duì)應(yīng)的輸出數(shù)量，是否對(duì)象中心 (1) + 區(qū)域偏移 (4) + 分類數(shù)量
    AnchorTotalOutputs = AnchorOutputs * len(AnchorAspects) # 每個(gè)錨點(diǎn)對(duì)應(yīng)的輸出數(shù)量
    ObjScoreThreshold = 0.9 # 認(rèn)為是對(duì)象中心所需要的最小分?jǐn)?shù)
    IOUMergeThreshold = 0.3 # 判斷是否應(yīng)該合并重疊區(qū)域的重疊率閾值

    def __init__(self):
        super().__init__()
        # 抽取圖片特征的 ResNet
        # 因?yàn)殄^點(diǎn)距離有三個(gè)，這里最后會(huì)輸出各個(gè)錨點(diǎn)距離對(duì)應(yīng)的特征
        self.previous_channels_out = 4
        self.resnet_models = nn.ModuleList([
            nn.Sequential(
                nn.Conv2d(3, self.previous_channels_out, kernel_size=3, stride=1, padding=1, bias=False),
                nn.BatchNorm2d(self.previous_channels_out),
                nn.ReLU(inplace=True),
                self._make_layer(BasicBlock, channels_out=16, num_blocks=2, stride=1),
                self._make_layer(BasicBlock, channels_out=32, num_blocks=2, stride=2),
                self._make_layer(BasicBlock, channels_out=64, num_blocks=2, stride=2),
                self._make_layer(BasicBlock, channels_out=128, num_blocks=2, stride=2),
                self._make_layer(BasicBlock, channels_out=256, num_blocks=2, stride=2)),
            self._make_layer(BasicBlock, channels_out=256, num_blocks=2, stride=2),
            self._make_layer(BasicBlock, channels_out=256, num_blocks=2, stride=2)
        ])
        # 根據(jù)各個(gè)錨點(diǎn)距離對(duì)應(yīng)的特征預(yù)測(cè)輸出的卷積層
        # 大的錨點(diǎn)距離抽取的特征會(huì)合并到小的錨點(diǎn)距離抽取的特征
        # 這里的三個(gè)子模型意義分別是:
        # - 計(jì)算用于合并的特征
        # - 放大特征
        # - 計(jì)算最終的預(yù)測(cè)輸出
        self.yolo_detectors = nn.ModuleList([
            nn.ModuleList([nn.Sequential(
                nn.Conv2d(256 if index == 0 else 512, 256, kernel_size=1, stride=1, padding=0, bias=True),
                nn.ReLU(inplace=True),
                nn.Conv2d(256, 512, kernel_size=3, stride=1, padding=1, bias=True),
                nn.ReLU(inplace=True),
                nn.Conv2d(512, 256, kernel_size=1, stride=1, padding=0, bias=True),
                nn.ReLU(inplace=True)),
            nn.Upsample(scale_factor=2, mode="nearest"),
            nn.Sequential(
                nn.Conv2d(256, 512, kernel_size=3, stride=1, padding=1, bias=True),
                nn.ReLU(inplace=True),
                nn.Conv2d(512, 256, kernel_size=3, stride=1, padding=1, bias=True),
                nn.ReLU(inplace=True),
                nn.Conv2d(256, MyModel.AnchorTotalOutputs, kernel_size=1, stride=1, padding=0, bias=True))])
            for index in range(len(self.resnet_models))
        ])
        # 處理結(jié)果范圍的函數(shù)
        self.sigmoid = nn.Sigmoid()

    def _make_layer(self, block_type, channels_out, num_blocks, stride):
        """創(chuàng)建 resnet 使用的層"""
        blocks = []
        # 添加第一個(gè)塊
        blocks.append(block_type(self.previous_channels_out, channels_out, stride))
        self.previous_channels_out = channels_out * block_type.expansion
        # 添加剩余的塊，剩余的塊固定處理間隔為 1，不會(huì)改變長(zhǎng)寬
        for _ in range(num_blocks-1):
            blocks.append(block_type(self.previous_channels_out, self.previous_channels_out, 1))
            self.previous_channels_out *= block_type.expansion
        return nn.Sequential(*blocks)

    @staticmethod
    def _generate_anchors():
        """根據(jù)錨點(diǎn)和形狀生成錨點(diǎn)范圍列表"""
        w, h = IMAGE_SIZE
        anchors = []
        for span in MyModel.AnchorSpans:
            for x in range(0, w, span):
                for y in range(0, h, span):
                    xcenter, ycenter = x + span / 2, y + span / 2
                    for ratio in MyModel.AnchorAspects:
                        ww = span * ratio[0]
                        hh = span * ratio[1]
                        xx = xcenter - ww / 2
                        yy = ycenter - hh / 2
                        xx = max(int(xx), 0)
                        yy = max(int(yy), 0)
                        ww = min(int(ww), w - xx)
                        hh = min(int(hh), h - yy)
                        anchors.append((xx, yy, ww, hh))
        return anchors

    def forward(self, x):
        # 抽取各個(gè)錨點(diǎn)距離對(duì)應(yīng)的特征
        # 維度分別是:
        # torch.Size([16, 256, 16, 12])
        # torch.Size([16, 256, 8, 6])
        # torch.Size([16, 256, 4, 3])
        features_list = []
        resnet_input = x
        for m in self.resnet_models:
            resnet_input = m(resnet_input)
            features_list.append(resnet_input)
        # 根據(jù)特征預(yù)測(cè)輸出
        # 維度分別是:
        # torch.Size([16, 16, 4, 3])
        # torch.Size([16, 16, 8, 6])
        # torch.Size([16, 16, 16, 12])
        # 16 是 (5 + 分類3) * 形狀2
        previous_upsampled_feature = None
        outputs = []
        for index, feature in enumerate(reversed(features_list)):
            if previous_upsampled_feature is not None:
                # 合并大的錨點(diǎn)距離抽取的特征到小的錨點(diǎn)距離抽取的特征
                feature = torch.cat((feature, previous_upsampled_feature), dim=1)
            # 計(jì)算用于合并的特征
            hidden = self.yolo_detectors[index][0](feature)
            # 放大特征 (用于下一次處理時(shí)合并)
            upsampled = self.yolo_detectors[index][1](hidden)
            # 計(jì)算最終的預(yù)測(cè)輸出
            output = self.yolo_detectors[index][2](hidden)
            previous_upsampled_feature = upsampled
            outputs.append(output)
        # 連接所有輸出
        # 注意順序需要與 Anchors 一致
        outputs_flatten = []
        for output in reversed(outputs):
            output = output.permute(0, 2, 3, 1)
            output = output.reshape(output.shape[0], -1, MyModel.AnchorOutputs)
            outputs_flatten.append(output)
        outputs_all = torch.cat(outputs_flatten, dim=1)
        # 是否對(duì)象中心應(yīng)該在 0 ~ 1 之間，使用 sigmoid 處理
        outputs_all[:,:,:1] = self.sigmoid(outputs_all[:,:,:1])
        # 分類應(yīng)該在 0 ~ 1 之間，使用 sigmoid 處理
        outputs_all[:,:,5:] = self.sigmoid(outputs_all[:,:,5:])
        return outputs_all

    @staticmethod
    def loss_function(predicted, actual):
        """YOLO 使用的多任務(wù)損失計(jì)算器"""
        result_tensor, result_isobject_masks, result_nonobject_masks = actual
        objectness_losses = []
        offsets_losses = []
        labels_losses = []
        for x in range(result_tensor.shape[0]):
            mask_positive = result_isobject_masks[x]
            mask_negative = result_nonobject_masks[x]
            # 計(jì)算是否對(duì)象中心的損失，分別針對(duì)正負(fù)樣本計(jì)算
            # 因?yàn)榇蟛糠謪^(qū)域不包含對(duì)象中心，這里減少負(fù)樣本的損失對(duì)調(diào)整參數(shù)的影響
            objectness_loss_positive = nn.functional.mse_loss(
                predicted[x,mask_positive,0], result_tensor[x,mask_positive,0])
            objectness_loss_negative = nn.functional.mse_loss(
                predicted[x,mask_negative,0], result_tensor[x,mask_negative,0]) * 0.5
            objectness_losses.append(objectness_loss_positive)
            objectness_losses.append(objectness_loss_negative)
            # 計(jì)算區(qū)域偏移的損失，只針對(duì)正樣本計(jì)算
            offsets_loss = nn.functional.mse_loss(
                predicted[x,mask_positive,1:5], result_tensor[x,mask_positive,1:5])
            offsets_losses.append(offsets_loss)
            # 計(jì)算標(biāo)簽分類的損失，分別針對(duì)正負(fù)樣本計(jì)算
            labels_loss_positive = nn.functional.binary_cross_entropy(
                predicted[x,mask_positive,5:], result_tensor[x,mask_positive,5:])
            labels_loss_negative = nn.functional.binary_cross_entropy(
                predicted[x,mask_negative,5:], result_tensor[x,mask_negative,5:]) * 0.5
            labels_losses.append(labels_loss_positive)
            labels_losses.append(labels_loss_negative)
        loss = (
            torch.mean(torch.stack(objectness_losses)) +
            torch.mean(torch.stack(offsets_losses)) +
            torch.mean(torch.stack(labels_losses)))
        return loss

    @staticmethod
    def calc_accuracy(actual, predicted):
        """YOLO 使用的正確率計(jì)算器，這里只計(jì)算是否對(duì)象中心與標(biāo)簽分類的正確率，區(qū)域偏移不計(jì)算"""
        result_tensor, result_isobject_masks, result_nonobject_masks = actual
        # 計(jì)算是否對(duì)象中心的正確率，正樣本和負(fù)樣本的正確率分別計(jì)算再平均
        a = result_tensor[:,:,0]
        p = predicted[:,:,0] > MyModel.ObjScoreThreshold
        obj_acc_positive = ((a == 1) & (p == 1)).sum().item() / ((a == 1).sum().item() + 0.00001)
        obj_acc_negative = ((a == 0) & (p == 0)).sum().item() / ((a == 0).sum().item() + 0.00001)
        obj_acc = (obj_acc_positive + obj_acc_negative) / 2
        # 計(jì)算標(biāo)簽分類的正確率
        cls_total = 0
        cls_correct = 0
        for x in range(result_tensor.shape[0]):
            mask = list(sorted(result_isobject_masks[x] + result_nonobject_masks[x]))
            actual_classes = result_tensor[x,mask,5:].max(dim=1).indices
            predicted_classes = predicted[x,mask,5:].max(dim=1).indices
            cls_total += len(mask)
            cls_correct += (actual_classes == predicted_classes).sum().item()
        cls_acc = cls_correct / cls_total
        return obj_acc, cls_acc

    @staticmethod
    def convert_predicted_result(predicted):
        """轉(zhuǎn)換預(yù)測(cè)結(jié)果到 (標(biāo)簽, 區(qū)域, 對(duì)象中心分?jǐn)?shù), 標(biāo)簽識(shí)別分?jǐn)?shù)) 的列表，重疊區(qū)域使用 NMS 算法合并"""
        # 記錄重疊的結(jié)果區(qū)域, 結(jié)果是 [ [(標(biāo)簽, 區(qū)域, RPN 分?jǐn)?shù), 標(biāo)簽識(shí)別分?jǐn)?shù))], ... ]
        final_result = []
        for anchor, tensor in zip(MyModel.Anchors, predicted):
            obj_score = tensor[0].item()
            if obj_score <= MyModel.ObjScoreThreshold:
                # 要求對(duì)象中心分?jǐn)?shù)超過(guò)一定值
                continue
            offset = tensor[1:5].tolist()
            offset[0] = max(min(offset[0], 1), 0) # 中心點(diǎn) x 的偏移應(yīng)該在 0 ~ 1 之間
            offset[1] = max(min(offset[1], 1), 0) # 中心點(diǎn) y 的偏移應(yīng)該在 0 ~ 1 之間
            box = adjust_box_by_offset(anchor, offset)
            label_max = tensor[5:].max(dim=0)
            cls_score = label_max.values.item()
            label = label_max.indices.item()
            if label == 0:
                # 跳過(guò)非對(duì)象分類
                continue
            for index in range(len(final_result)):
                exists_results = final_result[index]
                if any(calc_iou(box, r[1]) > MyModel.IOUMergeThreshold for r in exists_results):
                    exists_results.append((label, box, obj_score, cls_score))
                    break
            else:
                final_result.append([(label, box, obj_score, cls_score)])
        # 合并重疊的結(jié)果區(qū)域 (使用 對(duì)象中心分?jǐn)?shù) * 標(biāo)簽識(shí)別分?jǐn)?shù) 最高的區(qū)域?yàn)榻Y(jié)果區(qū)域)
        for index in range(len(final_result)):
            exists_results = final_result[index]
            exists_results.sort(key=lambda r: r[2]*r[3])
            final_result[index] = exists_results[-1]
        return final_result

    @staticmethod
    def fix_predicted_result_from_history(cls_result, history_results):
        """根據(jù)歷史結(jié)果減少預(yù)測(cè)結(jié)果中的誤判，適用于視頻識(shí)別，history_results 應(yīng)為指定了 maxlen 的 deque"""
        # 要求歷史結(jié)果中 50% 以上存在類似區(qū)域，并且選取歷史結(jié)果中最多的分類
        history_results.append(cls_result)
        final_result = []
        if len(history_results) < history_results.maxlen:
            # 歷史結(jié)果不足，不返回任何識(shí)別結(jié)果
            return final_result
        for label, box, rpn_score, cls_score in cls_result:
            # 查找歷史中的近似區(qū)域
            similar_results = []
            for history_result in history_results:
                history_result = [(calc_iou(r[1], box), r) for r in history_result]
                history_result.sort(key = lambda r: r[0])
                if history_result and history_result[-1][0] > MyModel.IOUMergeThreshold:
                    similar_results.append(history_result[-1][1])
            # 判斷近似區(qū)域數(shù)量是否過(guò)半
            if len(similar_results) < history_results.maxlen // 2:
                continue
            # 選取歷史結(jié)果中最多的分類
            cls_groups = defaultdict(lambda: [])
            for r in similar_results:
                cls_groups[r[0]].append(r)
            most_common = sorted(cls_groups.values(), key=len)[-1]
            # 添加最多的分類中的最新的結(jié)果
            final_result.append(most_common[-1])
        return final_result

MyModel.Anchors = MyModel._generate_anchors()

def save_tensor(tensor, path):
    """保存 tensor 對(duì)象到文件"""
    torch.save(tensor, gzip.GzipFile(path, "wb"))

def load_tensor(path):
    """從文件讀取 tensor 對(duì)象"""
    return torch.load(gzip.GzipFile(path, "rb"))

def calc_resize_parameters(sw, sh):
    """計(jì)算縮放圖片的參數(shù)"""
    sw_new, sh_new = sw, sh
    dw, dh = IMAGE_SIZE
    pad_w, pad_h = 0, 0
    if sw / sh < dw / dh:
        sw_new = int(dw / dh * sh)
        pad_w = (sw_new - sw) // 2 # 填充左右
    else:
        sh_new = int(dh / dw * sw)
        pad_h = (sh_new - sh) // 2 # 填充上下
    return sw_new, sh_new, pad_w, pad_h

def resize_image(img):
    """縮放圖片，比例不一致時(shí)填充"""
    sw, sh = img.size
    sw_new, sh_new, pad_w, pad_h = calc_resize_parameters(sw, sh)
    img_new = Image.new("RGB", (sw_new, sh_new))
    img_new.paste(img, (pad_w, pad_h))
    img_new = img_new.resize(IMAGE_SIZE)
    return img_new

def image_to_tensor(img):
    """轉(zhuǎn)換圖片對(duì)象到 tensor 對(duì)象"""
    arr = numpy.asarray(img)
    t = torch.from_numpy(arr)
    t = t.transpose(0, 2) # 轉(zhuǎn)換維度 H,W,C 到 C,W,H
    t = t / 255.0 # 正規(guī)化數(shù)值使得范圍在 0 ~ 1
    return t

def map_box_to_resized_image(box, sw, sh):
    """把原始區(qū)域轉(zhuǎn)換到縮放后的圖片對(duì)應(yīng)的區(qū)域"""
    x, y, w, h = box
    sw_new, sh_new, pad_w, pad_h = calc_resize_parameters(sw, sh)
    scale = IMAGE_SIZE[0] / sw_new
    x = int((x + pad_w) * scale)
    y = int((y + pad_h) * scale)
    w = int(w * scale)
    h = int(h * scale)
    if x + w > IMAGE_SIZE[0] or y + h > IMAGE_SIZE[1] or w == 0 or h == 0:
        return 0, 0, 0, 0
    return x, y, w, h

def map_box_to_original_image(box, sw, sh):
    """把縮放后圖片對(duì)應(yīng)的區(qū)域轉(zhuǎn)換到縮放前的原始區(qū)域"""
    x, y, w, h = box
    sw_new, sh_new, pad_w, pad_h = calc_resize_parameters(sw, sh)
    scale = IMAGE_SIZE[0] / sw_new
    x = int(x / scale - pad_w)
    y = int(y / scale - pad_h)
    w = int(w / scale)
    h = int(h / scale)
    if x + w > sw or y + h > sh or x < 0 or y < 0 or w == 0 or h == 0:
        return 0, 0, 0, 0
    return x, y, w, h

def calc_iou(rect1, rect2):
    """計(jì)算兩個(gè)區(qū)域重疊部分 / 合并部分的比率 (intersection over union)"""
    x1, y1, w1, h1 = rect1
    x2, y2, w2, h2 = rect2
    xi = max(x1, x2)
    yi = max(y1, y2)
    wi = min(x1+w1, x2+w2) - xi
    hi = min(y1+h1, y2+h2) - yi
    if wi > 0 and hi > 0: # 有重疊部分
        area_overlap = wi*hi
        area_all = w1*h1 + w2*h2 - area_overlap
        iou = area_overlap / area_all
    else: # 沒有重疊部分
        iou = 0
    return iou

def calc_box_offset(candidate_box, true_box):
    """計(jì)算候選區(qū)域與實(shí)際區(qū)域的偏移值，要求實(shí)際區(qū)域的中心點(diǎn)必須在候選區(qū)域中"""
    # 計(jì)算實(shí)際區(qū)域的中心點(diǎn)在候選區(qū)域中的位置，范圍會(huì)在 0 ~ 1 之間
    x1, y1, w1, h1 = candidate_box
    x2, y2, w2, h2 = true_box
    x_offset = ((x2 + w2 // 2) - x1) / w1
    y_offset = ((y2 + h2 // 2) - y1) / h1
    # 計(jì)算實(shí)際區(qū)域長(zhǎng)寬相對(duì)于候選區(qū)域長(zhǎng)寬的比例，使用 log 減少過(guò)大的值
    w_offset = math.log(w2 / w1)
    h_offset = math.log(h2 / h1)
    return (x_offset, y_offset, w_offset, h_offset)

def adjust_box_by_offset(candidate_box, offset):
    """根據(jù)偏移值調(diào)整候選區(qū)域"""
    x1, y1, w1, h1 = candidate_box
    x_offset, y_offset, w_offset, h_offset = offset
    w2 = math.exp(w_offset) * w1
    h2 = math.exp(h_offset) * h1
    x2 = x1 + w1 * x_offset - w2 // 2
    y2 = y1 + h1 * y_offset - h2 // 2
    x2 = min(IMAGE_SIZE[0]-1,  x2)
    y2 = min(IMAGE_SIZE[1]-1,  y2)
    w2 = min(IMAGE_SIZE[0]-x2, w2)
    h2 = min(IMAGE_SIZE[1]-y2, h2)
    return (x2, y2, w2, h2)

def prepare_save_batch(batch, image_tensors, result_tensors, result_isobject_masks, result_nonobject_masks):
    """準(zhǔn)備訓(xùn)練 - 保存單個(gè)批次的數(shù)據(jù)"""
    # 按索引值列表生成輸入和輸出 tensor 對(duì)象的函數(shù)
    def split_dataset(indices):
        indices_list = indices.tolist()
        image_tensors_splited = torch.stack([image_tensors[x] for x in indices_list])
        result_tensors_splited = torch.stack([result_tensors[x] for x in indices_list])
        result_isobject_masks_splited = [result_isobject_masks[x] for x in indices_list]
        result_nonobject_masks_splited = [result_nonobject_masks[x] for x in indices_list]
        return image_tensors_splited, (
            result_tensors_splited, result_isobject_masks_splited, result_nonobject_masks_splited)

    # 切分訓(xùn)練集 (80%)，驗(yàn)證集 (10%) 和測(cè)試集 (10%)
    random_indices = torch.randperm(len(image_tensors))
    training_indices = random_indices[:int(len(random_indices)*0.8)]
    validating_indices = random_indices[int(len(random_indices)*0.8):int(len(random_indices)*0.9):]
    testing_indices = random_indices[int(len(random_indices)*0.9):]
    training_set = split_dataset(training_indices)
    validating_set = split_dataset(validating_indices)
    testing_set = split_dataset(testing_indices)

    # 保存到硬盤
    save_tensor(training_set, f"data/training_set.{batch}.pt")
    save_tensor(validating_set, f"data/validating_set.{batch}.pt")
    save_tensor(testing_set, f"data/testing_set.{batch}.pt")
    print(f"batch {batch} saved")

def prepare():
    """準(zhǔn)備訓(xùn)練"""
    # 數(shù)據(jù)集轉(zhuǎn)換到 tensor 以后會(huì)保存在 data 文件夾下
    if not os.path.isdir("data"):
        os.makedirs("data")

    # 加載圖片和圖片對(duì)應(yīng)的區(qū)域與分類列表
    # { (路徑, 是否左右翻轉(zhuǎn)): [ 區(qū)域與分類, 區(qū)域與分類, .. ] }
    # 同一張圖片左右翻轉(zhuǎn)可以生成一個(gè)新的數(shù)據(jù)，讓數(shù)據(jù)量翻倍
    box_map = defaultdict(lambda: [])
    for filename in os.listdir(DATASET_1_IMAGE_DIR):
        # 從第一個(gè)數(shù)據(jù)集加載
        xml_path = os.path.join(DATASET_1_ANNOTATION_DIR, filename.split(".")[0] + ".xml")
        if not os.path.isfile(xml_path):
            continue
        tree = ET.ElementTree(file=xml_path)
        objects = tree.findall("object")
        path = os.path.join(DATASET_1_IMAGE_DIR, filename)
        for obj in objects:
            class_name = obj.find("name").text
            x1 = int(obj.find("bndbox/xmin").text)
            x2 = int(obj.find("bndbox/xmax").text)
            y1 = int(obj.find("bndbox/ymin").text)
            y2 = int(obj.find("bndbox/ymax").text)
            if class_name == "mask_weared_incorrect":
                # 佩戴口罩不正確的樣本數(shù)量太少 (只有 123)，模型無(wú)法學(xué)習(xí)，這里全合并到戴口罩的樣本
                class_name = "with_mask"
            box_map[(path, False)].append((x1, y1, x2-x1, y2-y1, CLASSES_MAPPING[class_name]))
            box_map[(path, True)].append((x1, y1, x2-x1, y2-y1, CLASSES_MAPPING[class_name]))
    df = pandas.read_csv(DATASET_2_BOX_CSV_PATH)
    for row in df.values:
        # 從第二個(gè)數(shù)據(jù)集加載，這個(gè)數(shù)據(jù)集只包含沒有帶口罩的圖片
        filename, width, height, x1, y1, x2, y2 = row[:7]
        path = os.path.join(DATASET_2_IMAGE_DIR, filename)
        box_map[(path, False)].append((x1, y1, x2-x1, y2-y1, CLASSES_MAPPING["without_mask"]))
        box_map[(path, True)].append((x1, y1, x2-x1, y2-y1, CLASSES_MAPPING["without_mask"]))
    # 打亂數(shù)據(jù)集 (因?yàn)榈诙€(gè)數(shù)據(jù)集只有不戴口罩的圖片)
    box_list = list(box_map.items())
    random.shuffle(box_list)
    print(f"found {len(box_list)} images")

    # 保存圖片和圖片對(duì)應(yīng)的分類與區(qū)域列表
    batch_size = 20
    batch = 0
    image_tensors = [] # 圖片列表
    result_tensors = [] # 圖片對(duì)應(yīng)的輸出結(jié)果列表，包含 [ 是否對(duì)象中心, 區(qū)域偏移, 各個(gè)分類的可能性 ]
    result_isobject_masks = [] # 各個(gè)圖片的包含對(duì)象的區(qū)域在 Anchors 中的索引
    result_nonobject_masks = [] # 各個(gè)圖片不包含對(duì)象的區(qū)域在 Anchors 中的索引 (重疊率低于閾值的區(qū)域)
    for (image_path, flip), original_boxes_labels in box_list:
        with Image.open(image_path) as img_original: # 加載原始圖片
            sw, sh = img_original.size # 原始圖片大小
            if flip:
                img = resize_image(img_original.transpose(Image.FLIP_LEFT_RIGHT)) # 翻轉(zhuǎn)然后縮放圖片
            else:
                img = resize_image(img_original) # 縮放圖片
            image_tensors.append(image_to_tensor(img)) # 添加圖片到列表
        # 生成輸出結(jié)果的 tensor
        result_tensor = torch.zeros((len(MyModel.Anchors), MyModel.AnchorOutputs), dtype=torch.float)
        result_tensor[:,5] = 1 # 默認(rèn)分類為 other
        result_tensors.append(result_tensor)
        # 包含對(duì)象的區(qū)域在 Anchors 中的索引
        result_isobject_mask = []
        result_isobject_masks.append(result_isobject_mask)
        # 不包含對(duì)象的區(qū)域在 Anchors 中的索引
        result_nonobject_mask = []
        result_nonobject_masks.append(result_nonobject_mask)
        # 根據(jù)真實(shí)區(qū)域定位所屬的錨點(diǎn)，然后設(shè)置輸出結(jié)果
        negative_mapping = [1] * len(MyModel.Anchors)
        for box_label in original_boxes_labels:
            x, y, w, h, label = box_label
            if flip: # 翻轉(zhuǎn)坐標(biāo)
                x = sw - x - w
            x, y, w, h = map_box_to_resized_image((x, y, w, h), sw, sh) # 縮放實(shí)際區(qū)域
            if w < 20 or h < 20:
                continue # 縮放后區(qū)域過(guò)小
            # 檢查計(jì)算是否有問(wèn)題
            # child_img = img.copy().crop((x, y, x+w, y+h))
            # child_img.save(f"{os.path.basename(image_path)}_{x}_{y}_{w}_{h}_{label}.png")
            # 定位所屬的錨點(diǎn)
            # 要求:
            # - 中心點(diǎn)落在錨點(diǎn)對(duì)應(yīng)的區(qū)域中
            # - 重疊率超過(guò)一定值
            x_center = x + w // 2
            y_center = y + h // 2
            matched_anchors = []
            for index, anchor in enumerate(MyModel.Anchors):
                ax, ay, aw, ah = anchor
                is_center = (x_center >= ax and x_center < ax + aw and
                    y_center >= ay and y_center < ay + ah)
                iou = calc_iou(anchor, (x, y, w, h))
                if is_center and iou > IOU_POSITIVE_THRESHOLD:
                    matched_anchors.append((index, anchor)) # 區(qū)域包含對(duì)象中心并且重疊率超過(guò)一定值
                    negative_mapping[index] = 0
                elif iou > IOU_NEGATIVE_THRESHOLD:
                    negative_mapping[index] = 0 # 區(qū)域與某個(gè)對(duì)象重疊率超過(guò)一定值，不應(yīng)該當(dāng)作負(fù)樣本
            for matched_index, matched_box in matched_anchors:
                # 計(jì)算區(qū)域偏移
                offset = calc_box_offset(matched_box, (x, y, w, h))
                # 修改輸出結(jié)果的 tensor
                result_tensor[matched_index] = torch.tensor((
                    1, # 是否對(duì)象中心
                    *offset, # 區(qū)域偏移
                    *[int(c == label) for c in range(len(CLASSES))] # 對(duì)應(yīng)分類
                ), dtype=torch.float)
                # 添加索引值
                # 注意如果兩個(gè)對(duì)象同時(shí)定位到相同的錨點(diǎn)，那么只有一個(gè)對(duì)象可以被識(shí)別，這里后面的對(duì)象會(huì)覆蓋前面的對(duì)象
                if matched_index not in result_isobject_mask:
                    result_isobject_mask.append(matched_index)
        # 沒有找到可識(shí)別的對(duì)象時(shí)跳過(guò)圖片
        if not result_isobject_mask:
            image_tensors.pop()
            result_tensors.pop()
            result_isobject_masks.pop()
            result_nonobject_masks.pop()
            continue
        # 添加不包含對(duì)象的區(qū)域在 Anchors 中的索引
        for index, value in enumerate(negative_mapping):
            if value:
                result_nonobject_mask.append(index)
        # 排序索引列表
        result_isobject_mask.sort()
        # 保存批次
        if len(image_tensors) >= batch_size:
            prepare_save_batch(batch, image_tensors, result_tensors,
                result_isobject_masks, result_nonobject_masks)
            image_tensors.clear()
            result_tensors.clear()
            result_isobject_masks.clear()
            result_nonobject_masks.clear()
            batch += 1
    # 保存剩余的批次
    if len(image_tensors) > 10:
        prepare_save_batch(batch, image_tensors, result_tensors,
            result_isobject_masks, result_nonobject_masks)

def train():
    """開始訓(xùn)練"""
    # 創(chuàng)建模型實(shí)例
    model = MyModel().to(device)

    # 創(chuàng)建多任務(wù)損失計(jì)算器
    loss_function = MyModel.loss_function

    # 創(chuàng)建參數(shù)調(diào)整器
    optimizer = torch.optim.Adam(model.parameters())

    # 記錄訓(xùn)練集和驗(yàn)證集的正確率變化
    training_obj_accuracy_history = []
    training_cls_accuracy_history = []
    validating_obj_accuracy_history = []
    validating_cls_accuracy_history = []

    # 記錄最高的驗(yàn)證集正確率
    validating_obj_accuracy_highest = -1
    validating_cls_accuracy_highest = -1
    validating_accuracy_highest = -1
    validating_accuracy_highest_epoch = 0

    # 讀取批次的工具函數(shù)
    def read_batches(base_path):
        for batch in itertools.count():
            path = f"{base_path}.{batch}.pt"
            if not os.path.isfile(path):
                break
            x, (y, mask1, mask2) = load_tensor(path)
            yield x.to(device), (y.to(device), mask1, mask2)

    # 計(jì)算正確率的工具函數(shù)
    calc_accuracy = MyModel.calc_accuracy

    # 開始訓(xùn)練過(guò)程
    for epoch in range(1, 10000):
        print(f"epoch: {epoch}")

        # 根據(jù)訓(xùn)練集訓(xùn)練并修改參數(shù)
        # 切換模型到訓(xùn)練模式，將會(huì)啟用自動(dòng)微分，批次正規(guī)化 (BatchNorm) 與 Dropout
        model.train()
        training_obj_accuracy_list = []
        training_cls_accuracy_list = []
        for batch_index, batch in enumerate(read_batches("data/training_set")):
            # 劃分輸入和輸出
            batch_x, batch_y = batch
            # 計(jì)算預(yù)測(cè)值
            predicted = model(batch_x)
            # 計(jì)算損失
            loss = loss_function(predicted, batch_y)
            # 從損失自動(dòng)微分求導(dǎo)函數(shù)值
            loss.backward()
            # 使用參數(shù)調(diào)整器調(diào)整參數(shù)
            optimizer.step()
            # 清空導(dǎo)函數(shù)值
            optimizer.zero_grad()
            # 記錄這一個(gè)批次的正確率，torch.no_grad 代表臨時(shí)禁用自動(dòng)微分功能
            with torch.no_grad():
                training_batch_obj_accuracy, training_batch_cls_accuracy = calc_accuracy(batch_y, predicted)
            # 輸出批次正確率
            training_obj_accuracy_list.append(training_batch_obj_accuracy)
            training_cls_accuracy_list.append(training_batch_cls_accuracy)
            print(f"epoch: {epoch}, batch: {batch_index}: " +
                f"batch obj accuracy: {training_batch_obj_accuracy}, cls accuracy: {training_batch_cls_accuracy}")
        training_obj_accuracy = sum(training_obj_accuracy_list) / len(training_obj_accuracy_list)
        training_cls_accuracy = sum(training_cls_accuracy_list) / len(training_cls_accuracy_list)
        training_obj_accuracy_history.append(training_obj_accuracy)
        training_cls_accuracy_history.append(training_cls_accuracy)
        print(f"training obj accuracy: {training_obj_accuracy}, cls accuracy: {training_cls_accuracy}")

        # 檢查驗(yàn)證集
        # 切換模型到驗(yàn)證模式，將會(huì)禁用自動(dòng)微分，批次正規(guī)化 (BatchNorm) 與 Dropout
        model.eval()
        validating_obj_accuracy_list = []
        validating_cls_accuracy_list = []
        for batch in read_batches("data/validating_set"):
            batch_x, batch_y = batch
            predicted = model(batch_x)
            validating_batch_obj_accuracy, validating_batch_cls_accuracy = calc_accuracy(batch_y, predicted)
            validating_obj_accuracy_list.append(validating_batch_obj_accuracy)
            validating_cls_accuracy_list.append(validating_batch_cls_accuracy)
            # 釋放 predicted 占用的顯存避免顯存不足的錯(cuò)誤
            predicted = None
        validating_obj_accuracy = sum(validating_obj_accuracy_list) / len(validating_obj_accuracy_list)
        validating_cls_accuracy = sum(validating_cls_accuracy_list) / len(validating_cls_accuracy_list)
        validating_obj_accuracy_history.append(validating_obj_accuracy)
        validating_cls_accuracy_history.append(validating_cls_accuracy)
        print(f"validating obj accuracy: {validating_obj_accuracy}, cls accuracy: {validating_cls_accuracy}")

        # 記錄最高的驗(yàn)證集正確率與當(dāng)時(shí)的模型狀態(tài)，判斷是否在 20 次訓(xùn)練后仍然沒有刷新記錄
        validating_accuracy = validating_obj_accuracy * validating_cls_accuracy
        if validating_accuracy > validating_accuracy_highest:
            validating_obj_accuracy_highest = validating_obj_accuracy
            validating_cls_accuracy_highest = validating_cls_accuracy
            validating_accuracy_highest = validating_accuracy
            validating_accuracy_highest_epoch = epoch
            save_tensor(model.state_dict(), "model.pt")
            print("highest validating accuracy updated")
        elif epoch - validating_accuracy_highest_epoch > 20:
            # 在 20 次訓(xùn)練后仍然沒有刷新記錄，結(jié)束訓(xùn)練
            print("stop training because highest validating accuracy not updated in 20 epoches")
            break

    # 使用達(dá)到最高正確率時(shí)的模型狀態(tài)
    print(f"highest obj validating accuracy: {validating_obj_accuracy_highest}",
        f"from epoch {validating_accuracy_highest_epoch}")
    print(f"highest cls validating accuracy: {validating_cls_accuracy_highest}",
        f"from epoch {validating_accuracy_highest_epoch}")
    model.load_state_dict(load_tensor("model.pt"))

    # 檢查測(cè)試集
    testing_obj_accuracy_list = []
    testing_cls_accuracy_list = []
    for batch in read_batches("data/testing_set"):
        batch_x, batch_y = batch
        predicted = model(batch_x)
        testing_batch_obj_accuracy, testing_batch_cls_accuracy = calc_accuracy(batch_y, predicted)
        testing_obj_accuracy_list.append(testing_batch_obj_accuracy)
        testing_cls_accuracy_list.append(testing_batch_cls_accuracy)
    testing_obj_accuracy = sum(testing_obj_accuracy_list) / len(testing_obj_accuracy_list)
    testing_cls_accuracy = sum(testing_cls_accuracy_list) / len(testing_cls_accuracy_list)
    print(f"testing obj accuracy: {testing_obj_accuracy}, cls accuracy: {testing_cls_accuracy}")

    # 顯示訓(xùn)練集和驗(yàn)證集的正確率變化
    pyplot.plot(training_obj_accuracy_history, label="training_obj_accuracy")
    pyplot.plot(training_cls_accuracy_history, label="training_cls_accuracy")
    pyplot.plot(validating_obj_accuracy_history, label="validating_obj_accuracy")
    pyplot.plot(validating_cls_accuracy_history, label="validating_cls_accuracy")
    pyplot.ylim(0, 1)
    pyplot.legend()
    pyplot.show()

def eval_model():
    """使用訓(xùn)練好的模型識(shí)別圖片"""
    # 創(chuàng)建模型實(shí)例，加載訓(xùn)練好的狀態(tài)，然后切換到驗(yàn)證模式
    model = MyModel().to(device)
    model.load_state_dict(load_tensor("model.pt"))
    model.eval()

    # 詢問(wèn)圖片路徑，并顯示所有可能是人臉的區(qū)域
    while True:
        try:
            image_path = input("Image path: ")
            if not image_path:
                continue
            # 構(gòu)建輸入
            with Image.open(image_path) as img_original: # 加載原始圖片
                sw, sh = img_original.size # 原始圖片大小
                img = resize_image(img_original) # 縮放圖片
                img_output = img_original.copy() # 復(fù)制圖片，用于后面添加標(biāo)記
                tensor_in = image_to_tensor(img)
            # 預(yù)測(cè)輸出
            predicted = model(tensor_in.unsqueeze(0).to(device))[0]
            final_result = MyModel.convert_predicted_result(predicted)
            # 標(biāo)記在圖片上
            draw = ImageDraw.Draw(img_output)
            for label, box, obj_score, cls_score in final_result:
                x, y, w, h = map_box_to_original_image(box, sw, sh)
                score = obj_score * cls_score
                color = "#00FF00" if CLASSES[label] == "with_mask" else "#FF0000"
                draw.rectangle((x, y, x+w, y+h), outline=color)
                draw.text((x, y-10), CLASSES[label], fill=color)
                draw.text((x, y+h), f"{score:.2f}", fill=color)
                print((x, y, w, h), CLASSES[label], obj_score, cls_score)
            img_output.save("img_output.png")
            print("saved to img_output.png")
            print()
        except Exception as e:
            print("error:", e)

def eval_video():
    """使用訓(xùn)練好的模型識(shí)別視頻"""
    # 創(chuàng)建模型實(shí)例，加載訓(xùn)練好的狀態(tài)，然后切換到驗(yàn)證模式
    model = MyModel().to(device)
    model.load_state_dict(load_tensor("model.pt"))
    model.eval()

    # 詢問(wèn)視頻路徑，給可能是人臉的區(qū)域添加標(biāo)記并保存新視頻
    import cv2
    font = ImageFont.truetype("FreeMonoBold.ttf", 20)
    while True:
        try:
            video_path = input("Video path: ")
            if not video_path:
                continue
            # 讀取輸入視頻
            video = cv2.VideoCapture(video_path)
            # 獲取每秒的幀數(shù)
            fps = int(video.get(cv2.CAP_PROP_FPS))
            # 獲取視頻長(zhǎng)寬
            size = (int(video.get(cv2.CAP_PROP_FRAME_WIDTH)), int(video.get(cv2.CAP_PROP_FRAME_HEIGHT)))
            # 創(chuàng)建輸出視頻
            video_output_path = os.path.join(
                os.path.dirname(video_path),
                os.path.splitext(os.path.basename(video_path))[0] + ".output.avi")
            result = cv2.VideoWriter(video_output_path, cv2.VideoWriter_fourcc(*"XVID"), fps, size)
            # 用于減少誤判的歷史結(jié)果
            history_results = deque(maxlen = fps // 2)
            # 逐幀處理
            count = 0
            while(True):
                ret, frame = video.read()
                if not ret:
                    break
                # opencv 使用的是 BGR, Pillow 使用的是 RGB, 需要轉(zhuǎn)換通道順序
                frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                # 構(gòu)建輸入
                img_original = Image.fromarray(frame_rgb) # 加載原始圖片
                sw, sh = img_original.size # 原始圖片大小
                img = resize_image(img_original) # 縮放圖片
                img_output = img_original.copy() # 復(fù)制圖片，用于后面添加標(biāo)記
                tensor_in = image_to_tensor(img)
                # 預(yù)測(cè)輸出
                predicted = model(tensor_in.unsqueeze(0).to(device))[0]
                cls_result = MyModel.convert_predicted_result(predicted)
                # 根據(jù)歷史結(jié)果減少誤判
                final_result = MyModel.fix_predicted_result_from_history(cls_result, history_results)
                # 標(biāo)記在圖片上
                draw = ImageDraw.Draw(img_output)
                for label, box, obj_score, cls_score in final_result:
                    x, y, w, h = map_box_to_original_image(box, sw, sh)
                    score = obj_score * cls_score
                    color = "#00FF00" if CLASSES[label] == "with_mask" else "#FF0000"
                    draw.rectangle((x, y, x+w, y+h), outline=color, width=3)
                    draw.text((x, y-20), CLASSES[label], fill=color, font=font)
                    draw.text((x, y+h), f"{score:.2f}", fill=color, font=font)
                # 寫入幀到輸出視頻
                frame_rgb_annotated = numpy.asarray(img_output)
                frame_bgr_annotated = cv2.cvtColor(frame_rgb_annotated, cv2.COLOR_RGB2BGR)
                result.write(frame_bgr_annotated)
                count += 1
                if count % fps == 0:
                    print(f"handled {count//fps}s")
            video.release()
            result.release()
            cv2.destroyAllWindows()
            print(f"saved to {video_output_path}")
            print()
        except Exception as e:
            raise
            print("error:", e)

def main():
    """主函數(shù)"""
    if len(sys.argv) < 2:
        print(f"Please run: {sys.argv[0]} prepare|train|eval")
        exit()

    # 給隨機(jī)數(shù)生成器分配一個(gè)初始值，使得每次運(yùn)行都可以生成相同的隨機(jī)數(shù)
    # 這是為了讓過(guò)程可重現(xiàn)，你也可以選擇不這樣做
    random.seed(0)
    torch.random.manual_seed(0)

    # 根據(jù)命令行參數(shù)選擇操作
    operation = sys.argv[1]
    if operation == "prepare":
        prepare()
    elif operation == "train":
        train()
    elif operation == "eval":
        eval_model()
    elif operation == "eval-video":
        eval_video()
    else:
        raise ValueError(f"Unsupported operation: {operation}")

if __name__ == "__main__":
    main()

?

2.5 檢測(cè)效果

檢測(cè)效果如下，可以看到效果還是很好的

實(shí)時(shí)檢測(cè)效果

3 口罩佩戴檢測(cè)算法評(píng)價(jià)指標(biāo)

目前， 口罩佩戴識(shí)別的研究已經(jīng)成為熱點(diǎn)， 算法也是越來(lái)越多， 所以要判斷算法的好壞還需要一些指標(biāo)作為參考 [36]。
大部分的評(píng)價(jià)指標(biāo)都來(lái)自對(duì)混淆矩陣（confusion matrix） 的計(jì)算。

其中 P（Positive） 表示預(yù)測(cè)值為正例。 N（Negative）表示預(yù)測(cè)值為反例。 T（True） 表示真實(shí)值與預(yù)測(cè)值相同。 預(yù)測(cè)值與真實(shí)值相反記為
F（False 。

TP 表示真實(shí)值是正樣本或者說(shuō)預(yù)測(cè)值為正樣本， 真實(shí)值和預(yù)測(cè)值相同。 TN 則為真實(shí)值為負(fù)樣本或者說(shuō)預(yù)測(cè)值為負(fù)樣本， 并且真實(shí)值和預(yù)測(cè)值相同。 FP
則為真實(shí)值為負(fù)樣本或者說(shuō)預(yù)測(cè)值為正樣本， 真實(shí)值和預(yù)測(cè)值不一樣。 FN 則為真實(shí)值為正樣本或者說(shuō)預(yù)測(cè)值為負(fù)樣本， 并且真實(shí)值和預(yù)測(cè)值不一樣。

3.1 準(zhǔn)確率（Accuracy）

表示的是模型在檢測(cè)時(shí)判斷為正樣本與所有樣本的比例，準(zhǔn)確率通常用來(lái)評(píng)價(jià)整個(gè)模型的準(zhǔn)確程度， 且不會(huì)包含太多信息， 因此不可能對(duì)模型的性能進(jìn)行綜合評(píng)價(jià)

3.2 精確率(Precision)和召回率(Recall)

精確率表示的是預(yù)測(cè)為正樣本中真正正樣本所占的比例， 它反映的是預(yù)測(cè)結(jié)果準(zhǔn)確與否。 精確率的計(jì)算公式如下所示：

召回率表明樣本中正例被預(yù)測(cè)正確的多少， 召回率主要看的是預(yù)測(cè)的結(jié)果是否全面。 召回率的計(jì)算公式如（2-3） 所示：

3.3 平均精度(Average precision AP)

表示每個(gè)類檢測(cè)好壞的結(jié)果。 召回率被當(dāng)作橫坐標(biāo)， 精確率被當(dāng)作縱坐標(biāo)， 表示 P-R 曲線下方的面積， AP 值越高表明模型的平均準(zhǔn)確率越高。 AP
的計(jì)算公式如下所示：

4 最后

?? 更多資料, 項(xiàng)目分享：

https://gitee.com/dancheng-senior/postgraduate

到了這里，關(guān)于競(jìng)賽項(xiàng)目 深度學(xué)習(xí)的口罩佩戴檢測(cè) - opencv 卷積神經(jīng)網(wǎng)絡(luò) 機(jī)器視覺 深度學(xué)習(xí)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！