Video Combine 🎥🅥🅗🅢¶
Documentation¶
- Class name:
VHS_VideoCombine - Category:
Video Helper Suite 🎥🅥🅗🅢 - Output node:
True
The VHS_VideoCombine node is designed for creating animated videos or GIFs by combining multiple images or latent representations. It supports various customization options such as frame rate adjustment, looping, format selection, and the inclusion of audio. This node is part of the Video Helper Suite, aimed at facilitating video processing and generation tasks.
Input types¶
Required¶
frame_rate- Specifies the frame rate for the video or GIF, affecting how smoothly the animation plays.
- Comfy dtype:
FLOAT - Python dtype:
float
loop_count- Determines how many times the animation will loop, with support for infinite looping.
- Comfy dtype:
INT - Python dtype:
int
filename_prefix- The prefix for the output file name, allowing for easy identification of generated files.
- Comfy dtype:
STRING - Python dtype:
str
format- The output format of the generated video or GIF, such as 'mp4' or 'gif'. Includes additional formats supported by ffmpeg.
- Comfy dtype:
COMBO[STRING] - Python dtype:
List[str]
pingpong- When enabled, the animation will play forwards and then backwards, creating a seamless looping effect.
- Comfy dtype:
BOOLEAN - Python dtype:
bool
save_output- Controls whether the generated video or GIF is saved to disk.
- Comfy dtype:
BOOLEAN - Python dtype:
bool
Optional¶
images- An optional list of images to be combined into a video or GIF. If not provided, latents may be used instead.
- Comfy dtype:
IMAGE - Python dtype:
List[torch.Tensor] or None
audio- An optional audio file to be included in the video, enhancing the multimedia experience.
- Comfy dtype:
VHS_AUDIO - Python dtype:
torch.Tensor or None
meta_batch- Metadata associated with the batch of images or latents being processed, useful for batch operations.
- Comfy dtype:
VHS_BatchManager - Python dtype:
Dict[str, Any] or None
vae- The VAE model used for decoding latent representations into images, necessary when latents are provided.
- Comfy dtype:
VAE - Python dtype:
torch.nn.Module or None
latents- An optional list of latent representations to be converted into images for the video or GIF. Requires a VAE model to decode.
- Comfy dtype:
LATENT - Python dtype:
List[torch.Tensor] or None
Output types¶
Filenames- Comfy dtype:
VHS_FILENAMES - The filenames of the generated videos or GIFs, indicating where the output files have been saved.
- Python dtype:
List[str]
- Comfy dtype:
Usage tips¶
- Infra type:
GPU - Common nodes: unknown
Source code¶
class VideoCombine:
@classmethod
def INPUT_TYPES(s):
ffmpeg_formats = get_video_formats()
return {
"required": {
"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": {
"images": ("IMAGE",),
"audio": ("VHS_AUDIO",),
"meta_batch": ("VHS_BatchManager",),
"vae": ("VAE",),
"latents": ("LATENT",),
},
"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,
frame_rate: int,
loop_count: int,
images=None,
latents=None,
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,
vae=None
):
if latents is not None:
images = latents
if images is None:
return ((save_output, []),)
if vae is not None:
if isinstance(images, dict):
images = images['samples']
else:
vae = None
if isinstance(images, torch.Tensor) and images.size(0) == 0:
return ((save_output, []),)
num_frames = len(images)
pbar = ProgressBar(num_frames)
if vae is not None:
downscale_ratio = getattr(vae, "downscale_ratio", 8)
width = images.size(3)*downscale_ratio
height = images.size(2)*downscale_ratio
frames_per_batch = (1920 * 1080 * 16) // (width * height) or 1
#Python 3.12 adds an itertools.batched, but it's easily replicated for legacy support
def batched(it, n):
while batch := tuple(itertools.islice(it, n)):
yield batch
def batched_encode(images, vae, frames_per_batch):
for batch in batched(iter(images), frames_per_batch):
image_batch = torch.from_numpy(np.array(batch))
yield from vae.decode(image_batch)
images = batched_encode(images, vae, frames_per_batch)
first_image = next(images)
#repush first_image
images = itertools.chain([first_image], images)
else:
first_image = images[0]
images = iter(images)
# get output information
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:
# comfy counter workaround
max_counter = 0
# Loop through the existing files
matcher = re.compile(f"{re.escape(filename)}_(\\d+)\\D*\\..+", re.IGNORECASE)
for existing_file in os.listdir(full_output_folder):
# Check if the file matches the expected format
match = matcher.fullmatch(existing_file)
if match:
# Extract the numeric portion of the filename
file_counter = int(match.group(1))
# Update the maximum counter value if necessary
if file_counter > max_counter:
max_counter = file_counter
# Increment the counter by 1 to get the next available value
counter = max_counter + 1
output_process = None
# save first frame as png to keep metadata
file = f"{filename}_{counter:05}.png"
file_path = os.path.join(full_output_folder, file)
Image.fromarray(tensor_to_bytes(first_image)).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":
#Save timestamp information
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)
# Use pillow directly to save an animated image
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:
# Use ffmpeg to save a video
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 imageio-ffmpeg with pip,\n- Place a ffmpeg executable in {os.path.abspath('')}, or\n- Install ffmpeg and add it to the system path.")
#Acquire additional format_widget values
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 = first_image.shape[-1] == 4
dim_alignment = video_format.get("dim_alignment", 8)
if (first_image.shape[1] % dim_alignment) or (first_image.shape[0] % dim_alignment):
#output frames must be padded
to_pad = (-first_image.shape[1] % dim_alignment,
-first_image.shape[0] % dim_alignment)
padding = (to_pad[0]//2, to_pad[0] - to_pad[0]//2,
to_pad[1]//2, to_pad[1] - to_pad[1]//2)
padfunc = torch.nn.ReplicationPad2d(padding)
def pad(image):
image = image.permute((2,0,1))#HWC to CHW
padded = padfunc(image.to(dtype=torch.float32))
return padded.permute((1,2,0))
images = map(pad, images)
new_dims = (-first_image.shape[1] % dim_alignment + first_image.shape[1],
-first_image.shape[0] % dim_alignment + first_image.shape[0])
dimensions = f"{new_dims[0]}x{new_dims[1]}"
logger.warn("Output images were not of valid resolution and have had padding applied")
else:
dimensions = f"{first_image.shape[1]}x{first_image.shape[0]}"
if loop_count > 0:
loop_args = ["-vf", "loop=loop=" + str(loop_count)+":size=" + str(num_frames)]
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
images = map(lambda x: x.tobytes(), images)
env=os.environ.copy()
if "environment" in video_format:
env.update(video_format["environment"])
if "pre_pass" in video_format:
if meta_batch is not None:
#Performing a prepass requires keeping access to all frames.
#Potential solutions include keeping just output frames in
#memory or using 3 passes with intermediate file, but
#very long gifs probably shouldn't be encouraged
raise Exception("Formats which require a pre_pass are incompatible with Batch Manager.")
images = [b''.join(images)]
os.makedirs(folder_paths.get_temp_directory(), exist_ok=True)
pre_pass_args = args[:13] + video_format['pre_pass']
try:
subprocess.run(pre_pass_args, input=images[0], env=env,
capture_output=True, check=True)
except subprocess.CalledProcessError as e:
raise Exception("An error occurred in the ffmpeg prepass:\n" \
+ e.stderr.decode("utf-8"))
if "inputs_main_pass" in video_format:
args = args[:13] + video_format['inputs_main_pass'] + args[13:]
if output_process is None:
if 'gifski_pass' in video_format:
output_process = gifski_process(args, video_format, file_path, env)
else:
args += video_format['main_pass'] + bitrate_arg
output_process = ffmpeg_process(args, video_format, video_metadata, file_path, env)
#Proceed to first yield
output_process.send(None)
if meta_batch is not None:
meta_batch.outputs[unique_id] = (counter, output_process)
for image in images:
pbar.update(1)
output_process.send(image)
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:
#Close pipe and wait for termination.
try:
total_frames_output = output_process.send(None)
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:
#batch is unfinished
#TODO: Check if empty output breaks other custom nodes
return {"ui": {"unfinished_batch": [True]}, "result": ((save_output, []),)}
output_files.append(file_path)
if audio is not None and audio() is not False:
# Create audio file if input was provided
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"]
# FFmpeg command with audio re-encoding
#TODO: expose audio quality options if format widgets makes it in
#Reconsider forcing apad/shortest
min_audio_dur = total_frames_output / frame_rate + 1
mux_args = [ffmpeg_path, "-v", "error", "-n", "-i", file_path,
"-i", "-", "-c:v", "copy"] \
+ video_format["audio_pass"] \
+ ["-af", "apad=whole_dur="+str(min_audio_dur),
"-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)
#Return this file with audio to the webui.
#It will be muted unless opened or saved with right click
file = output_file_with_audio
previews = [
{
"filename": file,
"subfolder": subfolder,
"type": "output" if save_output else "temp",
"format": format,
"frame_rate": frame_rate,
}
]
if num_frames == 1 and 'png' in format and '%03d' in file:
previews[0]['format'] = 'image/png'
previews[0]['filename'] = file.replace('%03d', '001')
return {"ui": {"gifs": previews}, "result": ((save_output, output_files),)}
@classmethod
def VALIDATE_INPUTS(self, format, **kwargs):
return True