Greg
277f12c213
- New ChainTextEmbeds node: chains up to 8 WanVideoTextEncode outputs into an ordered TEXT_EMBED_SEQUENCE for per-chunk prompting - WanChunkedI2VSampler: new optional text_embed_sequence input overrides single text_embeds when connected - Chunk N uses embed N from the sequence; last embed repeats for any extra chunks beyond the sequence length - Console output shows which embed index is used per chunk - README updated with per-chunk prompting workflow diagram
709 lines
28 KiB
Python
709 lines
28 KiB
Python
# MIT License — Copyright (c) 2026 Greg Tee — see LICENSE file
|
|
"""
|
|
ComfyUI_VideoChunkTools — Utility nodes for generating long videos
|
|
by splitting into overlapping chunks with rolling reference frames.
|
|
|
|
Designed to solve the "context window reversal" problem where video
|
|
generation models (like Wan/FantasyPortrait) revert to the starting
|
|
state after ~135 frames because the reference image embedding pulls
|
|
the generation back.
|
|
|
|
WORKFLOW (Rolling Reference for FantasyPortrait):
|
|
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
|
1. Load your full driving video (e.g., 300 frames)
|
|
2. ExtractVideoChunk(driving, chunk_index=0, chunk_frames=81, overlap=16)
|
|
→ driving_chunk_0 (frames 0-80)
|
|
3. Run FantasyPortrait pipeline with ORIGINAL portrait as reference
|
|
→ generated_chunk_0
|
|
4. GetFrameByIndex(generated_chunk_0, index=-1) → last_frame
|
|
5. ExtractVideoChunk(driving, chunk_index=1, ...) → driving_chunk_1
|
|
6. Run FantasyPortrait pipeline with last_frame as NEW reference
|
|
→ generated_chunk_1
|
|
7. BlendVideoChunks(generated_chunk_0, generated_chunk_1, overlap=16)
|
|
→ seamless output
|
|
8. Repeat steps 4-7 for additional chunks
|
|
|
|
The key insight: by using the LAST generated frame as the reference
|
|
for the next chunk, the model's identity stays consistent but the
|
|
pose/action can progress forward naturally without reversal.
|
|
"""
|
|
|
|
import math
|
|
import torch
|
|
|
|
|
|
class BlendVideoChunks:
|
|
"""
|
|
Crossfade-blend two video chunks that share overlapping frames
|
|
at the boundary into a single continuous video.
|
|
|
|
chunk_a ends where chunk_b begins — they share 'overlap_frames'
|
|
frames at the junction. This node smoothly transitions between them.
|
|
|
|
Example:
|
|
chunk_a = 81 frames (frames 0-80 of final video)
|
|
chunk_b = 81 frames (frames 65-145 of final video)
|
|
overlap = 16 frames
|
|
Result = 146 frames: A[0:65] + crossfade[65:81] + B[16:81]
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"chunk_a": ("IMAGE", {
|
|
"tooltip": "First video chunk (the earlier segment)"
|
|
}),
|
|
"chunk_b": ("IMAGE", {
|
|
"tooltip": "Second video chunk (continues from chunk_a). "
|
|
"The first 'overlap_frames' of chunk_b overlap "
|
|
"with the last 'overlap_frames' of chunk_a."
|
|
}),
|
|
"overlap_frames": ("INT", {
|
|
"default": 16, "min": 1, "max": 200, "step": 1,
|
|
"tooltip": "Number of frames shared between the two chunks. "
|
|
"These frames will be crossfaded."
|
|
}),
|
|
"blend_curve": (["linear", "ease_in_out", "sigmoid"], {
|
|
"default": "ease_in_out",
|
|
"tooltip": "Blend curve shape: linear = straight ramp, "
|
|
"ease_in_out = smooth S-curve (3t²-2t³, best for most cases), "
|
|
"sigmoid = steep S-curve (snappier transition)"
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("IMAGE", "INT",)
|
|
RETURN_NAMES = ("video", "total_frames",)
|
|
FUNCTION = "blend"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = ("Crossfade-blend two overlapping video chunks into a single "
|
|
"seamless video. Use with ExtractVideoChunk and GetFrameByIndex "
|
|
"for rolling-reference long video generation.")
|
|
|
|
def blend(self, chunk_a, chunk_b, overlap_frames, blend_curve):
|
|
a_len = chunk_a.shape[0]
|
|
b_len = chunk_b.shape[0]
|
|
|
|
if overlap_frames >= a_len:
|
|
raise ValueError(
|
|
f"overlap_frames ({overlap_frames}) must be less than "
|
|
f"chunk_a length ({a_len})"
|
|
)
|
|
if overlap_frames >= b_len:
|
|
raise ValueError(
|
|
f"overlap_frames ({overlap_frames}) must be less than "
|
|
f"chunk_b length ({b_len})"
|
|
)
|
|
|
|
# Split into non-overlapping + overlapping regions
|
|
a_unique = chunk_a[:a_len - overlap_frames] # frames only in A
|
|
a_overlap = chunk_a[a_len - overlap_frames:] # last N frames of A
|
|
b_overlap = chunk_b[:overlap_frames] # first N frames of B
|
|
b_unique = chunk_b[overlap_frames:] # frames only in B
|
|
|
|
# Build blend weights (0→1 = A→B transition)
|
|
t = torch.linspace(0.0, 1.0, overlap_frames, device=chunk_a.device,
|
|
dtype=chunk_a.dtype)
|
|
|
|
if blend_curve == "ease_in_out":
|
|
# Hermite smoothstep: 3t² - 2t³ (smooth acceleration/deceleration)
|
|
t = t * t * (3.0 - 2.0 * t)
|
|
elif blend_curve == "sigmoid":
|
|
# Steep logistic S-curve (sharper midpoint transition)
|
|
t = torch.sigmoid((t - 0.5) * 12.0)
|
|
t = (t - t[0]) / (t[-1] - t[0]) # renormalize to exact 0-1
|
|
|
|
# Reshape for broadcasting: (overlap_frames, 1, 1, 1) over (H, W, C)
|
|
weights = t.view(-1, 1, 1, 1)
|
|
|
|
# Crossfade: A*(1-w) + B*w
|
|
blended = a_overlap * (1.0 - weights) + b_overlap * weights
|
|
|
|
# Concatenate: [A unique] + [crossfade zone] + [B unique]
|
|
result = torch.cat([a_unique, blended, b_unique], dim=0)
|
|
|
|
return (result, result.shape[0],)
|
|
|
|
|
|
class ExtractVideoChunk:
|
|
"""
|
|
Extract a specific chunk from a longer video, with overlap support.
|
|
|
|
Divides a video into chunks of 'chunk_frames' length, where adjacent
|
|
chunks overlap by 'overlap_frames'. Use chunk_index to get each chunk.
|
|
|
|
Chunk layout (e.g., 300 frames, chunk=81, overlap=16):
|
|
Chunk 0: frames 0-80 (81 frames)
|
|
Chunk 1: frames 65-145 (81 frames, starts 16 before chunk 0 ends)
|
|
Chunk 2: frames 130-210 (81 frames)
|
|
Chunk 3: frames 195-275 (81 frames)
|
|
Chunk 4: frames 219-299 (last chunk, may start closer if near end)
|
|
|
|
For Wan models: chunk_frames should follow ((n-1)//4)*4+1 rule
|
|
(i.e., 5, 9, 13, ..., 77, 81, 85, ..., 129, 133, 137)
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"images": ("IMAGE", {
|
|
"tooltip": "Full driving video / image sequence to chunk"
|
|
}),
|
|
"chunk_index": ("INT", {
|
|
"default": 0, "min": 0, "max": 1000,
|
|
"tooltip": "Which chunk to extract (0-based)"
|
|
}),
|
|
"chunk_frames": ("INT", {
|
|
"default": 81, "min": 2, "max": 2000, "step": 1,
|
|
"tooltip": "Number of frames per chunk. "
|
|
"For Wan models, use values like 81, 97, 113, 129 "
|
|
"(must be 1 mod 4: ((n-1)//4)*4+1)"
|
|
}),
|
|
"overlap_frames": ("INT", {
|
|
"default": 16, "min": 0, "max": 200, "step": 1,
|
|
"tooltip": "Number of frames that overlap between adjacent chunks. "
|
|
"These will be crossfaded by BlendVideoChunks."
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("IMAGE", "INT", "INT", "BOOLEAN",)
|
|
RETURN_NAMES = ("chunk", "total_chunks", "chunk_index", "is_last_chunk",)
|
|
FUNCTION = "extract"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = ("Extract chunk N from a video for rolling-reference generation. "
|
|
"Chunks overlap by overlap_frames so they can be blended later.")
|
|
|
|
def extract(self, images, chunk_index, chunk_frames, overlap_frames):
|
|
total_frames = images.shape[0]
|
|
|
|
if overlap_frames >= chunk_frames:
|
|
raise ValueError(
|
|
f"overlap_frames ({overlap_frames}) must be less than "
|
|
f"chunk_frames ({chunk_frames})"
|
|
)
|
|
|
|
stride = chunk_frames - overlap_frames
|
|
|
|
# Calculate total number of chunks needed to cover all frames
|
|
if total_frames <= chunk_frames:
|
|
total_chunks = 1
|
|
else:
|
|
total_chunks = 1 + math.ceil((total_frames - chunk_frames) / stride)
|
|
|
|
# Calculate start index for this chunk
|
|
start = chunk_index * stride
|
|
end = start + chunk_frames
|
|
|
|
# Clamp to video bounds — last chunk may start earlier to stay full-size
|
|
if end > total_frames:
|
|
end = total_frames
|
|
start = max(0, end - chunk_frames)
|
|
|
|
is_last = chunk_index >= (total_chunks - 1)
|
|
|
|
chunk = images[start:end]
|
|
|
|
return (chunk, total_chunks, chunk_index, is_last,)
|
|
|
|
|
|
class GetFrameByIndex:
|
|
"""
|
|
Extract a single frame from a video / image batch by index.
|
|
|
|
Supports negative indexing:
|
|
-1 = last frame (default — use for rolling reference)
|
|
-2 = second-to-last
|
|
0 = first frame
|
|
5 = sixth frame
|
|
|
|
Primary use case: Get the last frame of a generated chunk
|
|
to use as the reference image for the next chunk.
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"images": ("IMAGE", {
|
|
"tooltip": "Image batch / video frames"
|
|
}),
|
|
"index": ("INT", {
|
|
"default": -1, "min": -10000, "max": 10000,
|
|
"tooltip": "Frame index to extract. Use -1 for last frame "
|
|
"(ideal for rolling reference), -2 for second-to-last, "
|
|
"0 for first frame, etc."
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("IMAGE",)
|
|
RETURN_NAMES = ("frame",)
|
|
FUNCTION = "get_frame"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = ("Get a single frame from a video batch by index. "
|
|
"Use -1 (last frame) as the reference for the next chunk "
|
|
"in a rolling-reference workflow.")
|
|
|
|
def get_frame(self, images, index):
|
|
total = images.shape[0]
|
|
|
|
# Handle negative indexing (Python-style)
|
|
if index < 0:
|
|
index = total + index
|
|
|
|
# Clamp to valid range
|
|
index = max(0, min(index, total - 1))
|
|
|
|
# Return as single-frame batch (1, H, W, C)
|
|
return (images[index:index + 1],)
|
|
|
|
|
|
class GetFrameRange:
|
|
"""
|
|
Extract a range of frames from a video / image batch.
|
|
|
|
Supports negative indexing for both start and end.
|
|
The end index is EXCLUSIVE (Python slice convention).
|
|
|
|
Examples:
|
|
start=0, end=16 → first 16 frames
|
|
start=-16, end=0 → last 16 frames (end=0 means "to the end")
|
|
start=10, end=30 → frames 10-29
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"images": ("IMAGE", {
|
|
"tooltip": "Image batch / video frames"
|
|
}),
|
|
"start": ("INT", {
|
|
"default": 0, "min": -10000, "max": 10000,
|
|
"tooltip": "Start index (inclusive). Negative = from end."
|
|
}),
|
|
"end": ("INT", {
|
|
"default": 0, "min": -10000, "max": 10000,
|
|
"tooltip": "End index (exclusive). 0 = to the end of the video. "
|
|
"Negative = from end."
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("IMAGE", "INT",)
|
|
RETURN_NAMES = ("frames", "frame_count",)
|
|
FUNCTION = "get_range"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = ("Extract a range of frames from a video batch. "
|
|
"Supports negative indexing. End=0 means 'to the end'.")
|
|
|
|
def get_range(self, images, start, end):
|
|
total = images.shape[0]
|
|
|
|
# Handle negative indexing
|
|
if start < 0:
|
|
start = total + start
|
|
start = max(0, min(start, total - 1))
|
|
|
|
# end=0 means "to the end"
|
|
if end == 0:
|
|
end = total
|
|
elif end < 0:
|
|
end = total + end
|
|
end = max(start + 1, min(end, total))
|
|
|
|
result = images[start:end]
|
|
return (result, result.shape[0],)
|
|
|
|
|
|
class VideoChunkPlanner:
|
|
"""
|
|
Calculate the chunking plan for a video — outputs how many chunks
|
|
are needed and the frame ranges, given chunk_frames and overlap.
|
|
|
|
Useful for planning how many times to run the pipeline and
|
|
understanding the layout before starting generation.
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"total_frames": ("INT", {
|
|
"default": 300, "min": 1, "max": 100000,
|
|
"tooltip": "Total number of frames in your driving video"
|
|
}),
|
|
"chunk_frames": ("INT", {
|
|
"default": 81, "min": 2, "max": 2000,
|
|
"tooltip": "Frames per chunk (e.g., 81 for Wan models)"
|
|
}),
|
|
"overlap_frames": ("INT", {
|
|
"default": 16, "min": 0, "max": 200,
|
|
"tooltip": "Frames of overlap between adjacent chunks"
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("INT", "STRING",)
|
|
RETURN_NAMES = ("total_chunks", "plan_info",)
|
|
FUNCTION = "plan"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = ("Plan the chunking strategy for long video generation. "
|
|
"Shows total chunks needed and frame ranges for each.")
|
|
|
|
def plan(self, total_frames, chunk_frames, overlap_frames):
|
|
if overlap_frames >= chunk_frames:
|
|
raise ValueError(
|
|
f"overlap_frames ({overlap_frames}) must be less than "
|
|
f"chunk_frames ({chunk_frames})"
|
|
)
|
|
|
|
stride = chunk_frames - overlap_frames
|
|
|
|
if total_frames <= chunk_frames:
|
|
total_chunks = 1
|
|
final_frames = total_frames
|
|
else:
|
|
total_chunks = 1 + math.ceil((total_frames - chunk_frames) / stride)
|
|
# output video length = total_frames without overlap duplication
|
|
final_frames = chunk_frames + (total_chunks - 1) * stride
|
|
|
|
lines = []
|
|
lines.append(f"=== Video Chunk Plan ===")
|
|
lines.append(f"Source video: {total_frames} frames")
|
|
lines.append(f"Chunk size: {chunk_frames} frames")
|
|
lines.append(f"Overlap: {overlap_frames} frames")
|
|
lines.append(f"Stride: {stride} frames")
|
|
lines.append(f"Total chunks needed: {total_chunks}")
|
|
lines.append(f"Output video length: ~{total_frames} frames")
|
|
lines.append(f"")
|
|
lines.append(f"--- Chunk Layout ---")
|
|
|
|
for i in range(total_chunks):
|
|
start = i * stride
|
|
end = start + chunk_frames
|
|
if end > total_frames:
|
|
end = total_frames
|
|
start = max(0, end - chunk_frames)
|
|
|
|
ref = "original reference" if i == 0 else f"last frame of chunk {i-1}"
|
|
lines.append(
|
|
f" Chunk {i}: frames {start:>4d}-{end-1:>4d} "
|
|
f"({end - start} frames) | ref: {ref}"
|
|
)
|
|
|
|
lines.append(f"")
|
|
lines.append(f"--- Workflow Steps ---")
|
|
lines.append(f"1. Load driving video ({total_frames} frames)")
|
|
for i in range(total_chunks):
|
|
start = i * stride
|
|
end = min(start + chunk_frames, total_frames)
|
|
if end > total_frames:
|
|
start = max(0, total_frames - chunk_frames)
|
|
if i == 0:
|
|
lines.append(f"2. ExtractVideoChunk(index=0) → Generate with original reference")
|
|
else:
|
|
lines.append(
|
|
f"{2+i}. GetFrameByIndex(chunk_{i-1}, -1) → New reference → "
|
|
f"ExtractVideoChunk(index={i}) → Generate"
|
|
)
|
|
lines.append(f"{2+total_chunks}. Chain BlendVideoChunks for each pair")
|
|
|
|
plan_text = "\n".join(lines)
|
|
return (total_chunks, plan_text,)
|
|
|
|
|
|
class BlendLatentChunks:
|
|
"""
|
|
Join two video chunk LATENTS along the temporal dimension BEFORE decoding.
|
|
|
|
The overlap_frames parameter is in LATENT temporal space (not pixel frames).
|
|
For Wan models with 4x temporal compression:
|
|
latent_overlap = pixel_overlap / 4
|
|
e.g. 16 pixel frames → 4 latent frames
|
|
|
|
Blend modes:
|
|
hard_cut — No blending. Cuts at the midpoint of the overlap region.
|
|
Best for rolling-reference workflows where both chunks
|
|
generated similar content in the overlap zone.
|
|
slerp — Spherical linear interpolation. Preserves latent vector
|
|
magnitude (standard for diffusion model interpolation).
|
|
Avoids the norm-collapse that causes washed-out frames.
|
|
linear — Simple linear crossfade in latent space.
|
|
ease_in_out — Hermite smooth-step crossfade.
|
|
sigmoid — Steep S-curve crossfade.
|
|
|
|
Wire BOTH sampler latent outputs into this node, then feed the result
|
|
into a SINGLE VAE decode for the final seamless video.
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"latent_a": ("LATENT",),
|
|
"latent_b": ("LATENT",),
|
|
"overlap_frames": ("INT", {
|
|
"default": 4,
|
|
"min": 1,
|
|
"max": 64,
|
|
"step": 1,
|
|
"tooltip": "Overlap in LATENT temporal frames. For Wan: pixel_overlap / 4 (e.g. 16px = 4 latent)"
|
|
}),
|
|
"blend_curve": (["hard_cut", "slerp", "linear", "ease_in_out", "sigmoid"], {
|
|
"default": "hard_cut",
|
|
"tooltip": "hard_cut = clean cut at midpoint (best for rolling reference). "
|
|
"slerp = spherical interpolation (preserves latent magnitude). "
|
|
"linear/ease_in_out/sigmoid = alpha-blend crossfades."
|
|
}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("LATENT", "INT",)
|
|
RETURN_NAMES = ("latent", "total_latent_frames",)
|
|
FUNCTION = "blend"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = (
|
|
"Join two video chunk latents along temporal dimension before decoding. "
|
|
"hard_cut (recommended): clean cut at overlap midpoint — no dissolve. "
|
|
"slerp: spherical interpolation preserving latent magnitude. "
|
|
"overlap_frames is in LATENT space (Wan: pixel_overlap / 4)."
|
|
)
|
|
|
|
@staticmethod
|
|
def _slerp_frame(t, a, b):
|
|
"""Spherical linear interpolation between two latent frames.
|
|
Preserves the norm of the vectors — standard for diffusion latent spaces.
|
|
a, b: [B, C, H, W] tensors. t: scalar 0-1."""
|
|
a_f = a.flatten().float()
|
|
b_f = b.flatten().float()
|
|
a_norm = torch.nn.functional.normalize(a_f, dim=0)
|
|
b_norm = torch.nn.functional.normalize(b_f, dim=0)
|
|
dot = (a_norm * b_norm).sum().clamp(-1.0, 1.0)
|
|
omega = torch.acos(dot)
|
|
# Fall back to lerp for nearly-parallel vectors
|
|
if omega.abs() < 1e-5:
|
|
return (a.float() * (1.0 - t) + b.float() * t).to(a.dtype)
|
|
sin_omega = torch.sin(omega)
|
|
wa = torch.sin((1.0 - t) * omega) / sin_omega
|
|
wb = torch.sin(t * omega) / sin_omega
|
|
return (a.float() * wa + b.float() * wb).to(a.dtype)
|
|
|
|
def blend(self, latent_a, latent_b, overlap_frames, blend_curve):
|
|
a = latent_a["samples"]
|
|
b = latent_b["samples"]
|
|
|
|
if a.dim() != 5 or b.dim() != 5:
|
|
raise ValueError(
|
|
f"Expected 5D video latents [B,C,T,H,W], "
|
|
f"got {a.dim()}D and {b.dim()}D. "
|
|
f"This node is for VIDEO latents, not image latents."
|
|
)
|
|
|
|
T_a = a.shape[2]
|
|
T_b = b.shape[2]
|
|
overlap = min(overlap_frames, T_a, T_b)
|
|
|
|
# Slice the latent tensors
|
|
a_keep = a[:, :, :T_a - overlap, :, :]
|
|
a_over = a[:, :, T_a - overlap:, :, :]
|
|
b_over = b[:, :, :overlap, :, :]
|
|
b_keep = b[:, :, overlap:, :, :]
|
|
|
|
if blend_curve == "hard_cut":
|
|
# Clean cut at the midpoint — no interpolation at all.
|
|
# With rolling reference, both chunks have similar content here,
|
|
# so a clean cut is virtually invisible.
|
|
mid = overlap // 2
|
|
blended = torch.cat([a_over[:, :, :mid, :, :],
|
|
b_over[:, :, mid:, :, :]], dim=2)
|
|
|
|
elif blend_curve == "slerp":
|
|
# Spherical linear interpolation — preserves latent vector norms.
|
|
# Standard technique for interpolating in diffusion latent spaces.
|
|
# Avoids the washed-out frames caused by lerp norm-collapse.
|
|
frames = []
|
|
for i in range(overlap):
|
|
t = i / max(overlap - 1, 1)
|
|
frame = self._slerp_frame(t, a_over[:, :, i, :, :],
|
|
b_over[:, :, i, :, :])
|
|
frames.append(frame.unsqueeze(2))
|
|
blended = torch.cat(frames, dim=2)
|
|
|
|
else:
|
|
# Alpha-blend crossfades (linear, ease_in_out, sigmoid)
|
|
weights = torch.linspace(0.0, 1.0, overlap,
|
|
device=a.device, dtype=a.dtype)
|
|
if blend_curve == "ease_in_out":
|
|
weights = weights * weights * (3.0 - 2.0 * weights)
|
|
elif blend_curve == "sigmoid":
|
|
weights = torch.sigmoid((weights - 0.5) * 10.0)
|
|
weights = (weights - weights[0]) / (weights[-1] - weights[0])
|
|
weights = weights.reshape(1, 1, -1, 1, 1)
|
|
blended = a_over * (1.0 - weights) + b_over * weights
|
|
|
|
# Concatenate: a_unique + overlap_region + b_unique
|
|
result = torch.cat([a_keep, blended, b_keep], dim=2)
|
|
|
|
# Preserve any extra keys from latent_a (noise_mask, batch_index, etc.)
|
|
out = {k: v for k, v in latent_a.items() if k != "samples"}
|
|
out["samples"] = result
|
|
|
|
return (out, int(result.shape[2]),)
|
|
|
|
|
|
class ConcatVideoChunks:
|
|
"""
|
|
Concatenate two decoded video chunks into a single continuous video.
|
|
|
|
For rolling-reference workflows where chunk B was generated using chunk A's
|
|
last frame as the CLIP Vision reference:
|
|
- Chunk B's FIRST frame closely matches chunk A's LAST frame
|
|
- Set trim_b_start=1 to drop chunk B's duplicate first frame
|
|
- Result: A[all frames] + B[frame 1 onward] — clean seamless join
|
|
|
|
No blending, no overlap, no dissolve — just a clean concatenation.
|
|
The Wan I2V model naturally makes chunk B's first frame match the reference,
|
|
so the boundary is virtually invisible.
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"chunk_a": ("IMAGE", {
|
|
"tooltip": "First video chunk (decoded pixels)"
|
|
}),
|
|
"chunk_b": ("IMAGE", {
|
|
"tooltip": "Second video chunk (decoded pixels). "
|
|
"Its first frame should match chunk_a's last frame "
|
|
"if using rolling reference."
|
|
}),
|
|
"trim_b_start": ("INT", {
|
|
"default": 1, "min": 0, "max": 100, "step": 1,
|
|
"tooltip": "Number of frames to trim from the START of chunk_b "
|
|
"before concatenating. Default 1 = drop B's first frame "
|
|
"(which duplicates A's last frame in rolling-reference workflows)."
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("IMAGE", "INT",)
|
|
RETURN_NAMES = ("video", "total_frames",)
|
|
FUNCTION = "concat"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = (
|
|
"Concatenate two decoded video chunks. Optionally trim B's first N frames "
|
|
"to remove the duplicate reference frame. Simplest join method — no blending needed "
|
|
"when using rolling reference (chunk B's first frame already matches A's last)."
|
|
)
|
|
|
|
def concat(self, chunk_a, chunk_b, trim_b_start):
|
|
if trim_b_start > 0:
|
|
chunk_b = chunk_b[trim_b_start:]
|
|
|
|
if chunk_b.shape[0] == 0:
|
|
return (chunk_a, chunk_a.shape[0],)
|
|
|
|
result = torch.cat([chunk_a, chunk_b], dim=0)
|
|
return (result, result.shape[0],)
|
|
|
|
|
|
class ChainTextEmbeds:
|
|
"""
|
|
Chains multiple pre-encoded text embeddings into an ordered sequence
|
|
for per-chunk text conditioning in WanChunkedI2VSampler.
|
|
|
|
Wire one WanVideoTextEncode output into each embed slot.
|
|
Chunk 1 uses embed_1, chunk 2 uses embed_2, and so on.
|
|
If you have fewer embeds than chunks, the last embed repeats
|
|
for all remaining chunks.
|
|
|
|
Example (3 chunks, 2 prompts):
|
|
embed_1 = encode("A cat sleeps on a sofa")
|
|
embed_2 = encode("The cat wakes up and stretches")
|
|
→ Chunk 1 gets embed_1, chunks 2-3 get embed_2
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(s):
|
|
return {
|
|
"required": {
|
|
"embed_1": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 1 (required)."
|
|
}),
|
|
},
|
|
"optional": {
|
|
"embed_2": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 2."
|
|
}),
|
|
"embed_3": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 3."
|
|
}),
|
|
"embed_4": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 4."
|
|
}),
|
|
"embed_5": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 5."
|
|
}),
|
|
"embed_6": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 6."
|
|
}),
|
|
"embed_7": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 7."
|
|
}),
|
|
"embed_8": ("WANVIDEOTEXTEMBEDS", {
|
|
"tooltip": "Text embedding for chunk 8."
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("TEXT_EMBED_SEQUENCE",)
|
|
RETURN_NAMES = ("embed_sequence",)
|
|
FUNCTION = "chain"
|
|
CATEGORY = "VideoChunkTools"
|
|
DESCRIPTION = (
|
|
"Chains multiple WanVideoTextEncode outputs into an ordered "
|
|
"sequence for per-chunk text conditioning.\n\n"
|
|
"Connect the output to the 'text_embed_sequence' input of "
|
|
"WanChunkedI2VSampler. If fewer embeds than chunks, the last "
|
|
"embed repeats for all remaining chunks."
|
|
)
|
|
|
|
def chain(self, embed_1, embed_2=None, embed_3=None, embed_4=None,
|
|
embed_5=None, embed_6=None, embed_7=None, embed_8=None):
|
|
sequence = [embed_1]
|
|
for e in [embed_2, embed_3, embed_4, embed_5,
|
|
embed_6, embed_7, embed_8]:
|
|
if e is not None:
|
|
sequence.append(e)
|
|
return (sequence,)
|
|
|
|
|
|
# ---------- Registration ----------
|
|
|
|
NODE_CLASS_MAPPINGS = {
|
|
"BlendVideoChunks": BlendVideoChunks,
|
|
"BlendLatentChunks": BlendLatentChunks,
|
|
"ConcatVideoChunks": ConcatVideoChunks,
|
|
"ExtractVideoChunk": ExtractVideoChunk,
|
|
"GetFrameByIndex": GetFrameByIndex,
|
|
"GetFrameRange": GetFrameRange,
|
|
"VideoChunkPlanner": VideoChunkPlanner,
|
|
"ChainTextEmbeds": ChainTextEmbeds,
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"BlendVideoChunks": "Blend Video Chunks (Crossfade)",
|
|
"BlendLatentChunks": "Blend Latent Chunks (Pre-Decode)",
|
|
"ConcatVideoChunks": "Concat Video Chunks",
|
|
"ExtractVideoChunk": "Extract Video Chunk",
|
|
"GetFrameByIndex": "Get Frame By Index",
|
|
"GetFrameRange": "Get Frame Range",
|
|
"VideoChunkPlanner": "Video Chunk Planner",
|
|
"ChainTextEmbeds": "Chain Text Embeds 🔗",
|
|
}
|