feat(ffmpeg): 实现智能硬件加速检测和编码器选择

添加智能硬件加速检测功能,支持多平台和渐进式降级
优化编码器选择逻辑,根据硬件类型自动选择最优编码器
增加测试视频生成和清理功能,用于硬件加速兼容性测试
支持强制软件编码模式,提供更可靠的备选方案
This commit is contained in:
linyq 2025-07-02 18:35:49 +08:00
parent 06148a6b8c
commit 57601e164f
4 changed files with 645 additions and 81 deletions

View File

@ -402,18 +402,36 @@ def save_clip_video(timestamp: str, origin_video: str, save_dir: str = "") -> st
ffmpeg_start_time = start_str.replace(',', '.') ffmpeg_start_time = start_str.replace(',', '.')
ffmpeg_end_time = end_str.replace(',', '.') ffmpeg_end_time = end_str.replace(',', '.')
# 构建FFmpeg命令 # 构建FFmpeg命令 - 使用新的智能编码器选择
encoder = ffmpeg_utils.get_optimal_ffmpeg_encoder()
ffmpeg_cmd = [ ffmpeg_cmd = [
"ffmpeg", "-y", *hwaccel_args, "ffmpeg", "-y", *hwaccel_args,
"-i", origin_video, "-i", origin_video,
"-ss", ffmpeg_start_time, "-ss", ffmpeg_start_time,
"-to", ffmpeg_end_time, "-to", ffmpeg_end_time,
"-c:v", "h264_videotoolbox" if hwaccel == "videotoolbox" else "libx264", "-c:v", encoder,
"-c:a", "aac", "-c:a", "aac",
"-strict", "experimental", "-strict", "experimental",
video_path video_path
] ]
# 根据编码器类型添加特定参数
if "nvenc" in encoder:
ffmpeg_cmd.insert(-1, "-preset")
ffmpeg_cmd.insert(-1, "medium")
elif "videotoolbox" in encoder:
ffmpeg_cmd.insert(-1, "-profile:v")
ffmpeg_cmd.insert(-1, "high")
elif "qsv" in encoder:
ffmpeg_cmd.insert(-1, "-preset")
ffmpeg_cmd.insert(-1, "medium")
elif encoder == "libx264":
ffmpeg_cmd.insert(-1, "-preset")
ffmpeg_cmd.insert(-1, "medium")
ffmpeg_cmd.insert(-1, "-crf")
ffmpeg_cmd.insert(-1, "23")
# 执行FFmpeg命令 # 执行FFmpeg命令
# logger.info(f"裁剪视频片段: {timestamp} -> {ffmpeg_start_time}到{ffmpeg_end_time}") # logger.info(f"裁剪视频片段: {timestamp} -> {ffmpeg_start_time}到{ffmpeg_end_time}")
# logger.debug(f"执行命令: {' '.join(ffmpeg_cmd)}") # logger.debug(f"执行命令: {' '.join(ffmpeg_cmd)}")

View File

