You've already forked FrameTour-RenderWorker
feat(video): 添加原始变速效果支持
- 在常量定义中新增 ospeed 效果类型用于兼容旧模板 - 在任务域中实现 get_ospeed_params 方法解析变速参数 - 修改视频渲染处理器合并 speed 与 ospeed 效果计算 - 更新时长计算逻辑以正确处理 ospeed 变速影响 - 新增 ospeed 参数验证和边界值处理机制 - 添加完整的 ospeed 效果单元测试覆盖各种场景
This commit is contained in:
@@ -39,6 +39,7 @@ EFFECT_TYPES = (
|
||||
'cameraShot', # 相机定格效果:在指定时间点冻结画面
|
||||
'zoom', # 缩放效果(预留)
|
||||
'blur', # 模糊效果(预留)
|
||||
'ospeed', # 原始变速效果(兼容旧模板)
|
||||
)
|
||||
|
||||
# 硬件加速类型
|
||||
|
||||
@@ -46,6 +46,7 @@ EFFECT_TYPES = {
|
||||
'cameraShot', # 相机定格效果
|
||||
'zoom', # 缩放效果(预留)
|
||||
'blur', # 模糊效果(预留)
|
||||
'ospeed', # 原始变速效果(兼容旧模板)
|
||||
}
|
||||
|
||||
|
||||
@@ -189,6 +190,20 @@ class Effect:
|
||||
|
||||
return (start_sec, scale_factor, duration_sec)
|
||||
|
||||
def get_ospeed_params(self) -> float:
|
||||
"""获取 ospeed 效果参数,返回 PTS 乘数(>0),无效时返回 1.0"""
|
||||
if self.effect_type != 'ospeed':
|
||||
return 1.0
|
||||
if not self.params:
|
||||
return 1.0
|
||||
try:
|
||||
factor = float(self.params.strip())
|
||||
except ValueError:
|
||||
return 1.0
|
||||
if not isfinite(factor) or factor <= 0:
|
||||
return 1.0
|
||||
return factor
|
||||
|
||||
|
||||
@dataclass
|
||||
class RenderSpec:
|
||||
|
||||
@@ -502,12 +502,20 @@ class RenderSegmentVideoHandler(BaseHandler):
|
||||
# 去掉末尾的逗号,作为第一个滤镜
|
||||
filters.append(hwaccel_prefix.rstrip(','))
|
||||
|
||||
# 1. 变速处理
|
||||
# 1. 变速处理(合并 RenderSpec.speed 与 ospeed 效果)
|
||||
speed = float(render_spec.speed) if render_spec.speed else 1.0
|
||||
if speed != 1.0 and speed > 0:
|
||||
# setpts 公式:PTS / speed
|
||||
pts_factor = 1.0 / speed
|
||||
filters.append(f"setpts={pts_factor}*PTS")
|
||||
if speed <= 0:
|
||||
speed = 1.0
|
||||
|
||||
ospeed_factor = 1.0
|
||||
for effect in effects:
|
||||
if effect.effect_type == 'ospeed':
|
||||
ospeed_factor = effect.get_ospeed_params()
|
||||
break
|
||||
|
||||
combined_pts_factor = (1.0 / speed) * ospeed_factor
|
||||
if combined_pts_factor != 1.0:
|
||||
filters.append(f"setpts={combined_pts_factor}*PTS")
|
||||
|
||||
# 2. LUT 调色
|
||||
if lut_file:
|
||||
@@ -569,10 +577,8 @@ class RenderSegmentVideoHandler(BaseHandler):
|
||||
# 计算是否需要额外的尾部冻结(源视频时长不足)
|
||||
extra_tail_freeze_sec = 0.0
|
||||
if source_duration_sec is not None:
|
||||
speed = float(render_spec.speed) if render_spec.speed else 1.0
|
||||
if speed > 0:
|
||||
# 计算变速后的有效时长
|
||||
effective_duration_sec = source_duration_sec / speed
|
||||
# 使用已计算的 combined_pts_factor
|
||||
effective_duration_sec = source_duration_sec * combined_pts_factor
|
||||
required_duration_sec = duration_ms / 1000.0
|
||||
|
||||
# 如果源视频时长不足,需要冻结最后一帧来补足
|
||||
@@ -732,9 +738,15 @@ class RenderSegmentVideoHandler(BaseHandler):
|
||||
extra_tail_freeze_sec = 0.0
|
||||
if source_duration_sec is not None and render_spec is not None and duration_ms > 0:
|
||||
speed = float(render_spec.speed) if render_spec.speed else 1.0
|
||||
if speed > 0:
|
||||
# 计算变速后的有效时长
|
||||
effective_duration_sec = source_duration_sec / speed
|
||||
if speed <= 0:
|
||||
speed = 1.0
|
||||
ospeed_factor = 1.0
|
||||
for effect in effects:
|
||||
if effect.effect_type == 'ospeed':
|
||||
ospeed_factor = effect.get_ospeed_params()
|
||||
break
|
||||
combined_pts_factor = (1.0 / speed) * ospeed_factor
|
||||
effective_duration_sec = source_duration_sec * combined_pts_factor
|
||||
required_duration_sec = duration_ms / 1000.0
|
||||
|
||||
# 如果源视频时长不足,需要冻结最后一帧来补足
|
||||
|
||||
@@ -82,3 +82,139 @@ def test_build_video_filters_zoom_and_camera_shot_stack_in_order(tmp_path):
|
||||
assert camera_shot_marker in filters
|
||||
assert zoom_marker in filters
|
||||
assert filters.index(camera_shot_marker) < filters.index(zoom_marker)
|
||||
|
||||
|
||||
# ---------- ospeed 测试 ----------
|
||||
|
||||
def test_get_ospeed_params_with_valid_values():
|
||||
effect = Effect.from_string('ospeed:2')
|
||||
assert effect is not None
|
||||
assert effect.get_ospeed_params() == pytest.approx(2.0)
|
||||
|
||||
effect2 = Effect.from_string('ospeed:0.5')
|
||||
assert effect2 is not None
|
||||
assert effect2.get_ospeed_params() == pytest.approx(0.5)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
'effect_str',
|
||||
[
|
||||
'ospeed:0',
|
||||
'ospeed:-1',
|
||||
'ospeed:nan',
|
||||
'ospeed:inf',
|
||||
'ospeed:abc',
|
||||
'ospeed:',
|
||||
],
|
||||
)
|
||||
def test_get_ospeed_params_invalid_values_fallback(effect_str):
|
||||
effect = Effect.from_string(effect_str)
|
||||
assert effect is not None
|
||||
assert effect.get_ospeed_params() == 1.0
|
||||
|
||||
|
||||
def test_get_ospeed_params_no_params():
|
||||
effect = Effect.from_string('ospeed')
|
||||
assert effect is not None
|
||||
assert effect.get_ospeed_params() == 1.0
|
||||
|
||||
|
||||
def test_ospeed_does_not_trigger_filter_complex(tmp_path):
|
||||
handler = _create_handler(tmp_path)
|
||||
render_spec = RenderSpec(effects='ospeed:2')
|
||||
output_spec = OutputSpec(width=1080, height=1920, fps=30)
|
||||
|
||||
command = handler._build_command(
|
||||
input_file='input.mp4',
|
||||
output_file='output.mp4',
|
||||
render_spec=render_spec,
|
||||
output_spec=output_spec,
|
||||
duration_ms=6000,
|
||||
)
|
||||
|
||||
assert '-vf' in command
|
||||
assert '-filter_complex' not in command
|
||||
# 验证 setpts 滤镜
|
||||
vf_idx = command.index('-vf')
|
||||
vf_value = command[vf_idx + 1]
|
||||
assert 'setpts=2.0*PTS' in vf_value
|
||||
|
||||
|
||||
def test_ospeed_combined_with_speed(tmp_path):
|
||||
handler = _create_handler(tmp_path)
|
||||
# speed=2 → 1/2=0.5, ospeed=3 → 0.5*3=1.5
|
||||
render_spec = RenderSpec(speed='2', effects='ospeed:3')
|
||||
output_spec = OutputSpec(width=1080, height=1920, fps=30)
|
||||
|
||||
filters = handler._build_video_filters(
|
||||
render_spec=render_spec,
|
||||
output_spec=output_spec,
|
||||
duration_ms=6000,
|
||||
source_duration_sec=10.0,
|
||||
)
|
||||
|
||||
assert 'setpts=1.5*PTS' in filters
|
||||
|
||||
|
||||
def test_ospeed_with_complex_effects(tmp_path):
|
||||
handler = _create_handler(tmp_path)
|
||||
render_spec = RenderSpec(effects='ospeed:2|zoom:1,1.2,2')
|
||||
output_spec = OutputSpec(width=1080, height=1920, fps=30)
|
||||
|
||||
filters = handler._build_video_filters(
|
||||
render_spec=render_spec,
|
||||
output_spec=output_spec,
|
||||
duration_ms=8000,
|
||||
source_duration_sec=10.0,
|
||||
)
|
||||
|
||||
# zoom 触发 filter_complex
|
||||
assert '[v_base]' in filters
|
||||
# setpts 在 base_chain 中
|
||||
assert 'setpts=2.0*PTS' in filters
|
||||
# zoom 正常处理
|
||||
assert "overlay=0:0:enable='between(t,1.0,3.0)'" in filters
|
||||
|
||||
|
||||
def test_only_first_ospeed_is_used(tmp_path):
|
||||
handler = _create_handler(tmp_path)
|
||||
render_spec = RenderSpec(effects='ospeed:2|ospeed:5')
|
||||
output_spec = OutputSpec(width=1080, height=1920, fps=30)
|
||||
|
||||
filters = handler._build_video_filters(
|
||||
render_spec=render_spec,
|
||||
output_spec=output_spec,
|
||||
duration_ms=6000,
|
||||
source_duration_sec=10.0,
|
||||
)
|
||||
|
||||
assert 'setpts=2.0*PTS' in filters
|
||||
assert 'setpts=5.0*PTS' not in filters
|
||||
|
||||
|
||||
def test_ospeed_affects_tpad_calculation(tmp_path):
|
||||
handler = _create_handler(tmp_path)
|
||||
# ospeed:2 使 10s 视频变为 20s 有效时长,目标 6s → 无需 tpad
|
||||
render_spec = RenderSpec(effects='ospeed:2')
|
||||
output_spec = OutputSpec(width=1080, height=1920, fps=30)
|
||||
|
||||
filters = handler._build_video_filters(
|
||||
render_spec=render_spec,
|
||||
output_spec=output_spec,
|
||||
duration_ms=6000,
|
||||
source_duration_sec=10.0,
|
||||
)
|
||||
|
||||
assert 'tpad' not in filters
|
||||
|
||||
# 对比:无 ospeed 时 10s 视频 → 目标 15s → 需要 5s tpad
|
||||
render_spec_no_ospeed = RenderSpec()
|
||||
filters_no_ospeed = handler._build_video_filters(
|
||||
render_spec=render_spec_no_ospeed,
|
||||
output_spec=output_spec,
|
||||
duration_ms=15000,
|
||||
source_duration_sec=10.0,
|
||||
)
|
||||
|
||||
assert 'tpad' in filters_no_ospeed
|
||||
assert 'stop_duration=5.0' in filters_no_ospeed
|
||||
|
||||
Reference in New Issue
Block a user