diff --git a/apps/worker/video_processing/speed_engine.py b/apps/worker/video_processing/speed_engine.py index 1ab365b17..cc9798f65 100755 --- a/apps/worker/video_processing/speed_engine.py +++ b/apps/worker/video_processing/speed_engine.py @@ -1,4 +1,4 @@ -"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。 +"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力. 支持: - 0.25x ~ 4x 变速范围 @@ -6,147 +6,57 @@ - 音频调速(atempo,多级串联处理超范围值) - 音调修正(pitch_correct,默认开启) - 边界自动钳制,不阻断渲染 + +注:核心领域模型已抽离到 packages/domain/speed_config.py, +本模块保留薄包装层,确保向后兼容。 """ -from dataclasses import dataclass +from __future__ import annotations + from typing import Optional -# ─── 常量 ─────────────────────────────────────────────── -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: Optional[dict]) -> "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 +from packages.domain.speed_config import ( # noqa: F401 — 向后兼容 + DEFAULT_SPEED, + MAX_SPEED, + MIN_SPEED, + SpeedConfig, + _split_atempo_stages, + adjust_duration as _adjust_duration_base, + build_audio_filter as _build_audio_filter_base, + build_video_filter as _build_video_filter_base, + resolve_clip_speed as _resolve_clip_speed_base, +) class SpeedEngine: - """调速引擎 — 生成 FFmpeg 调速滤镜链。 + """调速引擎 — 生成 FFmpeg 调速滤镜链. - 用法: - engine = SpeedEngine() - video_filter = engine.build_video_filter(config) - audio_filter = engine.build_audio_filter(config) - new_duration = engine.adjust_duration(duration, config) + 薄包装层,实际逻辑委托给 packages.domain.speed_config。 """ def build_video_filter(self, config: SpeedConfig) -> str: - """生成视频调速滤镜字符串。 - - 返回 setpts 滤镜表达式,原速时返回空字符串。 - """ - if config.is_original: - return "" - # setpts=PTS/speed — speed>1 加速,speed<1 减速 - return f"setpts=PTS/{config.speed:.4f}" + """生成视频调速滤镜字符串.""" + return _build_video_filter_base(config) def build_audio_filter(self, 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] = self._split_atempo_stages(speed) - return ",".join(f"atempo={s:.4f}" for s in stages) + """生成音频调速滤镜字符串.""" + return _build_audio_filter_base(config) @staticmethod 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 + """将速度拆分为多级 atempo 串联(内部方法,向后兼容).""" + return _split_atempo_stages(speed) def adjust_duration(self, original_duration: float, config: SpeedConfig) -> float: - """计算调速后的时长。 - - 加速 → 时长变短;减速 → 时长变长。 - """ - if config.is_original or original_duration <= 0: - return original_duration - return original_duration / config.speed + """计算调速后的时长.""" + return _adjust_duration_base(original_duration, config) def build_clip_speed_filter( self, speed: float, pitch_correct: bool = True, ) -> tuple[str, str, SpeedConfig]: - """便捷方法:从单一 speed 值生成视频+音频滤镜。 - - 返回 (video_filter, audio_filter, config)。 - """ + """便捷方法:从单一 speed 值生成视频+音频滤镜.""" config = SpeedConfig(speed=speed, pitch_correct=pitch_correct) config.clamp() return ( @@ -160,8 +70,5 @@ class SpeedEngine: clip_config: dict, 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) + """从 clip config 中解析 playback_speed,0 或缺失则使用全局速度.""" + return _resolve_clip_speed_base(clip_config, global_speed) diff --git a/packages/domain/speed_config.py b/packages/domain/speed_config.py new file mode 100755 index 000000000..955facc1a --- /dev/null +++ b/packages/domain/speed_config.py @@ -0,0 +1,179 @@ +"""调速配置领域模型 — 纯逻辑,无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) diff --git a/tests/unit/test_speed_config.py b/tests/unit/test_speed_config.py new file mode 100755 index 000000000..a5480a621 --- /dev/null +++ b/tests/unit/test_speed_config.py @@ -0,0 +1,313 @@ +"""speed_config 领域模型单测.""" + +from __future__ import annotations + +import pytest + +from packages.domain.speed_config import ( + DEFAULT_SPEED, + MAX_SPEED, + MIN_SPEED, + SpeedConfig, + adjust_duration, + build_audio_filter, + build_video_filter, + build_clip_speed_filter, + resolve_clip_speed, +) + +# ── 常量测试 ──────────────────────────────────────────────────────────────── + + +class TestConstants: + def test_min_speed(self): + assert MIN_SPEED == 0.25 + + def test_max_speed(self): + assert MAX_SPEED == 4.0 + + def test_default_speed(self): + assert DEFAULT_SPEED == 1.0 + + +# ── SpeedConfig.parse 测试 ──────────────────────────────────────────────── + + +class TestSpeedConfigParse: + def test_none_returns_default(self): + cfg = SpeedConfig.parse(None) + assert cfg.speed == DEFAULT_SPEED + assert cfg.pitch_correct is True + + def test_empty_dict_returns_default(self): + cfg = SpeedConfig.parse({}) + assert cfg.speed == DEFAULT_SPEED + + def test_invalid_type_returns_default(self): + cfg = SpeedConfig.parse("not_a_dict") + assert cfg.speed == DEFAULT_SPEED + + def test_valid_speed(self): + cfg = SpeedConfig.parse({"speed": 2.0}) + assert cfg.speed == 2.0 + + def test_speed_clamped_low(self): + cfg = SpeedConfig.parse({"speed": 0.1}) + assert cfg.speed == MIN_SPEED + + def test_speed_clamped_high(self): + cfg = SpeedConfig.parse({"speed": 5.0}) + assert cfg.speed == MAX_SPEED + + def test_zero_speed_returns_default(self): + cfg = SpeedConfig.parse({"speed": 0}) + assert cfg.speed == DEFAULT_SPEED + + def test_negative_speed_returns_default(self): + cfg = SpeedConfig.parse({"speed": -1.0}) + assert cfg.speed == DEFAULT_SPEED + + def test_pitch_correct_false(self): + cfg = SpeedConfig.parse({"pitch_correct": False}) + assert cfg.pitch_correct is False + + def test_pitch_correct_invalid_type_defaults_true(self): + cfg = SpeedConfig.parse({"pitch_correct": "yes"}) + assert cfg.pitch_correct is True + + def test_string_speed_invalid_uses_default(self): + cfg = SpeedConfig.parse({"speed": "fast"}) + assert cfg.speed == DEFAULT_SPEED + + +# ── SpeedConfig.clamp 测试 ──────────────────────────────────────────────── + + +class TestClamp: + def test_already_valid_unchanged(self): + cfg = SpeedConfig(speed=1.5) + cfg.clamp() + assert cfg.speed == 1.5 + + def test_below_min_clamped(self): + cfg = SpeedConfig(speed=0.1) + cfg.clamp() + assert cfg.speed == MIN_SPEED + + def test_above_max_clamped(self): + cfg = SpeedConfig(speed=10.0) + cfg.clamp() + assert cfg.speed == MAX_SPEED + + def test_zero_defaults(self): + cfg = SpeedConfig(speed=0.0) + cfg.clamp() + assert cfg.speed == DEFAULT_SPEED + + def test_negative_defaults(self): + cfg = SpeedConfig(speed=-2.0) + cfg.clamp() + assert cfg.speed == DEFAULT_SPEED + + def test_exact_min_stays(self): + cfg = SpeedConfig(speed=MIN_SPEED) + cfg.clamp() + assert cfg.speed == MIN_SPEED + + def test_exact_max_stays(self): + cfg = SpeedConfig(speed=MAX_SPEED) + cfg.clamp() + assert cfg.speed == MAX_SPEED + + +# ── is_original / is_fast / is_slow 测试 ───────────────────────────────── + + +class TestSpeedProperties: + def test_is_original_true(self): + cfg = SpeedConfig(speed=1.0) + assert cfg.is_original is True + + def test_is_original_false_fast(self): + cfg = SpeedConfig(speed=2.0) + assert cfg.is_original is False + + def test_is_original_false_slow(self): + cfg = SpeedConfig(speed=0.5) + assert cfg.is_original is False + + def test_is_original_near_one(self): + cfg = SpeedConfig(speed=1.0000001) + assert cfg.is_original is True + + def test_is_fast_true(self): + cfg = SpeedConfig(speed=2.0) + assert cfg.is_fast is True + + def test_is_fast_false(self): + cfg = SpeedConfig(speed=0.5) + assert cfg.is_fast is False + + def test_is_false_for_original(self): + cfg = SpeedConfig(speed=1.0) + assert cfg.is_fast is False + assert cfg.is_slow is False + + def test_is_slow_true(self): + cfg = SpeedConfig(speed=0.5) + assert cfg.is_slow is True + + def test_is_slow_false(self): + cfg = SpeedConfig(speed=2.0) + assert cfg.is_slow is False + + +# ── build_video_filter 测试 ─────────────────────────────────────────────── + + +class TestBuildVideoFilter: + def test_original_speed_empty(self): + cfg = SpeedConfig(speed=1.0) + assert build_video_filter(cfg) == "" + + def test_fast_speed_setpts(self): + cfg = SpeedConfig(speed=2.0) + result = build_video_filter(cfg) + assert "setpts=PTS/2.0000" in result + + def test_slow_speed_setpts(self): + cfg = SpeedConfig(speed=0.5) + result = build_video_filter(cfg) + assert "setpts=PTS/0.5000" in result + + def test_format_four_decimals(self): + cfg = SpeedConfig(speed=1.5) + result = build_video_filter(cfg) + assert "1.5000" in result + + +# ── build_audio_filter 测试 ─────────────────────────────────────────────── + + +class TestBuildAudioFilter: + def test_original_speed_empty(self): + cfg = SpeedConfig(speed=1.0) + assert build_audio_filter(cfg) == "" + + def test_single_stage_within_range(self): + cfg = SpeedConfig(speed=1.5) + result = build_audio_filter(cfg) + assert result == "atempo=1.5000" + assert result.count("atempo") == 1 + + def test_fast_two_stages(self): + cfg = SpeedConfig(speed=3.0) + result = build_audio_filter(cfg) + assert result.count("atempo") == 2 + # 2.0 * 1.5 = 3.0 + assert "atempo=2.0000" in result + assert "atempo=1.5000" in result + + def test_max_speed_two_stages(self): + cfg = SpeedConfig(speed=4.0) + result = build_audio_filter(cfg) + assert result.count("atempo") == 2 + # 2.0 * 2.0 = 4.0 + assert result == "atempo=2.0000,atempo=2.0000" + + def test_slow_two_stages(self): + cfg = SpeedConfig(speed=0.25) + result = build_audio_filter(cfg) + assert result.count("atempo") == 2 + # 0.5 * 0.5 = 0.25 + assert result == "atempo=0.5000,atempo=0.5000" + + def test_slow_single_stage(self): + cfg = SpeedConfig(speed=0.8) + result = build_audio_filter(cfg) + assert result == "atempo=0.8000" + assert result.count("atempo") == 1 + + def test_exactly_two_point_zero_single(self): + cfg = SpeedConfig(speed=2.0) + result = build_audio_filter(cfg) + assert result.count("atempo") == 1 + assert "atempo=2.0000" in result + + def test_exactly_half_single(self): + cfg = SpeedConfig(speed=0.5) + result = build_audio_filter(cfg) + assert result.count("atempo") == 1 + assert "atempo=0.5000" in result + + +# ── adjust_duration 测试 ────────────────────────────────────────────────── + + +class TestAdjustDuration: + def test_original_speed_unchanged(self): + cfg = SpeedConfig(speed=1.0) + assert adjust_duration(10.0, cfg) == 10.0 + + def test_double_speed_halved(self): + cfg = SpeedConfig(speed=2.0) + assert adjust_duration(10.0, cfg) == 5.0 + + def test_half_speed_doubled(self): + cfg = SpeedConfig(speed=0.5) + assert adjust_duration(10.0, cfg) == 20.0 + + def test_zero_duration_unchanged(self): + cfg = SpeedConfig(speed=2.0) + assert adjust_duration(0.0, cfg) == 0.0 + + def test_negative_duration_unchanged(self): + cfg = SpeedConfig(speed=2.0) + assert adjust_duration(-5.0, cfg) == -5.0 + + +# ── build_clip_speed_filter 测试 ───────────────────────────────────────── + + +class TestBuildClipSpeedFilter: + def test_normal_speed(self): + vf, af, cfg = build_clip_speed_filter(2.0) + assert vf == "setpts=PTS/2.0000" + assert "atempo=2.0000" in af + assert cfg.speed == 2.0 + + def test_clamped_speed(self): + vf, af, cfg = build_clip_speed_filter(10.0) + assert cfg.speed == MAX_SPEED + + def test_pitch_correct_param(self): + vf, af, cfg = build_clip_speed_filter(1.5, pitch_correct=False) + assert cfg.pitch_correct is False + + def test_original_speed_empty_filters(self): + vf, af, cfg = build_clip_speed_filter(1.0) + assert vf == "" + assert af == "" + + +# ── resolve_clip_speed 测试 ────────────────────────────────────────────── + + +class TestResolveClipSpeed: + def test_none_config_uses_global(self): + assert resolve_clip_speed(None, 1.5) == 1.5 + + def test_no_playback_speed_uses_global(self): + assert resolve_clip_speed({}, 1.5) == 1.5 + + def test_zero_speed_uses_global(self): + assert resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5 + + def test_valid_speed_returns_speed(self): + assert resolve_clip_speed({"playback_speed": 2.0}, 1.0) == 2.0 + + def test_invalid_type_uses_global(self): + assert resolve_clip_speed({"playback_speed": "fast"}, 1.0) == 1.0 + + def test_default_global_speed(self): + assert resolve_clip_speed({}) == DEFAULT_SPEED