实时面部识别是自动化安全部门仍然面临的问题。随着卷积神经网络的发展以及Region-CNN特别创新的方式, 已经证实, 利用我们当前的技术, 我们可以选择监督学习的选择, 例如FaceNet, YOLO, 以便在现实世界中快速, 实时地进行人脸识别。
为了训练监督模型，我们需要获取目标标签的数据集，这仍然是一项繁琐的工作。我们需要一个高效和自动化的解决方案，以最小的标记努力，用户干预的数据集生成。
拟议的解决方案–
介绍：我们提出了一个数据集生成管道, 该管道需要录像片段作为来源, 提取所有面部并将其聚类为代表不同人物的有限且准确的图像集。每个集合都可以轻松地由人为输入轻松标记。
技术细节:我们将使用opencv库从输入视频剪辑中每秒帧提取。1秒似乎适合于覆盖相关数据和处理的有限帧。
我们将使用face_recognition库(支持dlib)从框架中提取人脸，并对齐它们进行特征提取。
然后，我们将提取人类可观察的特征，并使用scikit-learn提供的DBSCAN聚类。
对于该解决方案，我们将裁剪所有的面，创建标签并将它们分组到文件夹中，以便用户将它们作为训练用例的数据集。
实施中的挑战：
对于更大的受众, 我们计划实施该解决方案以在CPU而非NVIDIA GPU中执行。使用NVIDIA GPU可以提高管道效率。
人脸嵌入提取的CPU实现非常慢(每幅图像30秒以上)。为了解决这个问题，我们用并行管道执行来实现它们(每幅图像约13秒)，然后合并它们的结果，以便进行进一步的聚类任务。我们引入了tqdm和PyPiper，用于进度更新和调整从输入视频中提取的帧的大小，用于流畅地执行管道。
Input: Footage.mp4
Output:必需的Python3模块：os, cv2, numpy, tensorflow, json, re, shutil, time, pickle, pyPiper, tqdm, imutils, face_recognition, dlib, warnings, sklearn
片段部分：
对于包含所有类定义的文件FaceClusteringLibrary.py的内容，下面是它们的代码片段和工作说明。
ResizeUtils的类实现提供了函数rescale_by_height和rescale_by_width。
" rescale_by_width "是一个以' image '和' target_width '作为输入的函数。它提高/降低图像的宽度尺寸以满足target_width。高度是自动计算的，因此长宽比保持不变。Rescale_by_height也是相同的，但它的目标不是宽度，而是高度。
'''
The ResizeUtils provides resizing function
to keep the aspect ratio intact
Credits: AndyP at StackOverflow'''
class ResizeUtils:
# Given a target height, adjust the image
# by calculating the width and resize
def rescale_by_height( self , image, target_height, method = cv2.INTER_LANCZOS4):
# Rescale `image` to `target_height`
# (preserving aspect ratio)
w = int ( round (target_height * image.shape[ 1 ] /image.shape[ 0 ]))
return (cv2.resize(image, (w, target_height), interpolation = method))
# Given a target width, adjust the image
# by calculating the height and resize
def rescale_by_width( self , image, target_width, method = cv2.INTER_LANCZOS4):
# Rescale `image` to `target_width`
# (preserving aspect ratio)
h = int ( round (target_width * image.shape[ 0 ] /image.shape[ 1 ]))
return (cv2.resize(image, (target_width, h), interpolation = method))以下是的定义FramesGenerator类。此类提供了通过顺序读取视频来提取jpg图像的功能。如果我们以输入视频文件为例, 它的帧速率约为30 fps。我们可以得出结论, 对于1秒钟的视频, 将有30张图像。即使是2分钟的视频, 要处理的图像数量也将是2 * 60 * 30 =3600。要处理的图像数量太多, 可能需要数小时才能完成流水线处理。
但是还有一个事实, 就是面孔和人在一秒钟之内可能不会改变。因此, 考虑一个2分钟的视频, 要在1秒钟内生成30张图像是麻烦且重复的。相反, 我们只能在1秒钟内拍摄1张图像。 " FramesGenerator"的实现每秒仅从视频剪辑中转储1张图像。
考虑到转储的图像受到face_recognition/dlib处理人脸提取时, 我们尝试将高度的阈值保持为不大于500, 宽度的上限为700。此限制是由"自动调整大小"函数施加的, 该函数会进一步调用rescale_by_heightorrescale_by_width如果达到极限但仍保留宽高比, 则可以减小图像的尺寸。
进入以下代码段, 自动调整大小函数尝试对给定图像的尺寸施加限制。如果宽度大于700, 我们将其缩小以保持宽度700并保持长宽比。此处设置的另一个限制是, 高度不能大于500。
# The FramesGenerator extracts image
# frames from the given video file
# The image frames are resized for
# face_recognition /dlib processing
class FramesGenerator:
def __init__( self , VideoFootageSource):
self .VideoFootageSource = VideoFootageSource
# Resize the given input to fit in a specified
# size for face embeddings extraction
def AutoResize( self , frame):
resizeUtils = ResizeUtils()
height, width, _ = frame.shape
if height> 500 :
frame = resizeUtils.rescale_by_height(frame, 500 )
self .AutoResize(frame)
if width> 700 :
frame = resizeUtils.rescale_by_width(frame, 700 )
self .AutoResize(frame)
return frame以下是的摘要生成框架功能。它查询fps以决定在多少帧中可以转储1张图像。我们清除输出目录并开始遍历所有帧。转储任何图像之前, 如果图像达到了指定的限制, 我们将调整图像的大小自动调整大小功能。
# Extract 1 frame from each second from video footage
# and save the frames to a specific folder
def GenerateFrames( self , OutputDirectoryName):
cap = cv2.VideoCapture( self .VideoFootageSource)
_, frame = cap.read()
fps = cap.get(cv2.CAP_PROP_FPS)
TotalFrames = cap.get(cv2.CAP_PROP_FRAME_COUNT)
print ( "[INFO] Total Frames " , TotalFrames, " @ " , fps, " fps" )
print ( "[INFO] Calculating number of frames per second" )
CurrentDirectory = os.path.curdir
OutputDirectoryPath = os.path.join(
CurrentDirectory, OutputDirectoryName)
if os.path.exists(OutputDirectoryPath):
shutil.rmtree(OutputDirectoryPath)
time.sleep( 0.5 )
os.mkdir(OutputDirectoryPath)
CurrentFrame = 1
fpsCounter = 0
FrameWrittenCount = 1
while CurrentFrame <TotalFrames:
_, frame = cap.read()
if (frame is None ):
continue
if fpsCounter> fps:
fpsCounter = 0
frame = self .AutoResize(frame)
filename = "frame_" + str (FrameWrittenCount) + ".jpg"
cv2.imwrite(os.path.join(
OutputDirectoryPath, filename), frame)
FrameWrittenCount + = 1
fpsCounter + = 1
CurrentFrame + = 1
print ( '[INFO] Frames extracted' )以下是的摘要FramesProvider类。它继承了"节点", 可用于构建图像处理管道。我们实现"设置"和"运行"功能。 " setup"函数中定义的任何参数都可以具有参数, 构造函数将在创建对象时将其作为参数。在这里, 我们可以通过sourcePath的参数FramesProvider目的。 "设置"功能仅运行一次。 "运行"功能运行并通过调用不断发出数据发射起到处理管道直到关函数被调用。
在这里, 在"设置"中, 我们接受sourcePath作为参数, 并遍历给定框架目录中的所有文件。无论文件的扩展名是.jpg(将由课程生成框架生成器), 我们将其添加到" filesList"列表中。
在通话期间运行函数, " filesList"中的所有jpg图像路径都打包有指定唯一" id"和" imagePath"作为对象的属性, 并发送到管道进行处理。
# Following are nodes for pipeline constructions.
# It will create and asynchronously execute threads
# for reading images, extracting facial features and
# storing them independently in different threads
# Keep emitting the filenames into
# the pipeline for processing
class FramesProvider(Node):
def setup( self , sourcePath):
self .sourcePath = sourcePath
self .filesList = []
for item in os.listdir( self .sourcePath):
_, fileExt = os.path.splitext(item)
if fileExt = = '.jpg' :
self .filesList.append(os.path.join(item))
self .TotalFilesCount = self .size = len ( self .filesList)
self .ProcessedFilesCount = self .pos = 0
# Emit each filename in the pipeline for parallel processing
def run( self , data):
if self .ProcessedFilesCount <self .TotalFilesCount:
self .emit({ 'id' : self .ProcessedFilesCount, 'imagePath' : os.path.join( self .sourcePath, self .filesList[ self .ProcessedFilesCount])})
self .ProcessedFilesCount + = 1
self .pos = self .ProcessedFilesCount
else :
self .close()以下是"
人脸编码器
继承了"节点", 并且可以在图像处理管道中推送。在"设置"功能中, 我们接受" face_recognition/dlib"面部识别器调用的" detection_method"值。它可以具有基于" cnn"的检测器或基于" hog"的检测器。
"运行"功能将输入的数据解压缩为" id"和" imagePath"。
随后, 它从" imagePath"中读取图像, 运行在" face_recognition/dlib"库中定义的" face_location", 以裁剪出对齐的面部图像, 这是我们感兴趣的区域。对齐的面部图像是矩形裁剪的图像, 其眼睛和嘴唇与图像中的特定位置对齐(注意：实现可能与其他库(例如opencv)不同)。
此外, 我们调用" face_recognition/dlib"中定义的" face_encodings"函数, 以从每个框中提取面部嵌入。嵌入浮点值可以帮助你在对齐的面部图像中找到特征的确切位置。
我们将变量" d"定义为一组盒子和各自的嵌入。现在, 我们将" id"和嵌入数组打包为对象中的" encoding"键, 并将其发送到图像处理管道。
# Encode the face embedding, reference path
# and location and emit to pipeline
class FaceEncoder(Node):
def setup( self , detection_method = 'cnn' ):
self .detection_method = detection_method
# detection_method can be cnn or hog
def run( self , data):
id = data[ 'id' ]
imagePath = data[ 'imagePath' ]
image = cv2.imread(imagePath)
rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
boxes = face_recognition.face_locations(
rgb, model = self .detection_method)
encodings = face_recognition.face_encodings(rgb, boxes)
d = [{ "imagePath" : imagePath, "loc" : box, "encoding" : enc}
for (box, enc) in zip (boxes, encodings)]
self .emit({ 'id' : id , 'encodings' : d})以下是一个实现
数据存储管理器
它再次继承自"节点", 并且可以插入图像处理管道。该类的目的是将" encodings"数组作为pickle文件转储, 并使用" id"参数唯一地命名pickle文件。我们希望管道运行多线程。
为了利用多线程来提高性能, 我们需要适当地分离异步任务, 并尝试避免任何同步需求。因此, 为了获得最佳性能, 我们独立让管道中的线程将数据写出到单独的单独文件中, 而不会干扰任何其他线程操作。
如果你正在考虑在不使用多线程的情况下在二手开发硬件中节省了多少时间, 则平均嵌入提取时间约为30秒。经过多线程管道之后(具有4个线程), 它减少到约10秒, 但代价是要占用大量CPU。
由于线程大约需要10秒钟, 因此不会发生频繁的磁盘写操作, 并且不会影响我们的多线程性能。
另一种情况, 如果你在考虑为什么使用pickle而不是JSON？事实是, JSON是泡菜的更好替代品。泡菜对于数据存储和通讯非常不安全。可以对咸菜进行恶意修改, 以将可执行代码嵌入Python中。 JSON文件易于阅读, 并且编码和解码速度更快。 pickle唯一擅长的是将python对象和内容无错误地转储到二进制文件中。
由于我们不打算存储和分发pickle文件, 并且为了实现无错误执行, 我们正在使用pickle。否则, 强烈建议你使用JSON和其他替代方法。
# Recieve the face embeddings for clustering and
# id for naming the distinct filename
class DatastoreManager(Node):
def setup( self , encodingsOutputPath):
self .encodingsOutputPath = encodingsOutputPath
def run( self , data):
encodings = data[ 'encodings' ]
id = data[ 'id' ]
with open (os.path.join( self .encodingsOutputPath, 'encodings_' + str ( id ) + '.pickle' ), 'wb' ) as f:
f.write(pickle.dumps(encodings))以下是实施类PickleListCollator。它旨在读取多个pickle文件中的对象数组, 合并为一个数组, 然后将合并的数组转储为单个pickle文件。
在这里, 只有一个功能GeneratePickle接受outputFilepath它指定将包含合并数组的单个输出pickle文件。
# PicklesListCollator takes multiple pickle
# files as input and merges them together
# It is made specifically to support use-case
# of merging distinct pickle files into one
class PicklesListCollator:
def __init__( self , picklesInputDirectory):
self .picklesInputDirectory = picklesInputDirectory
# Here we will list down all the pickles
# files generated from multiple threads, # read the list of results append them to a
# common list and create another pickle
# with combined list as content
def GeneratePickle( self , outputFilepath):
datastore = []
ListOfPickleFiles = []
for item in os.listdir( self .picklesInputDirectory):
_, fileExt = os.path.splitext(item)
if fileExt = = '.pickle' :
ListOfPickleFiles.append(os.path.join(
self .picklesInputDirectory, item))
for picklePath in ListOfPickleFiles:
with open (picklePath, "rb" ) as f:
data = pickle.loads(f.read())
datastore.extend(data)
with open (outputFilepath, 'wb' ) as f:
f.write(pickle.dumps(datastore))以下是执行
FaceClusterUtility类。定义了一个构造函数, 该构造函数将带有值的" EncodingFilePath"作为合并的泡菜文件的路径。我们从pickle文件中读取数组, 并尝试使用" scikit"库中的" DBSCAN"实现对它们进行集群。与k均值不同, DBSCAN扫描不需要簇数。簇数取决于阈值参数, 并将自动计算。
DBSCAN实现在" scikit"中提供, 并且也接受用于计算的线程数。
在这里, 我们有一个函数" Cluster", 该函数将被调用以从pickle文件中读取数组数据, 运行" DBSCAN", 将唯一的簇打印为唯一的面并返回标签。标签是代表类别的唯一值, 可用于识别数组中存在的面部的类别。 (数组内容来自pickle文件)。
# Face clustering functionality
class FaceClusterUtility:
def __init__( self , EncodingFilePath):
self .EncodingFilePath = EncodingFilePath
# Credits: Arian's pyimagesearch for the clustering code
# Here we are using the sklearn.DBSCAN functioanlity
# cluster all the facial embeddings to get clusters
# representing distinct people
def Cluster( self ):
InputEncodingFile = self .EncodingFilePath
if not (os.path.isfile(InputEncodingFile) and
os.access(InputEncodingFile, os.R_OK)):
print ( 'The input encoding file, ' +
str (InputEncodingFile) +
' does not exists or unreadable' )
exit()
NumberOfParallelJobs = - 1
# load the serialized face encodings
# + bounding box locations from disk, # then extract the set of encodings to
# so we can cluster on them
print ( "[INFO] Loading encodings" )
data = pickle.loads( open (InputEncodingFile, "rb" ).read())
data = np.array(data)
encodings = [d[ "encoding" ] for d in data]
# cluster the embeddings
print ( "[INFO] Clustering" )
clt = DBSCAN(eps = 0.5 , metric = "euclidean" , n_jobs = NumberOfParallelJobs)
clt.fit(encodings)
# determine the total number of
# unique faces found in the dataset
labelIDs = np.unique(clt.labels_)
numUniqueFaces = len (np.where(labelIDs> - 1 )[ 0 ])
print ( "[INFO] # unique faces: {}" . format (numUniqueFaces))
return clt.labels_下面是继承自“tqdm”的TqdmUpdate类的实现。tqdm是一个Python库，用于在控制台界面中可视化进度条。
变量" n"和"总计"由" tqdm"识别。这两个变量的值用于计算进度。
当在管道框架“PyPiper”中绑定更新事件时，“update”函数中的参数“done”和“total_size”会被提供值。super().refresh()调用“tqdm”类中的“refresh”函数的实现，该函数在控制台中可视化和更新进度条。
# Inherit class tqdm for visualization of progress
class TqdmUpdate(tqdm):
# This function will be passed as progress
# callback function. Setting the predefined
# variables for auto-updates in visualization
def update( self , done, total_size = None ):
if total_size is not None :
self .total = total_size
self .n = done
super ().refresh()以下是执行FaceImageGenerator类。此类提供的功能是根据聚类后产生的标签生成蒙太奇, 裁剪的人像图像和注释, 以备将来训练之用(例如Darknet YOLO)。
构造函数期望编码文件路径作为合并的pickle文件路径。它将用于加载所有面部编码。现在, 我们对生成图像的" imagePath"和面部坐标感兴趣。
对" GenerateImages"的调用完成了预期的工作。我们从合并的pickle文件中加载数组。我们对标签应用唯一的操作, 并遍历整个标签。在标签迭代中, 对于每个唯一标签, 我们列出了具有相同当前标签的所有数组索引。
再次迭代这些数组索引以处理每个面。
对于人脸处理, 我们使用索引来获取图像文件的路径和人脸坐标。
从图像文件的路径加载图像文件。人脸的坐标被扩展为人像形状(并且我们还确保其扩展不会超过图像的尺寸), 并将其裁剪并转储为人像图像。
我们再次从原始坐标开始, 并进行一些扩展以创建注释, 以用于将来受监督的训练选项, 从而提高识别能力。
对于注释, 我们只是为" Darknet YOLO"设计了它, 但是它也可以适用于任何其他框架。最后, 我们构建一个蒙太奇并将其写出到图像文件中。
class FaceImageGenerator:
def __init__( self , EncodingFilePath):
self .EncodingFilePath = EncodingFilePath
# Here we are creating montages for
# first 25 faces for each distinct face.
# We will also generate images for all
# the distinct faces by using the labels
# from clusters and image url from the
# encodings pickle file.
# The face bounding box is increased a
# little more for training purposes and
# we also created the exact annotation for
# each face image (similar to darknet YOLO)
# to easily adapt the annotation for future
# use in supervised training
def GenerateImages( self , labels, OutputFolderName = "ClusteredFaces" , MontageOutputFolder = "Montage" ):
output_directory = os.getcwd()
OutputFolder = os.path.join(output_directory, OutputFolderName)
if not os.path.exists(OutputFolder):
os.makedirs(OutputFolder)
else :
shutil.rmtree(OutputFolder)
time.sleep( 0.5 )
os.makedirs(OutputFolder)
MontageFolderPath = os.path.join(OutputFolder, MontageOutputFolder)
os.makedirs(MontageFolderPath)
data = pickle.loads( open ( self .EncodingFilePath, "rb" ).read())
data = np.array(data)
labelIDs = np.unique(labels)
# loop over the unique face integers
for labelID in labelIDs:
# find all indexes into the `data` array
# that belong to the current label ID, then
# randomly sample a maximum of 25 indexes
# from the set
print ( "[INFO] faces for face ID: {}" . format (labelID))
FaceFolder = os.path.join(OutputFolder, "Face_" + str (labelID))
os.makedirs(FaceFolder)
idxs = np.where(labels = = labelID)[ 0 ]
# initialize the list of faces to
# include in the montage
portraits = []
# loop over the sampled indexes
counter = 1
for i in idxs:
# load the input image and extract the face ROI
image = cv2.imread(data[i][ "imagePath" ])
(o_top, o_right, o_bottom, o_left) = data[i][ "loc" ]
height, width, channel = image.shape
widthMargin = 100
heightMargin = 150
top = o_top - heightMargin
if top <0 : top = 0
bottom = o_bottom + heightMargin
if bottom> height: bottom = height
left = o_left - widthMargin
if left <0 : left = 0
right = o_right + widthMargin
if right> width: right = width
portrait = image[top:bottom, left:right]
if len (portraits) <25 :
portraits.append(portrait)
resizeUtils = ResizeUtils()
portrait = resizeUtils.rescale_by_width(portrait, 400 )
FaceFilename = "face_" + str (counter) + ".jpg"
FaceImagePath = os.path.join(FaceFolder, FaceFilename)
cv2.imwrite(FaceImagePath, portrait)
widthMargin = 20
heightMargin = 20
top = o_top - heightMargin
if top <0 : top = 0
bottom = o_bottom + heightMargin
if bottom> height: bottom = height
left = o_left - widthMargin
if left <0 : left = 0
right = o_right + widthMargin
if right> width:
right = width
AnnotationFilename = "face_" + str (counter) + ".txt"
AnnotationFilePath = os.path.join(FaceFolder, AnnotationFilename)
f = open (AnnotationFilePath, 'w' )
f.write( str (labelID) + ' ' +
str (left) + ' ' + str (top) + ' ' +
str (right) + ' ' + str (bottom) + "\n" )
f.close()
counter + = 1
montage = build_montages(portraits, ( 96 , 120 ), ( 5 , 5 ))[ 0 ]
MontageFilenamePath = os.path.join(
MontageFolderPath, "Face_" + str (labelID) + ".jpg" )
cv2.imwrite(MontageFilenamePath, montage)将文件另存为FaceClusteringLibrary.py, 其中将包含所有类定义。
以下是文件驱动程序, 它调用功能来创建管道。
# importing all classes from above Python file
from FaceClusteringLibrary import *
if __name__ = = "__main__" :
# Generate the frames from given video footage
framesGenerator = FramesGenerator( "Footage.mp4" )
framesGenerator.GenerateFrames( "Frames" )
# Design and run the face clustering pipeline
CurrentPath = os.getcwd()
FramesDirectory = "Frames"
FramesDirectoryPath = os.path.join(CurrentPath, FramesDirectory)
EncodingsFolder = "Encodings"
EncodingsFolderPath = os.path.join(CurrentPath, EncodingsFolder)
if os.path.exists(EncodingsFolderPath):
shutil.rmtree(EncodingsFolderPath, ignore_errors = True )
time.sleep( 0.5 )
os.makedirs(EncodingsFolderPath)
pipeline = Pipeline(
FramesProvider( "Files source" , sourcePath = FramesDirectoryPath) |
FaceEncoder( "Encode faces" ) |
DatastoreManager( "Store encoding" , encodingsOutputPath = EncodingsFolderPath), n_threads = 3 , quiet = True )
pbar = TqdmUpdate()
pipeline.run(update_callback = pbar.update)
print ()
print ( '[INFO] Encodings extracted' )
# Merge all the encodings pickle files into one
CurrentPath = os.getcwd()
EncodingsInputDirectory = "Encodings"
EncodingsInputDirectoryPath = os.path.join(
CurrentPath, EncodingsInputDirectory)
OutputEncodingPickleFilename = "encodings.pickle"
if os.path.exists(OutputEncodingPickleFilename):
os.remove(OutputEncodingPickleFilename)
picklesListCollator = PicklesListCollator(
EncodingsInputDirectoryPath)
picklesListCollator.GeneratePickle(
OutputEncodingPickleFilename)
# To manage any delay in file writing
time.sleep( 0.5 )
# Start clustering process and generate
# output images with annotations
EncodingPickleFilePath = "encodings.pickle"
faceClusterUtility = FaceClusterUtility(EncodingPickleFilePath)
faceImageGenerator = FaceImageGenerator(EncodingPickleFilePath)
labelIDs = faceClusterUtility.Cluster()
faceImageGenerator.GenerateImages(
labelIDs, "ClusteredFaces" , "Montage" )蒙太奇输出：
故障排除 -
问题1：
提取面部嵌入时, 整个PC冻结。
解：
解决方案是从输入视频剪辑中提取帧时减小帧调整大小功能中的值。请记住, 将值减小太多将导致不正确的脸部聚类。除了调整框架的大小外, 我们可以引入一些正面人脸检测并裁剪正面以提高准确性。
问题2：
运行管道时, PC变慢。
解：
将最大程度地使用CPU。为了限制使用量, 你可以减少在管道构造函数中指定的线程数。
问题3：
输出聚类太不准确了。
解：
出现这种情况的唯一原因可能是从输入视频剪辑中提取的帧将具有非常小的分辨率的人脸, 或者帧数很少(大约7-8)。请获得一个带有明亮清晰面部表情的视频剪辑；对于后一种情况, 请获得一个2分钟的视频或带有源代码的mod, 用于提取视频帧。
请参阅Github链接以获取完整的代码和使用的其他文件：
https://github.com/cppxaxa/FaceRecognitionPipeline_GeeksForGeeks
参考文献：
1.阿德里安(Adrian)关于脸部聚类的博客文章
2. PyPiper指南
3. OpenCV手册
4. StackOverflow
首先, 你的面试准备可通过以下方式增强你的数据结构概念：Python DS课程。
![从字法上最小长度N的排列，使得对于正好为K个索引，a[i] a[i]+1](https://www.lsbin.com/wp-content/themes/begin%20lts/img/loading.png)
