"""视频调速引擎 — 基于 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 MAX_SPEED # noqa: F401 — 向后兼容 from packages.domain.speed_config import MIN_SPEED # noqa: F401 — 向后兼容 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_speed,0 或缺失则使用全局速度.""" return _resolve_clip_speed_base(clip_config, global_speed) # ── 向后兼容:模块级函数(重构前的 API) ───────────────────────── def build_video_filter(config): """向后兼容:模块级 build_video_filter.""" return _build_video_filter_base(config) def build_audio_filter(config): """向后兼容:模块级 build_audio_filter.""" return _build_audio_filter_base(config) def adjust_duration(original_duration, config): """向后兼容:模块级 adjust_duration.""" return _adjust_duration_base(original_duration, config) def resolve_clip_speed(clip_config, global_speed=DEFAULT_SPEED): """向后兼容:模块级 resolve_clip_speed.""" return _resolve_clip_speed_base(clip_config, global_speed) def build_clip_speed_filter(speed, pitch_correct=True): """向后兼容:模块级 build_clip_speed_filter.""" config = SpeedConfig(speed=speed, pitch_correct=pitch_correct) config.clamp() return ( build_video_filter(config), build_audio_filter(config), config, )