Files
CavinHuang-comfyui-nodes-docs/docs/LoadVideoFromUrl.md
2024-07-04 08:39:03 +08:00

8.3 KiB

Documentation

  • Class name: LoadVideoFromUrl
  • Category: Art Venture/Loaders
  • Output node: False

LoadVideoFromUrl节点旨在从URL加载视频内容,并将其转换为适合进一步处理或分析的格式。它支持处理各种URL格式,包括视频文件的直接链接,并将其转换为系统内使用的标准化视频格式。

Input types

Required

  • video
    • "video"参数是一个字符串,表示要加载的视频的URL。它在确定视频获取和处理的来源方面起着关键作用。
    • Comfy dtype: STRING
    • Python dtype: str
  • force_rate
    • 指定加载视频的目标帧率,允许控制播放速度或帧率标准化。
    • Comfy dtype: INT
    • Python dtype: Optional[float]
  • force_size
    • 定义加载视频的目标分辨率,实现调整大小或纵横比。
    • Comfy dtype: COMBO[STRING]
    • Python dtype: Optional[str]
  • custom_width
    • 为加载的视频设置自定义宽度,覆盖默认或指定的'force_size'宽度。
    • Comfy dtype: INT
    • Python dtype: Optional[int]
  • custom_height
    • 为加载的视频设置自定义高度,覆盖默认或指定的'force_size'高度。
    • Comfy dtype: INT
    • Python dtype: Optional[int]
  • frame_load_cap
    • 限制从视频中加载的帧数,适用于处理长视频或采样。
    • Comfy dtype: INT
    • Python dtype: Optional[int]
  • skip_first_frames
    • 跳过视频开头指定数量的帧,用于跳过不需要的内容或片头。
    • Comfy dtype: INT
    • Python dtype: Optional[int]
  • select_every_nth
    • 从视频中加载每第n帧,允许降低帧率或选择性帧分析。
    • Comfy dtype: INT
    • Python dtype: Optional[int]

Optional

  • meta_batch
    • 启用加载视频的元数据批处理,优化视频元数据至关重要的场景。
    • Comfy dtype: VHS_BatchManager
    • Python dtype: Optional[bool]

Output types

  • frames
    • 表示加载视频的张量列表。
    • Comfy dtype: IMAGE
    • Python dtype: List[torch.Tensor]
  • frame_count
    • 包含每个加载视频的帧数的列表。
    • Comfy dtype: INT
    • Python dtype: List[int]
  • has_video
    • 指示是否成功加载任何视频的布尔值。
    • Comfy dtype: BOOLEAN
    • Python dtype: bool

Usage tips

  • Infra type: CPU
  • Common nodes: unknown

