nodes/banodoco-steerable-motion
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
8e2a9a81301bb6ed721707882279b710d4a97ab1
banodoco-steerable-motion/SteerableMotion.py
T

1059 lines
61 KiB
Python
Raw Blame History

# Standard library imports
from ast import literal_eval
from io import BytesIO
import logging
import math
import gc
# Third-party library imports
import numpy as np
import torch
import torchvision.transforms as transforms
from PIL import Image
import matplotlib
import matplotlib.pyplot as plt
# Local application/library specific imports
from .imports.ComfyUI_IPAdapter_plus.IPAdapterPlus import IPAdapterBatchImport, IPAdapterTiledBatchImport, IPAdapterTiledImport, PrepImageForClipVisionImport, IPAdapterAdvancedImport, IPAdapterNoiseImport
from .imports.ComfyUI_Frame_Interpolation.vfi_models.film import FILM_VFIImport
from comfy.utils import common_upscale
try:
from .utils import log # If your .utils has a log object
except ImportError:
log = logging.getLogger(__name__) # Fallback to standard logging
class BatchCreativeInterpolationNode:
@classmethod
def IS_CHANGED(cls, **kwargs):
return float("NaN")
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"positive": ("CONDITIONING", ),
"negative": ("CONDITIONING", ),
"images": ("IMAGE", ),
"model": ("MODEL", ),
"ipadapter": ("IPADAPTER", ),
"clip_vision": ("CLIP_VISION",),
"type_of_frame_distribution": (["linear", "dynamic"],),
"linear_frame_distribution_value": ("INT", {"default": 16, "min": 4, "max": 64, "step": 1}),
"dynamic_frame_distribution_values": ("STRING", {"multiline": True, "default": "0,10,26,40"}),
"type_of_key_frame_influence": (["linear", "dynamic"],),
"linear_key_frame_influence_value": ("STRING", {"multiline": False, "default": "(1.0,1.0)"}),
"dynamic_key_frame_influence_values": ("STRING", {"multiline": True, "default": "(1.0,1.0),(1.0,1.5)(1.0,0.5)"}),
"type_of_strength_distribution": (["linear", "dynamic"],),
"linear_strength_value": ("STRING", {"multiline": False, "default": "(0.3,0.4)"}),
"dynamic_strength_values": ("STRING", {"multiline": True, "default": "(0.0,1.0),(0.0,1.0),(0.0,1.0),(0.0,1.0)"}),
"buffer": ("INT", {"default": 4, "min": 1, "max": 16, "step": 1}),
"high_detail_mode": ("BOOLEAN", {"default": True}),
},
"optional": {
"base_ipa_advanced_settings": ("ADVANCED_IPA_SETTINGS",),
"detail_ipa_advanced_settings": ("ADVANCED_IPA_SETTINGS",),
}
}
RETURN_TYPES = ("IMAGE","CONDITIONING","CONDITIONING","MODEL","STRING","INT", "INT", "STRING")
RETURN_NAMES = ("GRAPH","POSITIVE","NEGATIVE","MODEL","KEYFRAME_POSITIONS","BATCH_SIZE", "BUFFER","FRAMES_TO_DROP")
FUNCTION = "combined_function"
CATEGORY = "Steerable-Motion"
def combined_function(self,positive,negative,images,model,ipadapter,clip_vision,
type_of_frame_distribution,linear_frame_distribution_value,
dynamic_frame_distribution_values, type_of_key_frame_influence,linear_key_frame_influence_value,
dynamic_key_frame_influence_values,type_of_strength_distribution,
linear_strength_value,dynamic_strength_values,
buffer, high_detail_mode,base_ipa_advanced_settings=None,
detail_ipa_advanced_settings=None):
# set the matplotlib backend to 'Agg' to prevent crash on macOS
# 'Agg' is a non-interactive backend that can be used in a non-main thread
matplotlib.use('Agg')
def get_keyframe_positions(type_of_frame_distribution, dynamic_frame_distribution_values, images, linear_frame_distribution_value):
if type_of_frame_distribution == "dynamic":
# Check if the input is a string or a list
if isinstance(dynamic_frame_distribution_values, str):
# Parse the keyframe positions, sort them, and then increase each by 1 except the first
keyframes = sorted([int(kf.strip()) for kf in dynamic_frame_distribution_values.split(',')])
elif isinstance(dynamic_frame_distribution_values, list):
# Sort the list and then increase each by 1 except the first
keyframes = sorted(dynamic_frame_distribution_values)
else:
# Calculate the number of keyframes based on the total duration and linear_frames_per_keyframe
# Increase each by 1 except the first
keyframes = [(i * linear_frame_distribution_value) for i in range(len(images))]
# Increase all values by 1 except the first
if len(keyframes) > 1:
return [keyframes[0]] + [kf + 1 for kf in keyframes[1:]]
else:
return keyframes
def create_mask_batch(last_key_frame_position, weights, frames):
# Hardcoded dimensions
width, height = 512, 512
# Map frames to their corresponding reversed weights for easy lookup
frame_to_weight = {frame: weights[i] for i, frame in enumerate(frames)}
# Create masks for each frame up to last_key_frame_position
masks = []
for frame_number in range(last_key_frame_position):
# Determine the strength of the mask
strength = frame_to_weight.get(frame_number, 0.0)
# Create the mask with the determined strength
mask = torch.full((height, width), strength)
masks.append(mask)
# Convert list of masks to a single tensor
masks_tensor = torch.stack(masks, dim=0)
return masks_tensor
def create_weight_batch(last_key_frame_position, weights, frames):
# Map frames to their corresponding reversed weights for easy lookup
frame_to_weight = {frame: weights[i] for i, frame in enumerate(frames)}
# Create weights for each frame up to last_key_frame_position
weights = []
for frame_number in range(last_key_frame_position):
# Determine the strength of the weight
strength = frame_to_weight.get(frame_number, 0.0)
weights.append(strength)
return weights
def plot_weight_comparison(cn_frame_numbers, cn_weights, ipadapter_frame_numbers, ipadapter_weights, buffer):
plt.figure(figsize=(12, 8))
colors = ['b', 'g', 'r', 'c', 'm', 'y', 'k']
# Handle None values for frame numbers and weights
cn_frame_numbers = cn_frame_numbers if cn_frame_numbers is not None else []
cn_weights = cn_weights if cn_weights is not None else []
ipadapter_frame_numbers = ipadapter_frame_numbers if ipadapter_frame_numbers is not None else []
ipadapter_weights = ipadapter_weights if ipadapter_weights is not None else []
max_length = max(len(cn_frame_numbers), len(ipadapter_frame_numbers))
label_counter = 1 if buffer < 0 else 0
for i in range(max_length):
if i < len(cn_frame_numbers):
label = 'cn_strength_buffer' if (i == 0 and buffer > 0) else f'cn_strength_{label_counter}'
plt.plot(cn_frame_numbers[i], cn_weights[i], marker='o', color=colors[i % len(colors)], label=label)
if i < len(ipadapter_frame_numbers):
label = 'ipa_strength_buffer' if (i == 0 and buffer > 0) else f'ipa_strength_{label_counter}'
plt.plot(ipadapter_frame_numbers[i], ipadapter_weights[i], marker='x', linestyle='--', color=colors[i % len(colors)], label=label)
if label_counter == 0 or buffer < 0 or i > 0:
label_counter += 1
plt.legend()
# Adjusted generator expression for max_weight
all_weights = cn_weights + ipadapter_weights
max_weight = max(max(sublist) for sublist in all_weights if sublist) * 1.5
plt.ylim(0, max_weight)
buffer_io = BytesIO()
plt.savefig(buffer_io, format='png', bbox_inches='tight')
plt.close()
buffer_io.seek(0)
img = Image.open(buffer_io)
img_tensor = transforms.ToTensor()(img)
img_tensor = img_tensor.unsqueeze(0)
img_tensor = img_tensor.permute([0, 2, 3, 1])
return img_tensor,
def extract_strength_values(type_of_key_frame_influence, dynamic_key_frame_influence_values, keyframe_positions, linear_key_frame_influence_value):
if type_of_key_frame_influence == "dynamic":
# Process the dynamic_key_frame_influence_values depending on its format
if isinstance(dynamic_key_frame_influence_values, str):
dynamic_values = eval(dynamic_key_frame_influence_values)
else:
dynamic_values = dynamic_key_frame_influence_values
# Iterate through the dynamic values and convert tuples with two values to three values
dynamic_values_corrected = []
for value in dynamic_values:
if len(value) == 2:
value = (value[0], value[1], value[0])
dynamic_values_corrected.append(value)
return dynamic_values_corrected
else:
# Process for linear or other types
if len(linear_key_frame_influence_value) == 2:
linear_key_frame_influence_value = (linear_key_frame_influence_value[0], linear_key_frame_influence_value[1], linear_key_frame_influence_value[0])
return [linear_key_frame_influence_value for _ in range(len(keyframe_positions) - 1)]
def extract_influence_values(type_of_key_frame_influence, dynamic_key_frame_influence_values, keyframe_positions, linear_key_frame_influence_value):
# Check and convert linear_key_frame_influence_value if it's a float or string float
# if it's a string that starts with a parenthesis, convert it to a tuple
if isinstance(linear_key_frame_influence_value, str) and linear_key_frame_influence_value[0] == "(":
linear_key_frame_influence_value = eval(linear_key_frame_influence_value)
if not isinstance(linear_key_frame_influence_value, tuple):
if isinstance(linear_key_frame_influence_value, (float, str)):
try:
value = float(linear_key_frame_influence_value)
linear_key_frame_influence_value = (value, value)
except ValueError:
raise ValueError("linear_key_frame_influence_value must be a float or a string representing a float")
number_of_outputs = len(keyframe_positions) - 1
if type_of_key_frame_influence == "dynamic":
# Convert list of individual float values into tuples
if all(isinstance(x, float) for x in dynamic_key_frame_influence_values):
dynamic_values = [(value, value) for value in dynamic_key_frame_influence_values]
elif isinstance(dynamic_key_frame_influence_values[0], str) and dynamic_key_frame_influence_values[0] == "(":
string_representation = ''.join(dynamic_key_frame_influence_values)
dynamic_values = eval(f'[{string_representation}]')
else:
dynamic_values = dynamic_key_frame_influence_values if isinstance(dynamic_key_frame_influence_values, list) else [dynamic_key_frame_influence_values]
return dynamic_values[:number_of_outputs]
else:
return [linear_key_frame_influence_value for _ in range(number_of_outputs)]
def calculate_weights(batch_index_from, batch_index_to, strength_from, strength_to, interpolation,revert_direction_at_midpoint, last_key_frame_position,i, number_of_items,buffer):
# Initialize variables based on the position of the keyframe
range_start = batch_index_from
range_end = batch_index_to
# if it's the first value, set influence range from 1.0 to 0.0
if i == number_of_items - 1:
range_end = last_key_frame_position
steps = range_end - range_start
diff = strength_to - strength_from
# Calculate index for interpolation
index = np.linspace(0, 1, steps // 2 + 1) if revert_direction_at_midpoint else np.linspace(0, 1, steps)
# Calculate weights based on interpolation type
if interpolation == "linear":
weights = np.linspace(strength_from, strength_to, len(index))
elif interpolation == "ease-in":
weights = diff * np.power(index, 2) + strength_from
elif interpolation == "ease-out":
weights = diff * (1 - np.power(1 - index, 2)) + strength_from
elif interpolation == "ease-in-out":
weights = diff * ((1 - np.cos(index * np.pi)) / 2) + strength_from
if revert_direction_at_midpoint:
weights = np.concatenate([weights, weights[::-1]])
# Generate frame numbers
frame_numbers = np.arange(range_start, range_start + len(weights))
# "Dropper" component: For keyframes with negative start, drop the weights
if range_start < 0 and i > 0:
drop_count = abs(range_start)
weights = weights[drop_count:]
frame_numbers = frame_numbers[drop_count:]
# Dropper component: for keyframes a range_End is greater than last_key_frame_position, drop the weights
if range_end > last_key_frame_position and i < number_of_items - 1:
drop_count = range_end - last_key_frame_position
weights = weights[:-drop_count]
frame_numbers = frame_numbers[:-drop_count]
return weights, frame_numbers
def process_weights(frame_numbers, weights, multiplier):
# Multiply weights by the multiplier and apply the bounds of 0.0 and 1.0
adjusted_weights = [min(max(weight * multiplier, 0.0), 1.0) for weight in weights]
# Filter out frame numbers and weights where the weight is 0.0
filtered_frames_and_weights = [(frame, weight) for frame, weight in zip(frame_numbers, adjusted_weights) if weight > 0.0]
# Separate the filtered frame numbers and weights
filtered_frame_numbers, filtered_weights = zip(*filtered_frames_and_weights) if filtered_frames_and_weights else ([], [])
return list(filtered_frame_numbers), list(filtered_weights)
def calculate_influence_frame_number(key_frame_position, next_key_frame_position, distance):
# Calculate the absolute distance between key frames
key_frame_distance = abs(next_key_frame_position - key_frame_position)
# Apply the distance multiplier
extended_distance = key_frame_distance * distance
# Determine the direction of influence based on the positions of the key frames
if key_frame_position < next_key_frame_position:
# Normal case: influence extends forward
influence_frame_number = key_frame_position + extended_distance
else:
# Reverse case: influence extends backward
influence_frame_number = key_frame_position - extended_distance
# Return the result rounded to the nearest integer
return round(influence_frame_number)
# GET KEYFRAME POSITIONS
keyframe_positions = get_keyframe_positions(type_of_frame_distribution, dynamic_frame_distribution_values, images, linear_frame_distribution_value)
shifted_keyframes_position = [position + buffer - 2 for position in keyframe_positions]
shifted_keyframe_positions_string = ','.join(str(pos) for pos in shifted_keyframes_position)
if buffer > 0:
# add front buffer
keyframe_positions = [position + buffer - 1 for position in keyframe_positions]
keyframe_positions.insert(0, 0)
# add end buffer
last_position_with_buffer = keyframe_positions[-1] + buffer + 1
keyframe_positions.append(last_position_with_buffer)
# GET BASE ADVANCED SETTINGS OR SET DEFAULTS
if base_ipa_advanced_settings is None:
if high_detail_mode:
base_ipa_advanced_settings = {
"ipa_starts_at": 0.0,
"ipa_ends_at": 0.3,
"ipa_weight_type": "ease in-out",
"ipa_weight": 1.0,
"ipa_embeds_scaling": "V only",
"ipa_noise_strength": 0.0,
"use_image_for_noise": False,
"type_of_noise": "fade",
"noise_blur": 0,
}
else:
base_ipa_advanced_settings = {
"ipa_starts_at": 0.0,
"ipa_ends_at": 0.75,
"ipa_weight_type": "ease in-out",
"ipa_weight": 1.0,
"ipa_embeds_scaling": "V only",
"ipa_noise_strength": 0.0,
"use_image_for_noise": False,
"type_of_noise": "fade",
"noise_blur": 0,
}
# GET DETAILED ADVANCED SETTINGS OR SET DEFAULTS
if detail_ipa_advanced_settings is None:
if high_detail_mode:
detail_ipa_advanced_settings = {
"ipa_starts_at": 0.25,
"ipa_ends_at": 0.75,
"ipa_weight_type": "ease in-out",
"ipa_weight": 1.0,
"ipa_embeds_scaling": "V only",
"ipa_noise_strength": 0.0,
"use_image_for_noise": False,
"type_of_noise": "fade",
"noise_blur": 0,
}
strength_values = extract_strength_values(type_of_strength_distribution, dynamic_strength_values, keyframe_positions, linear_strength_value)
strength_values = [literal_eval(val) if isinstance(val, str) else val for val in strength_values]
corrected_strength_values = []
for val in strength_values:
if len(val) == 2:
val = (val[0], val[1], val[0])
corrected_strength_values.append(val)
strength_values = corrected_strength_values
# GET KEYFRAME INFLUENCE VALUES
key_frame_influence_values = extract_influence_values(type_of_key_frame_influence, dynamic_key_frame_influence_values, keyframe_positions, linear_key_frame_influence_value)
key_frame_influence_values = [literal_eval(val) if isinstance(val, str) else val for val in key_frame_influence_values]
# CALCULATE LAST KEYFRAME POSITION
if len(keyframe_positions) == 4:
last_key_frame_position = (keyframe_positions[-1]) - 1
else:
last_key_frame_position = (keyframe_positions[-1])
class IPBin:
def __init__(self):
self.indicies = []
self.image_schedule = []
self.weight_schedule = []
self.imageBatch = []
self.bigImageBatch = []
self.noiseBatch = []
self.bigNoiseBatch = []
def length(self):
return len(self.image_schedule)
def add(self, image, big_image, noise, big_noise, image_index, frame_numbers, weights):
# Map frames to their corresponding reversed weights for easy lookup
frame_to_weight = {frame: weights[i] for i, frame in enumerate(frame_numbers)}
# Search for image index, if it isn't there add the image
try:
index = self.indicies.index(image_index)
except ValueError:
self.imageBatch.append(image)
self.bigImageBatch.append(big_image)
if noise is not None: self.noiseBatch.append(noise)
if big_noise is not None: self.bigNoiseBatch.append(big_noise)
self.indicies.append(image_index)
index = self.indicies.index(image_index)
self.image_schedule.extend([index] * (frame_numbers[-1] + 1 - len(self.image_schedule)))
self.weight_schedule.extend([0] * (frame_numbers[0] - len(self.weight_schedule)))
self.weight_schedule.extend(frame_to_weight[frame] for frame in range(frame_numbers[0], frame_numbers[-1] + 1))
# CREATE LISTS FOR WEIGHTS AND FRAME NUMBERS
all_cn_frame_numbers = []
all_cn_weights = []
all_ipa_weights = []
all_ipa_frame_numbers = []
# Start with one bin
bins = [IPBin()]
for i in range(len(keyframe_positions)):
keyframe_position = keyframe_positions[i]
interpolation = "ease-in-out"
# strength_from = strength_to = 1.0
image_index = 0
if i == 0: # buffer
image = images[0]
image_index = 0
strength_from = strength_to = strength_values[0][1]
batch_index_from = 0
batch_index_to_excl = buffer
weights, frame_numbers = calculate_weights(batch_index_from, batch_index_to_excl, strength_from, strength_to, interpolation, False, last_key_frame_position, i, len(keyframe_positions), buffer)
elif i == 1: # first image
# GET IMAGE AND KEYFRAME INFLUENCE VALUES
image = images[i-1]
image_index = i-1
key_frame_influence_from, key_frame_influence_to = key_frame_influence_values[i-1]
start_strength, mid_strength, end_strength = strength_values[i-1]
keyframe_position = keyframe_positions[i] + 1
next_key_frame_position = keyframe_positions[i+1] + 1
batch_index_from = keyframe_position
batch_index_to_excl = calculate_influence_frame_number(keyframe_position, next_key_frame_position, key_frame_influence_to)
weights, frame_numbers = calculate_weights(batch_index_from, batch_index_to_excl, mid_strength, end_strength, interpolation, False, last_key_frame_position, i, len(keyframe_positions), buffer)
# interpolation = "ease-in"
elif i == len(keyframe_positions) - 2: # last image
# GET IMAGE AND KEYFRAME INFLUENCE VALUES
image = images[i-1]
image_index = i - 1
key_frame_influence_from,key_frame_influence_to = key_frame_influence_values[i-1]
start_strength, mid_strength, end_strength = strength_values[i-1]
if len(keyframe_positions) == 4:
keyframe_position = keyframe_positions[i] - 1
else:
keyframe_position = keyframe_positions[i]
previous_key_frame_position = keyframe_positions[i-1]
batch_index_from = calculate_influence_frame_number(keyframe_position, previous_key_frame_position, key_frame_influence_from)
batch_index_to_excl = keyframe_position + 1
weights, frame_numbers = calculate_weights(batch_index_from, batch_index_to_excl, start_strength, mid_strength, interpolation, False, last_key_frame_position, i, len(keyframe_positions), buffer)
# interpolation = "ease-out"
elif i == len(keyframe_positions) - 1: # buffer
image = images[i-2]
image_index = i - 2
strength_from = strength_to = strength_values[i-2][1]
if len(keyframe_positions) == 4:
batch_index_from = keyframe_positions[i-1]
batch_index_to_excl = last_key_frame_position - 1
else:
batch_index_from = keyframe_positions[i-1] + 1
batch_index_to_excl = last_key_frame_position
weights, frame_numbers = calculate_weights(batch_index_from, batch_index_to_excl, strength_from, strength_to, interpolation, False, last_key_frame_position, i, len(keyframe_positions), buffer)
else: # middle images
# GET IMAGE AND KEYFRAME INFLUENCE VALUES
image = images[i-1]
image_index = i - 1
key_frame_influence_from,key_frame_influence_to = key_frame_influence_values[i-1]
start_strength, mid_strength, end_strength = strength_values[i-1]
keyframe_position = keyframe_positions[i]
# CALCULATE WEIGHTS FOR FIRST HALF
previous_key_frame_position = keyframe_positions[i-1]
batch_index_from = calculate_influence_frame_number(keyframe_position, previous_key_frame_position, key_frame_influence_from)
batch_index_to_excl = keyframe_position + 1
first_half_weights, first_half_frame_numbers = calculate_weights(batch_index_from, batch_index_to_excl, start_strength, mid_strength, interpolation, False, last_key_frame_position, i, len(keyframe_positions), buffer)
# CALCULATE WEIGHTS FOR SECOND HALF
next_key_frame_position = keyframe_positions[i+1]
batch_index_from = keyframe_position
batch_index_to_excl = calculate_influence_frame_number(keyframe_position, next_key_frame_position, key_frame_influence_to) + 2
second_half_weights, second_half_frame_numbers = calculate_weights(batch_index_from, batch_index_to_excl, mid_strength, end_strength, interpolation, False, last_key_frame_position, i, len(keyframe_positions), buffer)
# COMBINE FIRST AND SECOND HALF
weights = np.concatenate([first_half_weights, second_half_weights])
frame_numbers = np.concatenate([first_half_frame_numbers, second_half_frame_numbers])
# PROCESS WEIGHTS
ipa_frame_numbers, ipa_weights = process_weights(frame_numbers, weights, 1.0)
prepare_for_clip_vision = PrepImageForClipVisionImport()
prepped_image, = prepare_for_clip_vision.prep_image(image=image.unsqueeze(0), interpolation="LANCZOS", crop_position="pad", sharpening=0.1)
if base_ipa_advanced_settings["ipa_noise_strength"] > 0:
if base_ipa_advanced_settings["use_image_for_noise"]:
noise_image = prepped_image
else:
noise_image = None
ipa_noise = IPAdapterNoiseImport()
negative_noise, = ipa_noise.make_noise(type=base_ipa_advanced_settings["type_of_noise"], strength=base_ipa_advanced_settings["ipa_noise_strength"], blur=base_ipa_advanced_settings["noise_blur"], image_optional=noise_image)
else:
negative_noise = None
if high_detail_mode and detail_ipa_advanced_settings["ipa_noise_strength"] > 0:
if detail_ipa_advanced_settings["use_image_for_noise"]:
noise_image = image.unsqueeze(0)
else:
noise_image = None
ipa_noise = IPAdapterNoiseImport()
big_negative_noise, = ipa_noise.make_noise(type=detail_ipa_advanced_settings["type_of_noise"], strength=detail_ipa_advanced_settings["ipa_noise_strength"], blur=detail_ipa_advanced_settings["noise_blur"], image_optional=noise_image)
else:
big_negative_noise = None
if len(ipa_frame_numbers) > 0:
# Fill up bins with image frames. Bins will automatically be created when needed but all the frames should be able to be packed into two bins
active_index = -1
# Find a bin that we can fit the next image into
for i, bin in enumerate(bins):
if bin.length() <= ipa_frame_numbers[0]:
active_index = i
break
# If we didn't find a suitable bin, add a new one
if active_index == -1:
bins.append(IPBin())
active_index = len(bins) - 1
# Add the image to the bin
bins[active_index].add(prepped_image, image.unsqueeze(0), negative_noise, big_negative_noise, image_index, ipa_frame_numbers, ipa_weights)
all_ipa_frame_numbers.append(ipa_frame_numbers)
all_ipa_weights.append(ipa_weights)
# Go through the bins and create IPAdapters for them
for i, bin in enumerate(bins):
ipadapter_application = IPAdapterBatchImport()
negative_noise = torch.cat(bin.noiseBatch, dim=0) if len(bin.noiseBatch) > 0 else None
model, *_ = ipadapter_application.apply_ipadapter(model=model, ipadapter=ipadapter, image=torch.cat(bin.imageBatch, dim=0), weight=[x * base_ipa_advanced_settings["ipa_weight"] for x in bin.weight_schedule], weight_type=base_ipa_advanced_settings["ipa_weight_type"], start_at=base_ipa_advanced_settings["ipa_starts_at"], end_at=base_ipa_advanced_settings["ipa_ends_at"], clip_vision=clip_vision,image_negative=negative_noise,embeds_scaling=base_ipa_advanced_settings["ipa_embeds_scaling"], encode_batch_size=1, image_schedule=bin.image_schedule)
if high_detail_mode:
tiled_ipa_application = IPAdapterTiledBatchImport()
negative_noise = torch.cat(bin.bigNoiseBatch, dim=0) if len(bin.bigNoiseBatch) > 0 else None
model, *_ = tiled_ipa_application.apply_tiled(model=model, ipadapter=ipadapter, image=torch.cat(bin.bigImageBatch, dim=0), weight=[x * detail_ipa_advanced_settings["ipa_weight"] for x in bin.weight_schedule], weight_type=detail_ipa_advanced_settings["ipa_weight_type"], start_at=detail_ipa_advanced_settings["ipa_starts_at"], end_at=detail_ipa_advanced_settings["ipa_ends_at"], clip_vision=clip_vision,sharpening=0.1,image_negative=negative_noise,embeds_scaling=detail_ipa_advanced_settings["ipa_embeds_scaling"], encode_batch_size=1, image_schedule=bin.image_schedule)
comparison_diagram, = plot_weight_comparison(all_cn_frame_numbers, all_cn_weights, all_ipa_frame_numbers, all_ipa_weights, buffer)
return comparison_diagram, positive, negative, model, shifted_keyframe_positions_string, last_key_frame_position, buffer, shifted_keyframes_position
class RemoveAndInterpolateFramesNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"images": ("IMAGE", ),
"frames_to_drop": ("STRING", {"multiline": True, "default": "[8, 16, 24]"}),
},
"optional": {}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("image",)
FUNCTION = "replace_and_interpolate_frames"
CATEGORY = "Steerable-Motion"
def replace_and_interpolate_frames(self, images: torch.Tensor, frames_to_drop: str):
if isinstance(frames_to_drop, str):
frames_to_drop = eval(frames_to_drop)
frames_to_drop = sorted(frames_to_drop, reverse=True)
# Create instance of FILM_VFI within the function
film_vfi = FILM_VFIImport() # Assuming FILM_VFI does not require any special setup
for index in frames_to_drop:
if 0 < index < images.shape[0] - 1:
# Extract the two surrounding frames
batch = images[index-1:index+2:2]
# Process through FILM_VFI
interpolated_frames = film_vfi.vfi(
ckpt_name='film_net_fp32.pt',
frames=batch,
clear_cache_after_n_frames=10,
multiplier=2
)[0] # Assuming vfi returns a tuple and the first element is the interpolated frames
# Replace the original frames at the location
images = torch.cat((images[:index-1], interpolated_frames, images[index+2:]))
return (images,)
class IpaConfigurationNode:
WEIGHT_TYPES = ["linear", "ease in", "ease out", 'ease in-out', 'reverse in-out', 'weak input', 'weak output', 'weak middle', 'strong middle']
IPA_EMBEDS_SCALING_OPTIONS = ["V only", "K+V", "K+V w/ C penalty", "K+mean(V) w/ C penalty"]
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"ipa_starts_at": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"ipa_ends_at": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"ipa_weight_type": (cls.WEIGHT_TYPES,),
"ipa_weight": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 2.0, "step": 0.01}),
"ipa_embeds_scaling": (cls.IPA_EMBEDS_SCALING_OPTIONS,),
"ipa_noise_strength": ("FLOAT", {"default": 0.3, "min": 0.0, "max": 1.0, "step": 0.01}),
"use_image_for_noise": ("BOOLEAN", {"default": False}),
"type_of_noise": (["fade", "dissolve", "gaussian", "shuffle"], ),
"noise_blur": ("INT", { "default": 0, "min": 0, "max": 32, "step": 1 }),
},
"optional": {}
}
FUNCTION = "process_inputs"
RETURN_TYPES = ("ADVANCED_IPA_SETTINGS",)
RETURN_NAMES = ("configuration",)
CATEGORY = "Steerable-Motion"
@classmethod
def process_inputs(cls, ipa_starts_at, ipa_ends_at, ipa_weight_type, ipa_weight, ipa_embeds_scaling, ipa_noise_strength, use_image_for_noise, type_of_noise, noise_blur):
return {
"ipa_starts_at": ipa_starts_at,
"ipa_ends_at": ipa_ends_at,
"ipa_weight_type": ipa_weight_type,
"ipa_weight": ipa_weight,
"ipa_embeds_scaling": ipa_embeds_scaling,
"ipa_noise_strength": ipa_noise_strength,
"use_image_for_noise": use_image_for_noise,
"type_of_noise": type_of_noise,
"noise_blur": noise_blur,
},
class VideoFrameExtractorAndMaskGenerator:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input_video_frames": ("IMAGE", {"tooltip": "Input video frames (IMAGE batch) to extract from."}),
"total_output_frames": ("INT", {"default": 81, "min": 1, "max": 10000, "step": 4, "tooltip": "Total number of frames for the output guidance video and masks. Must satisfy: (frames - 1) divisible by 4."}),
"frame_selection_string": ("STRING", {"default": "0, 10:20", "multiline": False, "tooltip": "Comma-separated integers or ranges (e.g., 0, 5, 10:15, 20) of frames to extract from input video. Takes precedence over depth_frames."}),
"empty_frame_fill_level": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Grayscale level (0.0 black, 1.0 white) for frames not explicitly selected or filled by depth."}),
},
"optional": {
"depth_video_frames": ("IMAGE", {"tooltip": "Optional depth frames (IMAGE batch). Placed if the slot is not already filled by frame_selection_string."}),
"master_inpaint_mask": ("MASK", {"tooltip": "Optional master inpaint mask. If provided, it defines the entire output mask, overriding masks for selected/depth frames."}),
},
}
RETURN_TYPES = ("IMAGE", "MASK",)
RETURN_NAMES = ("guidance_video_frames", "guidance_frame_masks",)
FUNCTION = "extract_frames_and_generate_masks"
CATEGORY = "Steerable-Motion"
DESCRIPTION = "Extracts/places frames from input/depth video into a new guidance video and generates corresponding masks. frame_selection_string takes precedence over depth frames."
def _parse_frame_selection_string(self, selection_string, max_frame_index_from_input):
selected_frame_indices = set()
selection_parts = selection_string.split(',')
for part in selection_parts:
part = part.strip()
if not part:
continue
if ':' in part:
try:
start_str, end_str = part.split(':')
start_frame = int(start_str)
end_frame = int(end_str)
if start_frame < 0 or end_frame < 0:
log.warning(f"Frame indices cannot be negative in '{part}'. Skipping.")
continue
if start_frame > end_frame:
log.warning(f"Range start {start_frame} is greater than end {end_frame} in '{part}'. Swapping.")
start_frame, end_frame = end_frame, start_frame
for frame_idx in range(start_frame, end_frame + 1): # Inclusive range
if 0 <= frame_idx <= max_frame_index_from_input:
selected_frame_indices.add(frame_idx)
else:
log.warning(f"Frame index {frame_idx} from range '{part}' is out of bounds for input video (0-{max_frame_index_from_input}). Skipping this specific index.")
except ValueError:
log.error(f"Invalid range format '{part}'. Skipping.")
else:
try:
frame_idx = int(part)
if frame_idx < 0:
log.warning(f"Frame index {frame_idx} cannot be negative. Skipping.")
continue
if 0 <= frame_idx <= max_frame_index_from_input:
selected_frame_indices.add(frame_idx)
else:
log.warning(f"Frame index {frame_idx} is out of bounds for input video (0-{max_frame_index_from_input}). Skipping.")
except ValueError:
log.error(f"Invalid frame index '{part}'. Skipping.")
return sorted(list(selected_frame_indices))
def extract_frames_and_generate_masks(self, input_video_frames, total_output_frames, frame_selection_string, empty_frame_fill_level, depth_video_frames=None, master_inpaint_mask=None):
# Convert string parameter to integer
total_output_frames = int(total_output_frames)
if (total_output_frames - 1) % 4 != 0:
raise ValueError("total_output_frames must satisfy (frames - 1) divisible by 4")
if input_video_frames is None or input_video_frames.shape[0] == 0:
log.error("Input video_frames is empty. Cannot proceed.")
dummy_height, dummy_width, dummy_channels = 64, 64, 3
return (torch.zeros((total_output_frames, dummy_height, dummy_width, dummy_channels), dtype=torch.float32),
torch.ones((total_output_frames, dummy_height, dummy_width), dtype=torch.float32))
device = input_video_frames.device
dtype = input_video_frames.dtype
batch_size_input, frame_height, frame_width, num_channels = input_video_frames.shape
max_input_frame_index = batch_size_input - 1
# Initialize guidance video with empty_frame_fill_level
guidance_video_output = torch.ones((total_output_frames, frame_height, frame_width, num_channels), device=device, dtype=dtype) * empty_frame_fill_level
# Initialize base masks: 1 for unknown/inpaint, 0 for known
base_frame_masks = torch.ones((total_output_frames, frame_height, frame_width), device=device, dtype=dtype)
# 1. Process frame_selection_string (highest priority)
selected_input_frame_indices = self._parse_frame_selection_string(frame_selection_string, max_input_frame_index)
log.info(f"Frames selected by 'frame_selection_string': {selected_input_frame_indices}")
for input_frame_index in selected_input_frame_indices:
# The selected_input_frame_indices are indices from the *input_video*.
# We place them at the *same index* in the output guidance_video if that index is valid.
target_output_frame_index = input_frame_index
if target_output_frame_index < total_output_frames:
guidance_video_output[target_output_frame_index] = input_video_frames[input_frame_index].clone()
base_frame_masks[target_output_frame_index] = 0.0 # This frame is now known and prioritized
log.debug(f"Placed frame {input_frame_index} from input_video_frames into guidance_video at index {target_output_frame_index}.")
else:
log.warning(f"Selected frame index {input_frame_index} from input video maps to target index {target_output_frame_index}, which is >= total_output_frames ({total_output_frames}). It won't be placed.")
# 2. Process depth_video_frames (second priority)
if depth_video_frames is not None and depth_video_frames.shape[0] > 0:
log.info(f"Processing {depth_video_frames.shape[0]} depth_video_frames.")
processed_depth_frames = depth_video_frames.clone().to(device=device, dtype=dtype)
# Resize depth_video_frames if their dimensions don't match input_video_frames
if processed_depth_frames.shape[1:] != (frame_height, frame_width, num_channels):
log.info(f"Resizing depth_video_frames from {processed_depth_frames.shape[1:]} to {(frame_height, frame_width, num_channels)} to match input_video_frames.")
resized_depth_frame_list = []
for frame_idx in range(processed_depth_frames.shape[0]):
# common_upscale expects (B, C, H, W) or (B, H, W)
# IMAGE is (B,H,W,C), so permute, upscale, permute back
frame_to_resize = processed_depth_frames[frame_idx:frame_idx+1].permute(0, 3, 1, 2) # (1, C, H_depth, W_depth)
resized_frame = common_upscale(frame_to_resize, frame_width, frame_height, "lanczos", "disabled") # (1, C, H, W)
resized_depth_frame_list.append(resized_frame.permute(0, 2, 3, 1)) # (1, H, W, C)
processed_depth_frames = torch.cat(resized_depth_frame_list, dim=0)
num_depth_frames_to_place = min(processed_depth_frames.shape[0], total_output_frames)
for frame_idx in range(num_depth_frames_to_place):
# Check if this slot in guidance_video is still an "empty" placeholder
# (i.e., its mask is still 1.0, meaning not filled by frame_selection_string)
if base_frame_masks[frame_idx].mean() > 0.99: # Check if it's still (mostly) 1.0
guidance_video_output[frame_idx] = processed_depth_frames[frame_idx].clone()
# Keep mask as 1.0 for depth frames (inpaint area) - don't set to 0.0
log.debug(f"Placed frame {frame_idx} from depth_video_frames into guidance_video at index {frame_idx} (keeping as inpaint area).")
else:
log.debug(f"Skipping depth_frame {frame_idx} as guidance_video index {frame_idx} was already filled by frame_selection_string.")
else:
log.info("No depth_video_frames provided or depth_video_frames is empty.")
# 3. Handle optional master_inpaint_mask (this will override base_frame_masks if provided)
final_frame_masks = base_frame_masks
if master_inpaint_mask is not None:
log.info("Processing provided master_inpaint_mask. This will override masks derived from frame/depth selection.")
processed_master_mask = master_inpaint_mask.clone().to(device=device, dtype=dtype)
if processed_master_mask.shape[1:] != (frame_height, frame_width):
log.info(f"Resizing master_inpaint_mask from {processed_master_mask.shape[1:]} to {(frame_height, frame_width)}.")
processed_master_mask = common_upscale(
processed_master_mask.unsqueeze(1),
frame_width, frame_height, "nearest-exact", "disabled"
).squeeze(1)
if processed_master_mask.shape[0] != total_output_frames:
log.info(f"Adjusting master_inpaint_mask frame count from {processed_master_mask.shape[0]} to {total_output_frames}.")
if processed_master_mask.shape[0] == 0:
log.error("Received an empty master_inpaint_mask after processing. Using base masks.")
elif processed_master_mask.shape[0] < total_output_frames:
num_mask_repeats = (total_output_frames + processed_master_mask.shape[0] - 1) // processed_master_mask.shape[0]
processed_master_mask = processed_master_mask.repeat(num_mask_repeats, 1, 1)[:total_output_frames]
else:
processed_master_mask = processed_master_mask[:total_output_frames]
final_frame_masks = processed_master_mask
return (guidance_video_output.cpu().float(), final_frame_masks.cpu().float())
class VideoContinuationGenerator:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"input_video_frames": ("IMAGE", {"tooltip": "Input video frames to create continuation from."}),
"total_output_frames": ("INT", {"default": 81, "min": 1, "max": 81, "step": 4, "tooltip": "Total number of frames for the output continuation video. Must satisfy: (frames - 1) divisible by 4."}),
"overlap_frames": ("INT", {"default": 3, "min": 1, "max": 50, "step": 1, "tooltip": "Number of frames from the end of input video to use as overlap at the start."}),
"empty_frame_fill_level": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01, "tooltip": "Grayscale level (0.0 black, 1.0 white) for empty continuation frames."}),
},
"optional": {
"end_frame": ("IMAGE", {"tooltip": "Optional single frame to place at the end of the continuation video."}),
"control_images": ("IMAGE", {"tooltip": "Optional control images to fill the empty frames."}),
"inpaint_mask": ("MASK", {"tooltip": "Optional inpaint mask to use for the empty frames, overriding the default mask."}),
"when_to_start_control_frames": (["beginning_of_generation", "after_overlap_frames"], {"default": "beginning_of_generation", "tooltip": "If beginning_of_generation is selected, aligns control frames with the start of the output (frame 0). Overlap frames from the input video will take priority, so control frames will become visible starting after the overlap period. If after_overlap_frames is selected, control frames will start being placed after the overlap frames from the input video."}),
"when_to_start_masks": (["beginning_of_generation", "after_overlap_frames"], {"default": "beginning_of_generation", "tooltip": "If beginning_of_generation is selected, aligns masks with the start of the output (frame 0). If after_overlap_frames is selected, masks will start being placed after the overlap frames from the input video."}),
},
}
RETURN_TYPES = ("IMAGE", "MASK",)
RETURN_NAMES = ("continuation_video_frames", "continuation_frame_masks",)
FUNCTION = "generate_continuation_video"
CATEGORY = "Steerable-Motion"
DESCRIPTION = "Creates a continuation video by placing overlap frames from the end of input video at the start, with optional end frame."
def generate_continuation_video(self, input_video_frames, total_output_frames, overlap_frames, empty_frame_fill_level, end_frame=None, control_images=None, inpaint_mask=None, when_to_start_control_frames="beginning_of_generation", when_to_start_masks="beginning_of_generation"):
# 1. Validation and Setup
total_output_frames = int(total_output_frames)
if (total_output_frames - 1) % 4 != 0:
raise ValueError("total_output_frames must satisfy (frames - 1) divisible by 4")
if input_video_frames is None or input_video_frames.shape[0] == 0:
log.error("Input video_frames is empty. Cannot proceed.")
dummy_height, dummy_width, dummy_channels = 64, 64, 3
return (torch.zeros((total_output_frames, dummy_height, dummy_width, dummy_channels), dtype=torch.float32),
torch.ones((total_output_frames, dummy_height, dummy_width), dtype=torch.float32))
device = input_video_frames.device
dtype = input_video_frames.dtype
batch_size_input, frame_height, frame_width, num_channels = input_video_frames.shape
# 2. Prepare Start Frames (from overlap)
actual_overlap_frames = min(overlap_frames, batch_size_input, total_output_frames)
if actual_overlap_frames < overlap_frames:
log.warning(f"Requested {overlap_frames} overlap frames but input video only has {batch_size_input} frames or total output is smaller. Using {actual_overlap_frames} instead.")
overlap_start_idx = batch_size_input - actual_overlap_frames
start_frames_part = input_video_frames[overlap_start_idx : overlap_start_idx + actual_overlap_frames].clone()
# 3. Prepare End Frame
end_frame_part = torch.empty((0, frame_height, frame_width, num_channels), device=device, dtype=dtype)
num_end_frames = 0
if end_frame is not None and end_frame.shape[0] > 0 and total_output_frames > actual_overlap_frames:
num_end_frames = 1
end_frame_processed = end_frame[0].clone().to(device=device, dtype=dtype)
if end_frame_processed.shape != (frame_height, frame_width, num_channels):
log.info(f"Resizing end_frame from {end_frame_processed.shape} to {(frame_height, frame_width, num_channels)}.")
frame_to_resize = end_frame_processed.unsqueeze(0).permute(0, 3, 1, 2)
resized_frame = common_upscale(frame_to_resize, frame_width, frame_height, "lanczos", "disabled")
end_frame_processed = resized_frame.permute(0, 2, 3, 1).squeeze(0)
end_frame_part = end_frame_processed.unsqueeze(0)
# 4. Prepare Middle Frames
num_middle_frames = total_output_frames - actual_overlap_frames - num_end_frames
middle_frames_part = torch.empty((0, frame_height, frame_width, num_channels), device=device, dtype=dtype)
if num_middle_frames > 0:
if control_images is not None:
log.info(f"Using 'control_images' to fill the {num_middle_frames} middle frames with '{when_to_start_control_frames}' mode.")
control_images_resized = common_upscale(control_images.movedim(-1, 1), frame_width, frame_height, "lanczos", "disabled").movedim(1, -1)
if when_to_start_control_frames == "beginning_of_generation":
# Skip the first overlap_frames control images to avoid duplication
duplicate_count = min(actual_overlap_frames, control_images_resized.shape[0])
available_after_dup = control_images_resized.shape[0] - duplicate_count
if available_after_dup < num_middle_frames:
log.info(f"After skipping {duplicate_count} control frames, only {available_after_dup} remain; padding {num_middle_frames - available_after_dup} frames with 'empty_frame_fill_level'.")
selected_control = control_images_resized[duplicate_count:]
padding_needed = num_middle_frames - selected_control.shape[0]
padding = torch.ones((padding_needed, frame_height, frame_width, num_channels), device=device, dtype=dtype) * empty_frame_fill_level
middle_frames_part = torch.cat([selected_control, padding], dim=0)
else:
middle_frames_part = control_images_resized[duplicate_count:duplicate_count + num_middle_frames].clone()
else: # "after_overlap_frames"
# Use control frames from the beginning of the sequence (C0, C1, C2...)
if control_images_resized.shape[0] < num_middle_frames:
log.warning(f"Provided 'control_images' have {control_images_resized.shape[0]} frames, less than needed ({num_middle_frames}). Padding with 'empty_frame_fill_level'.")
padding_needed = num_middle_frames - control_images_resized.shape[0]
padding = torch.ones((padding_needed, frame_height, frame_width, num_channels), device=device, dtype=dtype) * empty_frame_fill_level
middle_frames_part = torch.cat([control_images_resized, padding], dim=0)
else:
middle_frames_part = control_images_resized[:num_middle_frames].clone()
else:
log.info(f"No 'control_images', filling {num_middle_frames} middle frames with level {empty_frame_fill_level}.")
middle_frames_part = torch.ones((num_middle_frames, frame_height, frame_width, num_channels), device=device, dtype=dtype) * empty_frame_fill_level
# 5. Assemble Final Video
continuation_video_output = torch.cat([start_frames_part, middle_frames_part, end_frame_part], dim=0)
# 6. Create Mask
continuation_frame_masks = torch.ones((total_output_frames, frame_height, frame_width), device=device, dtype=dtype)
# Apply mask logic based on when_to_start_masks parameter
if when_to_start_masks == "beginning_of_generation":
# Set known frames (overlap and end) to 0.0, but also set middle section based on control frame logic
if actual_overlap_frames > 0:
continuation_frame_masks[0:actual_overlap_frames] = 0.0
if num_end_frames > 0:
continuation_frame_masks[-num_end_frames:] = 0.0
# For middle section, follow the same logic as control frames
if control_images is not None and num_middle_frames > 0:
duplicate_count = min(actual_overlap_frames, control_images.shape[0])
available_after_dup = control_images.shape[0] - duplicate_count
if available_after_dup >= num_middle_frames:
# If we have enough control frames after skipping, set those middle frames as known (0.0)
middle_start = actual_overlap_frames
middle_end = middle_start + num_middle_frames
continuation_frame_masks[middle_start:middle_end] = 0.0
else: # "after_overlap_frames"
# Set known frames (overlap and end) to 0.0, rest stay as 1.0 (inpaint)
if actual_overlap_frames > 0:
continuation_frame_masks[0:actual_overlap_frames] = 0.0
if num_end_frames > 0:
continuation_frame_masks[-num_end_frames:] = 0.0
# 7. Handle optional inpaint_mask override
if inpaint_mask is not None:
log.info("Processing provided 'inpaint_mask', which will override the automatically generated mask.")
processed_mask = common_upscale(inpaint_mask.unsqueeze(1), frame_width, frame_height, "nearest-exact", "disabled").squeeze(1).to(device)
if processed_mask.shape[0] != total_output_frames:
log.info(f"Adjusting inpaint_mask frame count from {processed_mask.shape[0]} to {total_output_frames}.")
if processed_mask.shape[0] < total_output_frames:
num_repeats = (total_output_frames + processed_mask.shape[0] - 1) // processed_mask.shape[0]
processed_mask = processed_mask.repeat(num_repeats, 1, 1)[:total_output_frames]
else:
processed_mask = processed_mask[:total_output_frames]
continuation_frame_masks = processed_mask.to(dtype=dtype)
log.info(f"Generated continuation video. Start: {actual_overlap_frames} frames, Middle: {num_middle_frames} frames, End: {num_end_frames} frames.")
return (continuation_video_output.cpu().float(), continuation_frame_masks.cpu().float())
class WanInputFrameNumber:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"frame_number": ("INT", {"default": 81, "min": 1, "max": 10000, "step": 4, "tooltip": "Frame number where (frames - 1) is divisible by 4."}),
},
}
RETURN_TYPES = ("INT",)
RETURN_NAMES = ("frame_number",)
FUNCTION = "get_frame_number"
CATEGORY = "Steerable-Motion"
DESCRIPTION = "Outputs a frame number that satisfies the WAN constraint: (frames - 1) divisible by 4."
def get_frame_number(self, frame_number):
frame_number = int(frame_number)
if (frame_number - 1) % 4 != 0:
raise ValueError("frame_number must satisfy (frame_number - 1) divisible by 4")
return (frame_number,)
class WanVideoBlenderNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"overlap_frames": ("INT", {"default": 10, "min": 1, "max": 1000, "step": 1}),
"video_1": ("IMAGE",),
"video_2": ("IMAGE",),
},
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("blended_video_frames",)
FUNCTION = "blend_videos"
CATEGORY = "Steerable-Motion"
DESCRIPTION = "Blends two input videos with a cross-fade. The resolution of the second clip is resized to match the first."
def _resize_video(self, video, target_height, target_width):
"""Resize a batch of frames (B,H,W,C) to (target_height,target_width) using Lanczos."""
if video.shape[1] == target_height and video.shape[2] == target_width:
return video
# (B, H, W, C) -> (B, C, H, W)
video_permuted = video.permute(0, 3, 1, 2)
resized = common_upscale(video_permuted, target_width, target_height, "lanczos", "disabled") # (B, C, H, W)
return resized.permute(0, 2, 3, 1)
def _cross_fade(self, tail, head, overlap_frames):
"""Blend two tensors of shape (overlap_frames,H,W,C) using linear alpha."""
device, dtype = tail.device, tail.dtype
alphas = torch.linspace(0, 1, overlap_frames, device=device, dtype=dtype).view(-1, 1, 1, 1)
blended = tail * (1 - alphas) + head * alphas
return blended
def blend_videos(self, overlap_frames, video_1, video_2):
if video_1 is None or video_2 is None:
raise ValueError("Both video_1 and video_2 are required.")
# Reference dimensions and properties from first video
ref_h, ref_w = video_1.shape[1:3]
# Ensure second video matches size
video_2_resized = self._resize_video(video_2, ref_h, ref_w)
if video_1.shape[0] < overlap_frames or video_2_resized.shape[0] < overlap_frames:
raise ValueError(f"One of the videos is shorter than overlap_frames={overlap_frames}.")
# Extract segments for blending
tail = video_1[-overlap_frames:]
head = video_2_resized[:overlap_frames]
blended = self._cross_fade(tail, head, overlap_frames)
# Assemble new timeline
final_video = torch.cat([
video_1[:-overlap_frames],
blended,
video_2_resized[overlap_frames:]
], dim=0)
return (final_video.cpu().float(),)
# NODE MAPPING
NODE_CLASS_MAPPINGS = {
"BatchCreativeInterpolation": BatchCreativeInterpolationNode,
"IpaConfiguration": IpaConfigurationNode,
"RemoveAndInterpolateFrames": RemoveAndInterpolateFramesNode,
"VideoFrameExtractorAndMaskGenerator": VideoFrameExtractorAndMaskGenerator,
"VideoContinuationGenerator": VideoContinuationGenerator,
"WanInputFrameNumber": WanInputFrameNumber,
"WanVideoBlender": WanVideoBlenderNode,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"BatchCreativeInterpolation": "Batch Creative Interpolation 🎞️🅢🅜",
"IpaConfiguration": "IP-Adapter Configuration 🎞️🅢🅜",
"RemoveAndInterpolateFrames": "Remove and Interpolate Frames 🎞️🅢🅜",
"VideoFrameExtractorAndMaskGenerator": "Video Frame Extractor & Mask Generator 🎞️🅢🅜",
"VideoContinuationGenerator": "Video Continuation Generator 🎞️🅢🅜",
"WanInputFrameNumber": "WAN Input Frame Number 🎞️🅢🅜",
"WanVideoBlender": "WAN Video Blender 🎞️🅢🅜",
}
Reference in New Issue View Git Blame Copy Permalink