人臉識別與檢測（保姆級教程--附帶源碼）

人臉識別與檢測（保姆級教程–附帶源碼）

項目背景

因項目需要招聘了一些日結工人，因此需要對工地現場的工人進行考勤管理，但工地只有海康攝像頭沒有專業考勤設備，因此需要基于視頻流開發人臉識別與檢測功能；因之前沒有做個做一塊所以再技術選型上面要具有，易用性、跨平臺、不用訓練，等等。因此我采用了face_recognition；

技術介紹

face_recognition 是一個流行的Python庫，專門用于執行面部識別任務。它是基于dlib的深度學習模型構建的，特別是基于深度卷積神經網絡的模型。

一、人臉錄入

要進行人臉識別與檢測，首先就是要將圖像中人像的面部編碼提取出來，并保存到數據庫中。

import os
import face_recognition
import mysql.connector
from mysql.connector import Error
import cv2
import numpy as np
import logginglogging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')model_path = "../ckpt/face/res10_300x300_ssd_iter_140000_fp16.caffemodel"
det = "../ckpt/face\deploy.prototxt"# 加載預訓練的人臉檢測模型
face_detector = cv2.dnn.readNetFromCaffe(det, model_path)# 數據庫配置
db_config = {'user': 'root','password': 'cqccc132645!!','host': '10.10.12.72','database': 'ai_platform','buffered': True
}def connect_to_database():try:connection = mysql.connector.connect(**db_config)if connection.is_connected():print("成功連接到MySQL數據庫")return connectionexcept Error as e:print(f"連接數據庫時出錯: {e}")return Nonedef check_table_structure(connection):try:cursor = connection.cursor()check_table_query = """SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = 'ai_platform' AND table_name = 'faces' AND column_name IN ('id', 'name', 'embedding', 'image')"""cursor.execute(check_table_query)count = cursor.fetchone()[0]if count == 4:print("faces表結構正確")else:print("faces表結構不正確，請檢查數據庫")return Falsereturn Trueexcept Error as e:print(f"檢查表結構時出錯: {e}")return Falsedef insert_face(connection, name, face_embedding, face_image):try:cursor = connection.cursor()insert_query = "INSERT INTO faces (name, embedding, image) VALUES (%s, %s, %s)"cursor.execute(insert_query, (name, face_embedding.tobytes(), face_image))connection.commit()print(f"成功插入 {name} 的人臉數據")except Error as e:print(f"插入人臉數據時出錯: {e}")def crop_face(image, face_location):top, right, bottom, left = face_locationreturn image[top:bottom, left:right]def load_known_faces(connection):known_face_encodings = []known_face_names = []try:cursor = connection.cursor()query = "SELECT name, embedding FROM faces"cursor.execute(query)for (name, face_encoding) in cursor:known_face_encodings.append(np.frombuffer(face_encoding, dtype=np.float64))known_face_names.append(name)logging.info(f"從數據庫加載了 {len(known_face_names)} 個人臉")except Error as e:logging.error(f"加載人臉數據時出錯: {e}")finally:cursor.close()return known_face_encodings, known_face_namesdef register_faces(image_folder):connection = connect_to_database()if connection is None:returnif not check_table_structure(connection):connection.close()returnknown_face_encodings, known_face_names = load_known_faces(connection)for filename in os.listdir(image_folder):if filename.endswith((".jpg", ".jpeg", ".png")):image_path = os.path.join(image_folder, filename)name = os.path.splitext(filename)[0]  # 使用文件名作為人名# 使用 face_recognition 庫讀取圖像，它可以處理中文路徑image = face_recognition.load_image_file(image_path)rgb_image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)face_locations = face_recognition.face_locations(rgb_image)face_encodings = face_recognition.face_encodings(rgb_image, face_locations)if face_encodings:face_encoding = face_encodings[0]  # 假設每張圖片只有一張臉face_location = face_locations[0]# 裁剪人臉區域face_image = crop_face(image, face_location)# 將圖像編碼為JPEG格式_, buffer = cv2.imencode('.jpg', face_image)face_image_bytes = buffer.tobytes()face_distances = face_recognition.face_distance(known_face_encodings, face_encoding)best_match_index = np.argmin(face_distances)if face_distances[best_match_index] < 0.5:name2 = known_face_names[best_match_index]print(f"{name} 與 {known_face_names[best_match_index]} 的距離為 {face_distances[best_match_index]}")print(f"{name} 已存在于數據庫中,{name2}")else:insert_face(connection, name, face_encoding, face_image_bytes)else:print(f"在 {filename} 中沒有檢測到人臉")connection.close()if __name__ == "__main__":image_folder = "../face/train/ces/"  # 替換為實際的圖片文件夾路徑register_faces(image_folder)

