BodyBalanceEvaluation/backend/test_opencv_backends.py

305 lines
10 KiB
Python
Raw Permalink Normal View History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
OpenCV后端性能测试脚本
测试不同OpenCV后端DirectShow vs MSMF的性能差异
"""
import sys
import os
import time
import logging
import cv2
# 添加项目路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_opencv_backend(backend_name, backend_id, device_index=0, width=1280, height=720, fps=30):
"""
测试指定OpenCV后端的性能
Args:
backend_name: 后端名称
backend_id: 后端ID
device_index: 设备索引
width: 宽度
height: 高度
fps: 帧率
Returns:
dict: 性能数据
"""
print(f"\n{'='*70}")
print(f"测试 {backend_name} 后端 (ID: {backend_id})")
print(f"分辨率: {width}x{height}, FPS: {fps}")
print(f"{'='*70}")
result = {
'backend_name': backend_name,
'backend_id': backend_id,
'success': False,
'init_time': -1,
'config_time': -1,
'first_frame_time': -1,
'total_time': -1,
'actual_resolution': 'N/A',
'error': None
}
cap = None
try:
# 1. 测试相机初始化时间
print(f"📷 初始化相机 (后端: {backend_name})...")
init_start = time.time()
# 创建VideoCapture对象并指定后端
cap = cv2.VideoCapture(device_index, backend_id)
if not cap.isOpened():
print(f"❌ 无法打开相机 (后端: {backend_name})")
result['error'] = f"无法打开相机 (后端: {backend_name})"
return result
init_time = (time.time() - init_start) * 1000
result['init_time'] = init_time
print(f"✅ 相机初始化成功: {init_time:.1f}ms")
# 2. 测试配置时间
print(f"⚙️ 配置相机参数...")
config_start = time.time()
# 设置分辨率和帧率
cap.set(cv2.CAP_PROP_FRAME_WIDTH, width)
cap.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
cap.set(cv2.CAP_PROP_FPS, fps)
# 设置缓冲区大小
cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
# 性能优化设置
try:
cap.set(cv2.CAP_PROP_AUTO_EXPOSURE, 0.25) # 手动曝光
cap.set(cv2.CAP_PROP_AUTO_WB, 0) # 禁用自动白平衡
cap.set(cv2.CAP_PROP_EXPOSURE, -6) # 设置曝光值
except Exception as e:
print(f"⚠️ 性能优化设置警告: {e}")
config_time = (time.time() - config_start) * 1000
result['config_time'] = config_time
print(f"✅ 配置完成: {config_time:.1f}ms")
# 3. 获取实际分辨率
actual_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
actual_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
actual_fps = cap.get(cv2.CAP_PROP_FPS)
result['actual_resolution'] = f"{actual_width}x{actual_height}@{actual_fps:.1f}fps"
print(f"🎯 实际参数: {actual_width}x{actual_height}, FPS: {actual_fps:.1f}")
# 4. 测试首帧获取时间
print(f"🖼️ 获取首帧...")
frame_start = time.time()
ret, frame = cap.read()
if ret and frame is not None:
first_frame_time = (time.time() - frame_start) * 1000
result['first_frame_time'] = first_frame_time
print(f"✅ 首帧获取成功: {first_frame_time:.1f}ms, 帧大小: {frame.shape}")
else:
print(f"❌ 首帧获取失败")
result['error'] = "首帧获取失败"
return result
# 5. 计算总时间
total_time = init_time + config_time + first_frame_time
result['total_time'] = total_time
result['success'] = True
print(f"📊 总耗时: {total_time:.1f}ms ({total_time/1000:.2f}秒)")
# 6. 测试连续帧获取性能
print(f"🎬 测试连续帧获取性能...")
frame_times = []
test_frames = 10
for i in range(test_frames):
frame_start = time.time()
ret, frame = cap.read()
if ret:
frame_time = (time.time() - frame_start) * 1000
frame_times.append(frame_time)
else:
break
if frame_times:
avg_frame_time = sum(frame_times) / len(frame_times)
max_frame_time = max(frame_times)
min_frame_time = min(frame_times)
print(f"📈 连续帧性能: 平均 {avg_frame_time:.1f}ms, 最快 {min_frame_time:.1f}ms, 最慢 {max_frame_time:.1f}ms")
result['avg_frame_time'] = avg_frame_time
result['max_frame_time'] = max_frame_time
result['min_frame_time'] = min_frame_time
return result
except Exception as e:
print(f"❌ 测试异常: {e}")
result['error'] = str(e)
return result
finally:
if cap:
cap.release()
print(f"🧹 相机资源已释放")
def get_available_backends():
"""
获取可用的OpenCV后端
Returns:
list: 可用后端列表
"""
backends = [
('DirectShow', cv2.CAP_DSHOW),
('MSMF', cv2.CAP_MSMF),
('V4L2', cv2.CAP_V4L2),
('GStreamer', cv2.CAP_GSTREAMER),
('Any', cv2.CAP_ANY)
]
available_backends = []
for name, backend_id in backends:
try:
# 尝试创建VideoCapture对象
cap = cv2.VideoCapture(0, backend_id)
if cap.isOpened():
available_backends.append((name, backend_id))
cap.release()
except Exception:
pass
return available_backends
def main():
"""
主测试函数
"""
print("🚀 OpenCV后端性能测试")
print(f"OpenCV版本: {cv2.__version__}")
# 获取可用后端
print("\n🔍 检测可用的OpenCV后端...")
available_backends = get_available_backends()
if not available_backends:
print("❌ 未找到可用的相机后端")
return
print(f"✅ 找到 {len(available_backends)} 个可用后端:")
for name, backend_id in available_backends:
print(f" - {name} (ID: {backend_id})")
# 测试参数
test_params = {
'device_index': 0,
'width': 1280,
'height': 720,
'fps': 30
}
print(f"\n📋 测试参数: {test_params['width']}x{test_params['height']}@{test_params['fps']}fps")
# 执行测试
results = []
for backend_name, backend_id in available_backends:
result = test_opencv_backend(
backend_name,
backend_id,
**test_params
)
results.append(result)
# 等待一下,避免设备冲突
time.sleep(2)
# 输出汇总结果
print(f"\n{'='*90}")
print("📈 OpenCV后端性能测试汇总")
print(f"{'='*90}")
print(f"{'后端':<12} {'状态':<8} {'初始化':<10} {'配置':<10} {'首帧':<10} {'总计':<10} {'实际分辨率':<20}")
print("-" * 90)
successful_results = []
for result in results:
status = "✅成功" if result['success'] else "❌失败"
init_time = f"{result['init_time']:.1f}ms" if result['init_time'] > 0 else "N/A"
config_time = f"{result['config_time']:.1f}ms" if result['config_time'] > 0 else "N/A"
frame_time = f"{result['first_frame_time']:.1f}ms" if result['first_frame_time'] > 0 else "N/A"
total_time = f"{result['total_time']:.1f}ms" if result['total_time'] > 0 else "N/A"
print(f"{result['backend_name']:<12} {status:<8} {init_time:<10} {config_time:<10} {frame_time:<10} {total_time:<10} {result['actual_resolution']:<20}")
if result['success']:
successful_results.append(result)
# 性能分析
if len(successful_results) >= 2:
print(f"\n📊 性能分析:")
# 找到最快和最慢的后端
fastest = min(successful_results, key=lambda x: x['total_time'])
slowest = max(successful_results, key=lambda x: x['total_time'])
print(f"🏆 最快后端: {fastest['backend_name']} - {fastest['total_time']:.1f}ms")
print(f"🐌 最慢后端: {slowest['backend_name']} - {slowest['total_time']:.1f}ms")
if slowest['total_time'] > fastest['total_time']:
improvement = ((slowest['total_time'] - fastest['total_time']) / slowest['total_time']) * 100
print(f"💡 性能提升: {improvement:.1f}% (使用最快后端)")
# 详细对比
print(f"\n📋 详细性能对比:")
for result in successful_results:
if result != fastest:
if result['total_time'] > fastest['total_time']:
slower = ((result['total_time'] - fastest['total_time']) / fastest['total_time']) * 100
print(f" {result['backend_name']}: 比最快后端慢 {slower:.1f}% ({result['total_time']:.1f}ms vs {fastest['total_time']:.1f}ms)")
elif len(successful_results) == 1:
result = successful_results[0]
print(f"\n📊 只有一个后端测试成功: {result['backend_name']} - {result['total_time']:.1f}ms")
# 推荐建议
print(f"\n🎯 建议:")
if successful_results:
fastest = min(successful_results, key=lambda x: x['total_time'])
print(f"✅ 推荐使用 {fastest['backend_name']} 后端以获得最佳性能")
print(f"📝 配置建议: 在相机初始化时指定后端 cv2.VideoCapture(device_index, cv2.CAP_{fastest['backend_name'].upper()})")
if fastest['total_time'] < 3000:
print(f"🚀 性能评级: 优秀 (< 3秒)")
elif fastest['total_time'] < 5000:
print(f"⚡ 性能评级: 良好 (< 5秒)")
else:
print(f"⚠️ 性能评级: 需要优化 (> 5秒)")
else:
print(f"❌ 所有后端测试都失败了,请检查相机连接和驱动")
print(f"\n{'='*90}")
print("测试完成")
print(f"{'='*90}")
if __name__ == "__main__":
main()