DamoFD开源镜像实操：定时任务批量处理监控录像帧-程序员充电站

DamoFD开源镜像实操：定时任务批量处理监控录像帧

你是不是也遇到过这样的烦恼？手头有一堆监控录像，想要快速找出里面出现的人脸，但一帧一帧地看，眼睛都快看花了。或者，你需要定期从海量的视频素材里提取人脸信息，手动操作不仅效率低，还容易出错。

今天，我就来分享一个非常实用的解决方案：利用DamoFD人脸检测模型，结合定时任务，实现监控录像帧的自动化批量处理。这个方案特别适合需要定期分析视频内容、进行人脸检索或者构建人脸库的场景。

简单来说，DamoFD是一个轻量级（仅0.5G）但效果出色的人脸检测与关键点定位模型。它能在一张图片里准确地框出人脸，并标出双眼、鼻尖和嘴角这五个关键点。而我们今天要做的，就是让它“动起来”，自动处理源源不断的视频文件。

下面，我就手把手带你搭建这套系统，让你彻底告别繁琐的手工操作。

1. 环境准备与快速上手

在开始自动化之前，我们先确保基础环境没问题，并跑通一个最简单的例子，建立信心。

1.1 启动与配置镜像

首先，你需要在CSDN星图镜像广场找到并启动“DamoFD人脸检测关键点模型-0.5G”这个镜像。启动后，系统已经为你准备好了所有环境，包括Python 3.7、PyTorch、CUDA以及预装好的模型。

镜像默认把代码放在系统盘。为了方便我们后续修改代码和添加自己的脚本，第一步是把代码复制到数据盘的工作空间：

cp -r /root/DamoFD /root/workspace/ cd /root/workspace/DamoFD

然后，激活项目专用的Python环境：

conda activate damofd

做完这两步，我们的工作舞台就搭好了。

1.2 跑通第一个检测例子

我们来试试模型的基本功能。镜像提供了两种运行方式：Python脚本和Jupyter Notebook。这里我们用脚本方式，更贴近我们后续的自动化场景。

用你喜欢的文本编辑器（比如vim或者Jupyter里的编辑器）打开DamoFD.py文件。找到下面这行：

img_path = 'https://modelscope.oss-cn-beijing.aliyuncs.com/test/images/mog_face_detection.jpg'

把这行里的图片地址，换成你自己的一张带人脸的图片的绝对路径。比如，如果你在/root/workspace下放了一张test.jpg，就改成：

img_path = '/root/workspace/test.jpg'

保存文件，然后在终端运行：

python DamoFD.py

稍等片刻，程序就会运行完毕。它会做两件事：

在终端打印检测结果：告诉你图片里发现了多少张脸，每张脸的坐标框和置信度是多少。
生成一张结果图片：在原图的基础上，用绿框把人脸框出来，并在同目录下保存为类似DamoFD_output.jpg的文件。

打开生成的结果图片看看，如果人脸被准确地框出来了，恭喜你！基础环境完全没问题，DamoFD模型已经成功运行起来了。

2. 核心思路：从单张图片到视频批量处理

单张图片检测很简单，但我们的目标是处理视频。视频本质上是由一帧帧图片连续播放组成的。所以，处理视频的思路就很清晰了：

拆解：把视频文件按帧拆分成一张张的图片。
处理：用DamoFD模型对每一张图片进行人脸检测。
收集：把每一帧的检测结果（比如人脸位置、数量）保存下来。
聚合：分析所有帧的结果，得到整个视频的人脸出现情况。

为了实现“批量”和“定时”，我们还需要两个帮手：

脚本封装：把上面的步骤写成一个完整的Python脚本，它接受一个视频文件路径，输出处理结果。
定时任务：使用Linux系统自带的cron服务，让这个脚本在指定的时间（比如每天凌晨2点）自动运行，处理指定文件夹下的所有新视频。

接下来，我们就来打造这两个核心部件。

3. 打造视频处理脚本

我们将在/root/workspace/DamoFD目录下，创建一个新的Python脚本，叫batch_process_video.py。这个脚本会包含我们需要的所有功能。

3.1 脚本框架与参数解析

首先，脚本要能接收外部参数，比如告诉我们处理哪个视频文件，结果存到哪里。

