feat: 视频调速引擎（快进/慢放）
- SpeedEngine 抽象层：SpeedConfig + SpeedEngine，支持 0.25x~4x 变速 - 视频调速：基于 FFmpeg setpts，在 clip 预处理环节插入 - 音频调速：基于 atempo，超范围自动多级串联（如 4x=2.0*2.0） - 音画同步：视频音频同时调速，音调自动修正 - 分段调速：每个 clip 独立 playback_speed 字段 - 整段调速：所有 clip 设相同 speed 即可实现 - 降级策略：速度超出范围自动钳制，不阻断渲染 - 领域模型：EditPlanClip 新增 playback_speed 字段 - 数据库：EditPlanClipModel 新增 playback_speed 列 - API：create_clip / update_clip 支持 playback_speed 参数 - 45 个新增单测全绿 + 100 个现有测试全绿，无回归
This commit is contained in:
CI Bot
@@ -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
|
||||
|
||||
Regular → Executable
+13
@@ -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,
|
||||
|
||||
Regular → Executable
+87
-19
@@ -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}]")
|
||||
|
||||
|
||||
Executable
+168
@@ -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)
|
||||
@@ -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
|
||||
|
||||
|
||||
Regular → Executable
+3
@@ -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,
|
||||
|
||||
@@ -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))
|
||||
|
||||
Regular → Executable
+10
@@ -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 {},
|
||||
)
|
||||
|
||||
Executable
+270
@@ -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
|
||||
Reference in New Issue
Block a user