Files
xiaoxia-saas/apps/worker/video_processing/speed_engine.py
T

71 lines
2.5 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力.
支持:
- 0.25x ~ 4x 变速范围
- 视频调速(setpts
- 音频调速(atempo,多级串联处理超范围值)
- 音调修正(pitch_correct,默认开启)
- 边界自动钳制,不阻断渲染
注:核心领域模型已抽离到 packages/domain/speed_config.py
本模块保留薄包装层,确保向后兼容。
"""
from __future__ import annotations
from packages.domain.speed_config import (
DEFAULT_SPEED,
SpeedConfig,
_split_atempo_stages,
)
from packages.domain.speed_config import adjust_duration as _adjust_duration_base # noqa: F401 — 向后兼容
from packages.domain.speed_config import build_audio_filter as _build_audio_filter_base
from packages.domain.speed_config import build_video_filter as _build_video_filter_base
from packages.domain.speed_config import resolve_clip_speed as _resolve_clip_speed_base
class SpeedEngine:
"""调速引擎 — 生成 FFmpeg 调速滤镜链.
薄包装层,实际逻辑委托给 packages.domain.speed_config。
"""
def build_video_filter(self, config: SpeedConfig) -> str:
"""生成视频调速滤镜字符串."""
return _build_video_filter_base(config)
def build_audio_filter(self, config: SpeedConfig) -> str:
"""生成音频调速滤镜字符串."""
return _build_audio_filter_base(config)
@staticmethod
def _split_atempo_stages(speed: float) -> list[float]:
"""将速度拆分为多级 atempo 串联(内部方法,向后兼容)."""
return _split_atempo_stages(speed)
def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float:
"""计算调速后的时长."""
return _adjust_duration_base(original_duration, config)
def build_clip_speed_filter(
self,
speed: float,
pitch_correct: bool = True,
) -> tuple[str, str, SpeedConfig]:
"""便捷方法:从单一 speed 值生成视频+音频滤镜."""
config = SpeedConfig(speed=speed, pitch_correct=pitch_correct)
config.clamp()
return (
self.build_video_filter(config),
self.build_audio_filter(config),
config,
)
@staticmethod
def resolve_clip_speed(
clip_config: dict,
global_speed: float = DEFAULT_SPEED,
) -> float:
"""从 clip config 中解析 playback_speed0 或缺失则使用全局速度."""
return _resolve_clip_speed_base(clip_config, global_speed)