diff --git a/alembic/versions/040_add_playback_speed_to_edit_plan_clips.py b/alembic/versions/040_add_playback_speed_to_edit_plan_clips.py new file mode 100755 index 000000000..21f6431cc --- /dev/null +++ b/alembic/versions/040_add_playback_speed_to_edit_plan_clips.py @@ -0,0 +1,29 @@ +"""add playback_speed to edit_plan_clips + +Revision ID: 040_playback_speed +Revises: 039_transition_duration +Create Date: 2026-07-14 10:00:00.000000 +""" + +from __future__ import annotations + +import sqlalchemy as sa + +from alembic import op + +# revision identifiers, used by Alembic. +revision = "040_playback_speed" +down_revision = "039_transition_duration" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "edit_plan_clips", + sa.Column("playback_speed", sa.Float(), nullable=False, server_default="1.0"), + ) + + +def downgrade() -> None: + op.drop_column("edit_plan_clips", "playback_speed") diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 1e124bcbc..a830cd193 100644 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -211,6 +211,7 @@ class _PlanClipItem(BaseModel): duration: float transition_effect: str transition_duration: float + playback_speed: float = 1.0 status: str config: Optional[dict[str, Any]] = None created_at: datetime diff --git a/apps/api/app/services/edit_plan_service.py b/apps/api/app/services/edit_plan_service.py index 80280eb07..6ff73b7a6 100755 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -282,6 +282,7 @@ class EditPlanService: duration: float = 0.0, transition_effect: str = "cut", transition_duration: float = 0.0, + playback_speed: float = 1.0, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """创建片段 @@ -303,6 +304,7 @@ class EditPlanService: duration=duration, transition_effect=transition_effect, transition_duration=transition_duration, + playback_speed=playback_speed, config=config, ) created = self._clip_repo.create(clip) @@ -327,6 +329,7 @@ class EditPlanService: duration: Optional[float] = None, transition_effect: Optional[str] = None, transition_duration: Optional[float] = None, + playback_speed: Optional[float] = None, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """更新片段 @@ -336,6 +339,15 @@ class EditPlanService: """ existing = self.get_clip_or_raise(clip_id) + # 速度边界钳制 + if playback_speed is not None: + if playback_speed <= 0: + playback_speed = 1.0 + elif playback_speed < 0.25: + playback_speed = 0.25 + elif playback_speed > 4.0: + playback_speed = 4.0 + updated = EditPlanClip( id=existing.id, plan_id=existing.plan_id, @@ -352,6 +364,7 @@ class EditPlanService: transition_duration=( transition_duration if transition_duration is not None else existing.transition_duration ), + playback_speed=playback_speed if playback_speed is not None else existing.playback_speed, status=existing.status, config=config if config is not None else existing.config, created_at=existing.created_at, diff --git a/apps/worker/video_processing/render_audio.py b/apps/worker/video_processing/render_audio.py index 0aa7f4415..536f08309 100755 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -23,6 +23,7 @@ from typing import TYPE_CHECKING from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_has_audio, run_ffmpeg from video_processing.reverse_engine import ReverseConfig, ReverseEngine +from video_processing.speed_engine import SpeedEngine if TYPE_CHECKING: from video_processing.unified_render_service import RenderLayer, ResolvedClip @@ -225,53 +226,118 @@ def concat_main_audio( clip = clips[0] effective_duration = clip_effective_duration(clip) trim_start = getattr(clip, "start_time", 0) or 0 - # 最终时长:取 clip 有效时长和视频总时长的较小值 - # (视频总时长由主图层决定,但单 clip 场景下两者应该一致,仍做保护) - final_duration = effective_duration + speed = getattr(clip, "playback_speed", 1.0) or 1.0 + if not isinstance(speed, (int, float)) or speed <= 0: + speed = 1.0 + + # 调速后时长 + adjusted_duration = effective_duration / speed if abs(speed - 1.0) >= 1e-6 else effective_duration + # 最终时长:取调速后时长和视频总时长的较小值 + final_duration = adjusted_duration if video_duration > 0 and (final_duration <= 0 or final_duration > video_duration): final_duration = video_duration # 音频倒放 reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) - af_filters = [] - if reverse_config.enabled and reverse_config.reverse_audio: - reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) - if reverse_filter: - af_filters.append(reverse_filter) + has_reverse = reverse_config.enabled and reverse_config.reverse_audio + has_speed = abs(speed - 1.0) >= 1e-6 - command = [ - FFMPEG_BIN, - "-y", - "-i", - str(clip.local_path), - "-vn", - "-acodec", - "aac", - "-b:a", - "128k", - ] - if trim_start > 0: - command.extend(["-ss", f"{trim_start:.3f}"]) - if af_filters: - command.extend(["-af", ",".join(af_filters)]) - if final_duration > 0: - command.extend(["-t", f"{final_duration:.3f}"]) - command.append(str(output_path)) - run_ffmpeg(command) + if not has_speed and not has_reverse: + # 无调速无倒放:简单命令行,-ss 裁剪更高效 + command = [ + FFMPEG_BIN, + "-y", + "-i", + str(clip.local_path), + "-vn", + "-acodec", + "aac", + "-b:a", + "128k", + ] + if trim_start > 0: + command.extend(["-ss", f"{trim_start:.3f}"]) + if final_duration > 0: + command.extend(["-t", f"{final_duration:.3f}"]) + command.append(str(output_path)) + run_ffmpeg(command) + else: + # 有调速或倒放:用 filter_complex + speed_engine = SpeedEngine() + audio_filters = [] + if effective_duration > 0: + audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}") + audio_filters.append("asetpts=PTS-STARTPTS") + + # 音频调速 + if has_speed: + from video_processing.speed_engine import SpeedConfig + + config = SpeedConfig(speed=float(speed)) + config.clamp() + atempo_filter = speed_engine.build_audio_filter(config) + if atempo_filter: + audio_filters.append(atempo_filter) + + # 音频倒放 + if has_reverse: + reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration) + if reverse_filter: + audio_filters.append(reverse_filter) + + filter_parts: list[str] = [f"[0:a]{','.join(audio_filters)}[outa]"] + if video_duration > 0 and final_duration < adjusted_duration: + filter_parts.append(f"[outa]atrim=0:{final_duration:.3f}[final_audio]") + final_label = "final_audio" + else: + final_label = "outa" + + filter_complex = ";".join(filter_parts) + command = [ + FFMPEG_BIN, + "-y", + "-i", + str(clip.local_path), + "-filter_complex", + filter_complex, + "-map", + f"[{final_label}]", + "-acodec", + "aac", + "-b:a", + "128k", + str(output_path), + ] + run_ffmpeg(command) return # 多 clip,用 filter_complex concat input_args: list[str] = [] filter_parts: list[str] = [] + speed_engine = SpeedEngine() for i, clip in enumerate(clips): input_args.extend(["-i", str(clip.local_path)]) effective_duration = clip_effective_duration(clip) trim_start = getattr(clip, "start_time", 0) or 0 + speed = getattr(clip, "playback_speed", 1.0) or 1.0 + if not isinstance(speed, (int, float)) or speed <= 0: + speed = 1.0 + audio_filters: list[str] = [] if effective_duration > 0: audio_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}") audio_filters.append("asetpts=PTS-STARTPTS") + + # 音频调速 — atempo 多级串联 + if abs(speed - 1.0) >= 1e-6: + from video_processing.speed_engine import SpeedConfig + + config = SpeedConfig(speed=float(speed)) + config.clamp() + atempo_filter = speed_engine.build_audio_filter(config) + if atempo_filter: + audio_filters.append(atempo_filter) else: audio_filters.append("asetpts=PTS-STARTPTS") diff --git a/apps/worker/video_processing/speed_engine.py b/apps/worker/video_processing/speed_engine.py new file mode 100755 index 000000000..1ab365b17 --- /dev/null +++ b/apps/worker/video_processing/speed_engine.py @@ -0,0 +1,167 @@ +"""视频调速引擎 — 基于 FFmpeg setpts + atempo 的速度调整能力。 + +支持: +- 0.25x ~ 4x 变速范围 +- 视频调速(setpts) +- 音频调速(atempo,多级串联处理超范围值) +- 音调修正(pitch_correct,默认开启) +- 边界自动钳制,不阻断渲染 +""" + +from dataclasses import dataclass +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 + + +class SpeedEngine: + """调速引擎 — 生成 FFmpeg 调速滤镜链。 + + 用法: + engine = SpeedEngine() + video_filter = engine.build_video_filter(config) + audio_filter = engine.build_audio_filter(config) + new_duration = engine.adjust_duration(duration, 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}" + + 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) + + @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 + + 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 + + def build_clip_speed_filter( + self, + 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 ( + 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 或缺失则使用全局速度。""" + 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/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py index 42d6f960b..c465277b0 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -45,6 +45,7 @@ from video_processing.pip_engine import PiPConfig, PiPEngine, PiPLayerConfig from video_processing.render_audio import RenderContext, merge_audio_video, mix_audio from video_processing.render_subtitles import generate_ass_subtitles from video_processing.reverse_engine import ReverseConfig, ReverseEngine +from video_processing.speed_engine import SpeedConfig, SpeedEngine from video_processing.sticker_engine import StickerEngine, parse_stickers_from_config from video_processing.subtitle_generator import generate_ass_from_timeline from video_processing.transition_engine import TransitionEngine @@ -73,6 +74,7 @@ class ResolvedClip: duration: float = 0.0 # 0 表示使用素材完整时长 transition_effect: str = "cut" transition_duration: float = 0.0 # 0 表示使用全局默认值 + playback_speed: float = 1.0 # 0 或 1.0 表示原速 config: dict[str, Any] = field(default_factory=dict) # 运行时填充 @@ -186,6 +188,7 @@ class UnifiedRenderService: self.asr_service = asr_service self.bgm_path = bgm_path self._transition_engine = TransitionEngine(default_duration=transition_duration) + self._speed_engine = SpeedEngine() def render(self) -> RenderResult: """执行渲染,返回 RenderResult. @@ -493,7 +496,7 @@ class UnifiedRenderService: if not main_layer or not main_layer.clips: return 0.0 - total = sum(UnifiedRenderService._clip_effective_duration(c) for c in main_layer.clips) + total = sum(UnifiedRenderService._clip_adjusted_duration(c) for c in main_layer.clips) # 减去转场重叠时间(粗略估算) n_clips = len(main_layer.clips) @@ -1197,6 +1200,7 @@ class UnifiedRenderService: duration=final_duration, transition_effect=clip.transition_effect or "cut", transition_duration=getattr(clip, "transition_duration", 0.0) or 0.0, + playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0, config=clip_config, actual_duration=actual_duration, trim_config=effective_trim, @@ -1294,6 +1298,11 @@ class UnifiedRenderService: filters.append(f"trim=duration={effective_duration:.3f}") filters.append("setpts=PTS-STARTPTS") + # 调速 — 基于 setpts 改变播放速度 + speed = UnifiedRenderService._clip_speed(clip) + if abs(speed - 1.0) >= 1e-6: + filters.append(f"setpts=PTS/{speed:.4f}") + # 倒放滤镜(在 trim 之后、scale 之前应用) reverse_config = ReverseConfig.from_dict(clip.config.get("reverse")) if reverse_config.enabled and reverse_config.reverse_video: @@ -1351,8 +1360,8 @@ class UnifiedRenderService: for layer in layers: layer_clip_indices = [all_clips.index(c) for c in layer.clips] layer_labels = [preprocessed_labels[i] for i in layer_clip_indices] - # 使用 trim 后的有效时长,与 Step 1 的 trim=duration 保持一致 - layer_durations = [UnifiedRenderService._clip_effective_duration(all_clips[i]) for i in layer_clip_indices] + # 使用调速后的实际时长,与 Step 1 的调速处理保持一致 + layer_durations = [UnifiedRenderService._clip_adjusted_duration(all_clips[i]) for i in layer_clip_indices] layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices] layer_transition_durations = [all_clips[i].transition_duration for i in layer_clip_indices] @@ -1597,7 +1606,7 @@ class UnifiedRenderService: @staticmethod def _clip_effective_duration(clip: ResolvedClip) -> float: - """计算 clip 的有效时长.""" + """计算 clip 的有效时长(原速 trim 后时长).""" if clip.duration > 0: return min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration return clip.actual_duration if clip.actual_duration > 0 else 0.0 @@ -1694,3 +1703,20 @@ class UnifiedRenderService: ) return new_filter, new_input_args + + @staticmethod + def _clip_speed(clip: ResolvedClip) -> float: + """获取 clip 的播放速度,无效值回退到 1.0.""" + speed = getattr(clip, "playback_speed", 1.0) + if not isinstance(speed, (int, float)) or speed <= 0: + return 1.0 + return float(speed) + + @staticmethod + def _clip_adjusted_duration(clip: ResolvedClip) -> float: + """计算调速后的 clip 实际时长(用于拼接计算).""" + base = UnifiedRenderService._clip_effective_duration(clip) + speed = UnifiedRenderService._clip_speed(clip) + if abs(speed - 1.0) < 1e-6: + return base + return base / speed diff --git a/docs/schema-metadata-snapshot.json b/docs/schema-metadata-snapshot.json index 81b20bec0..b12a3cf55 100644 --- a/docs/schema-metadata-snapshot.json +++ b/docs/schema-metadata-snapshot.json @@ -866,6 +866,14 @@ "type": "FLOAT", "unique": false }, + { + "index": false, + "name": "playback_speed", + "nullable": false, + "primary_key": false, + "type": "FLOAT", + "unique": false + }, { "index": true, "name": "status", diff --git a/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py b/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py index 8ae8a81ef..4c241269f 100755 --- a/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py +++ b/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py @@ -55,6 +55,7 @@ class SQLAlchemyEditPlanClipRepository: duration=clip.duration, transition_effect=clip.transition_effect, transition_duration=clip.transition_duration, + playback_speed=clip.playback_speed, status=clip.status, config=clip.config, ) @@ -78,6 +79,7 @@ class SQLAlchemyEditPlanClipRepository: model.duration = clip.duration model.transition_effect = clip.transition_effect model.transition_duration = clip.transition_duration + model.playback_speed = clip.playback_speed model.status = clip.status model.config = clip.config model.updated_at = clip.updated_at @@ -123,6 +125,7 @@ class SQLAlchemyEditPlanClipRepository: duration=model.duration or 0.0, transition_effect=model.transition_effect or "cut", transition_duration=getattr(model, "transition_duration", 0.0) or 0.0, + playback_speed=model.playback_speed or 1.0, status=EditPlanClipStatus(model.status) if model.status else EditPlanClipStatus.PENDING, config=model.config or {}, created_at=model.created_at, diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 00d4440b6..b7436c166 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -197,6 +197,7 @@ class EditPlanClipModel(Base): duration = Column(Float, nullable=False, default=0.0) transition_effect = Column(String(20), nullable=False, default="cut") transition_duration = Column(Float, nullable=False, default=0.0) + playback_speed = Column(Float, nullable=False, default=1.0) status = Column(String(20), nullable=False, default="pending", index=True) config = Column(JSON, nullable=False, default=dict) created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc)) diff --git a/packages/domain/edit_plan_clip.py b/packages/domain/edit_plan_clip.py index 8fe9e9051..748f2567a 100755 --- a/packages/domain/edit_plan_clip.py +++ b/packages/domain/edit_plan_clip.py @@ -51,6 +51,7 @@ class EditPlanClip: duration: float = 0.0 transition_effect: str = "cut" transition_duration: float = 0.0 # 0 表示使用全局默认值 + playback_speed: float = 1.0 # 0 或 1.0 表示原速,范围 0.25~4.0 status: EditPlanClipStatus = EditPlanClipStatus.PENDING config: dict[str, Any] = field(default_factory=dict) created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @@ -70,6 +71,7 @@ class EditPlanClip: duration: float = 0.0, transition_effect: str = "cut", transition_duration: float = 0.0, + playback_speed: float = 1.0, config: dict[str, Any] | None = None, ) -> EditPlanClip: """创建剪辑计划片段""" @@ -81,6 +83,13 @@ class EditPlanClip: raise ValueError("start_time 不能为负数") if duration < 0: raise ValueError("duration 不能为负数") + # 速度边界钳制 + if playback_speed <= 0: + playback_speed = 1.0 + elif playback_speed < 0.25: + playback_speed = 0.25 + elif playback_speed > 4.0: + playback_speed = 4.0 return cls( id=uuid4().hex, @@ -94,6 +103,7 @@ class EditPlanClip: duration=duration, transition_effect=transition_effect.strip() or "cut", transition_duration=max(0.0, transition_duration), + playback_speed=playback_speed, status=EditPlanClipStatus.PENDING, config=config or {}, ) diff --git a/tests/unit/test_speed_engine.py b/tests/unit/test_speed_engine.py new file mode 100755 index 000000000..6da62ef06 --- /dev/null +++ b/tests/unit/test_speed_engine.py @@ -0,0 +1,269 @@ +"""视频调速引擎单元测试.""" + +import pytest +from video_processing.speed_engine import ( + MAX_SPEED, + MIN_SPEED, + SpeedConfig, + SpeedEngine, +) + +# ─── SpeedConfig 解析与校验 ────────────────────────────────── + + +class TestSpeedConfig: + def test_default_values(self): + config = SpeedConfig() + assert config.speed == 1.0 + assert config.pitch_correct is True + + def test_parse_none(self): + config = SpeedConfig.parse(None) + assert config.speed == 1.0 + assert config.pitch_correct is True + + def test_parse_empty_dict(self): + config = SpeedConfig.parse({}) + assert config.speed == 1.0 + + def test_parse_valid_speed(self): + config = SpeedConfig.parse({"speed": 2.0}) + assert config.speed == 2.0 + + def test_parse_pitch_correct_false(self): + config = SpeedConfig.parse({"pitch_correct": False}) + assert config.pitch_correct is False + + def test_parse_invalid_speed_type(self): + config = SpeedConfig.parse({"speed": "fast"}) + assert config.speed == 1.0 + + def test_parse_invalid_pitch_type(self): + config = SpeedConfig.parse({"pitch_correct": "yes"}) + assert config.pitch_correct is True + + def test_clamp_below_min(self): + config = SpeedConfig(speed=0.1) + config.clamp() + assert config.speed == MIN_SPEED + + def test_clamp_zero(self): + config = SpeedConfig(speed=0) + config.clamp() + assert config.speed == 1.0 + + def test_clamp_negative(self): + config = SpeedConfig(speed=-1.0) + config.clamp() + assert config.speed == 1.0 + + def test_clamp_above_max(self): + config = SpeedConfig(speed=10.0) + config.clamp() + assert config.speed == MAX_SPEED + + def test_clamp_within_range(self): + config = SpeedConfig(speed=1.5) + config.clamp() + assert config.speed == 1.5 + + def test_is_original_true(self): + config = SpeedConfig(speed=1.0) + assert config.is_original is True + + def test_is_original_false(self): + config = SpeedConfig(speed=1.5) + assert config.is_original is False + + def test_parse_clamps_automatically(self): + """parse 方法应该自动调用 clamp.""" + config = SpeedConfig.parse({"speed": 100.0}) + assert config.speed == MAX_SPEED + + +# ─── SpeedEngine 视频滤镜 ──────────────────────────────────── + + +class TestSpeedEngineVideoFilter: + def setup_method(self): + self.engine = SpeedEngine() + + def test_original_speed_returns_empty(self): + config = SpeedConfig(speed=1.0) + assert self.engine.build_video_filter(config) == "" + + def test_double_speed(self): + config = SpeedConfig(speed=2.0) + result = self.engine.build_video_filter(config) + assert "setpts=PTS/2.0" in result + + def test_half_speed(self): + config = SpeedConfig(speed=0.5) + result = self.engine.build_video_filter(config) + assert "setpts=PTS/0.5" in result + + def test_quarter_speed(self): + config = SpeedConfig(speed=0.25) + result = self.engine.build_video_filter(config) + assert "setpts=PTS/0.25" in result + + def test_quad_speed(self): + config = SpeedConfig(speed=4.0) + result = self.engine.build_video_filter(config) + assert "setpts=PTS/4.0" in result + + +# ─── SpeedEngine 音频滤镜(atempo 多级串联) ───────────────── + + +class TestSpeedEngineAudioFilter: + def setup_method(self): + self.engine = SpeedEngine() + + def test_original_speed_returns_empty(self): + config = SpeedConfig(speed=1.0) + assert self.engine.build_audio_filter(config) == "" + + def test_double_speed_single_stage(self): + """2x 在 atempo 单级范围内,只需一个 atempo.""" + config = SpeedConfig(speed=2.0) + result = self.engine.build_audio_filter(config) + assert result == "atempo=2.0000" + + def test_half_speed_single_stage(self): + config = SpeedConfig(speed=0.5) + result = self.engine.build_audio_filter(config) + assert result == "atempo=0.5000" + + def test_quad_speed_two_stages(self): + """4x 需要两级 atempo: 2.0 * 2.0.""" + config = SpeedConfig(speed=4.0) + result = self.engine.build_audio_filter(config) + assert result == "atempo=2.0000,atempo=2.0000" + + def test_quarter_speed_two_stages(self): + """0.25x 需要两级 atempo: 0.5 * 0.5.""" + config = SpeedConfig(speed=0.25) + result = self.engine.build_audio_filter(config) + assert result == "atempo=0.5000,atempo=0.5000" + + def test_triple_speed_two_stages(self): + """3x: 2.0 * 1.5.""" + config = SpeedConfig(speed=3.0) + result = self.engine.build_audio_filter(config) + parts = result.split(",") + assert len(parts) == 2 + assert "atempo=2.0000" in parts + assert "atempo=1.5000" in parts + + def test_03_speed_two_stages(self): + """0.3x: 0.5 * 0.6.""" + config = SpeedConfig(speed=0.3) + result = self.engine.build_audio_filter(config) + parts = result.split(",") + assert len(parts) == 2 + assert "atempo=0.5000" in parts + assert "atempo=0.6000" in parts + + def test_split_atempo_inside_range(self): + """0.5~2.0 范围内只返回一级.""" + stages = SpeedEngine._split_atempo_stages(1.5) + assert len(stages) == 1 + assert stages[0] == 1.5 + + def test_split_atempo_boundary_min(self): + stages = SpeedEngine._split_atempo_stages(0.5) + assert len(stages) == 1 + assert stages[0] == 0.5 + + def test_split_atempo_boundary_max(self): + stages = SpeedEngine._split_atempo_stages(2.0) + assert len(stages) == 1 + assert stages[0] == 2.0 + + def test_split_atempo_product_equals_speed(self): + """所有级联的乘积应该等于原速度.""" + test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0] + for speed in test_cases: + stages = SpeedEngine._split_atempo_stages(speed) + product = 1.0 + for s in stages: + product *= s + assert abs(product - speed) < 1e-6, f"speed={speed}, stages={stages}, product={product}" + + def test_split_atempo_all_in_range(self): + """所有级都应该在 0.5~2.0 范围内.""" + test_cases = [0.25, 0.3, 0.5, 0.75, 1.0, 1.5, 2.0, 3.0, 4.0] + for speed in test_cases: + stages = SpeedEngine._split_atempo_stages(speed) + for s in stages: + assert 0.5 <= s <= 2.0, f"speed={speed}, stage={s} out of range" + + +# ─── SpeedEngine 时长计算 ──────────────────────────────────── + + +class TestSpeedEngineDuration: + def setup_method(self): + self.engine = SpeedEngine() + + def test_original_speed_same_duration(self): + config = SpeedConfig(speed=1.0) + assert self.engine.adjust_duration(10.0, config) == 10.0 + + def test_double_speed_half_duration(self): + config = SpeedConfig(speed=2.0) + assert self.engine.adjust_duration(10.0, config) == 5.0 + + def test_half_speed_double_duration(self): + config = SpeedConfig(speed=0.5) + assert self.engine.adjust_duration(10.0, config) == 20.0 + + def test_quad_speed_quarter_duration(self): + config = SpeedConfig(speed=4.0) + assert self.engine.adjust_duration(10.0, config) == 2.5 + + def test_zero_duration(self): + config = SpeedConfig(speed=2.0) + assert self.engine.adjust_duration(0.0, config) == 0.0 + + def test_negative_duration(self): + config = SpeedConfig(speed=2.0) + assert self.engine.adjust_duration(-1.0, config) == -1.0 + + +# ─── SpeedEngine 便捷方法 ──────────────────────────────────── + + +class TestSpeedEngineHelper: + def setup_method(self): + self.engine = SpeedEngine() + + def test_build_clip_speed_filter_original(self): + v_f, a_f, cfg = self.engine.build_clip_speed_filter(1.0) + assert v_f == "" + assert a_f == "" + assert cfg.speed == 1.0 + + def test_build_clip_speed_filter_2x(self): + v_f, a_f, cfg = self.engine.build_clip_speed_filter(2.0) + assert "setpts=PTS/2.0" in v_f + assert "atempo=2.0" in a_f + assert cfg.speed == 2.0 + + def test_build_clip_speed_clamped(self): + _, _, cfg = self.engine.build_clip_speed_filter(100.0) + assert cfg.speed == MAX_SPEED + + def test_resolve_clip_speed_default(self): + assert SpeedEngine.resolve_clip_speed({}) == 1.0 + assert SpeedEngine.resolve_clip_speed(None) == 1.0 + + def test_resolve_clip_speed_zero_uses_global(self): + assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 1.5) == 1.5 + + def test_resolve_clip_speed_custom(self): + assert SpeedEngine.resolve_clip_speed({"playback_speed": 2.0}) == 2.0 + + def test_resolve_clip_speed_invalid_type(self): + assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}) == 1.0