[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/>

Jacktang

将人脸识别和视频处理封装了下，采用多线程是希望能够加快处理速度，，

什么是将人脸识别和视频处理封装啊

lijinlei

膜拜大佬啊，太强了，识别效果好，题材（桃园三结义）也很棒

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

最新回复