#!/usr/bin/env python # -*- coding: utf-8 -*- """ 批量视频人脸检测脚本 功能：读取视频文件，逐帧进行人脸检测，统计并保存结果。 用法：python batch_process_video.py --input /path/to/video.mp4 --output /path/to/results/ """ import argparse import cv2 import os import sys import json from datetime import datetime # 添加当前目录到路径，以便导入DamoFD模型 sys.path.append('/root/workspace/DamoFD') from DamoFD import detect_faces # 假设我们将核心检测函数封装成了 detect_faces def parse_arguments(): parser = argparse.ArgumentParser(description='使用DamoFD批量处理视频帧进行人脸检测') parser.add_argument('--input', '-i', type=str, required=True, help='输入视频文件的路径') parser.add_argument('--output_dir', '-o', type=str, default='./video_results', help='输出结果目录，默认当前目录下的video_results') parser.add_argument('--frame_interval', '-f', type=int, default=1, help='帧采样间隔，例如5表示每5帧处理1帧，默认处理每一帧') parser.add_argument('--save_frames_with_faces', action='store_true', help='是否保存检测到人脸的帧图片') return parser.parse_args()

这段代码定义了脚本的“使用说明书”。运行脚本时，我们可以通过--input指定视频，--output_dir指定结果存放位置，--frame_interval可以设置跳帧处理（对于长时间视频，处理每一帧可能太慢，可以每隔几帧处理一次），--save_frames_with_faces则决定是否要把拍到人脸的画面存成图片。

3.2 视频读取与帧处理循环

这是脚本的核心部分，负责打开视频，一帧一帧地读取并处理。

def process_video(video_path, output_dir, frame_interval=1, save_frames=False): """ 处理单个视频文件 """ # 确保输出目录存在 os.makedirs(output_dir, exist_ok=True) if save_frames: frames_dir = os.path.join(output_dir, 'frames_with_faces') os.makedirs(frames_dir, exist_ok=True) # 打开视频文件 cap = cv2.VideoCapture(video_path) if not cap.isOpened(): print(f"错误：无法打开视频文件 {video_path}") return None total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = cap.get(cv2.CAP_PROP_FPS) print(f"开始处理视频: {os.path.basename(video_path)}") print(f"视频信息: 总帧数={total_frames}, 帧率={fps:.2f}, 采样间隔={frame_interval}") frame_count = 0 processed_count = 0 results = [] # 用于存储每一帧的结果 while True: ret, frame = cap.read() if not ret: break # 视频读取完毕 # 根据采样间隔决定是否处理当前帧 if frame_count % frame_interval == 0: # 将OpenCV的BGR格式转换为RGB格式（DamoFD模型通常需要RGB） frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) # 调用DamoFD进行人脸检测 # 注意：这里需要根据实际的DamoFD.py文件调整调用方式 # 以下是一个假设的调用接口，你可能需要适配 faces_info = detect_faces(frame_rgb) # 返回人脸框、关键点、置信度等列表 # 记录结果 frame_result = { 'frame_index': frame_count, 'timestamp': frame_count / fps, # 当前帧的时间戳（秒） 'face_count': len(faces_info), 'faces': faces_info # 每个人脸的详细信息 } results.append(frame_result) # 如果检测到人脸且需要保存图片 if len(faces_info) > 0 and save_frames: # 在图片上画框（可选，这里简单示例） for face in faces_info: bbox = face['bbox'] # 假设bbox是[x1, y1, x2, y2] cv2.rectangle(frame, (int(bbox[0]), int(bbox[1])), (int(bbox[2]), int(bbox[3])), (0, 255, 0), 2) # 保存图片 frame_filename = f"frame_{frame_count:06d}_faces_{len(faces_info)}.jpg" cv2.imwrite(os.path.join(frames_dir, frame_filename), frame) print(f" 保存带人脸帧: {frame_filename}") processed_count += 1 if processed_count % 50 == 0: print(f" 已处理 {processed_count}/{total_frames//frame_interval} 采样帧...") frame_count += 1 cap.release() print(f"视频处理完成。共处理 {processed_count} 个采样帧。") return results

这个函数就像一条流水线：抓取一帧图片 -> 转换颜色 -> 送入DamoFD模型检测 -> 记录结果。如果这帧里有人脸，并且我们要求保存图片，它还会把画好框的图片存起来。