二、人臉檢測

除了前面講到的使用了face_recognition，我還使用到了res10_300x300_ssd_iter_140000_fp16.caffemodel 深度學習模型文件，用于面部檢測，它是一個使用ResNet作為基礎網絡，輸入尺寸為300x300像素，經過140,000次迭代訓練，并以半精度浮點數格式存儲的SSD對象檢測模型，deploy.prototxt配合使用。

整個流程就是，先加載模型文件，再數據庫中取出之前錄入的人臉編碼，隨后是從視頻流中取出視頻幀（每隔10幀進行一次處理）進行檢測與數據庫中人類編碼進行對比，如果是數據庫中的人臉就將名字打印出來并進行標識，反之著直接跳過，命名為Unknown。其次為了加速檢測效率，因此我采用了多線程的每一個視頻流就添加一個線程。

import cv2
import face_recognition
import mysql.connector
import numpy as np
from mysql.connector import Error
from threading import Thread
import time
import logging
from PIL import Image,ImageDraw,ImageFont# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
# 數據庫配置
db_config = {'user': 'root','password': 'cqccc132645!!','host': '10.10.12.72','database': 'ai_platform','buffered': True
}# model_path = "D:\work\project\py\face_detection\ckpt\face\res10_300x300_ssd_iter_140000_fp16.caffemodel"
model_path = "../ckpt/face/res10_300x300_ssd_iter_140000_fp16.caffemodel"
det = "../ckpt/face\deploy.prototxt"# 加載預訓練的人臉檢測模型
face_detector = cv2.dnn.readNetFromCaffe(det, model_path)def connect_to_database():try:connection = mysql.connector.connect(**db_config)if connection.is_connected():logging.info("成功連接到MySQL數據庫")return connectionexcept Error as e:logging.error(f"連接數據庫時出錯: {e}")return Nonedef load_known_faces(connection):known_face_encodings = []known_face_names = []try:cursor = connection.cursor()query = "SELECT name, embedding FROM faces"cursor.execute(query)for (name, face_encoding) in cursor:known_face_encodings.append(np.frombuffer(face_encoding, dtype=np.float64))known_face_names.append(name)logging.info(f"從數據庫加載了 {len(known_face_names)} 個人臉")except Error as e:logging.error(f"加載人臉數據時出錯: {e}")finally:cursor.close()return known_face_encodings, known_face_namesdef detect_faces(frame, face_detector):start_time = time.time()height, width = frame.shape[:2]blob = cv2.dnn.blobFromImage(cv2.resize(frame, (300, 300)), 1.0, (300, 300), (104.0, 177.0, 123.0))face_detector.setInput(blob)detections = face_detector.forward()face_locations = []for i in range(detections.shape[2]):confidence = detections[0, 0, i, 2]if confidence > 0.8:box = detections[0, 0, i, 3:7] * np.array([width, height, width, height])(startX, startY, endX, endY) = box.astype("int")face_locations.append((startY, endX, endY, startX))end_time = time.time()logging.debug(f"Face detection time: {end_time - start_time:.4f} seconds")return face_locationsdef process_frames(frames_buffer, known_face_encodings, known_face_names, camera_id, face_detector):processed_frames = []faces_detected = Falsefor frame in frames_buffer:face_locations = detect_faces(frame, face_detector)if face_locations:face_encodings = face_recognition.face_encodings(frame, face_locations)for face_encoding in face_encodings:face_distances = face_recognition.face_distance(known_face_encodings, face_encoding)best_match_index = np.argmin(face_distances)if face_distances[best_match_index] < 0.6:name = known_face_names[best_match_index]else:name = "Unknown"logging.info(f"攝像頭 {camera_id} 檢測到 {name}")faces_detected = Truefor (top, right, bottom, left) in face_locations:cv2.rectangle(frame, (left, top), (right, bottom), (0, 0, 255), 2)cv2.rectangle(frame, (left, bottom - 35), (right, bottom), (0, 0, 255), cv2.FILLED)# 顯示中文# frame = cv2_img_add_text(frame, name, left + 6, bottom - 30, text_color=(255, 255, 255),#                          text_size=20)# font = cv2.FONT_HERSHEY_DUPLEX# cv2.putText(frame, name, (left + 6, bottom - 6), font, 0.5, (255, 255, 255), 1)processed_frames.append(frame)return processed_frames, faces_detecteddef video_stream_thread(stream_url, known_face_encodings, known_face_names):camera_id = stream_url.split('@')[1].split('/')[0]# camera_id = stream_url.split('_')[1]  # 提取IP地址作為攝像頭IDcap = cv2.VideoCapture(stream_url)if not cap.isOpened():logging.error(f"無法打開視頻流: {camera_id}")returnframe_count = 0start_time = time.time()faces_detected = Falseframes_buffer = []while True:ret, frame = cap.read()if not ret:logging.warning(f"無法從攝像頭 {camera_id} 獲取幀，嘗試重新連接...")cap.release()time.sleep(5)cap = cv2.VideoCapture(stream_url)if not cap.isOpened():logging.error(f"無法重新連接到攝像頭: {camera_id}")breakcontinueframe_count += 1frames_buffer.append(frame)if len(frames_buffer) == 10:  # 每10幀處理一次# processed_frames, faces_detected = process_frames(frames_buffer, known_face_encodings, known_face_names, camera_id)processed_frames, faces_detected = process_frames(frames_buffer, known_face_encodings, known_face_names,camera_id, face_detector)frames_buffer = []#     # 顯示最后一幀（如果需要的話）#     cv2.imshow(f'Stream: {camera_id}', processed_frames[-1])## if cv2.waitKey(1) & 0xFF == ord('q'):#     break# 每100幀計算和顯示FPSif frame_count % 100 == 0:end_time = time.time()fps = frame_count / (end_time - start_time)if faces_detected:logging.info(f"攝像頭 {camera_id}, FPS: {fps:.2f}, 檢測到人臉")else:logging.info(f"攝像頭 {camera_id}, FPS: {fps:.2f}, 未檢測到人臉")cap.release()cv2.destroyAllWindows()def cv2AddChineseText(img, text, position, textColor, textSize):if (isinstance(img, np.ndarray)):  # 判斷是否OpenCV圖片類型img = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))draw = ImageDraw.Draw(img)fontStyle = ImageFont.truetype("C:/WINDOWS/FONTS/ARLRDBD.TTF", textSize, encoding="utf-8")# 繪制文本draw.text(position, text, textColor, font=fontStyle)# 轉換回OpenCV格式return cv2.cvtColor(np.asarray(img), cv2.COLOR_RGB2BGR)def cv2_img_add_text(img, text, left, top, text_color=(255, 255, 255), text_size=20):if isinstance(img, np.ndarray):  # 判斷是否OpenCV圖片類型img = Image.fromarray(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))draw = ImageDraw.Draw(img)font_style = ImageFont.truetype(r"C:\ProgramData\kingsoft\office6\muifont\GBK\FZXBSK.TTF", text_size, encoding="utf-8")  # 請替換為你的中文字體路徑draw.text((left, top), text, text_color, font=font_style)return cv2.cvtColor(np.asarray(img), cv2.COLOR_RGB2BGR)def main():# 檢查CUDA是否可用，如果可用則使用GPUif cv2.cuda.getCudaEnabledDeviceCount() > 0:face_detector.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA)face_detector.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA)logging.info("使用CUDA進行GPU加速")print("使用CUDA進行GPU加速")else:logging.warning("CUDA不可用，使用CPU進行處理")connection = connect_to_database()if connection is None:returnknown_face_encodings, known_face_names = load_known_faces(connection)connection.close()video_streams = ["rtsp://xxxxxxxxx/cam/realmonitor?channel=1&subtype=1",# 添加更多視頻流URL]threads = []for stream_url in video_streams:thread = Thread(target=video_stream_thread, args=(stream_url, known_face_encodings, known_face_names))thread.start()threads.append(thread)for thread in threads:thread.join()if __name__ == "__main__":main()

三、加速

我發現在python中opencv-python并不支持GPU加速僅支持CPU,如果需要GPU加速需求自己下載opencv源碼用cmake和VS2019進行編譯，可以參考這篇文章https://blog.csdn.net/yangyu0515/article/details/133794355

至于有沒有成功，復制這個代碼就知道了，如果cv2.cuda.getCudaEnabledDeviceCount()大于0就是成功了

   import cv2# 檢查CUDA是否可用，如果可用則使用GPUif cv2.cuda.getCudaEnabledDeviceCount() > 0:print("使用CUDA進行GPU加速")else:logging.warning("CUDA不可用，使用CPU進行處理")

opencv源碼百度網盤地址：

鏈接：https://pan.baidu.com/s/1JzgFHmtnV6pScZLInoa6oA?pwd=u46d
提取碼：u46d

若有需求的可以聯系我
QQ: 2375255615