Source code

    class UtilLoadVideoFromUrl(LoadVideoPath):
        @classmethod
        def INPUT_TYPES(s):
            inputs = LoadVideoPath.INPUT_TYPES()
            inputs["required"]["video"] = ("STRING", {"default": "", "multiline": True, "dynamicPrompts": False})
            return inputs

        CATEGORY = "Art Venture/Loaders"
        FUNCTION = "load"
        RETURN_TYPES = ("IMAGE", "INT", "BOOLEAN")
        RETURN_NAMES = ("frames", "frame_count", "has_video")
        OUTPUT_IS_LIST = (True, True, False)

        def load_gif(
            self,
            gif_path: str,
            force_rate: int,
            force_size: str,
            skip_first_frames: int,
            frame_load_cap: int,
            select_every_nth: int,
        ):
            from PIL import Image, ImageSequence

            image = Image.open(gif_path)
            frames = []
            total_frames_evaluated = -1

            if force_rate != 0:
                print(f"Force rate is not supported for gifs/webps")
            if frame_load_cap == 0:
                frame_load_cap = 999999999

            for i, frame in enumerate(ImageSequence.Iterator(image)):
                if i < skip_first_frames:
                    continue
                elif i >= skip_first_frames + frame_load_cap:
                    break
                else:
                    total_frames_evaluated += 1
                    if total_frames_evaluated % select_every_nth == 0:
                        frames.append(pil2tensor(frame.copy().convert("RGB")))

            images = torch.cat(frames, dim=0)

            if force_size != "Disabled":
                height = images.shape[1]
                width = images.shape[2]
                new_size = target_size(width, height, force_size)
                if new_size[0] != width or new_size[1] != height:
                    s = images.movedim(-1, 1)
                    s = common_upscale(s, new_size[0], new_size[1], "lanczos", "disabled")
                    images = s.movedim(1, -1)

            return (images, len(frames))

        def load_url(self, video: str, **kwargs):
            url = video.strip('"')

            if url == "":
                return (None, 0)

            if os.path.isfile(url):
                pass
            elif url.startswith("file://"):
                url = url[7:]
                url = os.path.abspath(url)

                if not os.path.isfile(url):
                    raise Exception(f"File {url} does not exist")

                if url.startswith(input_dir):
                    video = url[len(input_dir) + 1 :] + " [input]"
                elif url.startswith(output_dir):
                    video = url[len(output_dir) + 1 :] + " [output]"
                elif url.startswith(temp_dir):
                    video = url[len(temp_dir) + 1 :] + " [temp]"
                else:
                    # move file to temp_dir
                    import shutil

                    tempdir = os.path.join(temp_dir, "video")
                    if not os.path.exists(tempdir):
                        os.makedirs(tempfile, exist_ok=True)

                    filename = os.path.basename(url)
                    filepath = os.path.join(tempdir, filename)

                    i = 1
                    split = os.path.splitext(filename)
                    while os.path.exists(filepath):
                        filename = f"{split[0]} ({i}){split[1]}"
                        filepath = os.path.join(tempdir, filename)
                        i += 1

                    shutil.copy(url, filepath)
                    video = "video/" + filename + " [temp]"
            elif url.startswith("http://") or url.startswith("https://"):
                from torch.hub import download_url_to_file
                from urllib.parse import urlparse

                parts = urlparse(url)
                filename = os.path.basename(parts.path)
                tempfile = os.path.join(temp_dir, "video")
                if not os.path.exists(tempfile):
                    os.makedirs(tempfile, exist_ok=True)
                tempfile = os.path.join(tempfile, filename)

                print(f'Downloading: "{url}" to {tempfile}\n')
                download_url_to_file(url, tempfile, progress=True)

                video = "video/" + filename + " [temp]"
            elif url.startswith("/view?"):
                from urllib.parse import parse_qs

                qs = parse_qs(url[6:])
                filename = qs.get("name", qs.get("filename", None))
                if filename is None:
                    raise Exception(f"Invalid url: {url}")

                filename = filename[0]
                subfolder = qs.get("subfolder", None)
                if subfolder is not None:
                    filename = os.path.join(subfolder[0], filename)

                dirtype = qs.get("type", ["input"])
                video = f"{filename} [{dirtype[0]}]"
            else:
                raise Exception(f"Invalid url: {url}")

            if ".gif [" in video.lower() or ".webp [" in video.lower():
                gif_path = folder_paths.get_annotated_filepath(video.strip('"'))
                res = self.load_gif(gif_path, **kwargs)
            else:
                res = self.load_video(video=video, **kwargs)

            return res

        def load(self, video: str, **kwargs):
            urls = video.strip().split("\n")

            videos = []
            frame_counts = []

            for url in urls:
                images, frame_count = self.load_url(url, **kwargs)
                if images is not None and frame_count > 0:
                    videos.append(images)
                    frame_counts.append(frame_count)

            has_video = len(videos) > 0
            if not has_video:
                image = torch.zeros((1, 64, 64, 3), dtype=torch.float32, device="cpu")
                videos.append(image)
                frame_counts.append(1)

            return (videos, frame_counts, has_video)

        @classmethod
        def IS_CHANGED(s, video: str, **kwargs):
            return video

        @classmethod
        def VALIDATE_INPUTS(s, **kwargs):
            return True