|
| 1 | +from __future__ import annotations |
| 2 | +import logging, math, torch |
| 3 | +from comfy_api.latest import io |
| 4 | + |
| 5 | +logger = logging.getLogger(__name__) |
| 6 | + |
| 7 | + |
| 8 | +class SwarmVideoResampleFPS(io.ComfyNode): |
| 9 | + MIN_FPS: float = 0.01 |
| 10 | + MAX_FPS: float = 99999.9 |
| 11 | + STEP_FPS: float = 1.0 |
| 12 | + DEFAULT_FPS_OUT: float = 24.0 |
| 13 | + METHOD_LINEAR: str = "linear" |
| 14 | + METHOD_NEAREST: str = "nearest" |
| 15 | + |
| 16 | + @classmethod |
| 17 | + def define_schema(cls) -> io.Schema: |
| 18 | + return io.Schema( |
| 19 | + node_id="SwarmVideoResampleFPS", |
| 20 | + display_name="Swarm Video Resample FPS", |
| 21 | + category="SwarmUI/video", |
| 22 | + description="Resample a video from fps_in to fps_out while preserving total duration.", |
| 23 | + inputs=[ |
| 24 | + io.Image.Input("images", tooltip="The images to resample."), |
| 25 | + io.Float.Input("fps_in", min=cls.MIN_FPS, max=cls.MAX_FPS, step=cls.STEP_FPS, tooltip="Source frame rate."), |
| 26 | + io.Float.Input("fps_out", default=cls.DEFAULT_FPS_OUT, min=cls.MIN_FPS, max=cls.MAX_FPS, step=cls.STEP_FPS, tooltip="Target frame rate."), |
| 27 | + io.Combo.Input("method", options=[cls.METHOD_LINEAR, cls.METHOD_NEAREST], default=cls.METHOD_LINEAR, |
| 28 | + tooltip=( |
| 29 | + "linear: each output frame is a linear blend of the two source frames bracketing its timestamp. Equivalent to ffmpeg's framerate filter. Slightly more expensive; avoids the duplicated-frame artifact. See https://ffmpeg.org/ffmpeg-filters.html#framerate\n" |
| 30 | + "nearest: each output frame is the source frame closest in time. Equivalent to ffmpeg's fps filter. Cheap; can produce visible judder on pans. See https://ffmpeg.org/ffmpeg-filters.html#fps-1" |
| 31 | + ), |
| 32 | + ), |
| 33 | + ], |
| 34 | + outputs=[io.Image.Output("images")], |
| 35 | + ) |
| 36 | + |
| 37 | + @classmethod |
| 38 | + @torch.inference_mode() |
| 39 | + def execute(cls, images: torch.Tensor, fps_in: float, fps_out: float, method: str) -> io.NodeOutput: |
| 40 | + if fps_in <= 0 or fps_out <= 0: |
| 41 | + raise ValueError(f"SwarmVideoResampleFPS: fps_in and fps_out must be positive (got {fps_in}, {fps_out})") |
| 42 | + |
| 43 | + frame_count_in = int(images.shape[0]) |
| 44 | + if frame_count_in <= 1 or math.isclose(fps_in, fps_out): |
| 45 | + return io.NodeOutput(images) |
| 46 | + |
| 47 | + # Compute output frame count and the fractional source-frame position for each output frame: 4 frames @ 2fps -> 4fps yields 8 frames at source positions [0, 0.5, 1.0, ..., 3.5] |
| 48 | + frame_count_out = max(1, round(frame_count_in / fps_in * fps_out)) |
| 49 | + source_positions = torch.arange(frame_count_out, dtype=torch.float64, device=images.device) / fps_out * fps_in |
| 50 | + |
| 51 | + if method == cls.METHOD_NEAREST: |
| 52 | + resampled = cls._sample_nearest(images, source_positions) |
| 53 | + else: |
| 54 | + resampled = cls._sample_linear(images, source_positions) |
| 55 | + |
| 56 | + logger.info(f"SwarmVideoResampleFPS: {frame_count_in} frames @ {fps_in} fps -> {frame_count_out} frames @ {fps_out} fps ({method})") |
| 57 | + return io.NodeOutput(resampled) |
| 58 | + |
| 59 | + @classmethod |
| 60 | + def _sample_nearest(cls, source_frames: torch.Tensor, source_positions: torch.Tensor) -> torch.Tensor: |
| 61 | + """Pick the closest source frame for each fractional position. |
| 62 | +
|
| 63 | + See https://ffmpeg.org/ffmpeg-filters.html#fps-1 |
| 64 | + """ |
| 65 | + nearest_idx = source_positions.round().long() |
| 66 | + last_valid_idx = source_frames.shape[0] - 1 |
| 67 | + nearest_idx = torch.clamp(nearest_idx, 0, last_valid_idx) |
| 68 | + return source_frames[nearest_idx].contiguous() |
| 69 | + |
| 70 | + @classmethod |
| 71 | + def _sample_linear(cls, source_frames: torch.Tensor, source_positions: torch.Tensor) -> torch.Tensor: |
| 72 | + """Linearly blend the two source frames bracketing each fractional position. |
| 73 | +
|
| 74 | + See https://ffmpeg.org/ffmpeg-filters.html#framerate |
| 75 | + """ |
| 76 | + last_valid_idx = source_frames.shape[0] - 1 |
| 77 | + lower_idx = torch.clamp(source_positions.floor().long(), 0, last_valid_idx) |
| 78 | + upper_idx = torch.clamp(lower_idx + 1, 0, last_valid_idx) |
| 79 | + blend_weight = (source_positions - lower_idx.to(torch.float64)).to(source_frames.dtype) |
| 80 | + while blend_weight.ndim < source_frames.ndim: |
| 81 | + blend_weight = blend_weight.unsqueeze(-1) |
| 82 | + |
| 83 | + lower_frames = source_frames[lower_idx] |
| 84 | + upper_frames = source_frames[upper_idx] |
| 85 | + return ((1.0 - blend_weight) * lower_frames + blend_weight * upper_frames).contiguous() |
| 86 | + |
| 87 | + |
| 88 | +NODE_CLASS_MAPPINGS = { |
| 89 | + "SwarmVideoResampleFPS": SwarmVideoResampleFPS, |
| 90 | +} |
0 commit comments