# Documentation - Class name: VideoCombine - Category: Video Helper Suite 🎥🅥🅗🅢 - Output node: True - Repo Ref: https://github.com/Kosinkadink/ComfyUI-VideoHelperSuite.git combine_video 方法旨在将一系列图像合并成视频格式。它处理各种参数，如帧率、循环计数和文件格式，以创建定制的视频输出。该方法能够以 GIF 和 WebP 等格式生成视频，并且还支持音频集成。 # Input types ## Required - images - images 参数是构成视频视觉内容的图像数据序列。它至关重要，因为它决定了最终视频输出的帧。 - Comfy dtype: IMAGE - Python dtype: List[torch.Tensor] - frame_rate - frame_rate 参数设置视频每秒的帧数，影响视频播放的速度。 - Comfy dtype: FLOAT - Python dtype: int - loop_count - loop_count 决定视频重复的次数。值为 0 表示视频不会循环。 - Comfy dtype: INT - Python dtype: int ## Optional - filename_prefix - filename_prefix 用于创建输出视频文件的唯一标识符，有助于文件管理。 - Comfy dtype: STRING - Python dtype: str - format - format 参数指定输出视频的格式。它可以是 GIF 或 WebP 这样的图像格式，或者是自定义的视频格式。 - Comfy dtype: COMBO[image/gif, image/webp, video/...] - Python dtype: str - pingpong - 当设置为 True 时，pingpong 参数通过反转交替帧的顺序来创建乒乓效应。 - Comfy dtype: BOOLEAN - Python dtype: bool - save_output - save_output 参数指示是否应将结果视频保存到输出目录。 - Comfy dtype: BOOLEAN - Python dtype: bool # Output types - Filenames - Filenames 输出提供了生成的视频文件的路径，允许用户根据需要访问和使用它们。 - Comfy dtype: VHS_FILENAMES - Python dtype: Tuple[bool, List[str]] # Usage tips - Infra type: CPU # Source code ``` class VideoCombine: @classmethod def INPUT_TYPES(s): ffmpeg_formats = get_video_formats() return {'required': {'images': ('IMAGE',), 'frame_rate': ('FLOAT', {'default': 8, 'min': 1, 'step': 1}), 'loop_count': ('INT', {'default': 0, 'min': 0, 'max': 100, 'step': 1}), 'filename_prefix': ('STRING', {'default': 'AnimateDiff'}), 'format': (['image/gif', 'image/webp'] + ffmpeg_formats,), 'pingpong': ('BOOLEAN', {'default': False}), 'save_output': ('BOOLEAN', {'default': True})}, 'optional': {'audio': ('VHS_AUDIO',), 'meta_batch': ('VHS_BatchManager',)}, 'hidden': {'prompt': 'PROMPT', 'extra_pnginfo': 'EXTRA_PNGINFO', 'unique_id': 'UNIQUE_ID'}} RETURN_TYPES = ('VHS_FILENAMES',) RETURN_NAMES = ('Filenames',) OUTPUT_NODE = True CATEGORY = 'Video Helper Suite 🎥🅥🅗🅢' FUNCTION = 'combine_video' def combine_video(self, images, frame_rate: int, loop_count: int, filename_prefix='AnimateDiff', format='image/gif', pingpong=False, save_output=True, prompt=None, extra_pnginfo=None, audio=None, unique_id=None, manual_format_widgets=None, meta_batch=None): output_dir = folder_paths.get_output_directory() if save_output else folder_paths.get_temp_directory() (full_output_folder, filename, _, subfolder, _) = folder_paths.get_save_image_path(filename_prefix, output_dir) output_files = [] metadata = PngInfo() video_metadata = {} if prompt is not None: metadata.add_text('prompt', json.dumps(prompt)) video_metadata['prompt'] = prompt if extra_pnginfo is not None: for x in extra_pnginfo: metadata.add_text(x, json.dumps(extra_pnginfo[x])) video_metadata[x] = extra_pnginfo[x] metadata.add_text('CreationTime', datetime.datetime.now().isoformat(' ')[:19]) if meta_batch is not None and unique_id in meta_batch.outputs: (counter, output_process) = meta_batch.outputs[unique_id] else: max_counter = 0 matcher = re.compile(f'{re.escape(filename)}_(\\d+)\\D*\\..+', re.IGNORECASE) for existing_file in os.listdir(full_output_folder): match = matcher.fullmatch(existing_file) if match: file_counter = int(match.group(1)) if file_counter > max_counter: max_counter = file_counter counter = max_counter + 1 output_process = None file = f'{filename}_{counter:05}.png' file_path = os.path.join(full_output_folder, file) Image.fromarray(tensor_to_bytes(images[0])).save(file_path, pnginfo=metadata, compress_level=4) output_files.append(file_path) (format_type, format_ext) = format.split('/') if format_type == 'image': if meta_batch is not None: raise Exception("Pillow('image/') formats are not compatible with batched output") image_kwargs = {} if format_ext == 'gif': image_kwargs['disposal'] = 2 if format_ext == 'webp': exif = Image.Exif() exif[ExifTags.IFD.Exif] = {36867: datetime.datetime.now().isoformat(' ')[:19]} image_kwargs['exif'] = exif file = f'{filename}_{counter:05}.{format_ext}' file_path = os.path.join(full_output_folder, file) if pingpong: images = to_pingpong(images) frames = map(lambda x: Image.fromarray(tensor_to_bytes(x)), images) next(frames).save(file_path, format=format_ext.upper(), save_all=True, append_images=frames, duration=round(1000 / frame_rate), loop=loop_count, compress_level=4, **image_kwargs) output_files.append(file_path) else: if ffmpeg_path is None: raise ProcessLookupError(f"ffmpeg is required for video outputs and could not be found.\nIn order to use video outputs, you must either:\n- Install imageoio-ffmpeg with pip,\n- Place a ffmpeg executable in {os.path.abspath('')}, or\n- Install ffmpeg and add it to the system path.") kwargs = None if manual_format_widgets is None: if prompt is not None: kwargs = prompt[unique_id]['inputs'] else: manual_format_widgets = {} if kwargs is None: kwargs = get_format_widget_defaults(format_ext) missing = {} for k in kwargs.keys(): if k in manual_format_widgets: kwargs[k] = manual_format_widgets[k] else: missing[k] = kwargs[k] if len(missing) > 0: logger.warn('Extra format values were not provided, the following defaults will be used: ' + str(kwargs) + '\nThis is likely due to usage of ComfyUI-to-python. These values can be manually set by supplying a manual_format_widgets argument') video_format = apply_format_widgets(format_ext, kwargs) has_alpha = images[0].shape[-1] == 4 dimensions = f'{len(images[0][0])}x{len(images[0])}' if loop_count > 0: loop_args = ['-vf', 'loop=loop=' + str(loop_count) + ':size=' + str(len(images))] else: loop_args = [] if pingpong: if meta_batch is not None: logger.error('pingpong is incompatible with batched output') images = to_pingpong(images) if video_format.get('input_color_depth', '8bit') == '16bit': images = map(tensor_to_shorts, images) if has_alpha: i_pix_fmt = 'rgba64' else: i_pix_fmt = 'rgb48' else: images = map(tensor_to_bytes, images) if has_alpha: i_pix_fmt = 'rgba' else: i_pix_fmt = 'rgb24' file = f"{filename}_{counter:05}.{video_format['extension']}" file_path = os.path.join(full_output_folder, file) bitrate_arg = [] bitrate = video_format.get('bitrate') if bitrate is not None: bitrate_arg = ['-b:v', str(bitrate) + 'M' if video_format.get('megabit') == 'True' else str(bitrate) + 'K'] args = [ffmpeg_path, '-v', 'error', '-f', 'rawvideo', '-pix_fmt', i_pix_fmt, '-s', dimensions, '-r', str(frame_rate), '-i', '-'] + loop_args + video_format['main_pass'] + bitrate_arg env = os.environ.copy() if 'environment' in video_format: env.update(video_format['environment']) if output_process is None: output_process = ffmpeg_process(args, video_format, video_metadata, file_path, env) output_process.send(None) if meta_batch is not None: meta_batch.outputs[unique_id] = (counter, output_process) for image in images: output_process.send(image.tobytes()) if meta_batch is not None: requeue_workflow((meta_batch.unique_id, not meta_batch.has_closed_inputs)) if meta_batch is None or meta_batch.has_closed_inputs: try: output_process.send(None) except StopIteration: pass if meta_batch is not None: meta_batch.outputs.pop(unique_id) if len(meta_batch.outputs) == 0: meta_batch.reset() else: return {'ui': {'unfinished_batch': [True]}, 'result': ((save_output, []),)} output_files.append(file_path) if 'gifski_pass' in video_format: gif_output = f'{filename}_{counter:05}.gif' gif_output_path = os.path.join(full_output_folder, gif_output) gifski_args = [gifski_path] + video_format['gifski_pass'] + ['-o', gif_output_path, file_path] try: res = subprocess.run(gifski_args, env=env, check=True, capture_output=True) except subprocess.CalledProcessError as e: raise Exception('An error occured in the gifski subprocess:\n' + e.stderr.decode('utf-8')) if res.stderr: print(res.stderr.decode('utf-8'), end='', file=sys.stderr) format = 'image/gif' output_files.append(gif_output_path) file = gif_output elif audio is not None and audio() is not False: output_file_with_audio = f"{filename}_{counter:05}-audio.{video_format['extension']}" output_file_with_audio_path = os.path.join(full_output_folder, output_file_with_audio) if 'audio_pass' not in video_format: logger.warn('Selected video format does not have explicit audio support') video_format['audio_pass'] = ['-c:a', 'libopus'] mux_args = [ffmpeg_path, '-v', 'error', '-n', '-i', file_path, '-i', '-', '-c:v', 'copy'] + video_format['audio_pass'] + ['-af', 'apad', '-shortest', output_file_with_audio_path] try: res = subprocess.run(mux_args, input=audio(), env=env, capture_output=True, check=True) except subprocess.CalledProcessError as e: raise Exception('An error occured in the ffmpeg subprocess:\n' + e.stderr.decode('utf-8')) if res.stderr: print(res.stderr.decode('utf-8'), end='', file=sys.stderr) output_files.append(output_file_with_audio_path) file = output_file_with_audio previews = [{'filename': file, 'subfolder': subfolder, 'type': 'output' if save_output else 'temp', 'format': format}] return {'ui': {'gifs': previews}, 'result': ((save_output, output_files),)} @classmethod def VALIDATE_INPUTS(self, format, **kwargs): return True ```