3.3 结果汇总与保存

所有帧处理完后，我们需要对结果做个总结，并保存到文件里，方便以后查看。

def save_results(results, output_dir, video_name): """ 将处理结果保存为JSON文件，并生成一个简明的统计报告 """ if not results: print("没有结果需要保存。") return # 1. 保存详细的每帧结果 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") detailed_json_path = os.path.join(output_dir, f"{video_name}_details_{timestamp}.json") with open(detailed_json_path, 'w', encoding='utf-8') as f: json.dump(results, f, indent=2, ensure_ascii=False) print(f"详细结果已保存至: {detailed_json_path}") # 2. 生成并保存统计报告 total_frames_with_faces = sum(1 for r in results if r['face_count'] > 0) max_faces_in_one_frame = max((r['face_count'] for r in results), default=0) face_occurrence_rate = (total_frames_with_faces / len(results)) * 100 if results else 0 # 找出人脸出现的时间段 face_timestamps = [] for r in results: if r['face_count'] > 0: face_timestamps.append(r['timestamp']) summary = { 'video_name': video_name, 'total_sampled_frames': len(results), 'frames_with_faces': total_frames_with_faces, 'face_occurrence_rate_percent': round(face_occurrence_rate, 2), 'max_faces_in_one_frame': max_faces_in_one_frame, 'processing_time': timestamp, 'face_timestamps_seconds': face_timestamps[:20] # 只保留前20个时间点，避免文件过大 } summary_json_path = os.path.join(output_dir, f"{video_name}_summary_{timestamp}.json") with open(summary_json_path, 'w', encoding='utf-8') as f: json.dump(summary, f, indent=2, ensure_ascii=False) # 3. 打印简易统计信息到屏幕 print("\n" + "="*50) print(f"视频分析总结 [{video_name}]") print("="*50) print(f"分析采样帧总数: {len(results)}") print(f"出现人脸的帧数: {total_frames_with_faces}") print(f"人脸出现频率: {face_occurrence_rate:.2f}%") print(f"单帧最大人脸数: {max_faces_in_one_frame}") if face_timestamps: print(f"人脸出现时间点(前5个): {[round(t,1) for t in face_timestamps[:5]]} 秒") print("="*50) print(f"统计报告已保存至: {summary_json_path}")

这个函数会生成两种文件：一个是包含每一帧详细数据的“流水账”，另一个是提炼了关键信息的“简报”。简报里会告诉你，视频里大概有多少比例的画面出现了人脸，最多同时出现了几张脸，以及人脸大概在哪些时间点出现，非常直观。

3.4 主函数与DamoFD调用适配

最后，我们把所有部分组装起来，并解决最关键的一步：如何正确调用DamoFD模型里的函数。

# 注意：你需要根据实际的 DamoFD.py 文件内容，来定义或导入 detect_faces 函数 # 以下是一个示例，展示了如何将原 DamoFD.py 中的检测逻辑封装成一个函数 def detect_faces(image_rgb): """ 封装DamoFD的人脸检测逻辑。 你需要根据 /root/workspace/DamoFD/DamoFD.py 的实际代码来调整此函数。 """ # 这里是模拟代码，你需要替换为实际的调用 # 通常，原 DamoFD.py 的检测部分可能依赖于一个全局模型和管道。 # 我们可以借鉴其方式，但避免重复初始化。 # 假设我们这样调用（请根据实际文件调整）： import torch from modelscope.pipelines import pipeline from modelscope.utils.constant import Tasks # 为了性能，模型应该只初始化一次。我们可以使用全局变量或单例。 if not hasattr(detect_faces, 'pipeline'): # 初始化人脸检测管道 model_id = 'iic/cv_ddsar_face-detection_iclr23-damofd' detect_faces.pipeline = pipeline(Tasks.face_detection, model=model_id, device='cuda') # 执行检测 result = detect_faces.pipeline(image_rgb) faces = result['boxes'] # 假设返回结果中包含'boxes'键 # 结果格式可能需要转换，例如转换为 [{'bbox': [x1,y1,x2,y2], 'score': s}, ...] formatted_faces = [] for i, box in enumerate(faces): # 这里需要根据实际返回数据结构调整 # 假设 box 是 [x1, y1, x2, y2, score] if len(box) >= 5: formatted_faces.append({ 'bbox': box[:4].tolist() if torch.is_tensor(box) else box[:4], 'score': float(box[4]) }) return formatted_faces def main(): args = parse_arguments() video_path = args.input output_dir = args.output_dir frame_interval = args.frame_interval save_frames = args.save_frames_with_faces # 提取视频文件名（不含后缀） video_name = os.path.splitext(os.path.basename(video_path))[0] # 为当前视频创建一个子目录 video_output_dir = os.path.join(output_dir, video_name) os.makedirs(video_output_dir, exist_ok=True) print("="*60) print("DamoFD 视频批量处理脚本启动") print("="*60) # 处理视频 results = process_video(video_path, video_output_dir, frame_interval, save_frames) # 保存结果 if results: save_results(results, video_output_dir, video_name) else: print("视频处理未产生结果。") print("批量处理脚本执行完毕！") if __name__ == '__main__': main()