@ -64,7 +64,7 @@ def get_hardware_acceleration_option() -> Optional[str]:
Returns: Returns:
Optional[str]: 硬件加速参数如果不支持则返回None Optional[str]: 硬件加速参数如果不支持则返回None
""" """
# 使用集中式硬件加速检测 # 使用新的硬件加速检测API
return ffmpeg_utils.get_ffmpeg_hwaccel_type() return ffmpeg_utils.get_ffmpeg_hwaccel_type()
@ -178,14 +178,20 @@ def process_single_video(
logger.warning(f"视频探测出错,禁用硬件加速: {str(e)}") logger.warning(f"视频探测出错,禁用硬件加速: {str(e)}")
hwaccel = None hwaccel = None
# 添加硬件加速参数(根据前面的安全检查可能已经被禁用 # 添加硬件加速参数(使用新的智能检测机制
if hwaccel: if hwaccel:
try: try:
# 使用集中式硬件加速参数 # 使用新的硬件加速检测API
hwaccel_args = ffmpeg_utils.get_ffmpeg_hwaccel_args() hwaccel_args = ffmpeg_utils.get_ffmpeg_hwaccel_args()
command.extend(hwaccel_args) if hwaccel_args:
command.extend(hwaccel_args)
logger.debug(f"应用硬件加速参数: {hwaccel_args}")
else:
logger.info("硬件加速不可用,将使用软件编码")
hwaccel = False # 标记为不使用硬件加速
except Exception as e: except Exception as e:
logger.warning(f"应用硬件加速参数时出错: {str(e)},将使用软件编码") logger.warning(f"应用硬件加速参数时出错: {str(e)},将使用软件编码")
hwaccel = False # 标记为不使用硬件加速
# 重置命令,移除可能添加了一半的硬件加速参数 # 重置命令,移除可能添加了一半的硬件加速参数
command = ['ffmpeg', '-y'] command = ['ffmpeg', '-y']
@ -212,41 +218,27 @@ def process_single_video(
'-r', '30', # 设置帧率为30fps '-r', '30', # 设置帧率为30fps
]) ])
# 选择编码器 - 考虑到Windows和特定硬件的兼容性 # 选择编码器 - 使用新的智能编码器选择
use_software_encoder = True encoder = ffmpeg_utils.get_optimal_ffmpeg_encoder()
if hwaccel: if hwaccel and encoder != "libx264":
# 获取硬件加速类型和编码器信息 logger.info(f"使用硬件编码器: {encoder}")
hwaccel_type = ffmpeg_utils.get_ffmpeg_hwaccel_type() command.extend(['-c:v', encoder])
hwaccel_encoder = ffmpeg_utils.get_ffmpeg_hwaccel_encoder()
if hwaccel_type == 'cuda' or hwaccel_type == 'nvenc': # 根据编码器类型添加特定参数
try: if "nvenc" in encoder:
# 检查NVENC编码器是否可用 command.extend(['-preset', 'p4', '-profile:v', 'high'])
encoders_cmd = subprocess.run( elif "videotoolbox" in encoder:
["ffmpeg", "-hide_banner", "-encoders"], command.extend(['-profile:v', 'high'])
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False elif "qsv" in encoder:
) command.extend(['-preset', 'medium'])
elif "vaapi" in encoder:
if "h264_nvenc" in encoders_cmd.stdout.lower(): command.extend(['-profile', '100'])
command.extend(['-c:v', 'h264_nvenc', '-preset', 'p4', '-profile:v', 'high']) elif "amf" in encoder:
use_software_encoder = False command.extend(['-quality', 'balanced'])
else: else:
logger.warning("NVENC编码器不可用将使用软件编码") command.extend(['-preset', 'medium', '-profile:v', 'high'])
except Exception as e: else:
logger.warning(f"NVENC编码器检测失败: {str(e)},将使用软件编码")
elif hwaccel_type == 'qsv':
command.extend(['-c:v', 'h264_qsv', '-preset', 'medium'])
use_software_encoder = False
elif hwaccel_type == 'videotoolbox': # macOS
command.extend(['-c:v', 'h264_videotoolbox', '-profile:v', 'high'])
use_software_encoder = False
elif hwaccel_type == 'vaapi': # Linux VA-API
command.extend(['-c:v', 'h264_vaapi', '-profile', '100'])
use_software_encoder = False
# 如果前面的条件未能应用硬件编码器,使用软件编码
if use_software_encoder:
logger.info("使用软件编码器(libx264)") logger.info("使用软件编码器(libx264)")
command.extend(['-c:v', 'libx264', '-preset', 'medium', '-profile:v', 'high']) command.extend(['-c:v', 'libx264', '-preset', 'medium', '-profile:v', 'high'])
@ -273,8 +265,11 @@ def process_single_video(
# 如果使用硬件加速失败,尝试使用软件编码 # 如果使用硬件加速失败,尝试使用软件编码
if hwaccel: if hwaccel:
logger.info("尝试使用软件编码作为备选方案") logger.info("硬件加速失败,尝试使用软件编码作为备选方案")
try: try:
# 强制使用软件编码
ffmpeg_utils.force_software_encoding()
# 构建新的命令,使用软件编码 # 构建新的命令,使用软件编码
fallback_cmd = ['ffmpeg', '-y', '-i', input_path] fallback_cmd = ['ffmpeg', '-y', '-i', input_path]
@ -302,14 +297,30 @@ def process_single_video(
output_path output_path
]) ])
logger.info(f"执行备选FFmpeg命令: {' '.join(fallback_cmd)}") logger.info("执行软件编码备选方案")
subprocess.run(fallback_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) subprocess.run(fallback_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logger.info(f"使用软件编码成功处理视频: {output_path}") logger.info(f"使用软件编码成功处理视频: {output_path}")
return output_path return output_path
except subprocess.CalledProcessError as fallback_error: except subprocess.CalledProcessError as fallback_error:
fallback_error_msg = fallback_error.stderr.decode() if fallback_error.stderr else str(fallback_error) fallback_error_msg = fallback_error.stderr.decode() if fallback_error.stderr else str(fallback_error)
logger.error(f"备选软件编码也失败: {fallback_error_msg}") logger.error(f"软件编码备选方案也失败: {fallback_error_msg}")
raise RuntimeError(f"无法处理视频 {input_path}: 硬件加速和软件编码都失败")
# 尝试最基本的编码参数
try:
logger.info("尝试最基本的编码参数")
basic_cmd = [
'ffmpeg', '-y', '-i', input_path,
'-c:v', 'libx264', '-preset', 'ultrafast',
'-crf', '23', '-pix_fmt', 'yuv420p',
output_path
]
subprocess.run(basic_cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
logger.info(f"使用基本编码参数成功处理视频: {output_path}")
return output_path
except subprocess.CalledProcessError as basic_error:
basic_error_msg = basic_error.stderr.decode() if basic_error.stderr else str(basic_error)
logger.error(f"基本编码参数也失败: {basic_error_msg}")
raise RuntimeError(f"无法处理视频 {input_path}: 所有编码方案都失败")
# 如果不是硬件加速导致的问题,或者备选方案也失败了,抛出原始错误 # 如果不是硬件加速导致的问题,或者备选方案也失败了,抛出原始错误
raise RuntimeError(f"处理视频失败: {error_msg}") raise RuntimeError(f"处理视频失败: {error_msg}")

View File

@ -366,15 +366,46 @@ def generate_video_v3(
else: else:
logger.warning("没有音频轨道需要合成") logger.warning("没有音频轨道需要合成")
# 导出视频 # 导出视频 - 使用优化的编码器
logger.info("开始导出视频...") # 调试信息 logger.info("开始导出视频...")
final_video.write_videofile(
output_path, # 获取最优编码器
codec='libx264', from app.utils import ffmpeg_utils
audio_codec='aac', optimal_encoder = ffmpeg_utils.get_optimal_ffmpeg_encoder()
fps=video.fps
) # 根据编码器类型设置参数
logger.info(f"视频已导出到: {output_path}") # 调试信息 ffmpeg_params = []
if "nvenc" in optimal_encoder:
ffmpeg_params = ['-preset', 'medium', '-profile:v', 'high']
elif "videotoolbox" in optimal_encoder:
ffmpeg_params = ['-profile:v', 'high']
elif "qsv" in optimal_encoder:
ffmpeg_params = ['-preset', 'medium']
elif "vaapi" in optimal_encoder:
ffmpeg_params = ['-profile', '100']
elif optimal_encoder == "libx264":
ffmpeg_params = ['-preset', 'medium', '-crf', '23']
try:
final_video.write_videofile(
output_path,
codec=optimal_encoder,
audio_codec='aac',
fps=video.fps,
ffmpeg_params=ffmpeg_params
)
logger.info(f"视频已导出到: {output_path} (使用编码器: {optimal_encoder})")
except Exception as e:
logger.warning(f"使用 {optimal_encoder} 编码器失败: {str(e)}, 尝试软件编码")
# 降级到软件编码
final_video.write_videofile(
output_path,
codec='libx264',
audio_codec='aac',
fps=video.fps,
ffmpeg_params=['-preset', 'medium', '-crf', '23']
)
logger.info(f"视频已导出到: {output_path} (使用软件编码)")
# 清理资源 # 清理资源
video.close() video.close()

View File

@ -1,9 +1,11 @@
""" """
FFmpeg 工具模块 - 提供 FFmpeg 相关的工具函数特别是硬件加速检测 FFmpeg 工具模块 - 提供 FFmpeg 相关的工具函数特别是硬件加速检测
优化多平台兼容性支持渐进式降级和智能错误处理
""" """
import os import os
import platform import platform
import subprocess import subprocess
import tempfile
from typing import Dict, List, Optional, Tuple, Union from typing import Dict, List, Optional, Tuple, Union
from loguru import logger from loguru import logger
@ -14,9 +16,104 @@ _FFMPEG_HW_ACCEL_INFO = {
"encoder": None, "encoder": None,
"hwaccel_args": [], "hwaccel_args": [],
"message": "", "message": "",
"is_dedicated_gpu": False "is_dedicated_gpu": False,
"fallback_available": False, # 是否有备用方案
"fallback_encoder": None, # 备用编码器
"platform": None, # 平台信息
"gpu_vendor": None, # GPU厂商
"tested_methods": [] # 已测试的方法
} }
# 硬件加速优先级配置按平台和GPU类型
HWACCEL_PRIORITY = {
"windows": {
"nvidia": ["cuda", "nvenc", "d3d11va", "dxva2"],
"amd": ["d3d11va", "dxva2", "amf"], # 不再完全禁用AMD
"intel": ["qsv", "d3d11va", "dxva2"],
"unknown": ["d3d11va", "dxva2"]
},
"darwin": {
"apple": ["videotoolbox"],
"nvidia": ["cuda", "videotoolbox"],
"amd": ["videotoolbox"],
"intel": ["videotoolbox"],
"unknown": ["videotoolbox"]
},
"linux": {
"nvidia": ["cuda", "nvenc", "vaapi"],
"amd": ["vaapi", "amf"],
"intel": ["qsv", "vaapi"],
"unknown": ["vaapi"]
}
}
# 编码器映射
ENCODER_MAPPING = {
"cuda": "h264_nvenc",
"nvenc": "h264_nvenc",
"videotoolbox": "h264_videotoolbox",
"qsv": "h264_qsv",
"vaapi": "h264_vaapi",
"amf": "h264_amf",
"d3d11va": "libx264", # D3D11VA只用于解码
"dxva2": "libx264", # DXVA2只用于解码
"software": "libx264"
}
def get_null_input() -> str:
"""
获取平台特定的空输入文件路径
Returns:
str: 平台特定的空输入路径
"""
system = platform.system().lower()
if system == "windows":
return "NUL"
else:
return "/dev/null"
def create_test_video() -> str:
"""
创建一个临时的测试视频文件用于硬件加速测试
Returns:
str: 临时测试视频文件路径
"""
try:
# 创建临时文件
temp_file = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
temp_path = temp_file.name
temp_file.close()
# 生成一个简单的测试视频1秒黑色画面
cmd = [
'ffmpeg', '-y', '-f', 'lavfi', '-i', 'color=black:size=320x240:duration=1',
'-c:v', 'libx264', '-pix_fmt', 'yuv420p', '-t', '1', temp_path
]
subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
return temp_path
except Exception as e:
logger.debug(f"创建测试视频失败: {str(e)}")
return get_null_input()
def cleanup_test_video(path: str) -> None:
"""
清理测试视频文件
Args:
path: 测试视频文件路径
"""
try:
if path != get_null_input() and os.path.exists(path):
os.unlink(path)
except Exception as e:
logger.debug(f"清理测试视频失败: {str(e)}")
def check_ffmpeg_installation() -> bool: def check_ffmpeg_installation() -> bool:
""" """
@ -38,9 +135,123 @@ def check_ffmpeg_installation() -> bool:
return False return False
def detect_gpu_vendor() -> str:
"""
检测GPU厂商
Returns:
str: GPU厂商 (nvidia, amd, intel, apple, unknown)
"""
system = platform.system().lower()
try:
if system == "windows":
gpu_info = _get_windows_gpu_info().lower()
if 'nvidia' in gpu_info or 'geforce' in gpu_info or 'quadro' in gpu_info:
return "nvidia"
elif 'amd' in gpu_info or 'radeon' in gpu_info:
return "amd"
elif 'intel' in gpu_info:
return "intel"
elif system == "darwin":
# macOS上检查是否为Apple Silicon
if platform.machine().lower() in ['arm64', 'aarch64']:
return "apple"
else:
# Intel Mac可能有独立显卡
gpu_info = _get_macos_gpu_info().lower()
if 'nvidia' in gpu_info:
return "nvidia"
elif 'amd' in gpu_info or 'radeon' in gpu_info:
return "amd"
else:
return "intel"
elif system == "linux":
gpu_info = _get_linux_gpu_info().lower()
if 'nvidia' in gpu_info:
return "nvidia"
elif 'amd' in gpu_info or 'radeon' in gpu_info:
return "amd"
elif 'intel' in gpu_info:
return "intel"
except Exception as e:
logger.debug(f"检测GPU厂商失败: {str(e)}")
return "unknown"
def test_hwaccel_method(method: str, test_input: str) -> bool:
"""
测试特定的硬件加速方法
Args:
method: 硬件加速方法名称
test_input: 测试输入文件路径
Returns:
bool: 是否支持该方法
"""
try:
# 构建测试命令
cmd = ["ffmpeg", "-hide_banner", "-loglevel", "error"]
# 添加硬件加速参数
if method == "cuda":
cmd.extend(["-hwaccel", "cuda", "-hwaccel_output_format", "cuda"])
elif method == "nvenc":
cmd.extend(["-hwaccel", "cuda"])
elif method == "videotoolbox":
cmd.extend(["-hwaccel", "videotoolbox"])
elif method == "qsv":
cmd.extend(["-hwaccel", "qsv"])
elif method == "vaapi":
# 尝试找到VAAPI设备
render_device = _find_vaapi_device()
if render_device:
cmd.extend(["-hwaccel", "vaapi", "-vaapi_device", render_device])
else:
cmd.extend(["-hwaccel", "vaapi"])
elif method == "d3d11va":
cmd.extend(["-hwaccel", "d3d11va"])
elif method == "dxva2":
cmd.extend(["-hwaccel", "dxva2"])
elif method == "amf":
cmd.extend(["-hwaccel", "auto"]) # AMF通常通过auto检测
else:
return False
# 添加输入和输出
cmd.extend(["-i", test_input, "-f", "null", "-t", "0.1", "-"])
# 执行测试
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
check=False,
timeout=10 # 10秒超时
)
success = result.returncode == 0
if success:
logger.debug(f"硬件加速方法 {method} 测试成功")
else:
logger.debug(f"硬件加速方法 {method} 测试失败: {result.stderr[:200]}")
return success
except subprocess.TimeoutExpired:
logger.debug(f"硬件加速方法 {method} 测试超时")
return False
except Exception as e:
logger.debug(f"硬件加速方法 {method} 测试异常: {str(e)}")
return False
def detect_hardware_acceleration() -> Dict[str, Union[bool, str, List[str], None]]: def detect_hardware_acceleration() -> Dict[str, Union[bool, str, List[str], None]]:
""" """
检测系统可用的硬件加速器并存储结果到全局变量 检测系统可用的硬件加速器使用渐进式检测和智能降级
Returns: Returns:
Dict: 包含硬件加速信息的字典 Dict: 包含硬件加速信息的字典
@ -56,45 +267,176 @@ def detect_hardware_acceleration() -> Dict[str, Union[bool, str, List[str], None
_FFMPEG_HW_ACCEL_INFO["message"] = "FFmpeg未安装或不在系统PATH中" _FFMPEG_HW_ACCEL_INFO["message"] = "FFmpeg未安装或不在系统PATH中"
return _FFMPEG_HW_ACCEL_INFO return _FFMPEG_HW_ACCEL_INFO
# 检测操作系统 # 检测平台和GPU信息
system = platform.system().lower() system = platform.system().lower()
logger.debug(f"检测硬件加速 - 操作系统: {system}") gpu_vendor = detect_gpu_vendor()
_FFMPEG_HW_ACCEL_INFO["platform"] = system
_FFMPEG_HW_ACCEL_INFO["gpu_vendor"] = gpu_vendor
logger.info(f"检测硬件加速 - 平台: {system}, GPU厂商: {gpu_vendor}")
# 获取FFmpeg支持的硬件加速器列表 # 获取FFmpeg支持的硬件加速器列表
try: try:
# 在Windows系统上使用UTF-8编码 hwaccels_cmd = subprocess.run(
is_windows = os.name == 'nt' ['ffmpeg', '-hide_banner', '-hwaccels'],
if is_windows: stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False
hwaccels_cmd = subprocess.run( )
['ffmpeg', '-hide_banner', '-hwaccels'], supported_hwaccels = hwaccels_cmd.stdout.lower() if hwaccels_cmd.returncode == 0 else ""
stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', text=True logger.debug(f"FFmpeg支持的硬件加速器: {supported_hwaccels}")
)
else:
hwaccels_cmd = subprocess.run(
['ffmpeg', '-hide_banner', '-hwaccels'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
supported_hwaccels = hwaccels_cmd.stdout.lower()
except Exception as e: except Exception as e:
logger.error(f"获取FFmpeg硬件加速器列表失败: {str(e)}") logger.warning(f"获取FFmpeg硬件加速器列表失败: {str(e)}")
supported_hwaccels = "" supported_hwaccels = ""
# 根据操作系统检测不同的硬件加速器 # 创建测试输入
if system == 'darwin': # macOS test_input = create_test_video()
_detect_macos_acceleration(supported_hwaccels)
elif system == 'windows': # Windows
_detect_windows_acceleration(supported_hwaccels)
elif system == 'linux': # Linux
_detect_linux_acceleration(supported_hwaccels)
else:
logger.warning(f"不支持的操作系统: {system}")
_FFMPEG_HW_ACCEL_INFO["message"] = f"不支持的操作系统: {system}"
# 记录检测结果已经在启动时输出,这里不再重复输出 try:
# 根据平台和GPU厂商获取优先级列表
priority_list = HWACCEL_PRIORITY.get(system, {}).get(gpu_vendor, [])
if not priority_list:
priority_list = HWACCEL_PRIORITY.get(system, {}).get("unknown", [])
logger.debug(f"硬件加速测试优先级: {priority_list}")
# 按优先级测试硬件加速方法
for method in priority_list:
# 检查FFmpeg是否支持该方法
if method not in supported_hwaccels and method != "nvenc": # nvenc可能不在hwaccels列表中
logger.debug(f"跳过不支持的硬件加速方法: {method}")
continue
_FFMPEG_HW_ACCEL_INFO["tested_methods"].append(method)
if test_hwaccel_method(method, test_input):
# 找到可用的硬件加速方法
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = method
_FFMPEG_HW_ACCEL_INFO["encoder"] = ENCODER_MAPPING.get(method, "libx264")
# 构建硬件加速参数
if method == "cuda":
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "cuda", "-hwaccel_output_format", "cuda"]
elif method == "nvenc":
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "cuda"]
elif method == "videotoolbox":
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "videotoolbox"]
elif method == "qsv":
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "qsv"]
elif method == "vaapi":
render_device = _find_vaapi_device()
if render_device:
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "vaapi", "-vaapi_device", render_device]
else:
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "vaapi"]
elif method in ["d3d11va", "dxva2"]:
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", method]
elif method == "amf":
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "auto"]
# 判断是否为独立GPU
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = gpu_vendor in ["nvidia", "amd"] or (gpu_vendor == "intel" and "arc" in _get_gpu_info().lower())
_FFMPEG_HW_ACCEL_INFO["message"] = f"使用 {method} 硬件加速 ({gpu_vendor} GPU)"
logger.info(f"硬件加速检测成功: {method} ({gpu_vendor})")
break
# 如果没有找到硬件加速,设置软件编码作为备用
if not _FFMPEG_HW_ACCEL_INFO["available"]:
_FFMPEG_HW_ACCEL_INFO["fallback_available"] = True
_FFMPEG_HW_ACCEL_INFO["fallback_encoder"] = "libx264"
_FFMPEG_HW_ACCEL_INFO["message"] = f"未找到可用的硬件加速,将使用软件编码 (平台: {system}, GPU: {gpu_vendor})"
logger.info("未检测到硬件加速,将使用软件编码")
finally:
# 清理测试文件
cleanup_test_video(test_input)
return _FFMPEG_HW_ACCEL_INFO return _FFMPEG_HW_ACCEL_INFO
def _get_gpu_info() -> str:
"""
获取GPU信息的统一接口
Returns:
str: GPU信息字符串
"""
system = platform.system().lower()
if system == "windows":
return _get_windows_gpu_info()
elif system == "darwin":
return _get_macos_gpu_info()
elif system == "linux":
return _get_linux_gpu_info()
else:
return "unknown"
def _get_macos_gpu_info() -> str:
"""
获取macOS系统的GPU信息
Returns:
str: GPU信息字符串
"""
try:
# 使用system_profiler获取显卡信息
result = subprocess.run(
['system_profiler', 'SPDisplaysDataType'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False
)
if result.returncode == 0:
return result.stdout
# 备用方法检查是否为Apple Silicon
if platform.machine().lower() in ['arm64', 'aarch64']:
return "Apple Silicon GPU"
else:
return "Intel Mac GPU"
except Exception as e:
logger.debug(f"获取macOS GPU信息失败: {str(e)}")
return "unknown"
def _find_vaapi_device() -> Optional[str]:
"""
查找可用的VAAPI设备
Returns:
Optional[str]: VAAPI设备路径如果没有找到则返回None
"""
try:
# 常见的VAAPI设备路径
possible_devices = [
"/dev/dri/renderD128",
"/dev/dri/renderD129",
"/dev/dri/card0",
"/dev/dri/card1"
]
for device in possible_devices:
if os.path.exists(device):
# 测试设备是否可用
test_cmd = subprocess.run(
["ffmpeg", "-hide_banner", "-loglevel", "error",
"-hwaccel", "vaapi", "-vaapi_device", device,
"-f", "lavfi", "-i", "color=black:size=64x64:duration=0.1",
"-f", "null", "-"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False
)
if test_cmd.returncode == 0:
logger.debug(f"找到可用的VAAPI设备: {device}")
return device
logger.debug("未找到可用的VAAPI设备")
return None
except Exception as e:
logger.debug(f"查找VAAPI设备失败: {str(e)}")
return None
def _detect_macos_acceleration(supported_hwaccels: str) -> None: def _detect_macos_acceleration(supported_hwaccels: str) -> None:
""" """
检测macOS系统的硬件加速 检测macOS系统的硬件加速
@ -511,3 +853,165 @@ def is_dedicated_gpu() -> bool:
detect_hardware_acceleration() detect_hardware_acceleration()
return _FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] return _FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"]
def get_optimal_ffmpeg_encoder() -> str:
"""
获取最优的FFmpeg编码器
Returns:
str: 编码器名称
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
if _FFMPEG_HW_ACCEL_INFO["available"]:
return _FFMPEG_HW_ACCEL_INFO["encoder"]
elif _FFMPEG_HW_ACCEL_INFO["fallback_available"]:
return _FFMPEG_HW_ACCEL_INFO["fallback_encoder"]
else:
return "libx264" # 默认软件编码器
def get_ffmpeg_command_with_hwaccel(input_path: str, output_path: str, **kwargs) -> List[str]:
"""
生成带有硬件加速的FFmpeg命令
Args:
input_path: 输入文件路径
output_path: 输出文件路径
**kwargs: 其他FFmpeg参数
Returns:
List[str]: FFmpeg命令列表
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
cmd = ["ffmpeg", "-y"]
# 添加硬件加速参数
if _FFMPEG_HW_ACCEL_INFO["available"]:
cmd.extend(_FFMPEG_HW_ACCEL_INFO["hwaccel_args"])
# 添加输入文件
cmd.extend(["-i", input_path])
# 添加编码器
encoder = get_optimal_ffmpeg_encoder()
cmd.extend(["-c:v", encoder])
# 添加其他参数
for key, value in kwargs.items():
if key.startswith("_"): # 跳过内部参数
continue
if isinstance(value, list):
cmd.extend(value)
else:
cmd.extend([f"-{key}", str(value)])
# 添加输出文件
cmd.append(output_path)
return cmd
def test_ffmpeg_compatibility() -> Dict[str, any]:
"""
测试FFmpeg兼容性并返回详细报告
Returns:
Dict: 兼容性测试报告
"""
report = {
"ffmpeg_installed": False,
"platform": platform.system().lower(),
"gpu_vendor": "unknown",
"hardware_acceleration": {
"available": False,
"type": None,
"encoder": None,
"tested_methods": []
},
"software_fallback": {
"available": False,
"encoder": "libx264"
},
"recommendations": []
}
# 检查FFmpeg安装
report["ffmpeg_installed"] = check_ffmpeg_installation()
if not report["ffmpeg_installed"]:
report["recommendations"].append("请安装FFmpeg并确保其在系统PATH中")
return report
# 检测硬件加速
hwaccel_info = detect_hardware_acceleration()
report["gpu_vendor"] = hwaccel_info.get("gpu_vendor", "unknown")
report["hardware_acceleration"]["available"] = hwaccel_info.get("available", False)
report["hardware_acceleration"]["type"] = hwaccel_info.get("type")
report["hardware_acceleration"]["encoder"] = hwaccel_info.get("encoder")
report["hardware_acceleration"]["tested_methods"] = hwaccel_info.get("tested_methods", [])
# 检查软件备用方案
report["software_fallback"]["available"] = hwaccel_info.get("fallback_available", True)
report["software_fallback"]["encoder"] = hwaccel_info.get("fallback_encoder", "libx264")
# 生成建议
if not report["hardware_acceleration"]["available"]:
if report["gpu_vendor"] == "nvidia":
report["recommendations"].append("建议安装NVIDIA驱动和CUDA工具包以启用硬件加速")
elif report["gpu_vendor"] == "amd":
report["recommendations"].append("AMD显卡硬件加速支持有限建议使用软件编码")
elif report["gpu_vendor"] == "intel":
report["recommendations"].append("建议更新Intel显卡驱动以启用QSV硬件加速")
else:
report["recommendations"].append("未检测到支持的GPU将使用软件编码")
return report
def force_software_encoding() -> None:
"""
强制使用软件编码禁用硬件加速
"""
global _FFMPEG_HW_ACCEL_INFO
_FFMPEG_HW_ACCEL_INFO.update({
"available": False,
"type": "software",
"encoder": "libx264",
"hwaccel_args": [],
"message": "强制使用软件编码",
"is_dedicated_gpu": False,
"fallback_available": True,
"fallback_encoder": "libx264"
})
logger.info("已强制切换到软件编码模式")
def reset_hwaccel_detection() -> None:
"""
重置硬件加速检测结果强制重新检测
"""
global _FFMPEG_HW_ACCEL_INFO
_FFMPEG_HW_ACCEL_INFO = {
"available": False,
"type": None,
"encoder": None,
"hwaccel_args": [],
"message": "",
"is_dedicated_gpu": False,
"fallback_available": False,
"fallback_encoder": None,
"platform": None,
"gpu_vendor": None,
"tested_methods": []
}
logger.info("已重置硬件加速检测结果")