YOLO12实时视频流处理方案：逐帧检测与API集成指南
·
YOLO12实时视频流处理方案：逐帧检测与API集成指南
实时视频流处理是计算机视觉领域的核心应用场景之一，从安防监控到智能交通，从工业质检到互动直播，都需要系统能够快速、准确地识别视频中的目标。YOLO12作为Ultralytics在2025年推出的最新实时目标检测模型，凭借其高达131 FPS的推理速度和优化的检测精度，成为了视频流处理的理想选择。
本文将手把手教你如何将YOLO12模型与实时视频流处理相结合，从单张图片检测扩展到连续视频帧分析，并提供完整的API集成方案，让你能够快速构建自己的视频分析应用。
1. 理解实时视频流处理的核心挑战
1.1 视频流处理的特殊性
与静态图片检测不同，视频流处理面临几个独特挑战：
- 连续性要求：视频是由连续的帧组成，处理必须跟上帧率，否则会出现延迟累积
- 实时性约束：很多应用场景要求毫秒级响应，如自动驾驶、安防报警
- 资源优化：连续处理需要高效利用计算资源，避免内存泄漏和性能下降
- 上下文关联：相邻帧之间的目标应该有连续性，避免检测结果跳跃
1.2 YOLO12的视频处理优势
YOLO12特别适合视频流处理的原因：
- 单阶段检测架构：端到端的前向传播，没有复杂的区域建议步骤，推理速度快
- 多规格模型：从nano到xlarge五种规格，可以根据硬件性能选择合适模型
- 轻量级设计：nano版仅5.6MB，370万参数，在边缘设备上也能流畅运行
- 高帧率支持：在RTX 4090上，nano版可达131 FPS，轻松处理30FPS的视频流
2. 搭建视频流处理基础环境
2.1 环境准备与YOLO12部署
首先按照标准流程部署YOLO12镜像：
- 在镜像市场选择
ins-yolo12-independent-v1 - 点击"部署实例"创建运行环境
- 等待1-2分钟初始化完成
- 确认实例状态变为"已启动"
部署完成后，你可以通过 http://<实例IP>:7860 访问Web界面进行基本测试，确保模型正常工作。
2.2 安装视频处理依赖库
虽然YOLO12镜像已经包含了核心依赖，但视频处理需要额外的库。通过SSH连接到实例后，安装必要的包：
# 连接到你的实例
ssh root@<实例IP>
# 安装视频处理相关库
pip install opencv-python-headless
pip install numpy
pip install requests
OpenCV是视频处理的核心库，headless版本不需要图形界面，更适合服务器环境。
2.3 验证API接口可用性
在开始视频处理前，先测试API接口是否正常工作：
# 测试单张图片检测
curl -X POST "http://localhost:8000/predict" \
-H "accept: application/json" \
-F "file=@/root/test_image.jpg"
如果返回JSON格式的检测结果，说明API服务运行正常。
3. 实现逐帧视频处理方案
3.1 基础视频处理脚本
下面是一个完整的视频逐帧处理脚本，展示了如何读取视频、逐帧调用YOLO12 API、保存处理结果：
import cv2
import requests
import json
import time
from datetime import datetime
class VideoProcessor:
def __init__(self, api_url="http://localhost:8000/predict"):
self.api_url = api_url
self.frame_count = 0
self.total_processing_time = 0
def process_video(self, video_path, output_path=None, frame_interval=1):
"""
处理视频文件，逐帧调用YOLO12进行目标检测
参数:
video_path: 输入视频文件路径
output_path: 输出视频文件路径（可选）
frame_interval: 处理间隔，1表示每帧都处理，2表示每隔一帧处理
"""
# 打开视频文件
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
print(f"无法打开视频文件: {video_path}")
return
# 获取视频基本信息
fps = int(cap.get(cv2.CAP_PROP_FPS))
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
print(f"视频信息: {width}x{height}, {fps}FPS, 总帧数: {total_frames}")
# 如果需要保存处理结果，创建VideoWriter
writer = None
if output_path:
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
# 逐帧处理
frame_index = 0
detection_results = []
while True:
ret, frame = cap.read()
if not ret:
break
# 按间隔处理帧
if frame_index % frame_interval == 0:
# 调用YOLO12 API进行检测
start_time = time.time()
detections = self.detect_frame(frame)
processing_time = time.time() - start_time
# 记录处理时间
self.total_processing_time += processing_time
self.frame_count += 1
# 在帧上绘制检测结果
processed_frame = self.draw_detections(frame, detections)
# 保存检测结果
frame_result = {
"frame_index": frame_index,
"timestamp": frame_index / fps,
"processing_time": processing_time,
"detections": detections
}
detection_results.append(frame_result)
# 显示处理进度
if frame_index % 30 == 0: # 每30帧显示一次进度
avg_time = self.total_processing_time / self.frame_count
remaining_frames = total_frames - frame_index
eta = remaining_frames * avg_time / frame_interval
print(f"处理进度: {frame_index}/{total_frames} | "
f"平均处理时间: {avg_time*1000:.1f}ms | "
f"预计剩余时间: {eta:.1f}秒")
# 更新当前帧为处理后的帧
frame = processed_frame
# 写入输出视频
if writer:
writer.write(frame)
frame_index += 1
# 释放资源
cap.release()
if writer:
writer.release()
# 输出统计信息
self.print_statistics(detection_results)
return detection_results
def detect_frame(self, frame):
"""调用YOLO12 API检测单帧"""
# 将帧编码为JPEG格式
_, img_encoded = cv2.imencode('.jpg', frame)
# 准备API请求
files = {'file': ('frame.jpg', img_encoded.tobytes(), 'image/jpeg')}
try:
response = requests.post(self.api_url, files=files, timeout=5)
if response.status_code == 200:
return response.json()
else:
print(f"API请求失败: {response.status_code}")
return {"predictions": []}
except Exception as e:
print(f"检测失败: {str(e)}")
return {"predictions": []}
def draw_detections(self, frame, detections):
"""在帧上绘制检测框和标签"""
predictions = detections.get("predictions", [])
# 为不同类别定义颜色
colors = {
"person": (0, 255, 0), # 绿色
"car": (255, 0, 0), # 蓝色
"bicycle": (0, 255, 255), # 黄色
"dog": (255, 0, 255), # 紫色
"cat": (255, 255, 0) # 青色
}
for pred in predictions:
class_name = pred.get("class", "unknown")
confidence = pred.get("confidence", 0)
bbox = pred.get("bbox", [])
if len(bbox) == 4:
x1, y1, x2, y2 = map(int, bbox)
# 获取类别颜色，默认白色
color = colors.get(class_name, (255, 255, 255))
# 绘制边界框
cv2.rectangle(frame, (x1, y1), (x2, y2), color, 2)
# 绘制标签背景
label = f"{class_name}: {confidence:.2f}"
label_size, baseline = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1)
cv2.rectangle(frame, (x1, y1 - label_size[1] - 5),
(x1 + label_size[0], y1), color, -1)
# 绘制标签文字
cv2.putText(frame, label, (x1, y1 - 5),
cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 0, 0), 1)
# 添加帧信息
cv2.putText(frame, f"Detections: {len(predictions)}", (10, 30),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (255, 255, 255), 2)
return frame
def print_statistics(self, detection_results):
"""输出处理统计信息"""
if not detection_results:
print("没有检测到任何结果")
return
total_detections = sum(len(frame["detections"].get("predictions", []))
for frame in detection_results)
avg_processing_time = self.total_processing_time / self.frame_count if self.frame_count > 0 else 0
avg_fps = 1 / avg_processing_time if avg_processing_time > 0 else 0
print("\n" + "="*50)
print("视频处理统计:")
print(f"处理总帧数: {self.frame_count}")
print(f"检测到目标总数: {total_detections}")
print(f"平均处理时间: {avg_processing_time*1000:.1f}ms")
print(f"平均处理帧率: {avg_fps:.1f} FPS")
print(f"总处理时间: {self.total_processing_time:.1f}秒")
print("="*50)
# 使用示例
if __name__ == "__main__":
processor = VideoProcessor()
# 处理本地视频文件
results = processor.process_video(
video_path="input_video.mp4",
output_path="output_video.mp4",
frame_interval=1 # 每帧都处理
)
# 保存检测结果到JSON文件
with open("detection_results.json", "w") as f:
json.dump(results, f, indent=2)
3.2 实时摄像头流处理
对于实时摄像头流，代码需要稍作调整：
class CameraProcessor(VideoProcessor):
def __init__(self, api_url="http://localhost:8000/predict", camera_id=0):
super().__init__(api_url)
self.camera_id = camera_id
self.running = False
def start_stream(self, duration_seconds=None, show_preview=True):
"""开始处理摄像头实时流"""
cap = cv2.VideoCapture(self.camera_id)
if not cap.isOpened():
print(f"无法打开摄像头 {self.camera_id}")
return
# 设置摄像头参数（可选）
cap.set(cv2.CAP_PROP_FRAME_WIDTH, 640)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 480)
cap.set(cv2.CAP_PROP_FPS, 30)
print(f"开始处理摄像头 {self.camera_id} 的实时流")
print("按 'q' 键退出")
self.running = True
start_time = time.time()
frame_count = 0
while self.running:
ret, frame = cap.read()
if not ret:
print("摄像头读取失败")
break
# 处理当前帧
processing_start = time.time()
detections = self.detect_frame(frame)
processing_time = time.time() - processing_start
# 绘制检测结果
processed_frame = self.draw_detections(frame, detections)
# 显示处理信息
fps_text = f"FPS: {1/processing_time:.1f}" if processing_time > 0 else "FPS: N/A"
cv2.putText(processed_frame, fps_text, (10, 60),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 255), 2)
# 显示预览
if show_preview:
cv2.imshow('YOLO12 Real-time Detection', processed_frame)
# 检查按键
if cv2.waitKey(1) & 0xFF == ord('q'):
break
frame_count += 1
# 检查处理时长
if duration_seconds and (time.time() - start_time) > duration_seconds:
break
# 清理资源
cap.release()
if show_preview:
cv2.destroyAllWindows()
print(f"处理完成，共处理 {frame_count} 帧")
def stop_stream(self):
"""停止摄像头流处理"""
self.running = False
# 使用示例
if __name__ == "__main__":
# 处理默认摄像头（ID=0）的实时流，持续30秒
processor = CameraProcessor(camera_id=0)
processor.start_stream(duration_seconds=30, show_preview=True)
4. 优化视频处理性能
4.1 多线程处理提升吞吐量
对于高帧率视频流，单线程处理可能成为瓶颈。下面是多线程处理的优化版本：
import threading
import queue
import time
from concurrent.futures import ThreadPoolExecutor
class MultiThreadVideoProcessor:
def __init__(self, api_url="http://localhost:8000/predict", num_workers=4):
self.api_url = api_url
self.num_workers = num_workers
self.frame_queue = queue.Queue(maxsize=50)
self.result_queue = queue.Queue()
self.running = False
def process_video_parallel(self, video_path, output_path=None):
"""多线程并行处理视频"""
# 打开视频文件
cap = cv2.VideoCapture(video_path)
fps = int(cap.get(cv2.CAP_PROP_FPS))
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
print(f"开始多线程处理，使用 {self.num_workers} 个工作线程")
# 启动工作线程
self.running = True
workers = []
for i in range(self.num_workers):
worker = threading.Thread(target=self.worker_thread, args=(i,))
worker.start()
workers.append(worker)
# 启动结果处理线程
result_thread = threading.Thread(target=self.result_handler,
args=(output_path, fps, total_frames))
result_thread.start()
# 读取视频帧并放入队列
frame_index = 0
read_start = time.time()
while True:
ret, frame = cap.read()
if not ret:
break
# 等待队列有空位
while self.frame_queue.full():
time.sleep(0.001)
# 放入帧数据
frame_data = {
"frame": frame.copy(),
"index": frame_index,
"timestamp": frame_index / fps
}
self.frame_queue.put(frame_data)
frame_index += 1
# 显示读取进度
if frame_index % 100 == 0:
elapsed = time.time() - read_start
print(f"已读取 {frame_index}/{total_frames} 帧 | "
f"读取速度: {frame_index/elapsed:.1f} FPS")
# 视频读取完成，发送结束信号
for _ in range(self.num_workers):
self.frame_queue.put(None)
# 等待所有工作线程完成
for worker in workers:
worker.join()
# 发送结果处理结束信号
self.result_queue.put(None)
result_thread.join()
cap.release()
print("多线程处理完成")
def worker_thread(self, worker_id):
"""工作线程：从队列获取帧并调用API"""
while self.running:
try:
frame_data = self.frame_queue.get(timeout=1)
if frame_data is None: # 结束信号
break
# 调用API进行检测
start_time = time.time()
detections = self.detect_frame(frame_data["frame"])
processing_time = time.time() - start_time
# 将结果放入结果队列
result = {
"frame": frame_data["frame"],
"index": frame_data["index"],
"timestamp": frame_data["timestamp"],
"detections": detections,
"processing_time": processing_time,
"worker_id": worker_id
}
self.result_queue.put(result)
except queue.Empty:
continue
except Exception as e:
print(f"工作线程 {worker_id} 出错: {str(e)}")
def result_handler(self, output_path, fps, total_frames):
"""结果处理线程：按顺序处理检测结果"""
# 初始化VideoWriter
writer = None
if output_path:
# 需要先获取视频尺寸，这里假设为640x480
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
writer = cv2.VideoWriter(output_path, fourcc, fps, (640, 480))
# 按帧索引排序的缓冲区
result_buffer = {}
next_index = 0
processed_count = 0
total_processing_time = 0
while True:
try:
result = self.result_queue.get(timeout=5)
if result is None: # 结束信号
break
# 存入缓冲区
result_buffer[result["index"]] = result
# 按顺序处理缓冲区的帧
while next_index in result_buffer:
result = result_buffer.pop(next_index)
# 绘制检测结果
processed_frame = self.draw_detections(
result["frame"],
result["detections"]
)
# 写入输出视频
if writer:
writer.write(processed_frame)
# 更新统计
total_processing_time += result["processing_time"]
processed_count += 1
next_index += 1
# 显示进度
if processed_count % 50 == 0:
avg_time = total_processing_time / processed_count
print(f"已处理 {processed_count}/{total_frames} 帧 | "
f"平均处理时间: {avg_time*1000:.1f}ms | "
f"队列大小: {self.frame_queue.qsize()}")
except queue.Empty:
if not self.running:
break
# 清理
if writer:
writer.release()
avg_fps = processed_count / total_processing_time if total_processing_time > 0 else 0
print(f"\n处理完成统计:")
print(f"总处理帧数: {processed_count}")
print(f"总处理时间: {total_processing_time:.1f}秒")
print(f"平均帧率: {avg_fps:.1f} FPS")
def detect_frame(self, frame):
"""检测单帧（与之前相同）"""
_, img_encoded = cv2.imencode('.jpg', frame)
files = {'file': ('frame.jpg', img_encoded.tobytes(), 'image/jpeg')}
try:
response = requests.post(self.api_url, files=files, timeout=2)
if response.status_code == 200:
return response.json()
except:
pass
return {"predictions": []}
def draw_detections(self, frame, detections):
"""绘制检测结果（与之前相同）"""
# 简化的绘制逻辑，实际使用时可以复用之前的draw_detections方法
return frame
# 使用示例
if __name__ == "__main__":
processor = MultiThreadVideoProcessor(num_workers=4)
processor.process_video_parallel(
video_path="input_video.mp4",
output_path="output_parallel.mp4"
)
4.2 帧采样与分辨率优化
对于实时性要求极高的应用，可以通过帧采样和降低分辨率来提升性能：
class OptimizedVideoProcessor(VideoProcessor):
def __init__(self, api_url="http://localhost:8000/predict"):
super().__init__(api_url)
def process_with_optimizations(self, video_path, output_path=None,
target_fps=15, target_width=640):
"""
优化版本：降低帧率和分辨率以提升处理速度
参数:
target_fps: 目标处理帧率（降低输入帧率）
target_width: 目标宽度（保持宽高比）
"""
cap = cv2.VideoCapture(video_path)
original_fps = int(cap.get(cv2.CAP_PROP_FPS))
original_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
original_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
# 计算缩放比例和目标尺寸
scale = target_width / original_width
target_height = int(original_height * scale)
# 计算帧采样间隔
frame_interval = max(1, int(original_fps / target_fps))
print(f"优化设置:")
print(f" 原始分辨率: {original_width}x{original_height}")
print(f" 目标分辨率: {target_width}x{target_height}")
print(f" 原始FPS: {original_fps}")
print(f" 目标FPS: {target_fps}")
print(f" 帧采样间隔: {frame_interval}")
frame_index = 0
processed_count = 0
total_time = 0
while True:
ret, frame = cap.read()
if not ret:
break
# 帧采样
if frame_index % frame_interval != 0:
frame_index += 1
continue
# 降低分辨率
if target_width != original_width:
frame = cv2.resize(frame, (target_width, target_height))
# 检测
start_time = time.time()
detections = self.detect_frame(frame)
processing_time = time.time() - start_time
total_time += processing_time
processed_count += 1
# 显示实时统计
if processed_count % 10 == 0:
current_fps = processed_count / total_time
print(f"已处理 {processed_count} 帧 | "
f"当前FPS: {current_fps:.1f} | "
f"最近处理时间: {processing_time*1000:.1f}ms")
frame_index += 1
cap.release()
avg_fps = processed_count / total_time if total_time > 0 else 0
print(f"\n优化处理完成:")
print(f" 处理总帧数: {processed_count}")
print(f" 总处理时间: {total_time:.1f}秒")
print(f) 平均帧率: {avg_fps:.1f} FPS")
print(f" 性能提升: {avg_fps/target_fps*100:.1f}% of target")
5. API集成与生产部署
5.1 构建完整的视频处理API服务
将视频处理功能封装为API服务，方便其他系统调用：
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
import uvicorn
import tempfile
import os
app = FastAPI(title="YOLO12 Video Processing API")
@app.post("/process_video")
async def process_video(
file: UploadFile = File(...),
frame_interval: int = 1,
confidence_threshold: float = 0.25,
output_format: str = "json"
):
"""
处理上传的视频文件
参数:
file: 视频文件
frame_interval: 帧采样间隔
confidence_threshold: 置信度阈值
output_format: 输出格式 (json/video)
"""
# 保存上传的文件
with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp_file:
content = await file.read()
tmp_file.write(content)
video_path = tmp_file.name
try:
# 创建处理器实例
processor = VideoProcessor()
# 处理视频
results = processor.process_video(
video_path=video_path,
frame_interval=frame_interval
)
# 根据阈值过滤结果
filtered_results = []
for frame_result in results:
filtered_detections = [
det for det in frame_result["detections"].get("predictions", [])
if det.get("confidence", 0) >= confidence_threshold
]
frame_result["detections"]["predictions"] = filtered_detections
filtered_results.append(frame_result)
# 返回结果
if output_format == "json":
return JSONResponse(content={
"status": "success",
"video_info": {
"frames_processed": len(filtered_results),
"total_detections": sum(
len(frame["detections"]["predictions"])
for frame in filtered_results
),
"processing_time": processor.total_processing_time
},
"detections": filtered_results
})
else:
# 生成处理后的视频文件
output_path = video_path.replace(".mp4", "_processed.mp4")
processor.process_video(
video_path=video_path,
output_path=output_path,
frame_interval=frame_interval
)
# 返回视频文件
def iterfile():
with open(output_path, mode="rb") as f:
yield from f
return StreamingResponse(
iterfile(),
media_type="video/mp4",
headers={"Content-Disposition": f"attachment; filename=processed_{file.filename}"}
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
# 清理临时文件
if os.path.exists(video_path):
os.unlink(video_path)
@app.post("/realtime_stream")
async def realtime_stream(
camera_url: str,
duration: int = 30,
model_size: str = "nano"
):
"""
处理实时视频流
参数:
camera_url: 摄像头RTSP地址或本地摄像头ID
duration: 处理时长（秒）
model_size: 模型规格 (nano/small/medium/large/xlarge)
"""
try:
# 根据模型规格选择API端点
api_url = f"http://localhost:8000/predict?model={model_size}"
processor = CameraProcessor(api_url=api_url)
# 启动实时处理
camera_id = int(camera_url) if camera_url.isdigit() else camera_url
processor.start_stream(duration_seconds=duration, show_preview=False)
return JSONResponse(content={
"status": "success",
"message": f"实时流处理已启动，将持续{duration}秒",
"camera": camera_url,
"model": model_size
})
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/system_status")
async def system_status():
"""获取系统状态信息"""
import psutil
import GPUtil
# 获取CPU和内存信息
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
# 获取GPU信息（如果可用）
gpu_info = []
try:
gpus = GPUtil.getGPUs()
for gpu in gpus:
gpu_info.append({
"name": gpu.name,
"load": gpu.load * 100,
"memory_used": gpu.memoryUsed,
"memory_total": gpu.memoryTotal,
"temperature": gpu.temperature
})
except:
gpu_info = []
return JSONResponse(content={
"cpu_usage": cpu_percent,
"memory_usage": memory.percent,
"memory_used_gb": memory.used / (1024**3),
"memory_total_gb": memory.total / (1024**3),
"gpus": gpu_info,
"service": "YOLO12 Video Processing API",
"status": "running"
})
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8001)
5.2 生产环境部署配置
对于生产环境，需要添加监控、日志和错误处理：
# config.py - 配置文件
import os
from datetime import datetime
class Config:
# API配置
API_HOST = os.getenv("API_HOST", "0.0.0.0")
API_PORT = int(os.getenv("API_PORT", 8001))
# YOLO12服务配置
YOLO_API_URL = os.getenv("YOLO_API_URL", "http://localhost:8000/predict")
# 视频处理配置
MAX_VIDEO_SIZE = int(os.getenv("MAX_VIDEO_SIZE", 100 * 1024 * 1024)) # 100MB
SUPPORTED_FORMATS = [".mp4", ".avi", ".mov", ".mkv"]
MAX_PROCESSING_TIME = int(os.getenv("MAX_PROCESSING_TIME", 300)) # 5分钟
# 性能配置
DEFAULT_FRAME_INTERVAL = int(os.getenv("DEFAULT_FRAME_INTERVAL", 1))
DEFAULT_CONFIDENCE = float(os.getenv("DEFAULT_CONFIDENCE", 0.25))
MAX_WORKERS = int(os.getenv("MAX_WORKERS", 4))
# 日志配置
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
LOG_FILE = os.getenv("LOG_FILE", f"logs/yolo12_video_{datetime.now().strftime('%Y%m%d')}.log")
# 存储配置
UPLOAD_FOLDER = os.getenv("UPLOAD_FOLDER", "/tmp/yolo12_uploads")
OUTPUT_FOLDER = os.getenv("OUTPUT_FOLDER", "/tmp/yolo12_outputs")
@classmethod
def validate(cls):
"""验证配置"""
os.makedirs(cls.UPLOAD_FOLDER, exist_ok=True)
os.makedirs(cls.OUTPUT_FOLDER, exist_ok=True)
os.makedirs(os.path.dirname(cls.LOG_FILE), exist_ok=True)
print(f"配置加载完成:")
print(f" API地址: {cls.API_HOST}:{cls.API_PORT}")
print(f" YOLO服务: {cls.YOLO_API_URL}")
print(f" 上传目录: {cls.UPLOAD_FOLDER}")
print(f) 输出目录: {cls.OUTPUT_FOLDER}")
# monitor.py - 监控模块
import time
import threading
import logging
from dataclasses import dataclass
from typing import Dict, List
from datetime import datetime
@dataclass
class ProcessingMetrics:
"""处理指标"""
start_time: float
end_time: float = 0
frames_processed: int = 0
total_detections: int = 0
avg_processing_time: float = 0
errors: List[str] = None
def __post_init__(self):
if self.errors is None:
self.errors = []
def complete(self):
"""标记处理完成"""
self.end_time = time.time()
@property
def duration(self):
"""处理时长"""
if self.end_time > 0:
return self.end_time - self.start_time
return time.time() - self.start_time
@property
def fps(self):
"""处理帧率"""
if self.duration > 0:
return self.frames_processed / self.duration
return 0
class PerformanceMonitor:
"""性能监控器"""
def __init__(self):
self.metrics: Dict[str, ProcessingMetrics] = {}
self.lock = threading.Lock()
self.logger = logging.getLogger(__name__)
def start_job(self, job_id: str):
"""开始监控任务"""
with self.lock:
self.metrics[job_id] = ProcessingMetrics(start_time=time.time())
self.logger.info(f"开始监控任务: {job_id}")
def update_job(self, job_id: str, frames: int = 1, detections: int = 0):
"""更新任务指标"""
with self.lock:
if job_id in self.metrics:
self.metrics[job_id].frames_processed += frames
self.metrics[job_id].total_detections += detections
def add_error(self, job_id: str, error: str):
"""记录错误"""
with self.lock:
if job_id in self.metrics:
self.metrics[job_id].errors.append(error)
self.logger.error(f"任务 {job_id} 错误: {error}")
def complete_job(self, job_id: str):
"""完成任务监控"""
with self.lock:
if job_id in self.metrics:
self.metrics[job_id].complete()
metrics = self.metrics[job_id]
self.logger.info(
f"任务完成: {job_id} | "
f"时长: {metrics.duration:.1f}s | "
f"帧数: {metrics.frames_processed} | "
f"FPS: {metrics.fps:.1f} | "
f"检测数: {metrics.total_detections}"
)
def get_metrics(self, job_id: str) -> Dict:
"""获取任务指标"""
with self.lock:
if job_id in self.metrics:
m = self.metrics[job_id]
return {
"job_id": job_id,
"duration": m.duration,
"frames_processed": m.frames_processed,
"total_detections": m.total_detections,
"fps": m.fps,
"errors": m.errors,
"is_completed": m.end_time > 0
}
return {}
def get_all_metrics(self) -> List[Dict]:
"""获取所有任务指标"""
with self.lock:
return [self.get_metrics(job_id) for job_id in self.metrics.keys()]
# 全局监控器实例
monitor = PerformanceMonitor()
6. 实际应用场景与优化建议
6.1 安防监控系统集成
对于安防监控场景，需要特别关注实时性和准确性：
class SecurityMonitor:
"""安防监控专用处理器"""
def __init__(self, api_url="http://localhost:8000/predict"):
self.api_url = api_url
self.alert_rules = {
"person": {"min_confidence": 0.7, "alert_message": "检测到人员入侵"},
"car": {"min_confidence": 0.8, "alert_message": "检测到车辆进入"},
"knife": {"min_confidence": 0.6, "alert_message": "检测到危险物品"},
"gun": {"min_confidence": 0.6, "alert_message": "检测到武器"}
}
self.alert_history = []
def process_security_feed(self, video_source, alert_callback=None):
"""
处理安防视频流，触发警报
参数:
video_source: 视频源（摄像头ID或RTSP地址）
alert_callback: 警报回调函数
"""
processor = CameraProcessor(api_url=self.api_url)
def process_frame_with_alerts(frame, detections):
"""处理帧并检查警报"""
alerts = []
predictions = detections.get("predictions", [])
for pred in predictions:
class_name = pred.get("class", "")
confidence = pred.get("confidence", 0)
# 检查是否符合警报规则
if class_name in self.alert_rules:
rule = self.alert_rules[class_name]
if confidence >= rule["min_confidence"]:
alert = {
"timestamp": datetime.now().isoformat(),
"class": class_name,
"confidence": confidence,
"message": rule["alert_message"],
"bbox": pred.get("bbox", [])
}
alerts.append(alert)
# 记录警报历史
self.alert_history.append(alert)
# 触发回调
if alert_callback:
alert_callback(alert)
# 在帧上绘制警报信息
if alerts:
frame = self.draw_alerts(frame, alerts)
return frame, alerts
# 修改CameraProcessor的绘制逻辑
original_draw = processor.draw_detections
def custom_draw(frame, detections):
frame, alerts = process_frame_with_alerts(frame, detections)
frame = original_draw(frame, detections)
# 在帧上添加警报计数
if alerts:
alert_text = f"Alerts: {len(alerts)}"
cv2.putText(frame, alert_text, (10, 90),
cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 0, 255), 2)
return frame
processor.draw_detections = custom_draw
# 开始处理
processor.start_stream(duration_seconds=None, show_preview=True)
def draw_alerts(self, frame, alerts):
"""在帧上绘制警报信息"""
for i, alert in enumerate(alerts):
# 绘制红色边界框
bbox = alert.get("bbox", [])
if len(bbox) == 4:
x1, y1, x2, y2 = map(int, bbox)
cv2.rectangle(frame, (x1, y1), (x2, y2), (0, 0, 255), 3)
# 添加警报文本
alert_text = f"ALERT: {alert['class']} ({alert['confidence']:.2f})"
cv2.putText(frame, alert_text, (x1, y1 - 10),
cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2)
return frame
def get_alert_summary(self, hours=24):
"""获取指定时间内的警报摘要"""
cutoff_time = datetime.now().timestamp() - (hours * 3600)
recent_alerts = [
alert for alert in self.alert_history
if datetime.fromisoformat(alert["timestamp"]).timestamp() > cutoff_time
]
summary = {}
for alert in recent_alerts:
class_name = alert["class"]
if class_name not in summary:
summary[class_name] = 0
summary[class_name] += 1
return {
"total_alerts": len(recent_alerts),
"by_class": summary,
"time_period_hours": hours
}
# 使用示例
def alert_handler(alert):
"""警报处理函数"""
print(f"[ALERT] {alert['timestamp']}: {alert['message']}")
print(f" 类别: {alert['class']}, 置信度: {alert['confidence']:.2f}")
# 这里可以添加实际警报逻辑，如：
# - 发送邮件通知
# - 触发声音警报
# - 保存截图
# - 调用其他系统API
if __name__ == "__main__":
monitor = SecurityMonitor()
monitor.process_security_feed(
video_source=0, # 默认摄像头
alert_callback=alert_handler
)
6.2 性能优化建议总结
根据实际使用经验，以下优化建议可以显著提升视频处理性能：
-
选择合适的模型规格
- 实时监控：使用nano或small版本
- 离线分析：根据精度要求选择medium/large版本
- 边缘设备：必须使用nano版本
-
调整处理参数
- 降低输入分辨率（如从1080p降到720p）
- 增加帧采样间隔（如每2帧处理1帧）
- 调整置信度阈值，减少后处理时间
-
硬件优化
- 使用GPU加速（确保CUDA正确配置）
- 增加系统内存，减少磁盘IO
- 使用SSD存储，加快视频读取速度
-
软件优化
- 使用多线程/多进程并行处理
- 批量处理API请求，减少网络开销
- 缓存检测结果，避免重复计算
7. 总结
通过本文的详细指南，你已经掌握了使用YOLO12进行实时视频流处理的核心技术。从基础的逐帧检测到高级的多线程优化，从简单的脚本到完整的API服务，这些方案可以满足不同场景的需求。
关键要点回顾：
- 逐帧处理是基础：通过OpenCV读取视频帧，调用YOLO12 API进行检测，再重新组合成视频
- 性能优化很重要：多线程处理、帧采样、分辨率调整都能显著提升处理速度
- API集成是关键：将视频处理功能封装为API服务，方便系统集成和扩展
- 实际应用需定制：不同场景（安防、质检、分析）需要不同的处理逻辑和优化策略
YOLO12的实时目标检测能力为视频分析应用提供了强大的基础。无论是构建安防监控系统、开发智能交通应用，还是创建内容分析平台，这套方案都能为你提供可靠的技术支持。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
9
0- 0
已为社区贡献114条内容
所有评论(0)