nodes/marcoc2-ComfyUI-AnotherUtils
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Align AnotherLTXSequencer and ImageListSampler with modular LTXV pipeline

Browse Source
This commit is contained in:
Marco
2026-04-16 00:49:16 -03:00
parent 63b3d2fc86
commit fa9e8d6a6d
4 changed files with 249 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
logic_management/image_list_sampler.py
+2 -6
View File
@@ -6,7 +6,7 @@ class ImageListSampler:
"images": ("IMAGE",),
"count": ("INT", {"default": 3, "min": 1, "max": 10000, "step": 1}),
"target_frames": ("INT", {"default": 0, "min": 0, "max": 10000, "step": 1,
"tooltip": "If > 0, outputs indices scaled to this length. Example: Use target_frames=97 for LTXV video generation."}),
"tooltip": "If > 0, outputs indices scaled to this length."}),
}
}
@@ -18,7 +18,7 @@ class ImageListSampler:
CATEGORY = "AnotherUtils/logic"
def sample_images(self, images, count, target_frames=0):
# Unwrap parameters that should be single values
# Unwrap parameters
cnt = count[0] if isinstance(count, list) else count
tgt_frames = target_frames[0] if isinstance(target_frames, list) else target_frames
@@ -44,14 +44,10 @@ class ImageListSampler:
if cnt == 1:
return ([all_frames[0]], [0])
# New math: Anchors to edges. 0 to total-1
step_img = (total - 1) / (cnt - 1) if (cnt - 1) > 0 else 0
img_indices = [int(round(i * step_img)) for i in range(cnt)]
# Clamp to avoid rounding out-of-bounds just in case
img_indices = [max(0, min(total - 1, idx)) for idx in img_indices]
# Determine output indices (scaled to target_frames if provided)
if tgt_frames is not None and tgt_frames > 0:
step_tgt = (tgt_frames - 1) / (cnt - 1) if (cnt - 1) > 0 else 0
out_indices = [int(round(i * step_tgt)) for i in range(cnt)]
logic_management/image_list_to_batch.py
+183
View File
@@ -0,0 +1,183 @@
import torch
import torch.nn.functional as F
import comfy.utils
import io
from PIL import Image
import numpy as np
class ImageListToBatch:
"""
Takes a Python List of images (e.g. from ImageListSampler) and converts it
into a single Batched Tensor [N, H, W, C] to be compatible with nodes
like LTXSequencer ('multi_input').
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE",),
"width": ("INT", {"default": 0, "min": 0, "max": 8192, "step": 1}),
"height": ("INT", {"default": 0, "min": 0, "max": 8192, "step": 1}),
"interpolation": (["lanczos", "nearest", "bilinear", "bicubic", "area", "nearest-exact"],),
"resize_method": (["keep proportion", "stretch", "pad", "crop"],),
"multiple_of": ("INT", {"default": 0, "min": 0, "max": 512, "step": 1}),
"img_compression": ("INT", {"default": 18, "min": 0, "max": 100, "step": 1}),
}
}
RETURN_TYPES = ("IMAGE", )
RETURN_NAMES = ("multi_output", )
INPUT_IS_LIST = True
FUNCTION = "convert"
CATEGORY = "AnotherUtils/logic"
def resize_image(self, image, width, height, resize_method="keep proportion", interpolation="nearest", multiple_of=0):
MAX_RESOLUTION = 8192
_, oh, ow, _ = image.shape
x = y = x2 = y2 = 0
pad_left = pad_right = pad_top = pad_bottom = 0
if multiple_of > 1:
width = width - (width % multiple_of)
height = height - (height % multiple_of)
if resize_method == 'keep proportion' or resize_method == 'pad':
if width == 0 and oh < height:
width = MAX_RESOLUTION
elif width == 0 and oh >= height:
width = ow
if height == 0 and ow < width:
height = MAX_RESOLUTION
elif height == 0 and ow >= width:
height = oh
ratio = min(width / ow, height / oh)
new_width = round(ow * ratio)
new_height = round(oh * ratio)
if resize_method == 'pad':
pad_left = (width - new_width) // 2
pad_right = width - new_width - pad_left
pad_top = (height - new_height) // 2
pad_bottom = height - new_height - pad_top
width = new_width
height = new_height
elif resize_method == 'crop':
width = width if width > 0 else ow
height = height if height > 0 else oh
ratio = max(width / ow, height / oh)
new_width = round(ow * ratio)
new_height = round(oh * ratio)
x = (new_width - width) // 2
y = (new_height - height) // 2
x2 = x + width
y2 = y + height
if x2 > new_width:
x -= (x2 - new_width)
if x < 0:
x = 0
if y2 > new_height:
y -= (y2 - new_height)
if y < 0:
y = 0
width = new_width
height = new_height
else:
width = width if width > 0 else ow
height = height if height > 0 else oh
# Always apply resize logic
outputs = image.permute(0, 3, 1, 2)
if interpolation == "lanczos":
outputs = comfy.utils.lanczos(outputs, width, height)
else:
outputs = F.interpolate(outputs, size=(height, width), mode=interpolation)
if resize_method == 'pad':
if pad_left > 0 or pad_right > 0 or pad_top > 0 or pad_bottom > 0:
outputs = F.pad(outputs, (pad_left, pad_right, pad_top, pad_bottom), value=0)
outputs = outputs.permute(0, 2, 3, 1)
if resize_method == 'crop':
if x > 0 or y > 0 or x2 > 0 or y2 > 0:
outputs = outputs[:, y:y2, x:x2, :]
if multiple_of > 1 and (outputs.shape[2] % multiple_of != 0 or outputs.shape[1] % multiple_of != 0):
width = outputs.shape[2]
height = outputs.shape[1]
x = (width % multiple_of) // 2
y = (height % multiple_of) // 2
x2 = width - ((width % multiple_of) - x)
y2 = height - ((height % multiple_of) - y)
outputs = outputs[:, y:y2, x:x2, :]
outputs = torch.clamp(outputs, 0, 1)
return outputs
def convert(self, images, width, height, interpolation, resize_method, multiple_of, img_compression):
# Unwrap list inputs (INPUT_IS_LIST=True)
width = width[0] if isinstance(width, list) else width
height = height[0] if isinstance(height, list) else height
interpolation = interpolation[0] if isinstance(interpolation, list) else interpolation
resize_method = resize_method[0] if isinstance(resize_method, list) else resize_method
multiple_of = multiple_of[0] if isinstance(multiple_of, list) else multiple_of
img_compression = img_compression[0] if isinstance(img_compression, list) else img_compression
if not images:
empty = torch.zeros((1, 64, 64, 3))
return (empty,)
# Unwrap list of lists if passed strangely
all_frames = []
for item in images:
if isinstance(item, torch.Tensor):
if len(item.shape) == 4:
for i in range(item.shape[0]):
all_frames.append(item[i:i+1])
else:
all_frames.append(item.unsqueeze(0) if len(item.shape) == 3 else item)
elif isinstance(item, list):
all_frames.extend(item)
if not all_frames:
empty = torch.zeros((1, 64, 64, 3))
return (empty,)
results = []
for img in all_frames:
# Apply Advanced Resize
img = self.resize_image(img, width, height, resize_method, interpolation, multiple_of)
# Compression (Applied after resize to accurately maintain the effect)
if img_compression > 0:
img_np = (img[0].numpy() * 255).clip(0, 255).astype(np.uint8)
img_pil = Image.fromarray(img_np)
img_byte_arr = io.BytesIO()
img_pil.save(img_byte_arr, format="JPEG", quality=max(1, 100 - img_compression))
img_pil = Image.open(img_byte_arr)
img = torch.from_numpy(np.array(img_pil).astype(np.float32) / 255.0)[None,]
results.append(img)
# Check dimension consistency to form a batch
first_shape = results[0].shape
all_same_shape = all(r.shape == first_shape for r in results)
if all_same_shape:
multi_output = torch.cat(results, dim=0)
else:
print("[ImageListToBatch] Warning: Images have different dimensions. Padding batch as zero tensor.")
multi_output = torch.zeros((1, 64, 64, 3))
return (multi_output,)
logic_management/indices_list_to_50.py
+44
View File
@@ -0,0 +1,44 @@
class IndicesListTo50:
"""
Takes a Python List of integers (e.g. from ImageListSampler's 'indices')
and splits it into 50 individual INT outputs.
This allows you to bypass manual typing in nodes that have 50 individual
widgets (like LTXSequencer). Just Right Click their node -> 'Convert Widget to Input'
and wire these outputs directly.
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"indices": ("INT",),
}
}
RETURN_TYPES = ("INT",) * 50
RETURN_NAMES = tuple(f"index_{i+1}" for i in range(50))
INPUT_IS_LIST = True
FUNCTION = "split"
CATEGORY = "AnotherUtils/logic"
def split(self, indices):
if not indices:
return (0,) * 50
# Unwrap list of lists if passed strangely
all_indices = []
for item in indices:
if isinstance(item, list):
all_indices.extend(item)
else:
all_indices.append(item)
if not all_indices:
return (0,) * 50
# Pad individual outputs to exactly 50
results = all_indices[:50]
padded_results = results + [0] * (50 - len(results))
return tuple(padded_results)
video/another_ltx_sequencer.py
+20 -60
View File
@@ -39,7 +39,7 @@ class AnotherLTXSequencer:
CATEGORY = "AnotherUtils/video"
def execute(self, positive, negative, vae, latent, multi_input, indices, num_images, insert_mode, frame_rate, strength):
from comfy_extras.nodes_lt import LTXVAddGuide, get_noise_mask, _append_guide_attention_entry
from comfy_extras.nodes_lt import LTXVAddGuide, get_noise_mask, _append_guide_attention_entry, get_keyframe_idxs
# Unwrap list inputs because INPUT_IS_LIST = True
positive = positive[0] if isinstance(positive, list) else positive
@@ -52,45 +52,25 @@ class AnotherLTXSequencer:
frame_rate = frame_rate[0] if isinstance(frame_rate, list) else frame_rate
strength = strength[0] if isinstance(strength, list) else strength
# Keep indices as a list
# Indices list
idx_list = indices if isinstance(indices, list) else [indices]
if len(idx_list) > 0 and isinstance(idx_list[0], list):
idx_list = idx_list[0]
scale_factors = vae.downscale_index_formula
# --- Extract video latent, separating from audio if nested ---
try:
from comfy.nested_tensor import NestedTensor
is_nested = isinstance(latent["samples"], NestedTensor)
except ImportError:
is_nested = False
if is_nested:
# Extract video-only portion for guide processing
latent_samples = latent["samples"].tensors[0].clone()
audio_samples = latent["samples"].tensors[1]
# Extract noise masks
noise_mask_raw = latent.get("noise_mask", None)
is_mask_nested = isinstance(noise_mask_raw, NestedTensor) if noise_mask_raw is not None else False
if is_mask_nested:
noise_mask = noise_mask_raw.tensors[0].clone()
audio_mask = noise_mask_raw.tensors[1]
else:
# Create default video noise mask
batch, _, lat_t, _, _ = latent_samples.shape
noise_mask = torch.ones((batch, 1, lat_t, 1, 1), dtype=torch.float32, device=latent_samples.device)
audio_mask = None
logger.info(f"[AnotherLTXSequencer] Nested input - Video: {latent_samples.shape}, Audio: {audio_samples.shape}")
else:
latent_samples = latent["samples"].clone()
noise_mask = get_noise_mask(latent).clone()
audio_samples = None
audio_mask = None
is_mask_nested = False
# Pure Video approach
latent_samples = latent["samples"]
noise_mask = get_noise_mask(latent).clone()
if len(latent_samples.shape) != 5:
# Check for NestedTensor (Audio+Video)
try:
from comfy.nested_tensor import NestedTensor
if isinstance(latent_samples, NestedTensor):
raise ValueError("[AnotherLTXSequencer] Este nó agora opera em modo modular (Apenas Vídeo). Use o nó 'Separate AV Latent' antes de conectar aqui, como no workflow oficial.")
except ImportError:
pass
_, _, latent_length, latent_height, latent_width = latent_samples.shape
batch_size = multi_input.shape[0] if multi_input is not None else 0
@@ -121,11 +101,8 @@ class AnotherLTXSequencer:
delta_t = t.shape[2]
if latent_idx + delta_t > latent_length:
logger.warning(f"[AnotherLTXSequencer] Guide at frame {f_idx} exceeds latent length. Skipping.")
continue
# append_keyframe concatenates guide frames onto the video latent (dim=2)
# In modular mode, we just append guides as instructed.
# If the user wants to CROP old guides, they use the modular LTXVCropGuides node.
cur_pos, cur_neg, latent_samples, noise_mask = LTXVAddGuide.append_keyframe(
cur_pos, cur_neg,
frame_idx,
@@ -146,26 +123,9 @@ class AnotherLTXSequencer:
except Exception as e:
logger.error(f"[AnotherLTXSequencer] Failed to append attention entry: {e}")
# --- Build output latent ---
# Build output latent
new_latent = latent.copy()
if is_nested:
# Re-wrap as NestedTensor. The audio is NOT padded because
# the sampler's pack/unpack handles mismatched time dimensions.
# CropGuides will trim the video back after sampling.
from comfy.nested_tensor import NestedTensor
new_latent["samples"] = NestedTensor((latent_samples, audio_samples))
if is_mask_nested and audio_mask is not None:
new_latent["noise_mask"] = NestedTensor((noise_mask, audio_mask))
else:
new_latent["noise_mask"] = noise_mask
logger.info(f"[AnotherLTXSequencer] Nested output - Video: {latent_samples.shape}, Audio: {audio_samples.shape}")
else:
new_latent["samples"] = latent_samples
new_latent["noise_mask"] = noise_mask
new_latent["samples"] = latent_samples
new_latent["noise_mask"] = noise_mask
return (cur_pos, cur_neg, new_latent)