重要提示：detect_faces函数是连接我们脚本和DamoFD模型的关键。上面给出的只是一个示例框架。你需要打开/root/workspace/DamoFD/DamoFD.py文件，仔细看它是如何加载模型、运行推理的，然后把那部分核心逻辑提取出来，封装成上面这样的函数。核心是确保模型只被加载一次，而不是每一帧都加载，这样速度才会快。

创建好这个脚本后，你可以先试运行一下：

cd /root/workspace/DamoFD conda activate damofd python batch_process_video.py --input /root/workspace/your_video.mp4 --output_dir ./my_results --frame_interval 10 --save_frames_with_faces

如果一切顺利，你会在my_results目录下看到以视频文件名命名的文件夹，里面包含了处理结果的JSON文件和可能的人脸帧图片。

4. 实现自动化定时任务

脚本能工作了，接下来就让它自动运行。我们使用Linux最经典的定时任务工具——cron。

4.1 创建监控与调度脚本

我们创建一个更高级的“调度员”脚本，比如叫auto_process_folder.py。它的任务是定期扫描某个文件夹（比如/root/workspace/videos_to_process），发现新的视频文件就调用我们刚才的batch_process_video.py去处理，处理完后把视频文件挪到另一个文件夹（比如/root/workspace/videos_processed），避免重复处理。

