diff --git a/apps/api/app/api/routes/edit_plans.py b/apps/api/app/api/routes/edit_plans.py index 85ba23b6a..1d38ef976 100644 --- a/apps/api/app/api/routes/edit_plans.py +++ b/apps/api/app/api/routes/edit_plans.py @@ -209,6 +209,7 @@ class _PlanClipItem(BaseModel): start_time: float duration: float transition_effect: str + 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 old mode 100644 new mode 100755 index bf8122705..077a8e8e7 --- a/apps/api/app/services/edit_plan_service.py +++ b/apps/api/app/services/edit_plan_service.py @@ -281,6 +281,7 @@ class EditPlanService: start_time: float = 0.0, duration: float = 0.0, transition_effect: str = "cut", + playback_speed: float = 1.0, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """创建片段 @@ -301,6 +302,7 @@ class EditPlanService: start_time=start_time, duration=duration, transition_effect=transition_effect, + playback_speed=playback_speed, config=config, ) created = self._clip_repo.create(clip) @@ -324,6 +326,7 @@ class EditPlanService: start_time: Optional[float] = None, duration: Optional[float] = None, transition_effect: Optional[str] = None, + playback_speed: Optional[float] = None, config: Optional[dict[str, Any]] = None, ) -> EditPlanClip: """更新片段 @@ -333,6 +336,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, @@ -346,6 +358,7 @@ class EditPlanService: transition_effect=( transition_effect.strip() if transition_effect is not None else existing.transition_effect ), + 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 old mode 100644 new mode 100755 index 286b404fd..f86deda78 --- a/apps/worker/video_processing/render_audio.py +++ b/apps/worker/video_processing/render_audio.py @@ -16,6 +16,8 @@ import subprocess from dataclasses import dataclass, field from pathlib import Path +from video_processing.speed_engine import SpeedEngine + # 延迟导入避免循环依赖：unified_render_service 定义 ResolvedClip / RenderLayer， # 本模块提供音频函数供 unified_render_service 调用。 # 使用 from __future__ import annotations + TYPE_CHECKING 解决类型引用。 @@ -145,38 +147,104 @@ def concat_main_audio( # 单 clip，直接提取音频，截断到 min(clip有效时长, 视频总时长) clip = clips[0] effective_duration = clip_effective_duration(clip) - # 最终时长：取 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 - command = [ - FFMPEG_BIN, - "-y", - "-i", - str(clip.local_path), - "-vn", - "-acodec", - "aac", - "-b:a", - "128k", - ] - if final_duration > 0: - command.extend(["-t", f"{final_duration:.3f}"]) - command.append(str(output_path)) - run_ffmpeg(command) + if abs(speed - 1.0) < 1e-6: + # 原速：简单命令行 + command = [ + FFMPEG_BIN, + "-y", + "-i", + str(clip.local_path), + "-vn", + "-acodec", + "aac", + "-b:a", + "128k", + ] + if final_duration > 0: + command.extend(["-t", f"{final_duration:.3f}"]) + command.append(str(output_path)) + run_ffmpeg(command) + else: + # 有调速：用 filter_complex + atempo + from video_processing.speed_engine import SpeedConfig + speed_engine = SpeedEngine() + config = SpeedConfig(speed=float(speed)) + config.clamp() + atempo_filter = speed_engine.build_audio_filter(config) + + filter_parts: list[str] = [] + audio_filters = [] + if effective_duration > 0: + audio_filters.append(f"atrim=0:{effective_duration:.3f}") + audio_filters.append("asetpts=PTS-STARTPTS") + if atempo_filter: + audio_filters.append(atempo_filter) + + filter_parts.append(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) + 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: - filter_parts.append(f"[{i}:a]atrim=0:{effective_duration:.3f},asetpts=PTS-STARTPTS[a{i}]") + audio_filters.append(f"atrim=0:{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) + + if audio_filters: + filter_parts.append(f"[{i}:a]{','.join(audio_filters)}[a{i}]") else: filter_parts.append(f"[{i}:a]asetpts=PTS-STARTPTS[a{i}]") diff --git a/apps/worker/video_processing/speed_engine.py b/apps/worker/video_processing/speed_engine.py new file mode 100755 index 000000000..ea21f6616 --- /dev/null +++ b/apps/worker/video_processing/speed_engine.py @@ -0,0 +1,168 @@ +"""视频调速引擎 — 基于 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 a02c5a626..c6b26e22e 100755 --- a/apps/worker/video_processing/unified_render_service.py +++ b/apps/worker/video_processing/unified_render_service.py @@ -45,7 +45,7 @@ from video_processing.render_audio import RenderContext, merge_audio_video, mix_ from video_processing.render_subtitles import generate_ass_subtitles from video_processing.subtitle_generator import generate_ass_from_timeline from video_processing.tts_engine import TtsEngine - +from video_processing.speed_engine import SpeedConfig, SpeedEngine from packages.domain.tts_config import TtsConfig logger = logging.getLogger(__name__) @@ -66,6 +66,7 @@ class ResolvedClip: start_time: float = 0.0 duration: float = 0.0 # 0 表示使用素材完整时长 transition_effect: str = "cut" + playback_speed: float = 1.0 # 0 或 1.0 表示原速 config: dict[str, Any] = field(default_factory=dict) # 运行时填充 @@ -175,6 +176,7 @@ class UnifiedRenderService: self.output_fps = output_fps self.transition_duration = transition_duration self.asr_service = asr_service + self._speed_engine = SpeedEngine() def render(self) -> RenderResult: """执行渲染，返回 RenderResult. @@ -352,7 +354,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) @@ -948,6 +950,7 @@ class UnifiedRenderService: start_time=clip.start_time, duration=clip.duration, transition_effect=clip.transition_effect or "cut", + playback_speed=getattr(clip, "playback_speed", 1.0) or 1.0, config=clip.config or {}, actual_duration=actual_duration, ) @@ -1040,6 +1043,11 @@ class UnifiedRenderService: filters.append(f"trim=duration={effective_duration}") 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}") + # scale if role in ("overlay", "corner_voice"): pip_w = int(self.output_width * _PIP_SCALE) @@ -1077,8 +1085,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] if len(layer_labels) == 1: @@ -1214,11 +1222,12 @@ 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 +<<<<<<< HEAD # ── 画中画（PiP）相关方法 ────────────────────────────────────────────────── def _resolve_pip_sources(self, pip_config: PiPConfig) -> list[tuple[str, PiPLayerConfig, Path]]: @@ -1311,3 +1320,21 @@ 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/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py b/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py old mode 100644 new mode 100755 index 5e51b9dcc..a27e87599 --- a/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py +++ b/packages/adapters/sqlalchemy_impl/edit_plan_clip_repository.py @@ -54,6 +54,7 @@ class SQLAlchemyEditPlanClipRepository: start_time=clip.start_time, duration=clip.duration, transition_effect=clip.transition_effect, + playback_speed=clip.playback_speed, status=clip.status, config=clip.config, ) @@ -76,6 +77,7 @@ class SQLAlchemyEditPlanClipRepository: model.start_time = clip.start_time model.duration = clip.duration model.transition_effect = clip.transition_effect + model.playback_speed = clip.playback_speed model.status = clip.status model.config = clip.config model.updated_at = clip.updated_at @@ -120,6 +122,7 @@ class SQLAlchemyEditPlanClipRepository: start_time=model.start_time or 0.0, duration=model.duration or 0.0, transition_effect=model.transition_effect or "cut", + 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 fb955846f..fb5a8ef23 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -196,6 +196,7 @@ class EditPlanClipModel(Base): start_time = Column(Float, nullable=False, default=0.0) duration = Column(Float, nullable=False, default=0.0) transition_effect = Column(String(20), nullable=False, default="cut") + 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 old mode 100644 new mode 100755 index 15171c23c..b8913ea95 --- a/packages/domain/edit_plan_clip.py +++ b/packages/domain/edit_plan_clip.py @@ -50,6 +50,7 @@ class EditPlanClip: start_time: float = 0.0 duration: float = 0.0 transition_effect: str = "cut" + 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)) @@ -68,6 +69,7 @@ class EditPlanClip: start_time: float = 0.0, duration: float = 0.0, transition_effect: str = "cut", + playback_speed: float = 1.0, config: dict[str, Any] | None = None, ) -> EditPlanClip: """创建剪辑计划片段""" @@ -79,6 +81,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, @@ -91,6 +100,7 @@ class EditPlanClip: start_time=start_time, duration=duration, transition_effect=transition_effect.strip() or "cut", + 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..cce4ff4cb --- /dev/null +++ b/tests/unit/test_speed_engine.py @@ -0,0 +1,270 @@ +"""视频调速引擎单元测试.""" + +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