[RV1106&InsightFace]--InsightFace实现视频中的人脸识别

许大锤

[RV1106&InsightFace]--InsightFace实现视频中的人脸识别 [复制链接]

前文已经实现了本地部署，也实现图片和视频的人脸检测。https://bbs.eeworld.com.cn/thread-1303465-1-1.html#pid3385431 接下来是实现视频中人脸识别。 ## 准备 ```shell pip install torch opencv-python insightface numpy scikit-learn pip install insightface pip install onnxruntime pip install onnxruntime-gpu pip install opencv-contrib-python ``` 为了让输出的视频包含音频，建议安装FFmpeg，具体如何安装网上有很多教程。 接下来是实现视频中人脸识别，但感觉处理速度挺慢的，所以每5帧进行一次识别，中间做的识别框就做插值处理了，还有感觉合成音频也蛮费时间的。 ## 构建人脸数据库 先准备几个人脸图片，修改image_path的路径，name变量是视频中标记的名字，建议用英文，中文可能会出问题。 ```python # 添加新的人脸 image_path = 'img/guanyu.jpg' # 已知人脸图片路径 name = "Guan Yu" # 对应的身份名称 ``` 完整代码： ```python import cv2 import insightface import pickle from insightface.app import FaceAnalysis # 初始化 ArcFace 模型 app = FaceAnalysis() app.prepare(ctx_id=0, det_size=(640, 640)) # 数据库路径 database_path = './face_database.pkl' try: with open(database_path, 'rb') as f: face_database = pickle.load(f) print("已加载人脸数据库") except FileNotFoundError: face_database = {} print("人脸数据库不存在，创建空数据库") # 添加新的人脸 image_path = 'img/guanyu.jpg' # 已知人脸图片路径 name = "Guan Yu" # 对应的身份名称 image = cv2.imread(image_path) faces = app.get(image) if len(faces) == 1: face = faces[0] embedding = face.embedding # 提取特征向量 face_database[name] = embedding print(f"已添加人脸：{name}") else: print("未检测到人脸或检测到多张人脸，请检查图片") # 保存数据库 with open(database_path, 'wb') as f: pickle.dump(face_database, f) print("人脸数据库已保存") ``` ## 视频中人脸识别 ### 代码 ```python import cv2 import pickle import numpy as np from insightface.app import FaceAnalysis from sklearn.metrics.pairwise import cosine_similarity import time import os import subprocess import threading from queue import Queue, Empty from tqdm import tqdm import torch # 检查是否可以使用GPU DEVICE = 'cuda:0' if torch.cuda.is_available() else 'cpu' # 设置 FFmpeg 路径 FFMPEG_PATH = r"C:\ProgramData\chocolatey\lib\ffmpeg\tools\ffmpeg\bin\ffmpeg.exe" class VideoProcessor: def __init__(self, input_path: str, output_path: str, face_database: dict, batch_size: int = 4, num_workers: int = 4): self.input_path = input_path self.output_path = output_path self.temp_output_path = output_path.replace('.mp4', '_temp.mp4') self.face_database = face_database self.batch_size = batch_size self.num_workers = num_workers # 初始化 InsightFace self.app = FaceAnalysis(providers=['CUDAExecutionProvider', 'CPUExecutionProvider']) self.app.prepare(ctx_id=0, det_size=(640, 640)) # 初始化队列 self.frame_queue = Queue(maxsize=30) self.result_queue = Queue(maxsize=30) # 获取视频信息 self.cap = cv2.VideoCapture(input_path) self.fps = self.cap.get(cv2.CAP_PROP_FPS) self.frame_width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH)) self.frame_height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT)) self.total_frames = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT)) # 创建视频写入器 self.writer = cv2.VideoWriter( self.temp_output_path, cv2.VideoWriter_fourcc(*'mp4v'), self.fps, (self.frame_width, self.frame_height) ) # 存储检测到的人脸信息 self.detected_faces = {} self.face_lock = threading.Lock() # 添加处理状态标志 self.processing = True def read_frames(self): """读取视频帧的线程函数""" frame_count = 0 pbar_read = tqdm(total=self.total_frames, desc="读取帧", position=0, leave=True) while self.processing: ret, frame = self.cap.read() if not ret: break self.frame_queue.put((frame_count, frame)) frame_count += 1 pbar_read.update(1) pbar_read.close() self.frame_queue.put(None) # 发送结束信号 def process_frame(self, frame_data): """处理单个视频帧""" if frame_data is None: return None frame_count, frame = frame_data current_time = time.time() # 每5帧处理一次人脸检测 if frame_count % 5 == 0: faces = self.app.get(frame) current_faces = {} for face in faces: face_embedding = face.normed_embedding bbox = face.bbox.astype(int) # 人脸识别 best_match = None best_similarity = -1 for name, db_embedding in self.face_database.items(): similarity = cosine_similarity([face_embedding], [db_embedding])[0][0] if similarity > best_similarity: best_similarity = similarity best_match = name if best_match is not None and best_similarity > 0.5: current_faces[best_match] = { 'bbox': bbox, 'last_seen': current_time } # 更新检测到的人脸信息 with self.face_lock: self.detected_faces.update(current_faces) # 移除超时的人脸 faces_to_remove = [] for name, info in self.detected_faces.items(): if current_time - info['last_seen'] > 0.5: # 0.5秒超时 faces_to_remove.append(name) for name in faces_to_remove: self.detected_faces.pop(name) # 绘制识别框 with self.face_lock: for name, info in self.detected_faces.items(): bbox = info['bbox'] cv2.rectangle(frame, (bbox[0], bbox[1]), (bbox[2], bbox[3]), (0, 255, 0), 2) cv2.putText(frame, name, (bbox[0], bbox[1] - 10), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 2) return frame_count, frame def process_frames(self): """处理视频帧的线程函数""" while self.processing: try: frame_data = self.frame_queue.get(timeout=1.0) if frame_data is None: self.result_queue.put(None) break result = self.process_frame(frame_data) if result is not None: self.result_queue.put(result) except Empty: continue except Exception as e: print(f"\n处理帧时出错: {str(e)}") break def write_frames(self): """写入处理后的帧的线程函数""" processed_frames = {} next_frame_to_write = 0 remaining_frames = self.total_frames timeout_counter = 0 pbar_write = tqdm(total=self.total_frames, desc="写入帧", position=1, leave=True) while remaining_frames > 0 and self.processing: try: result = self.result_queue.get(timeout=1.0) if result is None: if len(processed_frames) == 0: break timeout_counter += 1 if timeout_counter > 10: print("\n警告：部分帧可能未处理完成") break continue timeout_counter = 0 frame_count, frame = result processed_frames[frame_count] = frame while next_frame_to_write in processed_frames: self.writer.write(processed_frames[next_frame_to_write]) del processed_frames[next_frame_to_write] next_frame_to_write += 1 remaining_frames -= 1 pbar_write.update(1) except Empty: if len(processed_frames) > 0: available_frames = sorted(processed_frames.keys()) if available_frames[0] == next_frame_to_write: self.writer.write(processed_frames[next_frame_to_write]) del processed_frames[next_frame_to_write] next_frame_to_write += 1 remaining_frames -= 1 pbar_write.update(1) continue except Exception as e: print(f"\n写入帧时出错: {str(e)}") break if processed_frames: print("\n处理剩余帧...") for frame_count in sorted(processed_frames.keys()): self.writer.write(processed_frames[frame_count]) pbar_write.update(1) remaining_frames -= 1 pbar_write.close() def combine_audio(self): """合并视频和音频""" try: command = [ FFMPEG_PATH, '-i', self.temp_output_path, '-i', self.input_path, '-c:v', 'copy', '-c:a', 'aac', '-strict', 'experimental', '-y', self.output_path ] print("执行音频合并...") result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) if result.returncode == 0: print("音频合并成功！") return True else: print("FFmpeg 错误:") print(result.stderr) return False except Exception as e: print(f"音频合并出错: {str(e)}") print(f"临时文件保留在: {self.temp_output_path}") return False def process_video(self): """处理整个视频""" print(f"\n开始处理视频: {self.input_path}") print(f"总帧数: {self.total_frames}") start_time = time.time() try: # 创建线程 threads = [] # 读取帧的线程 read_thread = threading.Thread(target=self.read_frames) threads.append(read_thread) # 处理帧的线程池 process_threads = [] for _ in range(self.num_workers): t = threading.Thread(target=self.process_frames) process_threads.append(t) threads.append(t) # 写入帧的线程 write_thread = threading.Thread(target=self.write_frames) threads.append(write_thread) # 启动所有线程 for t in threads: t.start() # 设置超时时间（秒） timeout = 300 # 5分钟 start = time.time() # 等待所有线程完成 for t in threads: remaining = timeout - (time.time() - start) if remaining <= 0: print("\n处理超时，强制结束...") self.processing = False break t.join(timeout=remaining) # 检查是否所有线程都已结束 for t in threads: if t.is_alive(): print("\n强制结束未完成的线程...") self.processing = False except Exception as e: print(f"\n处理视频时出错: {str(e)}") self.processing = False finally: # 释放资源 self.cap.release() self.writer.release() print("\n开始合并音频...") # 添加音频 self.combine_audio() # 删除临时文件 if os.path.exists(self.temp_output_path): os.remove(self.temp_output_path) print("临时文件已删除") end_time = time.time() processing_time = end_time - start_time fps = self.total_frames / processing_time print(f"\n处理完成!") print(f"总用时: {processing_time:.2f} 秒") print(f"平均处理速度: {fps:.2f} FPS") print(f"输出文件: {self.output_path}") if __name__ == "__main__": try: # 加载人脸数据库 print("加载人脸数据库...") with open('face_database.pkl', 'rb') as f: face_database = pickle.load(f) # 设置输入输出路径 input_video = './video/Taoyuan.mp4' output_video = './video/Taoyuan_output.mp4' # 确保输出目录存在 os.makedirs(os.path.dirname(output_video), exist_ok=True) # 创建处理器并处理视频 processor = VideoProcessor( input_path=input_video, output_path=output_video, face_database=face_database, num_workers=4 # 根据CPU核心数调整 ) processor.process_video() except Exception as e: print(f"发生错误: {str(e)}") print(f"发生错误: {str(e)}") ``` 构建了一个视频处理类，将人脸识别和视频处理封装了下，采用多线程是希望能够加快处理速度，感觉还可以降低分辨率提高速度，但视频已经挺糊的就没这么处理。 ### 多线程处理： - 使用独立线程读取视频帧 - 使用线程池并行处理帧 - 使用独立线程写入处理后的帧 ### GPU 加速： - 检测是否可用 GPU - 使用 CUDA 加速（如果可用） ### 批处理： - 使用队列管理帧的处理 - 按顺序写入处理后的帧 ### 内存优化： - 使用队列限制内存使用 - 及时释放处理完的帧 ### 性能优化： - 降低人脸检测频率（每5帧检测一次） - 使用线程锁保护共享资源 ### 调整参数 - num_workers：根据CPU核心数调整 - batch_size：根据可用内存调整 - frame_interval：调整人脸检测频率 ## 输出视频结果 <div id="player_19b71298a98cd432d56b08ae30edf1cf0" class="localvideo-box"></div><div class="localvideo-title">Taoyuan_output</div><br/>