#!/usr/bin/env python # auto_process_folder.py import os import subprocess import shutil import time from datetime import datetime # 配置路径 WATCH_FOLDER = '/root/workspace/videos_to_process' PROCESSED_FOLDER = '/root/workspace/videos_processed' OUTPUT_BASE = '/root/workspace/processing_results' SCRIPT_PATH = '/root/workspace/DamoFD/batch_process_video.py' LOG_FILE = '/root/workspace/processing_results/processing.log' # 支持的视频格式 VIDEO_EXTENSIONS = ('.mp4', '.avi', '.mov', '.mkv', '.flv') def ensure_dirs(): """确保所有需要的目录都存在""" for folder in [WATCH_FOLDER, PROCESSED_FOLDER, OUTPUT_BASE]: os.makedirs(folder, exist_ok=True) def log_message(message): """记录日志""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_line = f"[{timestamp}] {message}\n" print(log_line, end='') with open(LOG_FILE, 'a') as f: f.write(log_line) def process_video_file(video_path): """处理单个视频文件""" video_filename = os.path.basename(video_path) log_message(f"开始处理视频: {video_filename}") # 构建输出目录，按日期组织 date_str = datetime.now().strftime("%Y%m%d") output_dir = os.path.join(OUTPUT_BASE, date_str, os.path.splitext(video_filename)[0]) # 调用批量处理脚本 # 这里设置每5帧处理1帧，并且保存带人脸的帧图片 cmd = [ 'python', SCRIPT_PATH, '--input', video_path, '--output_dir', output_dir, '--frame_interval', '5', '--save_frames_with_faces' ] try: # 运行命令，并捕获输出 result = subprocess.run(cmd, capture_output=True, text=True, timeout=3600) # 超时1小时 log_message(f"命令输出:\n{result.stdout}") if result.stderr: log_message(f"命令错误:\n{result.stderr}") if result.returncode == 0: log_message(f"视频处理成功: {video_filename}") return True else: log_message(f"视频处理失败，返回码 {result.returncode}: {video_filename}") return False except subprocess.TimeoutExpired: log_message(f"处理超时（1小时）: {video_filename}") return False except Exception as e: log_message(f"处理过程发生异常: {video_filename} - {str(e)}") return False def main_loop(): """主循环，扫描文件夹并处理视频""" ensure_dirs() log_message("=== 自动视频处理服务启动 ===") while True: try: # 扫描监控文件夹 files = os.listdir(WATCH_FOLDER) video_files = [f for f in files if f.lower().endswith(VIDEO_EXTENSIONS)] if video_files: log_message(f"发现 {len(video_files)} 个待处理视频文件。") for video_file in video_files: video_path = os.path.join(WATCH_FOLDER, video_file) success = process_video_file(video_path) # 无论成功与否，都移动文件，避免阻塞 processed_path = os.path.join(PROCESSED_FOLDER, video_file) if success: shutil.move(video_path, processed_path) log_message(f"已移动文件至已处理目录: {video_file}") else: # 失败的文件可以移动到一个专门的子目录，方便排查 failed_dir = os.path.join(PROCESSED_FOLDER, 'failed') os.makedirs(failed_dir, exist_ok=True) shutil.move(video_path, os.path.join(failed_dir, video_file)) log_message(f"处理失败，文件移至 failed 目录: {video_file}") else: # 没有文件时，休眠一段时间再检查 time.sleep(60) # 休眠60秒 except KeyboardInterrupt: log_message("收到中断信号，停止服务。") break except Exception as e: log_message(f"主循环发生异常: {str(e)}") time.sleep(60) if __name__ == '__main__': # 如果不想让脚本一直运行，而是由cron定时触发，可以去掉while循环，只执行一次扫描 # 这里提供两种模式：常驻服务模式 和 单次触发模式 # 模式一：常驻服务（作为后台守护进程） # main_loop() # 模式二：单次触发（由cron每分钟或每5分钟调用一次） ensure_dirs() log_message("=== 单次扫描触发 ===") # ... 这里放入 main_loop 中处理文件的代码块，但不包含最外层的 while True 循环 # 即，只执行一次扫描和处理逻辑 files = os.listdir(WATCH_FOLDER) video_files = [f for f in files if f.lower().endswith(VIDEO_EXTENSIONS)] if video_files: log_message(f"发现 {len(video_files)} 个待处理视频文件。") for video_file in video_files: video_path = os.path.join(WATCH_FOLDER, video_file) success = process_video_file(video_path) # ... 移动文件逻辑 else: log_message("未发现新的待处理视频文件。")

这个脚本提供了两种模式。如果你希望它24小时不间断地监控文件夹，就使用main_loop()模式。如果只是想每天固定时间（比如凌晨）处理一次，就更适合用下面的“单次触发”模式，然后通过cron来定时调用它。

4.2 配置Cron定时任务

假设我们采用“单次触发”模式，并希望每天凌晨2点处理一次积压的视频。我们需要配置cron。

首先，在终端输入：

crontab -e

这会打开你的用户cron配置表。在文件末尾添加一行：

0 2 * * * cd /root/workspace/DamoFD && /opt/conda/envs/damofd/bin/python /root/workspace/DamoFD/auto_process_folder.py >> /root/workspace/processing_results/cron.log 2>&1

这行命令的意思是：

0 2 * * *：每天凌晨2点0分执行。
cd /root/workspace/DamoFD：先切换到工作目录。
&&：如果上一步成功，则执行下一步。
/opt/conda/envs/damofd/bin/python：使用我们激活的damofd环境下的Python解释器（路径可能需要根据你的镜像实际情况调整）。
/root/workspace/DamoFD/auto_process_folder.py：执行我们的自动处理脚本。
>> /root/workspace/processing_results/cron.log 2>&1：把脚本的所有输出（包括正常信息和错误信息）都追加记录到cron.log文件里，方便以后查看。

保存并退出编辑器。cron服务会自动加载新的配置。

现在，整个系统就搭建完成了。你只需要把需要分析的监控录像视频，丢到/root/workspace/videos_to_process文件夹里，然后就可以去睡觉了。第二天早上，所有结果都会整齐地躺在/root/workspace/processing_results目录下，按日期和视频名分门别类地放好。