优化 ffmpeg 硬件加速兼容性

This commit is contained in:
linyq 2025-05-19 02:41:30 +08:00
parent 6356a140aa
commit 47cd4f145d
8 changed files with 596 additions and 393 deletions

View File

@ -13,6 +13,7 @@ from app.config import config
from app.models.exception import HttpException
from app.router import root_api_router
from app.utils import utils
from app.utils import ffmpeg_utils
def exception_handler(request: Request, e: HttpException):
@ -80,3 +81,10 @@ def shutdown_event():
@app.on_event("startup")
def startup_event():
logger.info("startup event")
# 检测FFmpeg硬件加速
hwaccel_info = ffmpeg_utils.detect_hardware_acceleration()
if hwaccel_info["available"]:
logger.info(f"FFmpeg硬件加速检测结果: 可用 | 类型: {hwaccel_info['type']} | 编码器: {hwaccel_info['encoder']} | 独立显卡: {hwaccel_info['is_dedicated_gpu']} | 参数: {hwaccel_info['hwaccel_args']}")
else:
logger.warning(f"FFmpeg硬件加速不可用: {hwaccel_info['message']}, 将使用CPU软件编码")

View File

@ -16,6 +16,8 @@ from loguru import logger
from typing import Dict, List, Optional
from pathlib import Path
from app.utils import ffmpeg_utils
def parse_timestamp(timestamp: str) -> tuple:
"""
@ -79,40 +81,8 @@ def check_hardware_acceleration() -> Optional[str]:
Returns:
Optional[str]: 硬件加速参数如果不支持则返回None
"""
# 检查NVIDIA GPU支持
try:
nvidia_check = subprocess.run(
["ffmpeg", "-hwaccel", "cuda", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if nvidia_check.returncode == 0:
return "cuda"
except Exception:
pass
# 检查MacOS videotoolbox支持
try:
videotoolbox_check = subprocess.run(
["ffmpeg", "-hwaccel", "videotoolbox", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if videotoolbox_check.returncode == 0:
return "videotoolbox"
except Exception:
pass
# 检查Intel Quick Sync支持
try:
qsv_check = subprocess.run(
["ffmpeg", "-hwaccel", "qsv", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if qsv_check.returncode == 0:
return "qsv"
except Exception:
pass
return None
# 使用集中式硬件加速检测
return ffmpeg_utils.get_ffmpeg_hwaccel_type()
def clip_video(
@ -152,12 +122,11 @@ def clip_video(
# 确保输出目录存在
Path(output_dir).mkdir(parents=True, exist_ok=True)
# 检查硬件加速支持
# 获取硬件加速支持
hwaccel = check_hardware_acceleration()
hwaccel_args = []
if hwaccel:
hwaccel_args = ["-hwaccel", hwaccel]
logger.info(f"使用硬件加速: {hwaccel}")
hwaccel_args = ffmpeg_utils.get_ffmpeg_hwaccel_args()
# 存储裁剪结果
result = {}

View File

@ -14,6 +14,7 @@ from moviepy.video.io.VideoFileClip import VideoFileClip
from app.config import config
from app.models.schema import VideoAspect, VideoConcatMode, MaterialInfo
from app.utils import utils
from app.utils import ffmpeg_utils
requested_count = 0
@ -314,40 +315,9 @@ def _detect_hardware_acceleration() -> Optional[str]:
Returns:
Optional[str]: 硬件加速参数如果不支持则返回None
"""
# 检查NVIDIA GPU支持
try:
nvidia_check = subprocess.run(
["ffmpeg", "-hwaccel", "cuda", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if nvidia_check.returncode == 0:
return "cuda"
except Exception:
pass
# 检查MacOS videotoolbox支持
try:
videotoolbox_check = subprocess.run(
["ffmpeg", "-hwaccel", "videotoolbox", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if videotoolbox_check.returncode == 0:
return "videotoolbox"
except Exception:
pass
# 检查Intel Quick Sync支持
try:
qsv_check = subprocess.run(
["ffmpeg", "-hwaccel", "qsv", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if qsv_check.returncode == 0:
return "qsv"
except Exception:
pass
return None
# 使用集中式硬件加速检测
hwaccel_type = ffmpeg_utils.get_ffmpeg_hwaccel_type()
return hwaccel_type
def save_clip_video(timestamp: str, origin_video: str, save_dir: str = "") -> str:
@ -422,12 +392,11 @@ def save_clip_video(timestamp: str, origin_video: str, save_dir: str = "") -> st
duration = end - start
# logger.info(f"开始剪辑视频: {format_timestamp(start)} - {format_timestamp(end)},时长 {format_timestamp(duration)}")
# 检测可用的硬件加速选项
# 获取硬件加速选项
hwaccel = _detect_hardware_acceleration()
hwaccel_args = []
if hwaccel:
hwaccel_args = ["-hwaccel", hwaccel]
logger.info(f"使用硬件加速: {hwaccel}")
hwaccel_args = ffmpeg_utils.get_ffmpeg_hwaccel_args()
# 转换为FFmpeg兼容的时间格式逗号替换为点
ffmpeg_start_time = start_str.replace(',', '.')

View File

@ -15,6 +15,8 @@ from enum import Enum
from typing import List, Optional, Tuple
from loguru import logger
from app.utils import ffmpeg_utils
class VideoAspect(Enum):
"""视频宽高比枚举"""
@ -62,60 +64,8 @@ def get_hardware_acceleration_option() -> Optional[str]:
Returns:
Optional[str]: 硬件加速参数如果不支持则返回None
"""
try:
# 检测操作系统
is_windows = os.name == 'nt'
# 检查NVIDIA GPU支持
nvidia_check = subprocess.run(
['ffmpeg', '-hide_banner', '-hwaccels'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True
)
output = nvidia_check.stdout.lower()
# 首先尝试获取系统信息Windows系统使用更安全的检测方法
if is_windows:
try:
# 尝试检测显卡信息
gpu_info = subprocess.run(
['wmic', 'path', 'win32_VideoController', 'get', 'name'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False
)
gpu_info_output = gpu_info.stdout.lower()
# 检测是否为AMD显卡
if 'amd' in gpu_info_output or 'radeon' in gpu_info_output:
logger.info("检测到AMD显卡为避免兼容性问题将使用软件编码")
return None
# 检测是否为集成显卡
if 'intel' in gpu_info_output and ('hd graphics' in gpu_info_output or 'uhd graphics' in gpu_info_output):
# 在Windows上Intel集成显卡可能不稳定建议使用软件编码
logger.info("检测到Intel集成显卡为避免兼容性问题将使用软件编码")
return None
except Exception as e:
logger.warning(f"获取显卡信息失败: {str(e)},将谨慎处理硬件加速")
# 根据ffmpeg支持的硬件加速器决定使用哪种
if 'cuda' in output and not is_windows:
# 在非Windows系统上使用CUDA
return 'cuda'
elif 'nvenc' in output and not is_windows:
# 在非Windows系统上使用NVENC
return 'nvenc'
elif 'qsv' in output and not (is_windows and ('amd' in gpu_info_output if 'gpu_info_output' in locals() else False)):
# 只有在非AMD系统上使用QSV
return 'qsv'
elif 'videotoolbox' in output: # macOS
return 'videotoolbox'
elif 'vaapi' in output and not is_windows: # Linux VA-API
return 'vaapi'
else:
logger.info("没有找到支持的硬件加速器或系统不兼容,将使用软件编码")
return None
except Exception as e:
logger.warning(f"检测硬件加速器时出错: {str(e)},将使用软件编码")
return None
# 使用集中式硬件加速检测
return ffmpeg_utils.get_ffmpeg_hwaccel_type()
def check_video_has_audio(video_path: str) -> bool:
@ -231,15 +181,9 @@ def process_single_video(
# 添加硬件加速参数(根据前面的安全检查可能已经被禁用)
if hwaccel:
try:
if hwaccel == 'cuda' or hwaccel == 'nvenc':
command.extend(['-hwaccel', 'cuda'])
elif hwaccel == 'qsv':
command.extend(['-hwaccel', 'qsv'])
elif hwaccel == 'videotoolbox':
command.extend(['-hwaccel', 'videotoolbox'])
elif hwaccel == 'vaapi':
command.extend(['-hwaccel', 'vaapi', '-vaapi_device', '/dev/dri/renderD128'])
logger.info(f"应用硬件加速: {hwaccel}")
# 使用集中式硬件加速参数
hwaccel_args = ffmpeg_utils.get_ffmpeg_hwaccel_args()
command.extend(hwaccel_args)
except Exception as e:
logger.warning(f"应用硬件加速参数时出错: {str(e)},将使用软件编码")
# 重置命令,移除可能添加了一半的硬件加速参数

419
app/utils/ffmpeg_utils.py Normal file
View File

@ -0,0 +1,419 @@
"""
FFmpeg 工具模块 - 提供 FFmpeg 相关的工具函数特别是硬件加速检测
"""
import os
import platform
import subprocess
from typing import Dict, List, Optional, Tuple, Union
from loguru import logger
# 全局变量,存储检测到的硬件加速信息
_FFMPEG_HW_ACCEL_INFO = {
"available": False,
"type": None,
"encoder": None,
"hwaccel_args": [],
"message": "",
"is_dedicated_gpu": False
}
def check_ffmpeg_installation() -> bool:
"""
检查ffmpeg是否已安装
Returns:
bool: 如果安装则返回True否则返回False
"""
try:
subprocess.run(['ffmpeg', '-version'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True)
return True
except (subprocess.SubprocessError, FileNotFoundError):
logger.error("ffmpeg未安装或不在系统PATH中请安装ffmpeg")
return False
def detect_hardware_acceleration() -> Dict[str, Union[bool, str, List[str], None]]:
"""
检测系统可用的硬件加速器并存储结果到全局变量
Returns:
Dict: 包含硬件加速信息的字典
"""
global _FFMPEG_HW_ACCEL_INFO
# 如果已经检测过,直接返回结果
if _FFMPEG_HW_ACCEL_INFO["type"] is not None:
return _FFMPEG_HW_ACCEL_INFO
# 检查ffmpeg是否已安装
if not check_ffmpeg_installation():
_FFMPEG_HW_ACCEL_INFO["message"] = "FFmpeg未安装或不在系统PATH中"
return _FFMPEG_HW_ACCEL_INFO
# 检测操作系统
system = platform.system().lower()
logger.debug(f"检测硬件加速 - 操作系统: {system}")
# 获取FFmpeg支持的硬件加速器列表
try:
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:
logger.error(f"获取FFmpeg硬件加速器列表失败: {str(e)}")
supported_hwaccels = ""
# 根据操作系统检测不同的硬件加速器
if system == 'darwin': # macOS
_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}"
# 记录检测结果已经在启动时输出,这里不再重复输出
return _FFMPEG_HW_ACCEL_INFO
def _detect_macos_acceleration(supported_hwaccels: str) -> None:
"""
检测macOS系统的硬件加速
Args:
supported_hwaccels: FFmpeg支持的硬件加速器列表
"""
global _FFMPEG_HW_ACCEL_INFO
if 'videotoolbox' in supported_hwaccels:
# 测试videotoolbox
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "videotoolbox", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "videotoolbox"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_videotoolbox"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "videotoolbox"]
# macOS的Metal GPU加速通常是集成GPU
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = False
return
except Exception as e:
logger.debug(f"测试videotoolbox失败: {str(e)}")
_FFMPEG_HW_ACCEL_INFO["message"] = "macOS系统未检测到可用的videotoolbox硬件加速"
def _detect_windows_acceleration(supported_hwaccels: str) -> None:
"""
检测Windows系统的硬件加速
Args:
supported_hwaccels: FFmpeg支持的硬件加速器列表
"""
global _FFMPEG_HW_ACCEL_INFO
# 在Windows上首先检查显卡信息
gpu_info = _get_windows_gpu_info()
# 检查是否为AMD显卡
if 'amd' in gpu_info.lower() or 'radeon' in gpu_info.lower():
logger.info("检测到AMD显卡为避免兼容性问题将使用软件编码")
_FFMPEG_HW_ACCEL_INFO["message"] = "检测到AMD显卡为避免兼容性问题将使用软件编码"
return
# 检查是否为Intel集成显卡
is_intel_integrated = False
if 'intel' in gpu_info.lower() and ('hd graphics' in gpu_info.lower() or 'uhd graphics' in gpu_info.lower()):
logger.info("检测到Intel集成显卡")
is_intel_integrated = True
# 检测NVIDIA CUDA支持
if 'cuda' in supported_hwaccels and 'nvidia' in gpu_info.lower():
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "cuda", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "cuda"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_nvenc"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "cuda"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
return
except Exception as e:
logger.debug(f"测试CUDA失败: {str(e)}")
# 检测Intel QSV支持如果是Intel显卡
if 'qsv' in supported_hwaccels and 'intel' in gpu_info.lower():
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "qsv", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "qsv"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_qsv"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "qsv"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = not is_intel_integrated
return
except Exception as e:
logger.debug(f"测试QSV失败: {str(e)}")
# 检测D3D11VA支持
if 'd3d11va' in supported_hwaccels:
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "d3d11va", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "d3d11va"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264" # D3D11VA只用于解码编码仍使用软件编码器
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "d3d11va"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = not is_intel_integrated
return
except Exception as e:
logger.debug(f"测试D3D11VA失败: {str(e)}")
# 检测DXVA2支持
if 'dxva2' in supported_hwaccels:
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "dxva2", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "dxva2"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264" # DXVA2只用于解码编码仍使用软件编码器
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "dxva2"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = not is_intel_integrated
return
except Exception as e:
logger.debug(f"测试DXVA2失败: {str(e)}")
_FFMPEG_HW_ACCEL_INFO["message"] = f"Windows系统未检测到可用的硬件加速显卡信息: {gpu_info}"
def _detect_linux_acceleration(supported_hwaccels: str) -> None:
"""
检测Linux系统的硬件加速
Args:
supported_hwaccels: FFmpeg支持的硬件加速器列表
"""
global _FFMPEG_HW_ACCEL_INFO
# 获取Linux显卡信息
gpu_info = _get_linux_gpu_info()
is_nvidia = 'nvidia' in gpu_info.lower()
is_intel = 'intel' in gpu_info.lower()
is_amd = 'amd' in gpu_info.lower() or 'radeon' in gpu_info.lower()
# 检测NVIDIA CUDA支持
if 'cuda' in supported_hwaccels and is_nvidia:
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "cuda", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "cuda"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_nvenc"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "cuda"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = True
return
except Exception as e:
logger.debug(f"测试CUDA失败: {str(e)}")
# 检测VAAPI支持
if 'vaapi' in supported_hwaccels:
# 检查是否存在渲染设备
render_devices = ['/dev/dri/renderD128', '/dev/dri/renderD129']
render_device = None
for device in render_devices:
if os.path.exists(device):
render_device = device
break
if render_device:
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "vaapi", "-vaapi_device", render_device,
"-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "vaapi"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_vaapi"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "vaapi", "-vaapi_device", render_device]
# 根据显卡类型判断是否为独立显卡
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = is_nvidia or (is_amd and not is_intel)
return
except Exception as e:
logger.debug(f"测试VAAPI失败: {str(e)}")
# 检测Intel QSV支持
if 'qsv' in supported_hwaccels and is_intel:
try:
test_cmd = subprocess.run(
["ffmpeg", "-hwaccel", "qsv", "-i", "/dev/null", "-f", "null", "-"],
stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, check=False
)
if test_cmd.returncode == 0:
_FFMPEG_HW_ACCEL_INFO["available"] = True
_FFMPEG_HW_ACCEL_INFO["type"] = "qsv"
_FFMPEG_HW_ACCEL_INFO["encoder"] = "h264_qsv"
_FFMPEG_HW_ACCEL_INFO["hwaccel_args"] = ["-hwaccel", "qsv"]
_FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"] = False # Intel QSV通常是集成GPU
return
except Exception as e:
logger.debug(f"测试QSV失败: {str(e)}")
_FFMPEG_HW_ACCEL_INFO["message"] = f"Linux系统未检测到可用的硬件加速显卡信息: {gpu_info}"
def _get_windows_gpu_info() -> str:
"""
获取Windows系统的显卡信息
Returns:
str: 显卡信息字符串
"""
try:
gpu_info = subprocess.run(
['wmic', 'path', 'win32_VideoController', 'get', 'name'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False
)
return gpu_info.stdout
except Exception as e:
logger.warning(f"获取Windows显卡信息失败: {str(e)}")
return "Unknown GPU"
def _get_linux_gpu_info() -> str:
"""
获取Linux系统的显卡信息
Returns:
str: 显卡信息字符串
"""
try:
# 尝试使用lspci命令
gpu_info = subprocess.run(
['lspci', '-v', '-nn', '|', 'grep', '-i', 'vga\\|display'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, shell=True, check=False
)
if gpu_info.stdout:
return gpu_info.stdout
# 如果lspci命令失败尝试使用glxinfo
gpu_info = subprocess.run(
['glxinfo', '|', 'grep', '-i', 'vendor\\|renderer'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, shell=True, check=False
)
if gpu_info.stdout:
return gpu_info.stdout
return "Unknown GPU"
except Exception as e:
logger.warning(f"获取Linux显卡信息失败: {str(e)}")
return "Unknown GPU"
def get_ffmpeg_hwaccel_args() -> List[str]:
"""
获取FFmpeg硬件加速参数
Returns:
List[str]: FFmpeg硬件加速参数列表
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
return _FFMPEG_HW_ACCEL_INFO["hwaccel_args"]
def get_ffmpeg_hwaccel_type() -> Optional[str]:
"""
获取FFmpeg硬件加速类型
Returns:
Optional[str]: 硬件加速类型如果不支持则返回None
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
return _FFMPEG_HW_ACCEL_INFO["type"] if _FFMPEG_HW_ACCEL_INFO["available"] else None
def get_ffmpeg_hwaccel_encoder() -> Optional[str]:
"""
获取FFmpeg硬件加速编码器
Returns:
Optional[str]: 硬件加速编码器如果不支持则返回None
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
return _FFMPEG_HW_ACCEL_INFO["encoder"] if _FFMPEG_HW_ACCEL_INFO["available"] else None
def get_ffmpeg_hwaccel_info() -> Dict[str, Union[bool, str, List[str], None]]:
"""
获取FFmpeg硬件加速信息
Returns:
Dict: 包含硬件加速信息的字典
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
return _FFMPEG_HW_ACCEL_INFO
def is_ffmpeg_hwaccel_available() -> bool:
"""
检查是否有可用的FFmpeg硬件加速
Returns:
bool: 如果有可用的硬件加速则返回True否则返回False
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
return _FFMPEG_HW_ACCEL_INFO["available"]
def is_dedicated_gpu() -> bool:
"""
检查是否使用独立显卡进行硬件加速
Returns:
bool: 如果使用独立显卡则返回True否则返回False
"""
# 如果还没有检测过,先进行检测
if _FFMPEG_HW_ACCEL_INFO["type"] is None:
detect_hardware_acceleration()
return _FFMPEG_HW_ACCEL_INFO["is_dedicated_gpu"]

View File

@ -19,6 +19,8 @@ from typing import List, Dict
from loguru import logger
from tqdm import tqdm
from app.utils import ffmpeg_utils
class VideoProcessor:
def __init__(self, video_path: str):
@ -115,14 +117,8 @@ class VideoProcessor:
# 确定硬件加速器选项
hw_accel = []
if use_hw_accel:
# 尝试检测可用的硬件加速器
hw_accel_options = self._detect_hw_accelerator()
if hw_accel_options:
hw_accel = hw_accel_options
logger.info(f"使用硬件加速: {' '.join(hw_accel)}")
else:
logger.warning("未检测到可用的硬件加速器,使用软件解码")
if use_hw_accel and ffmpeg_utils.is_ffmpeg_hwaccel_available():
hw_accel = ffmpeg_utils.get_ffmpeg_hwaccel_args()
# 提取帧
frame_numbers = []
@ -173,119 +169,9 @@ class VideoProcessor:
Returns:
List[str]: 硬件加速器ffmpeg命令参数
"""
# 检测操作系统
import platform
system = platform.system().lower()
# 测试不同的硬件加速器
accelerators = []
if system == 'darwin': # macOS
# 测试 videotoolbox (Apple 硬件加速)
test_cmd = [
"ffmpeg",
"-hide_banner",
"-loglevel", "error",
"-hwaccel", "videotoolbox",
"-i", self.video_path,
"-t", "0.1",
"-f", "null",
"-"
]
try:
subprocess.run(test_cmd, capture_output=True, check=True)
return ["-hwaccel", "videotoolbox"]
except subprocess.CalledProcessError:
pass
elif system == 'linux':
# 测试 VAAPI
test_cmd = [
"ffmpeg",
"-hide_banner",
"-loglevel", "error",
"-hwaccel", "vaapi",
"-i", self.video_path,
"-t", "0.1",
"-f", "null",
"-"
]
try:
subprocess.run(test_cmd, capture_output=True, check=True)
return ["-hwaccel", "vaapi"]
except subprocess.CalledProcessError:
pass
# 尝试 CUDA
test_cmd = [
"ffmpeg",
"-hide_banner",
"-loglevel", "error",
"-hwaccel", "cuda",
"-i", self.video_path,
"-t", "0.1",
"-f", "null",
"-"
]
try:
subprocess.run(test_cmd, capture_output=True, check=True)
return ["-hwaccel", "cuda"]
except subprocess.CalledProcessError:
pass
elif system == 'windows':
# 测试 CUDA
test_cmd = [
"ffmpeg",
"-hide_banner",
"-loglevel", "error",
"-hwaccel", "cuda",
"-i", self.video_path,
"-t", "0.1",
"-f", "null",
"-"
]
try:
subprocess.run(test_cmd, capture_output=True, check=True)
return ["-hwaccel", "cuda"]
except subprocess.CalledProcessError:
pass
# 测试 D3D11VA
test_cmd = [
"ffmpeg",
"-hide_banner",
"-loglevel", "error",
"-hwaccel", "d3d11va",
"-i", self.video_path,
"-t", "0.1",
"-f", "null",
"-"
]
try:
subprocess.run(test_cmd, capture_output=True, check=True)
return ["-hwaccel", "d3d11va"]
except subprocess.CalledProcessError:
pass
# 测试 DXVA2
test_cmd = [
"ffmpeg",
"-hide_banner",
"-loglevel", "error",
"-hwaccel", "dxva2",
"-i", self.video_path,
"-t", "0.1",
"-f", "null",
"-"
]
try:
subprocess.run(test_cmd, capture_output=True, check=True)
return ["-hwaccel", "dxva2"]
except subprocess.CalledProcessError:
pass
# 如果没有找到可用的硬件加速器
# 使用集中式硬件加速检测
if ffmpeg_utils.is_ffmpeg_hwaccel_available():
return ffmpeg_utils.get_ffmpeg_hwaccel_args()
return []
def process_video_pipeline(self,

View File

@ -1,5 +1,5 @@
[app]
project_version="0.6.1"
project_version="0.6.2"
# 支持视频理解的大模型提供商
# gemini (谷歌, 需要 VPN)
# siliconflow (硅基流动)

View File

@ -7,6 +7,7 @@ from webui.components import basic_settings, video_settings, audio_settings, sub
review_settings, merge_settings, system_settings
# from webui.utils import cache, file_utils
from app.utils import utils
from app.utils import ffmpeg_utils
from app.models.schema import VideoClipParams, VideoAspect
@ -193,6 +194,13 @@ def main():
init_log()
init_global_state()
# 检测FFmpeg硬件加速
hwaccel_info = ffmpeg_utils.detect_hardware_acceleration()
if hwaccel_info["available"]:
logger.info(f"FFmpeg硬件加速检测结果: 可用 | 类型: {hwaccel_info['type']} | 编码器: {hwaccel_info['encoder']} | 独立显卡: {hwaccel_info['is_dedicated_gpu']} | 参数: {hwaccel_info['hwaccel_args']}")
else:
logger.warning(f"FFmpeg硬件加速不可用: {hwaccel_info['message']}, 将使用CPU软件编码")
# 仅初始化基本资源避免过早地加载依赖PyTorch的资源
# 检查是否能分解utils.init_resources()为基本资源和高级资源(如依赖PyTorch的资源)
try: