"""调速配置领域模型 — 纯逻辑,无FFmpeg依赖. 抽离自 speed_engine.py,包含: - SpeedConfig 数据类(解析/钳制/原速判断) - 视频/音频调速滤镜构建 - atempo 多级拆分算法 - 时长计算 """ from __future__ import annotations from dataclasses import dataclass from typing import Any # ─── 常量 ─────────────────────────────────────────────── MIN_SPEED = 0.25 MAX_SPEED = 4.0 DEFAULT_SPEED = 1.0 # atempo 单级有效范围 _ATEMPO_MIN = 0.5 _ATEMPO_MAX = 2.0 @dataclass class SpeedConfig: """调速配置. Attributes: speed: 播放速度,0.25~4.0,1.0 为原速 pitch_correct: 是否保持音调(默认 True,用 atempo 时间拉伸算法) """ speed: float = DEFAULT_SPEED pitch_correct: bool = True @classmethod def parse(cls, data: dict[str, Any] | None) -> SpeedConfig: """从 dict 解析配置,无效值回退到默认.""" if not data or not isinstance(data, dict): return cls() speed = data.get("speed", DEFAULT_SPEED) if not isinstance(speed, (int, float)): speed = DEFAULT_SPEED pitch_correct = data.get("pitch_correct", True) if not isinstance(pitch_correct, bool): pitch_correct = True config = cls(speed=float(speed), pitch_correct=pitch_correct) config.clamp() return config def clamp(self) -> None: """将速度钳制到合法范围.""" if self.speed <= 0: self.speed = DEFAULT_SPEED elif self.speed < MIN_SPEED: self.speed = MIN_SPEED elif self.speed > MAX_SPEED: self.speed = MAX_SPEED @property def is_original(self) -> bool: """是否原速(无需调速).""" return abs(self.speed - 1.0) < 1e-6 @property def is_fast(self) -> bool: """是否加速播放.""" return self.speed > 1.0 @property def is_slow(self) -> bool: """是否减速播放.""" return self.speed < 1.0 # ── 滤镜构建 ──────────────────────────────────────────────────────────────── def build_video_filter(config: SpeedConfig) -> str: """生成视频调速滤镜字符串. 返回 setpts 滤镜表达式,原速时返回空字符串。 """ if config.is_original: return "" # setpts=PTS/speed — speed>1 加速,speed<1 减速 return f"setpts=PTS/{config.speed:.4f}" def build_audio_filter(config: SpeedConfig) -> str: """生成音频调速滤镜字符串. atempo 单级范围 0.5~2.0,超出范围时自动多级串联: - 0.25x → atempo=0.5,atempo=0.5 - 4x → atempo=2.0,atempo=2.0 - 0.3x → atempo=0.5,atempo=0.6 - 3x → atempo=2.0,atempo=1.5 原速时返回空字符串。 """ if config.is_original: return "" speed = config.speed stages: list[float] = _split_atempo_stages(speed) return ",".join(f"atempo={s:.4f}" for s in stages) def _split_atempo_stages(speed: float) -> list[float]: """将速度拆分为多级 atempo 串联,每级都在 [0.5, 2.0] 范围内.""" if _ATEMPO_MIN <= speed <= _ATEMPO_MAX: return [speed] stages: list[float] = [] remaining = speed # 加速场景(speed > 2.0) if speed > _ATEMPO_MAX: while remaining > _ATEMPO_MAX: stages.append(_ATEMPO_MAX) remaining /= _ATEMPO_MAX stages.append(remaining) # 减速场景(speed < 0.5) else: while remaining < _ATEMPO_MIN: stages.append(_ATEMPO_MIN) remaining /= _ATEMPO_MIN stages.append(remaining) return stages # ── 时长计算 ──────────────────────────────────────────────────────────────── def adjust_duration(original_duration: float, config: SpeedConfig) -> float: """计算调速后的时长. 加速 → 时长变短;减速 → 时长变长。 """ if config.is_original or original_duration <= 0: return original_duration return original_duration / config.speed # ── 便捷方法 ──────────────────────────────────────────────────────────────── def build_clip_speed_filter( speed: float, pitch_correct: bool = True, ) -> tuple[str, str, SpeedConfig]: """便捷方法:从单一 speed 值生成视频+音频滤镜. 返回 (video_filter, audio_filter, config)。 """ config = SpeedConfig(speed=speed, pitch_correct=pitch_correct) config.clamp() return ( build_video_filter(config), build_audio_filter(config), config, ) def resolve_clip_speed( clip_config: dict[str, Any] | None, global_speed: float = DEFAULT_SPEED, ) -> float: """从 clip config 中解析 playback_speed,0 或缺失则使用全局速度.""" speed = clip_config.get("playback_speed", 0) if clip_config else 0 if not isinstance(speed, (int, float)) or speed <= 0: return global_speed return float(speed)