Files
2026-08-04 20:28:12 +02:00

1538 lines
62 KiB
Python

# iamccs_ltx2_tools.py
# ===============================================================
# IAMCCS LTX-2 helpers
# - FrameRate sync (INT + FLOAT)
# - Validator / autofix for LTX-2 constraints
# - Simple control-image preprocess helper (aux) for IC-LoRA style workflows
# ===============================================================
from __future__ import annotations
import logging
import math
from typing import Any, Mapping, Tuple
import torch
from .iamccs_flexible_inputs import FlexibleOptionalInputType, any_type
_F = torch.nn.functional
_log = logging.getLogger("IAMCCS.LTX2.Tools")
def _audio_duration_seconds(audio: Any) -> float:
if not isinstance(audio, Mapping):
raise ValueError("audio input is required")
waveform = audio.get("waveform")
sample_rate = int(audio.get("sample_rate", 0) or 0)
if waveform is None or sample_rate <= 0:
raise ValueError("audio input must contain waveform and sample_rate")
try:
total_samples = int(waveform.shape[-1])
except Exception as exc:
raise ValueError("audio waveform has no sample dimension") from exc
if total_samples <= 0:
raise ValueError("audio waveform is empty")
return float(total_samples) / float(sample_rate)
def _to_grayscale(image: torch.Tensor) -> torch.Tensor:
# image: [B,H,W,C] float in [0,1]
if image.shape[-1] == 1:
return image
r = image[..., 0:1]
g = image[..., 1:2]
b = image[..., 2:3]
return r * 0.299 + g * 0.587 + b * 0.114
def _sobel_edges(gray: torch.Tensor) -> torch.Tensor:
# gray: [B,H,W,1]
x = gray.permute(0, 3, 1, 2) # [B,1,H,W]
kx = torch.tensor([[-1, 0, 1], [-2, 0, 2], [-1, 0, 1]], dtype=x.dtype, device=x.device).view(1, 1, 3, 3)
ky = torch.tensor([[-1, -2, -1], [0, 0, 0], [1, 2, 1]], dtype=x.dtype, device=x.device).view(1, 1, 3, 3)
gx = torch.nn.functional.conv2d(x, kx, padding=1)
gy = torch.nn.functional.conv2d(x, ky, padding=1)
mag = torch.sqrt(torch.clamp(gx * gx + gy * gy, min=0.0))
mag = mag / (mag.amax(dim=(2, 3), keepdim=True).clamp(min=1e-6))
return mag.permute(0, 2, 3, 1) # [B,H,W,1]
class IAMCCS_LTX2_FrameRateSync:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"fps": ("FLOAT", {"default": 24.0, "min": 1.0, "max": 240.0, "step": 0.01}),
"int_mode": (["round", "floor", "ceil", "fixed"], {"default": "round"}),
}
}
RETURN_TYPES = ("INT", "FLOAT", "STRING")
RETURN_NAMES = ("fps_int", "fps_float", "report")
FUNCTION = "sync"
CATEGORY = "IAMCCS/LTX-2"
def sync(self, fps: float, int_mode: str):
fps_in = float(fps)
if int_mode == "floor":
fps_int = int(math.floor(fps_in))
fps_float = fps_in
elif int_mode == "ceil":
fps_int = int(math.ceil(fps_in))
fps_float = fps_in
elif int_mode == "fixed":
# "fixed" = force perfect INT/FLOAT agreement by snapping the float output
# to the derived integer (useful when downstream nodes require exact match).
fps_int = int(round(fps_in))
fps_float = float(fps_int)
else:
fps_int = int(round(fps_in))
fps_float = fps_in
fps_int = max(1, fps_int)
delta = fps_in - float(fps_int)
report = f"fps_in={fps_in:.3f} -> fps_int={fps_int}, fps_float={fps_float:.3f} (mode={int_mode}, delta={delta:+.3f})"
if int_mode == "fixed" and abs(delta) > 1e-6:
_log.warning("[IAMCCS_LTX2_FrameRateSync] fixed mode snapped float to int. %s", report)
return (fps_int, fps_float, report)
class IAMCCS_LTX2_Validator:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
# Wraps EmptyImage-like behavior
"width": ("INT", {"default": 768, "min": 64, "max": 8192, "step": 16}),
"height": ("INT", {"default": 512, "min": 64, "max": 8192, "step": 16}),
"batch_size": ("INT", {"default": 1, "min": 1, "max": 4096, "step": 1}),
"color": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
# Duration widgets are both visible for UX.
# The frontend JS keeps seconds <-> length in sync using IAMCCS_LTX2_FrameRateSync.
"seconds": ("FLOAT", {"default": 4.0, "min": 0.01, "max": 3600.0, "step": 0.01}),
"length": ("INT", {"default": 81, "min": 1, "max": 16385, "step": 1, "control_after_generate": True}),
# Validation controls
"autofix": ("BOOLEAN", {"default": True}),
"length_fix": (["up", "down", "nearest"], {"default": "up"}),
},
# Optional pass-through: if provided, we will make the IMAGE batch frame-count
# match the validated length (8n+1) by padding/cropping. This is the workflow-safe
# way to guarantee the VAE encode constraint without adding extra nodes.
"optional": {
"images": ("IMAGE", {}),
"images_mode": (["pad_repeat_last", "crop_end"], {"default": "pad_repeat_last"}),
},
}
RETURN_TYPES = ("IMAGE", "INT", "STRING")
RETURN_NAMES = ("image", "length", "report")
FUNCTION = "validate"
CATEGORY = "IAMCCS/LTX-2"
def _fix_multiple(self, value: int, multiple: int) -> Tuple[int, int]:
value = int(value)
if multiple <= 0:
return value, 0
rem = value % multiple
if rem == 0:
return value, 0
fixed = value + (multiple - rem)
return fixed, fixed - value
def _frames_rule_fix(self, frames: int, mode: str) -> Tuple[int, int]:
# LTX-2 guideline: frame count should be 8n + 1.
frames = int(frames)
if frames < 1:
frames = 1
rem = (frames - 1) % 8
if rem == 0:
return frames, 0
down = frames - rem
up = frames + (8 - rem)
if mode == "down":
fixed = max(1, down)
elif mode == "nearest":
fixed = up if (up - frames) <= (frames - down) else max(1, down)
else:
fixed = up
return fixed, fixed - frames
def validate(
self,
width: int,
height: int,
batch_size: int,
color: int,
seconds: float,
length: int,
autofix: bool,
length_fix: str,
images: torch.Tensor | None = None,
images_mode: str = "pad_repeat_last",
):
width_in = int(width)
height_in = int(height)
batch_in = int(batch_size)
color_in = int(color)
seconds_in = float(seconds)
length_in = int(length)
# Spatial rule:
# - We keep this relatively permissive (16px multiple) because many valid LTX-2
# workflows use resolutions like 1280x720 (720 is not divisible by 32).
# - If a downstream node truly requires 32, it should validate there.
spatial_multiple = 16
w_ok = (width_in % spatial_multiple) == 0
h_ok = (height_in % spatial_multiple) == 0
l_ok = ((length_in - 1) % 8) == 0
width_fixed, pad_w = self._fix_multiple(width_in, spatial_multiple)
height_fixed, pad_h = self._fix_multiple(height_in, spatial_multiple)
length_fixed, pad_len = self._frames_rule_fix(length_in, length_fix)
if not autofix:
width_fixed, height_fixed, length_fixed = width_in, height_in, length_in
pad_w, pad_h, pad_len = 0, 0, 0
batch_fixed = max(1, batch_in)
color_fixed = max(0, min(255, color_in))
def _pad_repeat_last_frames(x: torch.Tensor, pad: int) -> torch.Tensor:
if pad <= 0:
return x
last = x[-1:, ...].repeat(int(pad), 1, 1, 1)
return torch.cat([x, last], dim=0)
# If an IMAGE batch is provided, enforce that its frames match the validated length.
# This is the actual guarantee needed by LTX VAE encode.
if images is not None:
img_in = images
if not torch.is_floating_point(img_in):
img_in = img_in.float() / 255.0
# Expect ComfyUI IMAGE: [frames,H,W,C]
if img_in.ndim != 4:
raise ValueError("images must be a ComfyUI IMAGE tensor with shape [frames,H,W,C]")
frames_in = int(img_in.shape[0])
frames_out = int(length_fixed)
if not autofix:
frames_out = frames_in
mode = str(images_mode or "pad_repeat_last")
if mode not in ("pad_repeat_last", "crop_end"):
mode = "pad_repeat_last"
if frames_out == frames_in:
img = img_in
elif frames_out < frames_in:
# Crop end to match target length
img = img_in[:frames_out, ...]
else:
# Pad by repeating last frame
img = _pad_repeat_last_frames(img_in, frames_out - frames_in)
# Note: we do NOT resize spatially here; this node's width/height validation is
# meant for parameter sanity and EmptyImage-like generation. Spatial resizing remains
# the responsibility of the workflow.
else:
# EmptyImage-compatible IMAGE tensor: [B,H,W,3] float in [0,1]
fill = float(color_fixed) / 255.0
img = torch.full(
(batch_fixed, int(height_fixed), int(width_fixed), 3),
fill,
dtype=torch.float32,
device="cpu",
)
modified = (width_fixed != width_in) or (height_fixed != height_in) or (length_fixed != length_in) or (batch_fixed != batch_in) or (color_fixed != color_in)
if images is not None and isinstance(images, torch.Tensor) and images.ndim == 4:
modified = modified or (int(images.shape[0]) != int(img.shape[0]))
implied_fps_str = "n/a"
if seconds_in > 0:
implied_fps = (float(length_fixed) - 1.0) / seconds_in
implied_fps_str = f"{implied_fps:.3f}"
images_note = ""
if images is not None and isinstance(images, torch.Tensor) and images.ndim == 4:
images_note = f" | images_frames: in={int(images.shape[0])} -> out={int(img.shape[0])} (mode={images_mode}, autofix={autofix})"
report = (
f"input: {width_in}x{height_in}, batch={batch_in}, color={color_in} | "
f"seconds_input={seconds_in:.3f}, length_input={length_in} | "
f"rules: w%{spatial_multiple}={w_ok}, h%{spatial_multiple}={h_ok}, length=8n+1={l_ok} | "
f"output: {width_fixed}x{height_fixed}, length={length_fixed}, batch={batch_fixed}, color={color_fixed} | "
f"implied_fps={implied_fps_str} | "
f"autofix={autofix} (len_fix={length_fix}) | modified={modified} | "
f"delta: +{pad_w}w, +{pad_h}h, {pad_len:+d} length"
f"{images_note}"
)
if not (w_ok and h_ok and l_ok):
_log.warning("[IAMCCS_LTX2_Validator] %s", report)
return (img, int(length_fixed), report)
class IAMCCS_LTX2_TimeFrameCount:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
# Duration widgets are both visible for UX.
# The frontend JS keeps seconds <-> length in sync using IAMCCS_LTX2_FrameRateSync.
"seconds": ("FLOAT", {"default": 4.0, "min": 0.01, "max": 3600.0, "step": 0.01}),
"length": ("INT", {"default": 81, "min": 1, "max": 16385, "step": 1, "control_after_generate": True}),
}
}
RETURN_TYPES = ("INT", "FLOAT", "STRING")
RETURN_NAMES = ("length", "seconds", "report")
FUNCTION = "timeframe"
CATEGORY = "IAMCCS/LTX-2"
def timeframe(self, seconds: float, length: int):
seconds_in = float(seconds)
length_in = int(length)
length_fixed = max(1, length_in)
# LTX-2 encode constraint: frames must be 8n + 1.
# Round UP to avoid shortening the requested duration.
pad = 0
rem = (length_fixed - 1) % 8
if rem != 0:
pad = 8 - rem
length_fixed = length_fixed + pad
seconds_fixed = max(0.01, seconds_in)
implied_fps_str = "n/a"
if seconds_fixed > 0:
implied_fps = (float(length_fixed) - 1.0) / seconds_fixed
implied_fps_str = f"{implied_fps:.3f}"
snap = "ok" if pad == 0 else f"up(+{pad})"
report = (
f"seconds_input={seconds_in:.3f}, length_input={length_in} -> "
f"seconds={seconds_fixed:.3f}, length={length_fixed} | implied_fps={implied_fps_str} | ltx2_8n+1={snap}"
)
return (int(length_fixed), float(seconds_fixed), report)
class IAMCCS_LTX2_ImageBatchPadReflect:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE", {}),
"pad_x": ("INT", {"default": 16, "min": 0, "max": 512, "step": 1}),
"pad_y": ("INT", {"default": 16, "min": 0, "max": 512, "step": 1}),
"pad_mode": (["reflect", "replicate"], {"default": "reflect"}),
}
}
RETURN_TYPES = ("IMAGE", "INT", "INT", "STRING")
RETURN_NAMES = ("images", "pad_x", "pad_y", "report")
FUNCTION = "pad"
CATEGORY = "IAMCCS/LTX-2"
def pad(self, images: torch.Tensor, pad_x: int, pad_y: int, pad_mode: str):
# images: [B,H,W,C] float
if images is None:
raise ValueError("images is required")
b, h, w, c = images.shape
px = max(0, int(pad_x))
py = max(0, int(pad_y))
# reflect requires pad < dim; clamp to avoid runtime errors
px_eff = min(px, max(0, w - 1))
py_eff = min(py, max(0, h - 1))
if px_eff == 0 and py_eff == 0:
return (images, 0, 0, f"PadReflect: no-op (input {w}x{h})")
mode = str(pad_mode or "reflect")
if mode not in ("reflect", "replicate"):
mode = "reflect"
x = images.permute(0, 3, 1, 2) # [B,C,H,W]
# pad format: (left, right, top, bottom)
x = _F.pad(x, (px_eff, px_eff, py_eff, py_eff), mode=mode)
out = x.permute(0, 2, 3, 1).contiguous()
report = (
f"PadReflect: mode={mode}, requested=({px},{py}), used=({px_eff},{py_eff}) | "
f"{w}x{h} -> {int(out.shape[2])}x{int(out.shape[1])}"
)
return (out, int(px_eff), int(py_eff), report)
class IAMCCS_LTX2_ImageBatchCropByPad:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE", {}),
"pad_x": ("INT", {"default": 16, "min": 0, "max": 512, "step": 1}),
"pad_y": ("INT", {"default": 16, "min": 0, "max": 512, "step": 1}),
}
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("images", "report")
FUNCTION = "crop"
CATEGORY = "IAMCCS/LTX-2"
def crop(self, images: torch.Tensor, pad_x: int, pad_y: int):
if images is None:
raise ValueError("images is required")
b, h, w, c = images.shape
px = max(0, int(pad_x))
py = max(0, int(pad_y))
if px == 0 and py == 0:
return (images, f"CropByPad: no-op (input {w}x{h})")
# Clamp so we never invert the crop
px_eff = min(px, max(0, (w - 1) // 2))
py_eff = min(py, max(0, (h - 1) // 2))
out = images[:, py_eff : h - py_eff, px_eff : w - px_eff, :]
report = (
f"CropByPad: requested=({px},{py}), used=({px_eff},{py_eff}) | "
f"{w}x{h} -> {int(out.shape[2])}x{int(out.shape[1])}"
)
return (out, report)
class IAMCCS_LTX2_EnsureFrames8nPlus1:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE", {}),
# LTX-2 encode constraint: frames must be 1 + 8*k.
# "pad" is workflow-safe (never shortens), "crop" is deterministic.
"mode": (["pad_repeat_last", "crop_end"], {"default": "pad_repeat_last"}),
"fix": (["up", "down", "nearest"], {"default": "up"}),
}
}
RETURN_TYPES = ("IMAGE", "INT", "STRING")
RETURN_NAMES = ("images", "frames", "report")
FUNCTION = "ensure"
CATEGORY = "IAMCCS/LTX-2"
def _frames_rule_fix(self, frames: int, mode: str) -> tuple[int, int]:
frames = int(frames)
if frames < 1:
frames = 1
rem = (frames - 1) % 8
if rem == 0:
return frames, 0
down = frames - rem
up = frames + (8 - rem)
if mode == "down":
fixed = max(1, down)
elif mode == "nearest":
fixed = up if (up - frames) <= (frames - down) else max(1, down)
else:
fixed = up
return fixed, fixed - frames
def ensure(self, images: torch.Tensor, mode: str, fix: str):
if images is None:
raise ValueError("images is required")
if not isinstance(images, torch.Tensor) or images.ndim != 4:
raise ValueError("images must be a ComfyUI IMAGE tensor with shape [frames,H,W,C]")
frames_in = int(images.shape[0])
frames_fixed, delta = self._frames_rule_fix(frames_in, str(fix or "up"))
if frames_fixed == frames_in:
return (images, frames_in, f"EnsureFrames8n+1: ok ({frames_in})")
mode = str(mode or "pad_repeat_last")
if mode == "crop_end":
# For crop mode, prefer shortening, regardless of requested 'fix'.
rem = (frames_in - 1) % 8
frames_fixed = frames_in if rem == 0 else max(1, frames_in - rem)
out = images[:frames_fixed, ...]
return (out, frames_fixed, f"EnsureFrames8n+1: crop_end {frames_in} -> {frames_fixed}")
# pad_repeat_last (workflow-safe)
if frames_fixed < frames_in:
# If user selected fix=down/nearest and it resulted in fewer frames,
# still keep behavior consistent with padding node: do a crop.
out = images[:frames_fixed, ...]
return (out, frames_fixed, f"EnsureFrames8n+1: crop_end {frames_in} -> {frames_fixed} (fix={fix})")
pad = int(frames_fixed - frames_in)
last = images[-1:, ...].repeat(pad, 1, 1, 1)
out = torch.cat([images, last], dim=0)
return (out, frames_fixed, f"EnsureFrames8n+1: pad_repeat_last {frames_in} -> {frames_fixed} (pad={pad}, fix={fix})")
class IAMCCS_LTX2_EnsureMinFrames:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"images": ("IMAGE", {}),
"min_frames": ("INT", {"default": 25, "min": 1, "max": 1000000, "step": 1}),
"mode": (["repeat_last", "error"], {"default": "repeat_last"}),
"ltx_frames_fix": (["none", "up", "down", "nearest"], {"default": "up"}),
}
}
RETURN_TYPES = ("IMAGE", "INT", "STRING")
RETURN_NAMES = ("images", "frames", "report")
FUNCTION = "ensure"
CATEGORY = "IAMCCS/LTX-2"
def _fix_ltx_frames(self, frames: int, mode: str) -> int:
frames = max(1, int(frames))
rem = (frames - 1) % 8
if rem == 0:
return frames
down = max(1, frames - rem)
up = frames + (8 - rem)
if mode == "down":
return down
if mode == "nearest":
return up if (up - frames) <= (frames - down) else down
return up
def ensure(self, images: torch.Tensor, min_frames: int, mode: str, ltx_frames_fix: str):
if images is None:
raise ValueError("images is required")
if not isinstance(images, torch.Tensor) or images.ndim != 4:
raise ValueError("images must be a ComfyUI IMAGE tensor with shape [frames,H,W,C]")
frames_in = int(images.shape[0])
min_frames = max(1, int(min_frames))
target_frames = max(frames_in, min_frames)
ltx_mode = str(ltx_frames_fix or "none")
if ltx_mode != "none":
target_frames = self._fix_ltx_frames(target_frames, ltx_mode)
if frames_in >= target_frames:
return (images, frames_in, f"EnsureMinFrames: ok ({frames_in}) | min={min_frames} | ltx_fix={ltx_mode}")
mode = str(mode or "repeat_last")
if mode == "error":
raise ValueError(f"EnsureMinFrames: got {frames_in} frames, required at least {target_frames}")
pad = target_frames - frames_in
tail = images[-1:, ...].repeat((pad, 1, 1, 1))
out = torch.cat([images, tail], dim=0)
report = f"EnsureMinFrames: repeat_last {frames_in} -> {target_frames} | min={min_frames} | ltx_fix={ltx_mode}"
return (out, int(target_frames), report)
class IAMCCS_LTX2_ControlPreprocess:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"image": ("IMAGE",),
"mode": ([
"none",
"grayscale",
"invert",
"threshold",
"edges_sobel",
"edges_canny (opencv_if_available)",
], {"default": "edges_sobel"}),
"threshold": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01}),
"canny_low": ("INT", {"default": 100, "min": 0, "max": 1024, "step": 1}),
"canny_high": ("INT", {"default": 200, "min": 0, "max": 2048, "step": 1}),
}
}
RETURN_TYPES = ("IMAGE", "STRING")
RETURN_NAMES = ("image", "report")
FUNCTION = "process"
CATEGORY = "IAMCCS/LTX-2"
def process(self, image: torch.Tensor, mode: str, threshold: float, canny_low: int, canny_high: int):
if mode == "none":
return (image, "mode=none")
img = image
# ComfyUI IMAGE is typically float in [0,1], but in practice some custom nodes
# provide float in [0,255] or even [-1,1]. If we only clamp, we can destroy
# the signal (all-white / crushed blacks). Normalize defensively.
if not torch.is_floating_point(img):
img = img.float() / 255.0
else:
# Heuristic 1: float in [0,255]
# Using a threshold > 1 to avoid touching regular [0,1] images.
max_v = float(img.detach().amax().item()) if img.numel() else 0.0
if max_v > 1.5:
img = img / 255.0
# Heuristic 2: float in [-1,1]
min_v = float(img.detach().amin().item()) if img.numel() else 0.0
if min_v < -0.01 and max_v <= 1.01:
img = (img + 1.0) * 0.5
img = img.clamp(0.0, 1.0)
if mode == "invert":
out = 1.0 - img
return (out, "mode=invert")
gray = _to_grayscale(img)
if mode == "grayscale":
out = gray.repeat(1, 1, 1, 3)
return (out, "mode=grayscale")
if mode == "threshold":
t = float(threshold)
out1 = (gray > t).to(gray.dtype)
out = out1.repeat(1, 1, 1, 3)
return (out, f"mode=threshold t={t:.3f}")
if mode == "edges_sobel":
edges = _sobel_edges(gray)
out = edges.repeat(1, 1, 1, 3)
return (out, "mode=edges_sobel")
# canny via opencv if available; fall back to sobel
try:
import cv2 # type: ignore
import numpy as np # type: ignore
# Work on CPU per-frame. Convert to uint8 grayscale.
gray_cpu = (gray.detach().to("cpu") * 255.0).clamp(0, 255).to(torch.uint8)
b, h, w, _ = gray_cpu.shape
out_frames = []
for i in range(b):
g = gray_cpu[i, :, :, 0].numpy()
e = cv2.Canny(g, int(canny_low), int(canny_high))
out_frames.append(e)
edges_np = np.stack(out_frames, axis=0).astype(np.float32) / 255.0
edges = torch.from_numpy(edges_np).to(img.device, dtype=img.dtype).view(-1, h, w, 1)
out = edges.repeat(1, 1, 1, 3)
return (out, f"mode=edges_canny low={canny_low} high={canny_high}")
except Exception:
edges = _sobel_edges(gray)
out = edges.repeat(1, 1, 1, 3)
return (out, "mode=edges_canny fallback=edges_sobel")
class IAMCCS_SegmentPlanner:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"song_duration_s": ("FLOAT", {"default": 180.0, "min": 0.01, "max": 36000.0, "step": 0.01}),
"fps": ("FLOAT", {"default": 24.0, "min": 0.001, "max": 240.0, "step": 0.01}),
"segment_duration_s": ("FLOAT", {"default": 10.0, "min": 0.01, "max": 3600.0, "step": 0.01}),
"planning_mode": (["manual_segment_seconds", "explicit_preset_seconds"], {"default": "manual_segment_seconds"}),
"segment_preset": (["5sec", "10sec", "15sec", "20sec", "videoclip", "monologue"], {"default": "15sec"}),
"overlap_frames": ("INT", {"default": 9, "min": 0, "max": 4096, "step": 1}),
"ltx_round_mode": (["up", "nearest", "down"], {"default": "up"}),
"segment_index": ("INT", {"default": 0, "min": 0, "max": 100000, "step": 1}),
}
}
RETURN_TYPES = (
"INT", "INT", "INT", "INT", "INT", "INT", "INT", "STRING",
"INT", "INT", "INT", "INT", "INT", "INT", "FLOAT", "FLOAT", "STRING", "FLOAT",
"INT", "FLOAT", "STRING", "STRING", "STRING",
)
RETURN_NAMES = (
"total_frames",
"unique_segment_frames",
"first_segment_raw_frames",
"continuation_raw_frames",
"estimated_segments",
"continuation_loops",
"last_segment_unique_frames",
"report",
"segment_index_out",
"current_segment_raw_frames",
"current_segment_unique_frames",
"current_segment_start_frames",
"current_segment_end_frames",
"current_remaining_frames_after",
"current_segment_start_s",
"current_segment_end_s",
"current_segment_report",
"fps_out",
"recommended_overlap_frames",
"recommended_audio_left_context_s",
"recommended_extension_preset",
"effective_planning_mode",
"planning_profile_report",
)
FUNCTION = "plan"
CATEGORY = "IAMCCS/LTX-2"
def _fix_ltx_frames(self, frames: int, mode: str) -> int:
frames = max(1, int(frames))
rem = (frames - 1) % 8
if rem == 0:
return frames
down = max(1, frames - rem)
up = frames + (8 - rem)
if mode == "down":
return down
if mode == "nearest":
return up if (up - frames) <= (frames - down) else down
return up
def _normalize_segment_preset(self, segment_preset: str) -> str:
value = str(segment_preset or "15sec")
if value == "videoclip":
return "10sec"
if value == "monologue":
return "15sec"
if value in {"5sec", "10sec", "15sec", "20sec"}:
return value
return "15sec"
def _normalize_planning_mode(self, planning_mode: str) -> str:
value = str(planning_mode or "manual_segment_seconds")
if value == "auto_profile":
return "explicit_preset_seconds"
if value in {"manual_segment_seconds", "explicit_preset_seconds"}:
return value
return "manual_segment_seconds"
def _recommend_profile(self, segment_preset: str):
preset = self._normalize_segment_preset(segment_preset)
if preset == "5sec":
return {
"segment_seconds": 5.0,
"overlap_frames": 9,
"audio_left_context_s": 0.25,
"extension_preset": "videoclip_audio_24fps",
}
if preset == "20sec":
return {
"segment_seconds": 20.0,
"overlap_frames": 9,
"audio_left_context_s": 1.0,
"extension_preset": "monologue_audio_24fps",
}
if preset == "15sec":
return {
"segment_seconds": 15.0,
"overlap_frames": 9,
"audio_left_context_s": 0.75,
"extension_preset": "monologue_audio_24fps",
}
return {
"segment_seconds": 10.0,
"overlap_frames": 9,
"audio_left_context_s": 0.5,
"extension_preset": "videoclip_audio_24fps",
}
def _recommend_profile_for_segment_seconds(self, segment_duration_s: float):
duration = float(segment_duration_s)
for preset, target_seconds in (("5sec", 5.0), ("10sec", 10.0), ("15sec", 15.0), ("20sec", 20.0)):
if abs(duration - target_seconds) <= 0.1:
return preset, self._recommend_profile(preset)
return None, None
def plan(self, song_duration_s: float, fps: float, segment_duration_s: float, planning_mode: str, segment_preset: str, overlap_frames: int, ltx_round_mode: str, segment_index: int):
song_duration_s = max(0.01, float(song_duration_s))
fps = max(0.001, float(fps))
segment_duration_s = max(0.01, float(segment_duration_s))
overlap_frames = max(0, int(overlap_frames))
segment_index = max(0, int(segment_index))
planning_mode = self._normalize_planning_mode(planning_mode)
segment_preset = self._normalize_segment_preset(segment_preset)
rec = self._recommend_profile(segment_preset)
auto_duration_profile = None
effective_planning_mode = planning_mode
if planning_mode == "explicit_preset_seconds":
segment_duration_s = max(0.01, float(rec["segment_seconds"]))
else:
auto_duration_profile, auto_rec = self._recommend_profile_for_segment_seconds(segment_duration_s)
if auto_rec is not None:
rec = auto_rec
total_frames = max(1, int(round(song_duration_s * fps)))
unique_segment_frames = max(1, int(round(segment_duration_s * fps)))
first_segment_raw_frames = self._fix_ltx_frames(unique_segment_frames, str(ltx_round_mode))
continuation_raw_frames = self._fix_ltx_frames(unique_segment_frames + overlap_frames, str(ltx_round_mode))
estimated_segments = max(1, int(math.ceil(float(total_frames) / float(unique_segment_frames))))
continuation_loops = max(0, estimated_segments - 1)
remainder = total_frames - unique_segment_frames * max(0, estimated_segments - 1)
last_segment_unique_frames = unique_segment_frames if remainder <= 0 else int(remainder)
last_segment_raw_frames = first_segment_raw_frames if estimated_segments <= 1 else self._fix_ltx_frames(last_segment_unique_frames + overlap_frames, str(ltx_round_mode))
clamped_segment_index = min(segment_index, max(0, estimated_segments - 1))
current_segment_start_frames = unique_segment_frames * clamped_segment_index
current_segment_unique_frames = max(0, min(unique_segment_frames, total_frames - current_segment_start_frames))
if current_segment_unique_frames <= 0:
current_segment_unique_frames = last_segment_unique_frames if clamped_segment_index == max(0, estimated_segments - 1) else unique_segment_frames
current_segment_start_frames = min(current_segment_start_frames, max(0, total_frames - current_segment_unique_frames))
current_segment_end_frames = min(total_frames, current_segment_start_frames + current_segment_unique_frames)
current_remaining_frames_after = max(0, total_frames - current_segment_end_frames)
if clamped_segment_index == 0:
current_segment_raw_frames = first_segment_raw_frames
elif clamped_segment_index >= estimated_segments - 1:
current_segment_raw_frames = last_segment_raw_frames
else:
current_segment_raw_frames = continuation_raw_frames
current_segment_start_s = float(current_segment_start_frames) / float(fps)
current_segment_end_s = float(current_segment_end_frames) / float(fps)
recommended_overlap_frames = int(rec["overlap_frames"])
recommended_audio_left_context_s = float(rec["audio_left_context_s"])
recommended_extension_preset = str(rec["extension_preset"])
report = (
f"song={song_duration_s:.3f}s @ {fps:.3f}fps -> total={total_frames}f | "
f"segment={segment_duration_s:.3f}s -> unique={unique_segment_frames}f | "
f"overlap={overlap_frames}f | first_raw={first_segment_raw_frames}f | "
f"continuation_raw={continuation_raw_frames}f | segments={estimated_segments} | "
f"loops={continuation_loops} | last_unique={last_segment_unique_frames}f | ltx_round={ltx_round_mode} | "
f"planning={effective_planning_mode} | segment_preset={segment_preset}"
)
current_segment_report = (
f"segment_index={clamped_segment_index} | raw={current_segment_raw_frames}f | unique={current_segment_unique_frames}f | "
f"range=[{current_segment_start_frames}..{current_segment_end_frames}) | remaining_after={current_remaining_frames_after}f | "
f"time=[{current_segment_start_s:.3f}s..{current_segment_end_s:.3f}s]"
)
planning_profile_report = (
f"segment_preset={segment_preset} | planning_mode={effective_planning_mode} | segment_duration_s={segment_duration_s:.3f} | "
f"recommended_overlap={recommended_overlap_frames}f | recommended_left_context={recommended_audio_left_context_s:.2f}s | "
f"recommended_extension_preset={recommended_extension_preset} | auto_duration_profile={auto_duration_profile or 'none'}"
)
return (
int(total_frames),
int(unique_segment_frames),
int(first_segment_raw_frames),
int(continuation_raw_frames),
int(estimated_segments),
int(continuation_loops),
int(last_segment_unique_frames),
report,
int(clamped_segment_index),
int(current_segment_raw_frames),
int(current_segment_unique_frames),
int(current_segment_start_frames),
int(current_segment_end_frames),
int(current_remaining_frames_after),
float(current_segment_start_s),
float(current_segment_end_s),
current_segment_report,
float(fps),
int(recommended_overlap_frames),
float(recommended_audio_left_context_s),
recommended_extension_preset,
effective_planning_mode,
planning_profile_report,
)
class IAMCCS_SegmentPlannerSettings(IAMCCS_SegmentPlanner):
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"segment_duration_s": ("FLOAT", {"default": 10.0, "min": 0.01, "max": 3600.0, "step": 0.01}),
"planning_mode": (["manual_segment_seconds", "explicit_preset_seconds"], {"default": "manual_segment_seconds"}),
"segment_preset": (["5sec", "10sec", "15sec", "20sec", "videoclip", "monologue"], {"default": "10sec"}),
"overlap_frames": ("INT", {"default": 9, "min": 0, "max": 4096, "step": 1}),
"auto_sync_overlap": ("BOOLEAN", {"default": False}),
},
"optional": {
"segment_duration_s_in": ("FLOAT", {"forceInput": True, "default": 0.0, "min": 0.0, "max": 3600.0, "step": 0.01}),
"planning_mode_in": ("STRING", {"forceInput": True, "default": ""}),
"segment_preset_in": ("STRING", {"forceInput": True, "default": ""}),
"overlap_frames_in": ("INT", {"forceInput": True, "default": -1, "min": -1, "max": 4096, "step": 1}),
}
}
RETURN_TYPES = ("FLOAT", "STRING", "STRING", "INT", "STRING")
RETURN_NAMES = ("segment_duration_s", "planning_mode", "segment_preset", "overlap_frames", "report")
FUNCTION = "build"
CATEGORY = "IAMCCS/LTX-2"
def build(
self,
segment_duration_s: float,
planning_mode: str,
segment_preset: str,
overlap_frames: int,
auto_sync_overlap: bool,
segment_duration_s_in: float = 0.0,
planning_mode_in: str = "",
segment_preset_in: str = "",
overlap_frames_in: int = -1,
):
if segment_duration_s_in is not None and float(segment_duration_s_in) > 0.0:
segment_duration_s = float(segment_duration_s_in)
if planning_mode_in:
planning_mode = str(planning_mode_in)
if segment_preset_in:
segment_preset = str(segment_preset_in)
if overlap_frames_in is not None and int(overlap_frames_in) >= 0:
overlap_frames = int(overlap_frames_in)
segment_duration_s = max(0.01, float(segment_duration_s))
overlap_frames = max(0, int(overlap_frames))
planning_mode = self._normalize_planning_mode(planning_mode)
segment_preset = self._normalize_segment_preset(segment_preset)
rec = self._recommend_profile(segment_preset)
auto_duration_profile = None
if planning_mode == "explicit_preset_seconds":
segment_duration_s = max(0.01, float(rec["segment_seconds"]))
elif auto_sync_overlap:
auto_duration_profile, auto_rec = self._recommend_profile_for_segment_seconds(segment_duration_s)
if auto_rec is not None:
rec = auto_rec
report = (
f"segment_duration_s={segment_duration_s:.3f} | planning_mode={planning_mode} | "
f"segment_preset={segment_preset} | overlap_frames={overlap_frames} | "
f"auto_sync_overlap={bool(auto_sync_overlap)} | auto_duration_profile={auto_duration_profile or 'none'} | "
"overlap_policy=manual/default_9"
)
return (
float(segment_duration_s),
str(planning_mode),
str(segment_preset),
int(overlap_frames),
report,
)
class IAMCCS_AudioSegmentAutoPlanner(IAMCCS_SegmentPlanner):
SPLIT_BY_SEGMENTS = "choose segment count (audio / segments)"
SPLIT_BY_SECONDS = "choose seconds per segment (auto count)"
SEGMENT_COUNT_CHOICES = (1, 2, 3, 4, 5, 6, 8, 10, 12)
SEGMENT_SECONDS_CHOICES = (5.0, 8.0, 10.0, 12.0, 15.0, 20.0, 30.0)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio": ("AUDIO",),
"fps": ("FLOAT", {"default": 25.0, "min": 0.001, "max": 240.0, "step": 0.01}),
"split_mode": ([cls.SPLIT_BY_SEGMENTS, cls.SPLIT_BY_SECONDS], {"default": cls.SPLIT_BY_SEGMENTS}),
"segment_count": (list(cls.SEGMENT_COUNT_CHOICES), {"default": 2}),
"target_segment_seconds": (list(cls.SEGMENT_SECONDS_CHOICES), {"default": 10.0}),
"overlap_frames": ("INT", {"default": 9, "min": 0, "max": 4096, "step": 1}),
"ltx_round_mode": (["up", "nearest", "down"], {"default": "up"}),
}
}
RETURN_TYPES = (
"FLOAT", "FLOAT", "STRING", "STRING", "INT",
"INT", "INT", "INT", "INT", "INT", "INT", "FLOAT", "STRING",
)
RETURN_NAMES = (
"audio_duration_s",
"segment_duration_s",
"planning_mode",
"segment_preset",
"overlap_frames",
"segment_count",
"total_frames",
"unique_segment_frames",
"first_segment_raw_frames",
"continuation_raw_frames",
"last_segment_unique_frames",
"fps_out",
"report",
)
FUNCTION = "plan_audio_segments"
CATEGORY = "IAMCCS/LTX-2"
def _preset_from_seconds(self, seconds: float) -> str:
preset, _rec = self._recommend_profile_for_segment_seconds(float(seconds))
if preset is not None:
return str(preset)
choices = ((5.0, "5sec"), (10.0, "10sec"), (15.0, "15sec"), (20.0, "20sec"))
return min(choices, key=lambda item: abs(float(seconds) - item[0]))[1]
@staticmethod
def _first_number(value: Any, default: float) -> float:
if isinstance(value, (int, float)):
return float(value)
text = str(value or "").strip()
token = ""
seen_decimal = False
for ch in text:
if ch.isdigit():
token += ch
elif ch == "." and token and not seen_decimal:
token += ch
seen_decimal = True
elif token:
break
if not token or token == ".":
return float(default)
try:
return float(token)
except Exception:
return float(default)
def _normalize_split_mode(self, split_mode: Any) -> tuple[str, str]:
text = str(split_mode or "").strip().lower()
if text in ("fixed_segment_seconds", "choose_seconds_per_segment_auto_count"):
return "fixed_segment_seconds", self.SPLIT_BY_SECONDS
if "seconds per segment" in text or "auto count" in text:
return "fixed_segment_seconds", self.SPLIT_BY_SECONDS
return "fixed_segment_count", self.SPLIT_BY_SEGMENTS
def plan_audio_segments(
self,
audio,
fps: float,
split_mode: str,
segment_count: int,
target_segment_seconds: float,
overlap_frames: int,
ltx_round_mode: str,
):
audio_duration_s = max(0.01, _audio_duration_seconds(audio))
fps = max(0.001, float(fps))
split_mode_key, split_mode_label = self._normalize_split_mode(split_mode)
requested_segment_count = max(1, int(round(self._first_number(segment_count, 2))))
target_segment_seconds = max(0.01, float(self._first_number(target_segment_seconds, 10.0)))
if split_mode_key == "fixed_segment_seconds":
segment_duration_s = target_segment_seconds
else:
segment_duration_s = max(0.01, audio_duration_s / float(requested_segment_count))
segment_preset = self._preset_from_seconds(segment_duration_s)
planning_mode = "manual_segment_seconds"
planner_result = self.plan(
audio_duration_s,
fps,
segment_duration_s,
planning_mode,
segment_preset,
int(overlap_frames),
str(ltx_round_mode),
0,
)
computed_segment_count = int(planner_result[4])
report = (
f"audio_duration={audio_duration_s:.3f}s @ {fps:.3f}fps | split_mode={split_mode_key} ({split_mode_label}) | "
f"requested_segments={requested_segment_count} | computed_segments={computed_segment_count} | "
f"segment_duration={segment_duration_s:.3f}s | total={int(planner_result[0])}f | "
f"unique={int(planner_result[1])}f | first_raw={int(planner_result[2])}f | "
f"continuation_raw={int(planner_result[3])}f | last_unique={int(planner_result[6])}f | "
f"preset={segment_preset} | overlap={int(overlap_frames)}f | ltx_round={ltx_round_mode}"
)
_log.info("[AudioSegmentAutoPlanner] %s", report)
return (
float(audio_duration_s),
float(segment_duration_s),
planning_mode,
segment_preset,
int(overlap_frames),
computed_segment_count,
int(planner_result[0]),
int(planner_result[1]),
int(planner_result[2]),
int(planner_result[3]),
int(planner_result[6]),
float(planner_result[17]),
report,
)
class IAMCCS_SegmentPlannerLinked(IAMCCS_SegmentPlanner):
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"song_duration_s": ("FLOAT", {"default": 180.0, "min": 0.01, "max": 36000.0, "step": 0.01}),
"fps": ("FLOAT", {"default": 24.0, "min": 0.001, "max": 240.0, "step": 0.01}),
"segment_duration_s": ("FLOAT", {"default": 10.0, "min": 0.01, "max": 3600.0, "step": 0.01, "forceInput": True}),
"planning_mode_in": ("STRING", {"default": "manual_segment_seconds", "forceInput": True}),
"segment_preset_in": ("STRING", {"default": "10sec", "forceInput": True}),
"overlap_frames_in": ("INT", {"default": 9, "min": 0, "max": 4096, "step": 1, "forceInput": True}),
"ltx_round_mode": (["up", "nearest", "down"], {"default": "up"}),
"segment_index": ("INT", {"default": 0, "min": 0, "max": 100000, "step": 1}),
},
"optional": {
"ltx_round_mode_in": ("STRING", {"default": "", "forceInput": True}),
}
}
RETURN_TYPES = IAMCCS_SegmentPlanner.RETURN_TYPES
RETURN_NAMES = IAMCCS_SegmentPlanner.RETURN_NAMES
FUNCTION = "plan_linked"
CATEGORY = "IAMCCS/LTX-2"
def plan_linked(
self,
song_duration_s: float,
fps: float,
segment_duration_s: float,
planning_mode_in: str,
segment_preset_in: str,
overlap_frames_in: int,
ltx_round_mode: str,
segment_index: int,
ltx_round_mode_in: str = "",
):
effective_round_mode = str(ltx_round_mode_in or ltx_round_mode or "up")
return self.plan(
song_duration_s,
fps,
segment_duration_s,
str(planning_mode_in or "manual_segment_seconds"),
str(segment_preset_in or "10sec"),
int(overlap_frames_in),
effective_round_mode,
segment_index,
)
class IAMCCS_SegmentPlannerFramesLinked(IAMCCS_SegmentPlanner):
"""Segment planner whose authoritative total length is expressed in frames."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"total_generation_frames": ("INT", {"default": 720, "min": 1, "max": 10000000, "step": 1, "forceInput": True}),
"fps": ("FLOAT", {"default": 24.0, "min": 0.001, "max": 240.0, "step": 0.01}),
"segment_duration_s": ("FLOAT", {"default": 10.0, "min": 0.01, "max": 3600.0, "step": 0.01, "forceInput": True}),
"planning_mode_in": ("STRING", {"default": "manual_segment_seconds", "forceInput": True}),
"segment_preset_in": ("STRING", {"default": "10sec", "forceInput": True}),
"overlap_frames_in": ("INT", {"default": 9, "min": 0, "max": 4096, "step": 1, "forceInput": True}),
"ltx_round_mode": (["up", "nearest", "down"], {"default": "up"}),
"segment_index": ("INT", {"default": 0, "min": 0, "max": 100000, "step": 1}),
},
"optional": {
"ltx_round_mode_in": ("STRING", {"default": "", "forceInput": True}),
},
}
RETURN_TYPES = IAMCCS_SegmentPlanner.RETURN_TYPES
RETURN_NAMES = IAMCCS_SegmentPlanner.RETURN_NAMES
FUNCTION = "plan_frames_linked"
CATEGORY = "IAMCCS/LTX-2"
def plan_frames_linked(
self,
total_generation_frames: int,
fps: float,
segment_duration_s: float,
planning_mode_in: str,
segment_preset_in: str,
overlap_frames_in: int,
ltx_round_mode: str,
segment_index: int,
ltx_round_mode_in: str = "",
):
fps_value = max(0.001, float(fps))
total_frames = max(1, int(total_generation_frames))
total_duration_s = float(total_frames) / fps_value
effective_round_mode = str(ltx_round_mode_in or ltx_round_mode or "up")
return self.plan(
total_duration_s,
fps_value,
segment_duration_s,
str(planning_mode_in or "manual_segment_seconds"),
str(segment_preset_in or "10sec"),
int(overlap_frames_in),
effective_round_mode,
segment_index,
)
class IAMCCS_SegmentPlanFromPlanner:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"fps": ("FLOAT", {"default": 24.0, "min": 0.001, "max": 240.0, "step": 0.01}),
"total_frames": ("INT", {"default": 4320, "min": 1, "max": 10000000, "step": 1}),
"unique_segment_frames": ("INT", {"default": 240, "min": 1, "max": 1000000, "step": 1}),
"first_segment_raw_frames": ("INT", {"default": 241, "min": 1, "max": 1000000, "step": 1}),
"continuation_raw_frames": ("INT", {"default": 249, "min": 1, "max": 1000000, "step": 1}),
"estimated_segments": ("INT", {"default": 18, "min": 1, "max": 100000, "step": 1}),
"last_segment_unique_frames": ("INT", {"default": 240, "min": 1, "max": 1000000, "step": 1}),
"segment_index": ("INT", {"default": 0, "min": 0, "max": 100000, "step": 1}),
}
}
RETURN_TYPES = ("INT", "INT", "INT", "INT", "INT", "INT", "FLOAT", "FLOAT", "STRING")
RETURN_NAMES = (
"segment_index_out",
"current_segment_raw_frames",
"current_segment_unique_frames",
"current_segment_start_frames",
"current_segment_end_frames",
"current_remaining_frames_after",
"current_segment_start_s",
"current_segment_end_s",
"current_segment_report",
)
FUNCTION = "derive"
CATEGORY = "IAMCCS/LTX-2"
def derive(
self,
fps: float,
total_frames: int,
unique_segment_frames: int,
first_segment_raw_frames: int,
continuation_raw_frames: int,
estimated_segments: int,
last_segment_unique_frames: int,
segment_index: int,
):
fps = max(0.001, float(fps))
total_frames = max(1, int(total_frames))
unique_segment_frames = max(1, int(unique_segment_frames))
first_segment_raw_frames = max(1, int(first_segment_raw_frames))
continuation_raw_frames = max(1, int(continuation_raw_frames))
estimated_segments = max(1, int(estimated_segments))
last_segment_unique_frames = max(1, int(last_segment_unique_frames))
segment_index = max(0, min(int(segment_index), estimated_segments - 1))
overlap_hint_frames = max(0, continuation_raw_frames - unique_segment_frames)
current_segment_start_frames = unique_segment_frames * segment_index
if segment_index >= estimated_segments - 1:
current_segment_unique_frames = last_segment_unique_frames
else:
current_segment_unique_frames = unique_segment_frames
current_segment_end_frames = min(total_frames, current_segment_start_frames + current_segment_unique_frames)
current_remaining_frames_after = max(0, total_frames - current_segment_end_frames)
if segment_index == 0:
current_segment_raw_frames = first_segment_raw_frames
elif segment_index >= estimated_segments - 1:
current_segment_raw_frames = max(1, current_segment_unique_frames + overlap_hint_frames)
current_segment_raw_frames = 1 + 8 * max(0, int(math.ceil(float(current_segment_raw_frames - 1) / 8.0)))
else:
current_segment_raw_frames = continuation_raw_frames
current_segment_start_s = float(current_segment_start_frames) / float(fps)
current_segment_end_s = float(current_segment_end_frames) / float(fps)
current_segment_report = (
f"segment_index={segment_index} | raw={current_segment_raw_frames}f | unique={current_segment_unique_frames}f | "
f"range=[{current_segment_start_frames}..{current_segment_end_frames}) | remaining_after={current_remaining_frames_after}f | "
f"time=[{current_segment_start_s:.3f}s..{current_segment_end_s:.3f}s]"
)
return (
int(segment_index),
int(current_segment_raw_frames),
int(current_segment_unique_frames),
int(current_segment_start_frames),
int(current_segment_end_frames),
int(current_remaining_frames_after),
float(current_segment_start_s),
float(current_segment_end_s),
current_segment_report,
)
class IAMCCS_SourceRangeFromSegmentPlan:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"segment_index": ("INT", {"default": 0, "min": 0, "max": 100000, "step": 1}),
"current_segment_raw_frames": ("INT", {"default": 241, "min": 1, "max": 1000000, "step": 1}),
"current_segment_unique_frames": ("INT", {"default": 240, "min": 1, "max": 1000000, "step": 1}),
"current_segment_start_frames": ("INT", {"default": 0, "min": 0, "max": 10000000, "step": 1}),
}
}
RETURN_TYPES = ("INT", "INT", "INT", "STRING")
RETURN_NAMES = ("range_start_index", "range_end_index", "range_count", "report")
FUNCTION = "derive"
CATEGORY = "IAMCCS/LTX-2"
def derive(
self,
segment_index: int,
current_segment_raw_frames: int,
current_segment_unique_frames: int,
current_segment_start_frames: int,
):
segment_index = max(0, int(segment_index))
current_segment_raw_frames = max(1, int(current_segment_raw_frames))
current_segment_unique_frames = max(1, int(current_segment_unique_frames))
current_segment_start_frames = max(0, int(current_segment_start_frames))
overlap_delta = max(0, current_segment_raw_frames - current_segment_unique_frames)
overlap_backtrack = max(0, overlap_delta - 1)
if segment_index == 0:
overlap_backtrack = 0
range_start_index = max(0, current_segment_start_frames - overlap_backtrack)
range_end_index = range_start_index + current_segment_raw_frames
range_count = max(1, range_end_index - range_start_index)
report = (
f"segment_index={segment_index} | raw={current_segment_raw_frames}f | unique={current_segment_unique_frames}f | "
f"start_unique={current_segment_start_frames}f | backtrack={overlap_backtrack}f | "
f"range=[{range_start_index}..{range_end_index}) count={range_count}"
)
return (int(range_start_index), int(range_end_index), int(range_count), report)
class IAMCCS_SegmentSwitch:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"default_value": (any_type,),
"mode": (["single_prompt", "by_segment"], {"default": "single_prompt"}),
"segment_index": ("INT", {"default": 0, "min": 0, "max": 100000, "step": 1}),
},
"optional": FlexibleOptionalInputType(
any_type,
data={
"segment_01": (any_type,),
"segment_02": (any_type,),
"segment_03": (any_type,),
"segment_04": (any_type,),
"segment_05": (any_type,),
"segment_06": (any_type,),
},
),
}
RETURN_TYPES = (any_type, "INT", "STRING")
RETURN_NAMES = ("value", "active_slot", "report")
FUNCTION = "select"
CATEGORY = "IAMCCS/LTX-2"
def select(self, default_value, mode: str, segment_index: int, **kwargs):
if str(mode) == "single_prompt":
return (default_value, 0, "mode=single_prompt -> default_value")
slot = max(1, int(segment_index) + 1)
key = f"segment_{slot:02d}"
value = kwargs.get(key, None)
if value is None:
return (default_value, 0, f"mode=by_segment | segment_index={int(segment_index)} -> fallback=default_value")
return (value, slot, f"mode=by_segment | segment_index={int(segment_index)} -> {key}")
class IAMCCS_TwoSegmentPlanner:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"total_frames": ("INT", {"default": 481, "min": 1, "max": 10000000, "step": 1}),
"fps": ("FLOAT", {"default": 24.0, "min": 0.001, "max": 240.0, "step": 0.01}),
"first_segment_raw_frames": ("INT", {"default": 241, "min": 1, "max": 1000000, "step": 1}),
"second_segment_raw_frames": ("INT", {"default": 249, "min": 1, "max": 1000000, "step": 1}),
}
}
RETURN_TYPES = (
"INT", "INT", "INT",
"INT", "INT", "INT",
"INT", "FLOAT", "STRING",
)
RETURN_NAMES = (
"seg0_start_frames",
"seg0_end_frames",
"seg0_count_frames",
"seg1_start_frames",
"seg1_end_frames",
"seg1_count_frames",
"overlap_frames",
"total_duration_s",
"report",
)
FUNCTION = "plan"
CATEGORY = "IAMCCS/LTX-2"
def plan(self, total_frames: int, fps: float, first_segment_raw_frames: int, second_segment_raw_frames: int):
total_frames = max(1, int(total_frames))
fps = max(0.001, float(fps))
first_segment_raw_frames = max(1, int(first_segment_raw_frames))
second_segment_raw_frames = max(1, int(second_segment_raw_frames))
seg0_start_frames = 0
seg0_end_frames = min(total_frames, first_segment_raw_frames)
seg0_count_frames = max(1, seg0_end_frames - seg0_start_frames)
seg1_count_frames = min(total_frames, second_segment_raw_frames)
seg1_end_frames = total_frames
seg1_start_frames = max(0, seg1_end_frames - seg1_count_frames)
overlap_frames = max(0, seg0_count_frames + seg1_count_frames - total_frames)
total_duration_s = float(total_frames) / float(fps)
report = (
f"total={total_frames}f @ {fps:.3f}fps ({total_duration_s:.3f}s) | "
f"seg0=[{seg0_start_frames}..{seg0_end_frames}) count={seg0_count_frames} | "
f"seg1=[{seg1_start_frames}..{seg1_end_frames}) count={seg1_count_frames} | "
f"overlap={overlap_frames}f"
)
return (
int(seg0_start_frames),
int(seg0_end_frames),
int(seg0_count_frames),
int(seg1_start_frames),
int(seg1_end_frames),
int(seg1_count_frames),
int(overlap_frames),
float(total_duration_s),
report,
)
class IAMCCS_ThreeSegmentPlanner:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"total_frames": ("INT", {"default": 721, "min": 1, "max": 10000000, "step": 1}),
"fps": ("FLOAT", {"default": 24.0, "min": 0.001, "max": 240.0, "step": 0.01}),
"first_segment_raw_frames": ("INT", {"default": 241, "min": 1, "max": 1000000, "step": 1}),
"continuation_raw_frames": ("INT", {"default": 249, "min": 1, "max": 1000000, "step": 1}),
}
}
RETURN_TYPES = (
"INT", "INT", "INT",
"INT", "INT", "INT",
"INT", "INT", "INT",
"INT", "INT", "FLOAT", "STRING",
)
RETURN_NAMES = (
"seg0_start_frames",
"seg0_end_frames",
"seg0_count_frames",
"seg1_start_frames",
"seg1_end_frames",
"seg1_count_frames",
"seg2_start_frames",
"seg2_end_frames",
"seg2_count_frames",
"overlap_frames",
"unique_segment_frames",
"total_duration_s",
"report",
)
FUNCTION = "plan"
CATEGORY = "IAMCCS/LTX-2"
def plan(self, total_frames: int, fps: float, first_segment_raw_frames: int, continuation_raw_frames: int):
total_frames = max(1, int(total_frames))
fps = max(0.001, float(fps))
first_segment_raw_frames = max(1, int(first_segment_raw_frames))
continuation_raw_frames = max(1, int(continuation_raw_frames))
unique_segment_frames = max(1, first_segment_raw_frames - 1)
overlap_frames = max(0, continuation_raw_frames - unique_segment_frames)
overlap_backtrack = max(0, overlap_frames - 1)
def segment_range(index: int):
if index == 0:
start_frames = 0
count_frames = min(total_frames, first_segment_raw_frames)
else:
unique_start_frames = unique_segment_frames * index
start_frames = max(0, unique_start_frames - overlap_backtrack)
count_frames = max(1, min(continuation_raw_frames, total_frames - start_frames))
end_frames = min(total_frames, start_frames + count_frames)
count_frames = max(1, end_frames - start_frames)
return int(start_frames), int(end_frames), int(count_frames)
seg0_start_frames, seg0_end_frames, seg0_count_frames = segment_range(0)
seg1_start_frames, seg1_end_frames, seg1_count_frames = segment_range(1)
seg2_start_frames, seg2_end_frames, seg2_count_frames = segment_range(2)
total_duration_s = float(total_frames) / float(fps)
report = (
f"total={total_frames}f @ {fps:.3f}fps ({total_duration_s:.3f}s) | "
f"unique={unique_segment_frames}f | overlap={overlap_frames}f | "
f"seg0=[{seg0_start_frames}..{seg0_end_frames}) count={seg0_count_frames} | "
f"seg1=[{seg1_start_frames}..{seg1_end_frames}) count={seg1_count_frames} | "
f"seg2=[{seg2_start_frames}..{seg2_end_frames}) count={seg2_count_frames}"
)
return (
int(seg0_start_frames),
int(seg0_end_frames),
int(seg0_count_frames),
int(seg1_start_frames),
int(seg1_end_frames),
int(seg1_count_frames),
int(seg2_start_frames),
int(seg2_end_frames),
int(seg2_count_frames),
int(overlap_frames),
int(unique_segment_frames),
float(total_duration_s),
report,
)
NODE_CLASS_MAPPINGS = {
"IAMCCS_LTX2_FrameRateSync": IAMCCS_LTX2_FrameRateSync,
"IAMCCS_LTX2_Validator": IAMCCS_LTX2_Validator,
"IAMCCS_LTX2_TimeFrameCount": IAMCCS_LTX2_TimeFrameCount,
"IAMCCS_LTX2_EnsureFrames8nPlus1": IAMCCS_LTX2_EnsureFrames8nPlus1,
"IAMCCS_LTX2_ControlPreprocess": IAMCCS_LTX2_ControlPreprocess,
"IAMCCS_SegmentPlanner": IAMCCS_SegmentPlanner,
"IAMCCS_SegmentPlanFromPlanner": IAMCCS_SegmentPlanFromPlanner,
"IAMCCS_SourceRangeFromSegmentPlan": IAMCCS_SourceRangeFromSegmentPlan,
"IAMCCS_TwoSegmentPlanner": IAMCCS_TwoSegmentPlanner,
"IAMCCS_ThreeSegmentPlanner": IAMCCS_ThreeSegmentPlanner,
"IAMCCS_SegmentSwitch": IAMCCS_SegmentSwitch,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"IAMCCS_LTX2_FrameRateSync": "LTX-2 FrameRate Sync (int+float)",
"IAMCCS_LTX2_Validator": "LTX-2 Validator (32px, 8n +1)",
"IAMCCS_LTX2_TimeFrameCount": "LTX-2 TimeFrameCount",
"IAMCCS_LTX2_EnsureFrames8nPlus1": "LTX-2 Ensure Frames (8n + 1)",
"IAMCCS_LTX2_ControlPreprocess": "LTX-2 Control Preprocess (aux)",
"IAMCCS_SegmentPlanner": "Segment Planner (song -> LTX frames)",
"IAMCCS_SegmentPlanFromPlanner": "Segment Plan From Planner (per index)",
"IAMCCS_SourceRangeFromSegmentPlan": "Source Range From Segment Plan",
"IAMCCS_TwoSegmentPlanner": "Two Segment Planner (stable 2SEG)",
"IAMCCS_ThreeSegmentPlanner": "Three Segment Planner (stable 3SEG)",
"IAMCCS_SegmentSwitch": "Segment Switch (by segment_index)",
}