1065 lines
44 KiB
Python
1065 lines
44 KiB
Python
import logging
|
|
import math
|
|
import os
|
|
import av
|
|
import numpy as np
|
|
import torch
|
|
import comfy
|
|
import comfy.sd
|
|
import comfy.utils
|
|
import folder_paths
|
|
import node_helpers
|
|
from comfy_extras import nodes_lt
|
|
from .ltx_director import _resize_image, _execute_comfy_node, _unpack
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# --- Helper Functions from Nghtdrp ---
|
|
|
|
def _get_guide_attention_entries(conditioning):
|
|
for item in conditioning:
|
|
entries = item[1].get("guide_attention_entries", None)
|
|
if entries is not None:
|
|
return entries
|
|
return []
|
|
|
|
def _set_guide_attention_entries(conditioning, entries):
|
|
return node_helpers.conditioning_set_values(
|
|
conditioning, {"guide_attention_entries": entries}
|
|
)
|
|
|
|
def _append_guide_attention_entry(
|
|
conditioning,
|
|
pre_filter_count,
|
|
latent_shape,
|
|
attention_strength=1.0,
|
|
attention_mask=None,
|
|
):
|
|
entries = [*_get_guide_attention_entries(conditioning)]
|
|
entries.append(
|
|
{
|
|
"pre_filter_count": int(pre_filter_count),
|
|
"strength": float(attention_strength),
|
|
"pixel_mask": attention_mask,
|
|
"latent_shape": list(latent_shape),
|
|
}
|
|
)
|
|
return _set_guide_attention_entries(conditioning, entries)
|
|
|
|
def _clone_noise_mask(latent, latent_image):
|
|
if "noise_mask" in latent and latent["noise_mask"] is not None:
|
|
return latent["noise_mask"].clone()
|
|
batch, _, frames, _, _ = latent_image.shape
|
|
return torch.ones(
|
|
(batch, 1, frames, 1, 1),
|
|
dtype=torch.float32,
|
|
device=latent_image.device,
|
|
)
|
|
|
|
def _safe_json_loads(raw_value, default=None):
|
|
import json
|
|
|
|
if isinstance(raw_value, dict):
|
|
return raw_value
|
|
if not raw_value:
|
|
return {} if default is None else default
|
|
try:
|
|
return json.loads(raw_value)
|
|
except Exception:
|
|
return {} if default is None else default
|
|
|
|
def _resize_latent_spatial(latent_image, noise_mask, width, height, method):
|
|
b, c, f, h, w = latent_image.shape
|
|
if width == w and height == h:
|
|
return latent_image, noise_mask
|
|
|
|
latent_4d = latent_image.permute(0, 2, 1, 3, 4).reshape(b * f, c, h, w)
|
|
latent_4d = comfy.utils.common_upscale(latent_4d, width, height, method, "disabled")
|
|
latent_image = latent_4d.reshape(b, f, c, height, width).permute(0, 2, 1, 3, 4)
|
|
|
|
if noise_mask is not None and (noise_mask.shape[-1] > 1 or noise_mask.shape[-2] > 1):
|
|
mb, mc, mf, mh, mw = noise_mask.shape
|
|
mask_4d = noise_mask.permute(0, 2, 1, 3, 4).reshape(mb * mf, mc, mh, mw)
|
|
mask_4d = comfy.utils.common_upscale(mask_4d, width, height, method, "disabled")
|
|
noise_mask = mask_4d.reshape(mb, mf, mc, height, width).permute(0, 2, 1, 3, 4)
|
|
|
|
return latent_image, noise_mask
|
|
|
|
def _ceil_to_multiple(value, multiple):
|
|
multiple = max(1, int(multiple))
|
|
return int(math.ceil(value / multiple) * multiple)
|
|
|
|
def _normalize_resize_method_for_encode(resize_method):
|
|
if resize_method == "maintain aspect ratio":
|
|
return "pad"
|
|
return resize_method
|
|
|
|
def _append_latent_frames(latent_image, noise_mask, extra_frames, mask_fill=1.0):
|
|
extra_frames = int(max(0, extra_frames))
|
|
if extra_frames == 0:
|
|
return latent_image, noise_mask
|
|
|
|
batch, channels, _, height, width = latent_image.shape
|
|
latent_tail = torch.zeros(
|
|
(batch, channels, extra_frames, height, width),
|
|
dtype=latent_image.dtype,
|
|
device=latent_image.device,
|
|
)
|
|
latent_image = torch.cat([latent_image, latent_tail], dim=2)
|
|
|
|
mask_batch, mask_channels, _, mask_height, mask_width = noise_mask.shape
|
|
mask_tail = torch.full(
|
|
(mask_batch, mask_channels, extra_frames, mask_height, mask_width),
|
|
float(mask_fill),
|
|
dtype=noise_mask.dtype,
|
|
device=noise_mask.device,
|
|
)
|
|
noise_mask = torch.cat([noise_mask, mask_tail], dim=2)
|
|
return latent_image, noise_mask
|
|
|
|
def _encode_resized_video_frames(
|
|
vae,
|
|
frames,
|
|
target_width,
|
|
target_height,
|
|
resize_method,
|
|
time_scale_factor,
|
|
use_tiled_encode,
|
|
tile_size,
|
|
tile_overlap,
|
|
):
|
|
pixels = _resize_image(
|
|
frames,
|
|
target_width,
|
|
target_height,
|
|
_normalize_resize_method_for_encode(resize_method),
|
|
divisible_by=1,
|
|
)
|
|
num_frames_to_keep = ((pixels.shape[0] - 1) // time_scale_factor) * time_scale_factor + 1
|
|
encode_src = pixels[:num_frames_to_keep, :, :, :3]
|
|
if use_tiled_encode:
|
|
encoded = vae.encode_tiled(
|
|
encode_src,
|
|
tile_x=tile_size,
|
|
tile_y=tile_size,
|
|
overlap=tile_overlap,
|
|
)
|
|
else:
|
|
encoded = vae.encode(encode_src)
|
|
return pixels, encoded
|
|
|
|
def _snap_latent_to_downscale(latent_image, noise_mask, downscale_factor, method):
|
|
factor = int(max(1, round(float(downscale_factor))))
|
|
if factor <= 1:
|
|
return latent_image, noise_mask
|
|
|
|
_, _, _, h, w = latent_image.shape
|
|
new_w = _ceil_to_multiple(w, factor)
|
|
new_h = _ceil_to_multiple(h, factor)
|
|
if new_w == w and new_h == h:
|
|
return latent_image, noise_mask
|
|
|
|
log.warning(
|
|
"[LTXDirectorGuide] Auto-snapping latent grid from %sx%s to %sx%s for IC-LoRA downscale factor %s.",
|
|
w, h, new_w, new_h, factor,
|
|
)
|
|
return _resize_latent_spatial(latent_image, noise_mask, new_w, new_h, method)
|
|
|
|
def _load_lora_model_only(model, ic_lora_name, strength_model):
|
|
lora_path = folder_paths.get_full_path_or_raise("loras", ic_lora_name)
|
|
lora, metadata = comfy.utils.load_torch_file(
|
|
lora_path, safe_load=True, return_metadata=True,
|
|
)
|
|
|
|
try:
|
|
latent_downscale_factor = float(metadata["reference_downscale_factor"])
|
|
except Exception:
|
|
latent_downscale_factor = 1.0
|
|
log.warning(
|
|
"[LTXDirectorGuide] Could not read reference_downscale_factor from %s, using 1.0.",
|
|
ic_lora_name,
|
|
)
|
|
|
|
if strength_model == 0:
|
|
return model, latent_downscale_factor
|
|
|
|
model_lora, _ = comfy.sd.load_lora_for_models(
|
|
model, None, lora, strength_model, 0,
|
|
)
|
|
return model_lora, latent_downscale_factor
|
|
|
|
def _encode_video_iclora_guide(vae, latent_width, latent_height, images, scale_factors, latent_downscale_factor, crop, use_tiled_encode, tile_size, tile_overlap, resize_method="crop"):
|
|
time_scale_factor, width_scale_factor, height_scale_factor = scale_factors
|
|
|
|
num_frames_to_keep = ((images.shape[0] - 1) // time_scale_factor) * time_scale_factor + 1
|
|
images = images[:num_frames_to_keep]
|
|
|
|
target_width = int(latent_width * width_scale_factor / latent_downscale_factor)
|
|
target_height = int(latent_height * height_scale_factor / latent_downscale_factor)
|
|
target_width = max(8, target_width)
|
|
target_height = max(8, target_height)
|
|
|
|
# For guides, we must match the exact target latent shape.
|
|
# "maintain aspect ratio" doesn't pad, which causes shape mismatch in VAE encode / concatenation.
|
|
# So we fallback "maintain aspect ratio" to "pad".
|
|
if resize_method == "maintain aspect ratio":
|
|
resize_method = "pad"
|
|
|
|
pixels = _resize_image(images, target_width, target_height, resize_method, divisible_by=1)
|
|
|
|
encode_pixels = pixels[:, :, :, :3]
|
|
if use_tiled_encode:
|
|
guide_latent = vae.encode_tiled(encode_pixels, tile_x=tile_size, tile_y=tile_size, overlap=tile_overlap)
|
|
else:
|
|
guide_latent = vae.encode(encode_pixels)
|
|
|
|
return pixels, guide_latent
|
|
|
|
def _dilate_latent(latent: dict, horizontal_scale: int, vertical_scale: int) -> dict:
|
|
if horizontal_scale == 1 and vertical_scale == 1:
|
|
return latent
|
|
|
|
samples = latent["samples"]
|
|
mask = latent.get("noise_mask", None)
|
|
|
|
dilated_shape = samples.shape[:3] + (
|
|
samples.shape[3] * vertical_scale,
|
|
samples.shape[4] * horizontal_scale,
|
|
)
|
|
dilated_samples = torch.zeros(dilated_shape, device=samples.device, dtype=samples.dtype, requires_grad=False)
|
|
dilated_samples[..., ::vertical_scale, ::horizontal_scale] = samples
|
|
|
|
dilated_mask_shape = (
|
|
dilated_samples.shape[0], 1, dilated_samples.shape[2],
|
|
dilated_samples.shape[3], dilated_samples.shape[4],
|
|
)
|
|
dilated_mask = torch.full(dilated_mask_shape, -1.0, device=samples.device, dtype=samples.dtype, requires_grad=False)
|
|
dilated_mask[..., ::vertical_scale, ::horizontal_scale] = (mask if mask is not None else 1.0)
|
|
|
|
return {"samples": dilated_samples, "noise_mask": dilated_mask}
|
|
|
|
def _resolve_input_video_path(video_file):
|
|
if os.path.isabs(str(video_file)) and os.path.exists(str(video_file)):
|
|
return str(video_file)
|
|
input_dir = folder_paths.get_input_directory()
|
|
candidate = os.path.join(input_dir, str(video_file))
|
|
if os.path.exists(candidate):
|
|
return candidate
|
|
try:
|
|
annotated = folder_paths.get_annotated_filepath(str(video_file))
|
|
if annotated and os.path.exists(annotated):
|
|
return annotated
|
|
except Exception:
|
|
pass
|
|
raise FileNotFoundError(f"Could not find motion guide video: {video_file}")
|
|
|
|
class ResampleGuideFrames:
|
|
def execute(self, images, source_fps, target_fps, target_num_frames, mode):
|
|
if images is None: return images
|
|
frames = images
|
|
n = int(frames.shape[0])
|
|
target_num_frames = int(target_num_frames)
|
|
if n <= 1:
|
|
if target_num_frames > 1 and n == 1:
|
|
return frames.repeat(target_num_frames, 1, 1, 1)
|
|
return frames
|
|
source_fps = float(max(0.001, source_fps))
|
|
target_fps = float(max(0.001, target_fps))
|
|
if target_num_frames <= 0:
|
|
duration = (n - 1) / source_fps
|
|
target_num_frames = max(1, int(round(duration * target_fps)) + 1)
|
|
if target_num_frames == n and abs(target_fps - source_fps) < 1e-6:
|
|
return frames
|
|
positions = torch.linspace(0, n - 1, target_num_frames, device=frames.device, dtype=torch.float32)
|
|
if mode == "nearest":
|
|
idx = torch.round(positions).long().clamp(0, n - 1)
|
|
return frames.index_select(0, idx)
|
|
idx0 = torch.floor(positions).long().clamp(0, n - 1)
|
|
idx1 = torch.ceil(positions).long().clamp(0, n - 1)
|
|
alpha = (positions - idx0.to(positions.dtype)).view(-1, 1, 1, 1)
|
|
f0 = frames.index_select(0, idx0).to(torch.float32)
|
|
f1 = frames.index_select(0, idx1).to(torch.float32)
|
|
return (f0 * (1.0 - alpha) + f1 * alpha).to(frames.dtype)
|
|
|
|
def _load_motion_video_frames(video_file, trim_start_frames, length_frames, director_fps, resample_mode="nearest"):
|
|
path = _resolve_input_video_path(video_file)
|
|
target_fps = max(1.0, float(director_fps))
|
|
start_s = max(0.0, float(trim_start_frames) / target_fps)
|
|
dur_s = max(0.0, float(length_frames) / target_fps)
|
|
end_s = start_s + dur_s if dur_s > 0 else None
|
|
|
|
log.info(f"[LTXDirectorGuide] Loading video frames from {path}. start_s: {start_s:.3f}, dur_s: {dur_s:.3f}, end_s: {end_s}")
|
|
|
|
container = av.open(path)
|
|
stream = container.streams.video[0]
|
|
stream.thread_type = "AUTO" # Enable multi-threaded decoding
|
|
|
|
try:
|
|
source_fps = float(stream.average_rate) if stream.average_rate else float(stream.base_rate)
|
|
except Exception:
|
|
source_fps = target_fps
|
|
if source_fps <= 0: source_fps = target_fps
|
|
|
|
# Seek to keyframe slightly before target start_s to optimize loading speed and memory
|
|
if start_s > 0:
|
|
try:
|
|
if stream.time_base:
|
|
seek_pts = int(max(0, start_s - 0.5) / float(stream.time_base))
|
|
else:
|
|
seek_pts = int(max(0, start_s - 0.5) * av.time_base)
|
|
container.seek(seek_pts, stream=stream, backward=True)
|
|
log.info(f"[LTXDirectorGuide] PyAV seeked stream to pts={seek_pts} (approx {max(0, start_s - 0.5):.3f}s)")
|
|
except Exception as seek_err:
|
|
log.warning(f"[LTXDirectorGuide] Seek failed: {seek_err}, decoding from beginning.")
|
|
|
|
frames = []
|
|
decoded_count = 0
|
|
for frame in container.decode(stream):
|
|
if frame.time is not None:
|
|
t = float(frame.time)
|
|
elif frame.pts is not None and stream.time_base is not None:
|
|
t = float(frame.pts * stream.time_base)
|
|
else:
|
|
t = float(decoded_count / source_fps)
|
|
decoded_count += 1
|
|
|
|
if t < start_s - 0.01: continue
|
|
if end_s is not None and t >= end_s: break
|
|
|
|
# Append raw uint8 numpy arrays to minimize CPU allocation overhead
|
|
frames.append(frame.to_ndarray(format="rgb24"))
|
|
container.close()
|
|
|
|
if not frames: raise ValueError(f"No frames decoded for motion guide segment: {video_file}")
|
|
|
|
# Convert all frames to float32 at once to optimize memory allocation
|
|
frames_np = np.array(frames, dtype=np.float32) / 255.0
|
|
images = torch.from_numpy(frames_np)
|
|
|
|
target_count = max(1, int(round(float(length_frames))))
|
|
images = ResampleGuideFrames().execute(images, source_fps, target_fps, target_count, resample_mode)
|
|
return images
|
|
|
|
# --- Main Class ---
|
|
|
|
class LTXDirectorGuide:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
loras = folder_paths.get_filename_list("loras")
|
|
if not loras: loras = ["put_ic_lora_in_ComfyUI_models_loras"]
|
|
return {
|
|
"required": {
|
|
"positive": ("CONDITIONING", {"tooltip": "Positive conditioning to add guide keyframe info to."}),
|
|
"negative": ("CONDITIONING", {"tooltip": "Negative conditioning to add guide keyframe info to."}),
|
|
"vae": ("VAE", {"tooltip": "Video VAE used to encode the guide images."}),
|
|
"latent": ("LATENT", {"tooltip": "Video latent — guides are inserted into this latent."}),
|
|
"guide_data": ("GUIDE_DATA", {"tooltip": "Guide data produced by Prompt Relay Encode (Timeline)."}),
|
|
},
|
|
"optional": {
|
|
"motion_guide_data": ("MOTION_GUIDE_DATA", {"tooltip": "Connect motion guide data from the timeline node to use IC-LoRA video guidance."}),
|
|
"model": ("MODEL", {"tooltip": "Connect model if using IC-LoRA for motion guidance."}),
|
|
"ic_lora_name": (["None"] + loras, {"default": "None", "tooltip": "Select the IC-LoRA model to use for motion guidance."}),
|
|
"ic_lora_strength": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01}),
|
|
"scale_by": ("FLOAT", {"default": 0.5, "min": 0.01, "max": 8.0, "step": 0.01, "tooltip": "Scale the latent by this factor. 0.5 is recommended for IC-LoRA."}),
|
|
"upscale_method": (["nearest-exact", "bilinear", "area", "bicubic", "bislerp"], {"default": "bicubic", "tooltip": "Method used to upscale/downscale the latent."}),
|
|
"image_attention_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
|
"crop": (["disabled", "center"], {"default": "center"}),
|
|
"auto_snap_ic_grid": ("BOOLEAN", {"default": True}),
|
|
"use_tiled_encode": ("BOOLEAN", {"default": False}),
|
|
"tile_size": ("INT", {"default": 256, "min": 64, "max": 512, "step": 32}),
|
|
"tile_overlap": ("INT", {"default": 64, "min": 16, "max": 256, "step": 16}),
|
|
"retake_mode": ("BOOLEAN", {"default": False, "tooltip": "Force Retake Mode. If false, it will still auto-detect Retake Mode from the timeline data."}),
|
|
"msr_strength": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.05, "tooltip": "Licon MSR only: per-stage reference strength override. 0 = use the Director's value (full pull, right for stage 1). On a refinement/upscale stage set ~0.4 to hold detail without the references repainting the opening (fixes stage-2 mist/ghosting)."}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("CONDITIONING", "CONDITIONING", "LATENT", "MODEL", "FLOAT")
|
|
RETURN_NAMES = ("positive", "negative", "latent", "model", "latent_downscale_factor")
|
|
FUNCTION = "execute"
|
|
|
|
@classmethod
|
|
def _maybe_add_attention(cls, positive, negative, is_lora_active, tokens_added, guide_orig_shape, attention_strength):
|
|
if not is_lora_active:
|
|
return positive, negative
|
|
positive = _append_guide_attention_entry(
|
|
positive,
|
|
tokens_added,
|
|
guide_orig_shape,
|
|
attention_strength=attention_strength,
|
|
)
|
|
negative = _append_guide_attention_entry(
|
|
negative,
|
|
tokens_added,
|
|
guide_orig_shape,
|
|
attention_strength=attention_strength,
|
|
)
|
|
return positive, negative
|
|
|
|
@classmethod
|
|
def _apply_crop_count(cls, positive, negative, crop_frames):
|
|
crop_frames = max(0, int(crop_frames))
|
|
values = {"nghtdrp_guide_crop_latent_frames": crop_frames}
|
|
positive = node_helpers.conditioning_set_values(positive, values)
|
|
negative = node_helpers.conditioning_set_values(negative, values)
|
|
return positive, negative
|
|
|
|
@classmethod
|
|
def _load_cached_motion_video_frames(cls, video_cache, video_file, trim_start_frames, length_frames, director_fps, resample_mode):
|
|
cache_key = (
|
|
str(video_file),
|
|
int(trim_start_frames),
|
|
int(length_frames),
|
|
float(director_fps),
|
|
str(resample_mode),
|
|
)
|
|
cached = video_cache.get(cache_key)
|
|
if cached is None:
|
|
cached = _load_motion_video_frames(
|
|
video_file,
|
|
trim_start_frames=trim_start_frames,
|
|
length_frames=length_frames,
|
|
director_fps=director_fps,
|
|
resample_mode=resample_mode,
|
|
)
|
|
video_cache[cache_key] = cached
|
|
return cached
|
|
|
|
@classmethod
|
|
def _encode_image_guide(cls, vae, latent_width, latent_height, img_tensor, scale_factors, target_pix_w, target_pix_h, upscale_method):
|
|
_, height, width, _ = img_tensor.shape
|
|
if target_pix_w != width or target_pix_h != height:
|
|
img_nchw = img_tensor.permute(0, 3, 1, 2)
|
|
img_resized = comfy.utils.common_upscale(
|
|
img_nchw,
|
|
target_pix_w,
|
|
target_pix_h,
|
|
upscale_method,
|
|
"disabled",
|
|
)
|
|
img_tensor = img_resized.permute(0, 2, 3, 1)
|
|
return nodes_lt.LTXVAddGuide.encode(
|
|
vae,
|
|
latent_width,
|
|
latent_height,
|
|
img_tensor,
|
|
scale_factors,
|
|
)
|
|
|
|
@classmethod
|
|
def _apply_image_guides(
|
|
cls,
|
|
positive,
|
|
negative,
|
|
vae,
|
|
latent_image,
|
|
noise_mask,
|
|
images,
|
|
insert_frames,
|
|
strengths,
|
|
latent_width,
|
|
latent_height,
|
|
latent_length,
|
|
scale_factors,
|
|
image_attention_strength,
|
|
is_lora_active,
|
|
upscale_method,
|
|
):
|
|
target_pix_w = int(latent_width * 32)
|
|
target_pix_h = int(latent_height * 32)
|
|
for idx, img_tensor in enumerate(images):
|
|
frame_value = insert_frames[idx] if idx < len(insert_frames) else 0
|
|
strength = float(strengths[idx] if idx < len(strengths) else 1.0)
|
|
if strength <= 0.0:
|
|
continue
|
|
|
|
image_pixels, guide_latent = cls._encode_image_guide(
|
|
vae,
|
|
latent_width,
|
|
latent_height,
|
|
img_tensor,
|
|
scale_factors,
|
|
target_pix_w,
|
|
target_pix_h,
|
|
upscale_method,
|
|
)
|
|
frame_idx, latent_idx = nodes_lt.LTXVAddGuide.get_latent_index(
|
|
positive,
|
|
latent_length,
|
|
len(image_pixels),
|
|
int(frame_value),
|
|
scale_factors,
|
|
)
|
|
if latent_idx >= latent_length:
|
|
continue
|
|
|
|
max_frames = latent_length - latent_idx
|
|
if guide_latent.shape[2] > max_frames:
|
|
guide_latent = guide_latent[:, :, :max_frames]
|
|
|
|
tokens_added = guide_latent.shape[2] * guide_latent.shape[3] * guide_latent.shape[4]
|
|
guide_orig_shape = list(guide_latent.shape[2:])
|
|
positive, negative, latent_image, noise_mask = nodes_lt.LTXVAddGuide.append_keyframe(
|
|
positive,
|
|
negative,
|
|
frame_idx,
|
|
latent_image,
|
|
noise_mask,
|
|
guide_latent,
|
|
strength,
|
|
scale_factors,
|
|
)
|
|
positive, negative = cls._maybe_add_attention(
|
|
positive,
|
|
negative,
|
|
is_lora_active,
|
|
tokens_added,
|
|
guide_orig_shape,
|
|
image_attention_strength,
|
|
)
|
|
|
|
return positive, negative, latent_image, noise_mask
|
|
|
|
@classmethod
|
|
def _apply_motion_segment_guides(
|
|
cls,
|
|
positive,
|
|
negative,
|
|
vae,
|
|
latent_image,
|
|
noise_mask,
|
|
segments,
|
|
director_fps,
|
|
latent_length,
|
|
latent_width,
|
|
latent_height,
|
|
time_scale_factor,
|
|
scale_factors,
|
|
latent_downscale_factor,
|
|
crop,
|
|
use_tiled_encode,
|
|
tile_size,
|
|
tile_overlap,
|
|
active_resize_method,
|
|
is_lora_active,
|
|
video_cache,
|
|
):
|
|
for seg in segments:
|
|
try:
|
|
video_file = seg.get("videoFile")
|
|
if not video_file:
|
|
continue
|
|
|
|
start_frame = int(seg.get("start", 0))
|
|
length_frames = int(seg.get("length", 1))
|
|
trim_start = int(seg.get("trimStart", 0))
|
|
video_strength = float(seg.get("videoStrength", 1.0))
|
|
video_attention_strength = float(seg.get("videoAttentionStrength", 0.65))
|
|
if length_frames <= 0 or video_strength <= 0.0:
|
|
continue
|
|
|
|
video_frames = cls._load_cached_motion_video_frames(
|
|
video_cache,
|
|
video_file,
|
|
trim_start,
|
|
length_frames,
|
|
director_fps,
|
|
seg.get("resampleMode", "nearest"),
|
|
)
|
|
num_frames_to_keep = ((video_frames.shape[0] - 1) // time_scale_factor) * time_scale_factor + 1
|
|
video_frames = video_frames[:num_frames_to_keep]
|
|
causal_fix = start_frame == 0 or num_frames_to_keep == 1
|
|
encode_frames = video_frames if causal_fix else torch.cat([video_frames[:1], video_frames], dim=0)
|
|
|
|
_, guide_latent = _encode_video_iclora_guide(
|
|
vae,
|
|
latent_width,
|
|
latent_height,
|
|
encode_frames,
|
|
scale_factors,
|
|
latent_downscale_factor,
|
|
crop,
|
|
use_tiled_encode,
|
|
tile_size,
|
|
tile_overlap,
|
|
resize_method=active_resize_method,
|
|
)
|
|
if not causal_fix:
|
|
guide_latent = guide_latent[:, :, 1:, :, :]
|
|
|
|
frame_idx = start_frame
|
|
latent_idx = (frame_idx + time_scale_factor - 1) // time_scale_factor if frame_idx > 0 else 0
|
|
if latent_idx >= latent_length:
|
|
continue
|
|
|
|
if start_frame > 0 and guide_latent.shape[2] > 1:
|
|
guide_latent = guide_latent[:, :, 1:, :, :]
|
|
frame_idx += time_scale_factor
|
|
latent_idx += 1
|
|
if latent_idx >= latent_length:
|
|
continue
|
|
|
|
max_frames = latent_length - latent_idx
|
|
if guide_latent.shape[2] > max_frames:
|
|
guide_latent = guide_latent[:, :, :max_frames]
|
|
|
|
guide_orig_shape = list(guide_latent.shape[2:])
|
|
batch, _, frames, height, width = guide_latent.shape
|
|
guide_mask = torch.ones(
|
|
(batch, 1, frames, height, width),
|
|
device=guide_latent.device,
|
|
dtype=guide_latent.dtype,
|
|
)
|
|
if start_frame > 0:
|
|
ramp_steps = [0.25, 0.65]
|
|
for i, step in enumerate(ramp_steps):
|
|
if i < frames:
|
|
guide_mask[:, :, i, :, :] = 1.0 + video_strength * (1.0 - step)
|
|
|
|
ldf = int(max(1, round(float(latent_downscale_factor))))
|
|
if ldf > 1:
|
|
dilated = _dilate_latent(
|
|
{"samples": guide_latent, "noise_mask": guide_mask},
|
|
horizontal_scale=ldf,
|
|
vertical_scale=ldf,
|
|
)
|
|
guide_latent = dilated["samples"]
|
|
guide_mask = dilated["noise_mask"]
|
|
|
|
tokens_added = guide_latent.shape[2] * guide_latent.shape[3] * guide_latent.shape[4]
|
|
positive, negative, latent_image, noise_mask = nodes_lt.LTXVAddGuide.append_keyframe(
|
|
positive,
|
|
negative,
|
|
frame_idx,
|
|
latent_image,
|
|
noise_mask,
|
|
guide_latent,
|
|
video_strength,
|
|
scale_factors,
|
|
guide_mask=guide_mask,
|
|
latent_downscale_factor=float(latent_downscale_factor),
|
|
causal_fix=causal_fix,
|
|
)
|
|
positive, negative = cls._maybe_add_attention(
|
|
positive,
|
|
negative,
|
|
is_lora_active,
|
|
tokens_added,
|
|
guide_orig_shape,
|
|
video_attention_strength,
|
|
)
|
|
except Exception as e:
|
|
raise RuntimeError(f"LTX Director Guide motion segment failed for {seg}: {e}") from e
|
|
|
|
return positive, negative, latent_image, noise_mask
|
|
|
|
@classmethod
|
|
def _execute_retake_mode(
|
|
cls,
|
|
positive,
|
|
negative,
|
|
vae,
|
|
latent_image,
|
|
noise_mask,
|
|
guide_data,
|
|
tdata,
|
|
segments,
|
|
model,
|
|
latent_downscale_factor,
|
|
director_fps,
|
|
latent_length,
|
|
latent_width,
|
|
latent_height,
|
|
time_scale_factor,
|
|
active_resize_method,
|
|
use_tiled_encode,
|
|
tile_size,
|
|
tile_overlap,
|
|
is_empty_latent,
|
|
initial_latent_length,
|
|
video_cache,
|
|
):
|
|
print(
|
|
f"[LTXDirectorGuide] Retake Mode active. Preserving base latent, masking selected regions. "
|
|
f"is_empty_latent: {is_empty_latent}"
|
|
)
|
|
target_width = latent_width * 32
|
|
target_height = latent_height * 32
|
|
retake_start = int(tdata.get("retakeStart", 0))
|
|
retake_len = int(tdata.get("retakeLength", 0))
|
|
retake_strength = float(tdata.get("retakeStrength", 1.0))
|
|
|
|
start_frame = int(guide_data.get("start_frame", 0))
|
|
relative_start = max(0, retake_start - start_frame)
|
|
l_start = min(max(0, relative_start // time_scale_factor), latent_length)
|
|
l_end = min(
|
|
max(0, int(math.ceil((relative_start + retake_len) / time_scale_factor))),
|
|
latent_length,
|
|
)
|
|
|
|
need_base_video = is_empty_latent or l_start > 0 or l_end < latent_length
|
|
if not need_base_video:
|
|
print(
|
|
"[LTXDirectorGuide] Stage 2: Retake region covers the entire generated range. "
|
|
"Skipping base video loading and VAE encoding."
|
|
)
|
|
|
|
retake_vid_info = tdata.get("retakeVideo") or {}
|
|
video_file = retake_vid_info.get("imageFile", "") if isinstance(retake_vid_info, dict) else ""
|
|
if not video_file and not retake_vid_info and segments:
|
|
video_file = segments[0].get("videoFile", "")
|
|
|
|
if need_base_video and not video_file:
|
|
if retake_vid_info and not retake_vid_info.get("imageFile"):
|
|
raise ValueError(
|
|
"Retake Mode is active, but the base video file upload is still in progress (or failed). "
|
|
"Please wait for the 'Uploading base video...' overlay on the timeline to disappear before queuing the prompt."
|
|
)
|
|
raise ValueError(
|
|
"Retake Mode is active, but no base video has been selected on the timeline. "
|
|
"Please drag and drop or upload a base video on the timeline first."
|
|
)
|
|
|
|
if need_base_video and video_file:
|
|
try:
|
|
ltxv_length = (latent_length - 1) * time_scale_factor + 1
|
|
print(
|
|
f"[LTXDirectorGuide] Loading and encoding base video file: {video_file} "
|
|
f"starting at frame {start_frame} for length {ltxv_length} at resolution "
|
|
f"{target_width}x{target_height}"
|
|
)
|
|
video_frames = cls._load_cached_motion_video_frames(
|
|
video_cache,
|
|
video_file,
|
|
start_frame,
|
|
ltxv_length,
|
|
director_fps,
|
|
"nearest",
|
|
)
|
|
_, base_latent = _encode_resized_video_frames(
|
|
vae,
|
|
video_frames,
|
|
target_width,
|
|
target_height,
|
|
active_resize_method,
|
|
time_scale_factor,
|
|
use_tiled_encode,
|
|
tile_size,
|
|
tile_overlap,
|
|
)
|
|
base_latent = base_latent.to(device=latent_image.device, dtype=latent_image.dtype)
|
|
|
|
paste_len = min(base_latent.shape[2], latent_length)
|
|
if is_empty_latent:
|
|
latent_image[:, :, :paste_len] = base_latent[:, :, :paste_len]
|
|
else:
|
|
print(
|
|
f"[LTXDirectorGuide] Stage 2: Copying high-resolution base latent for preserved "
|
|
f"regions (0-{l_start} and {l_end}-{paste_len})"
|
|
)
|
|
if l_start > 0:
|
|
latent_image[:, :, :l_start] = base_latent[:, :, :l_start]
|
|
if l_end < paste_len:
|
|
latent_image[:, :, l_end:paste_len] = base_latent[:, :, l_end:paste_len]
|
|
except Exception as e:
|
|
print(f"[LTXDirectorGuide] Failed to load/encode base video: {e}. Falling back to input latent.")
|
|
|
|
noise_mask = torch.zeros_like(noise_mask)
|
|
if l_end > l_start:
|
|
noise_mask[:, :, l_start:l_end] = retake_strength
|
|
print(f"[LTXDirectorGuide] noise_mask slice: {noise_mask[0, 0, :, 0, 0].tolist()}")
|
|
|
|
exact_crop_frames = max(0, int(latent_image.shape[2]) - initial_latent_length)
|
|
positive, negative = cls._apply_crop_count(positive, negative, exact_crop_frames)
|
|
return (
|
|
positive,
|
|
negative,
|
|
{"samples": latent_image, "noise_mask": noise_mask},
|
|
model,
|
|
float(latent_downscale_factor),
|
|
)
|
|
|
|
@classmethod
|
|
def execute(cls, positive, negative, vae, latent, guide_data, motion_guide_data=None, model=None, ic_lora_name="None", ic_lora_strength=1.0, scale_by=1.0, upscale_method="bicubic", image_attention_strength=1.0, crop="center", auto_snap_ic_grid=True, use_tiled_encode=False, tile_size=256, tile_overlap=64, retake_mode=False, msr_strength=0.0):
|
|
motion_segments = (motion_guide_data or {}).get("segments", []) if motion_guide_data else []
|
|
image_guides_count = len(guide_data.get("images", [])) if guide_data else 0
|
|
print(f"[LTXDirectorGuide] execute started. motion_segments: {len(motion_segments)}, image_guides: {image_guides_count}, ic_lora_name: {ic_lora_name}, model connected: {model is not None}, retake_mode: {retake_mode}")
|
|
video_cache = {}
|
|
|
|
active_resize_method = guide_data.get("resize_method") if guide_data else None
|
|
if not active_resize_method:
|
|
active_resize_method = motion_guide_data.get("resize_method") if motion_guide_data else None
|
|
if not active_resize_method:
|
|
active_resize_method = "crop" if crop == "center" else "stretch to fit"
|
|
|
|
latent_downscale_factor = 1.0
|
|
if model is not None and ic_lora_name != "None":
|
|
model, latent_downscale_factor = _load_lora_model_only(model, ic_lora_name, ic_lora_strength)
|
|
|
|
scale_factors = vae.downscale_index_formula
|
|
latent_image = latent["samples"].clone()
|
|
noise_mask = _clone_noise_mask(latent, latent_image)
|
|
|
|
if scale_by != 1.0:
|
|
_, _, _, h, w = latent_image.shape
|
|
target_w = max(1, round(w * scale_by))
|
|
target_h = max(1, round(h * scale_by))
|
|
latent_image, noise_mask = _resize_latent_spatial(latent_image, noise_mask, target_w, target_h, upscale_method)
|
|
|
|
if auto_snap_ic_grid and model is not None and ic_lora_name != "None":
|
|
latent_image, noise_mask = _snap_latent_to_downscale(latent_image, noise_mask, latent_downscale_factor, upscale_method)
|
|
|
|
# Ghost Mask: the Director kept the latent clean (so it stays length-matched with the audio
|
|
# latent) and asked us to pre-extend it here by one hidden frame per reference. The tail
|
|
# sits past the clean region; the reference images (in guide_data["images"], anchored at
|
|
# (clean + i)) are written into it by the loop below, then dropped downstream by Clean
|
|
# Latent Slice (length = clean_latent_frames).
|
|
_ghost_pre_extend = int(guide_data.get("ghost_pre_extend", 0)) if guide_data else 0
|
|
if _ghost_pre_extend > 0:
|
|
latent_image, noise_mask = _append_latent_frames(latent_image, noise_mask, _ghost_pre_extend, mask_fill=1.0)
|
|
|
|
_, _, latent_length, latent_height, latent_width = latent_image.shape
|
|
initial_latent_length = int(latent_length)
|
|
|
|
# Licon MSR (Prefix) mode: the Director handed us raw references via guide_data["msr"].
|
|
# Inject them at THIS node's resolution (so a half-res Stage 1 and full-res Stage 2 each
|
|
# stay self-consistent), then return early — the standard guide paths don't apply.
|
|
msr = guide_data.get("msr") if guide_data else None
|
|
if msr is not None:
|
|
return cls._inject_msr(positive, negative, vae, latent_image, noise_mask, msr, model, latent_downscale_factor, msr_strength)
|
|
|
|
tdata = _safe_json_loads(guide_data.get("timeline_data", "{}") if guide_data else "{}")
|
|
|
|
is_retake_active = bool(retake_mode) or tdata.get("retakeMode", False)
|
|
is_empty_latent = (latent_image.abs().max().item() < 1e-5)
|
|
|
|
# Director image guides and motion video segments
|
|
images = guide_data.get("images", []) if guide_data else []
|
|
insert_frames = guide_data.get("insert_frames", []) if guide_data else []
|
|
strengths = guide_data.get("strengths", []) if guide_data else []
|
|
|
|
director_fps = float((motion_guide_data or {}).get("frame_rate", guide_data.get("frame_rate", 24) if guide_data else 24))
|
|
segments = (motion_guide_data or {}).get("segments", [])
|
|
|
|
is_lora_active = (model is not None) and (ic_lora_name != "None")
|
|
time_scale_factor = scale_factors[0]
|
|
ltxv_length = (latent_length - 1) * time_scale_factor + 1
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Retake Mode Branch:
|
|
# Load single base video, encode continuously, apply temporal mask.
|
|
# -----------------------------------------------------------------------
|
|
if is_retake_active:
|
|
return cls._execute_retake_mode(
|
|
positive,
|
|
negative,
|
|
vae,
|
|
latent_image,
|
|
noise_mask,
|
|
guide_data,
|
|
tdata,
|
|
segments,
|
|
model,
|
|
latent_downscale_factor,
|
|
director_fps,
|
|
latent_length,
|
|
latent_width,
|
|
latent_height,
|
|
time_scale_factor,
|
|
active_resize_method,
|
|
use_tiled_encode,
|
|
tile_size,
|
|
tile_overlap,
|
|
is_empty_latent,
|
|
initial_latent_length,
|
|
video_cache,
|
|
)
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Standard Timeline Keyframe Guidance:
|
|
# Appends image segments as keyframes (and motion segments if present)
|
|
# to the latent stream using standard LTX-Video cross-attention conditioning.
|
|
# Registers guide attention entries if IC-LoRA is active.
|
|
# -----------------------------------------------------------------------
|
|
if len(images) > 0 or len(segments) > 0:
|
|
print(f"[LTXDirectorGuide] Using Appended Keyframe Guidance. is_lora_active: {is_lora_active}")
|
|
|
|
positive, negative, latent_image, noise_mask = cls._apply_image_guides(
|
|
positive,
|
|
negative,
|
|
vae,
|
|
latent_image,
|
|
noise_mask,
|
|
images,
|
|
insert_frames,
|
|
strengths,
|
|
latent_width,
|
|
latent_height,
|
|
latent_length,
|
|
scale_factors,
|
|
image_attention_strength,
|
|
is_lora_active,
|
|
upscale_method,
|
|
)
|
|
positive, negative, latent_image, noise_mask = cls._apply_motion_segment_guides(
|
|
positive,
|
|
negative,
|
|
vae,
|
|
latent_image,
|
|
noise_mask,
|
|
segments,
|
|
director_fps,
|
|
latent_length,
|
|
latent_width,
|
|
latent_height,
|
|
time_scale_factor,
|
|
scale_factors,
|
|
latent_downscale_factor,
|
|
crop,
|
|
use_tiled_encode,
|
|
tile_size,
|
|
tile_overlap,
|
|
active_resize_method,
|
|
is_lora_active,
|
|
video_cache,
|
|
)
|
|
|
|
else:
|
|
print("[LTXDirectorGuide] No timeline guides present. Passing through.")
|
|
|
|
exact_crop_frames = max(0, int(latent_image.shape[2]) - initial_latent_length)
|
|
positive, negative = cls._apply_crop_count(positive, negative, exact_crop_frames)
|
|
|
|
return (positive, negative, {"samples": latent_image, "noise_mask": noise_mask}, model, float(latent_downscale_factor))
|
|
|
|
@classmethod
|
|
def _inject_msr(cls, positive, negative, vae, latent_image, noise_mask, msr, model, latent_downscale_factor, msr_strength=0.0):
|
|
"""Inject the Licon MSR references at this node's current resolution.
|
|
|
|
Returns this node's 5-tuple (positive, negative, latent, model, latent_downscale_factor).
|
|
"""
|
|
from nodes import NODE_CLASS_MAPPINGS
|
|
|
|
IC = NODE_CLASS_MAPPINGS.get("LTXAddVideoICLoRAGuide")
|
|
AG = NODE_CLASS_MAPPINGS.get("LTXVAddGuide")
|
|
if IC is None or AG is None:
|
|
raise ValueError("MSR mode needs LTXAddVideoICLoRAGuide (ComfyUI-LTXVideo) and LTXVAddGuide (core LTXV nodes).")
|
|
|
|
def clamp01(v):
|
|
return max(0.0, min(1.0, float(v)))
|
|
|
|
slideshow = msr["slideshow"]
|
|
keyframes = msr["keyframes"]
|
|
prefix_latents = int(msr["prefix_latents"])
|
|
# Per-stage override: msr_strength > 0 wins (set it low on a refinement stage so the
|
|
# references hold detail without repainting the opening); 0 = use the Director's value.
|
|
strength = clamp01(msr_strength) if float(msr_strength) > 0.0 else clamp01(msr["strength"])
|
|
downscale = float(msr["downscale"])
|
|
|
|
initial_latent_length = int(latent_image.shape[2])
|
|
|
|
# Pad the generated region so (pad + keyframes) == prefix_latents, so the downstream crop
|
|
# (which trims the slideshow's temporal footprint) lands on the true clean length.
|
|
pad_latents = max(0, prefix_latents - len(keyframes))
|
|
if pad_latents > 0:
|
|
latent_image, noise_mask = _append_latent_frames(latent_image, noise_mask, pad_latents, mask_fill=1.0)
|
|
|
|
latent = {"samples": latent_image, "noise_mask": noise_mask}
|
|
|
|
# CONDITIONING PATH: slideshow -> IC-LoRA -> keep conditioning, discard latent.
|
|
cp, cn, _ = _unpack(_execute_comfy_node(
|
|
IC, positive=positive, negative=negative, vae=vae, latent=latent, image=slideshow,
|
|
frame_idx=0, strength=strength, latent_downscale_factor=downscale,
|
|
crop="center", use_tiled_encode=False, tile_size=256, tile_overlap=64,
|
|
))[:3]
|
|
positive, negative = cp, cn
|
|
|
|
# LATENT PATH: one frozen keyframe per reference at frame 0 -> keep latent, drop cond.
|
|
for kf in keyframes:
|
|
unpacked = _unpack(_execute_comfy_node(
|
|
AG, positive=positive, negative=negative, vae=vae, latent=latent,
|
|
image=kf, frame_idx=0, strength=strength,
|
|
))
|
|
latent = unpacked[2]
|
|
|
|
# NOTE: MSR is a PREFIX. We deliberately do NOT set nghtdrp_guide_crop_latent_frames
|
|
# (the new LTXDirectorCropGuides trims from the END, which would cut clean frames).
|
|
# The IC/AG nodes write position-aware keyframe metadata into the conditioning, so the
|
|
# MSR branch of the workflow should crop with the STOCK position-aware LTXVCropGuides
|
|
# node (exactly as the old MSR pipeline did).
|
|
return (positive, negative, latent, model, float(latent_downscale_factor))
|
|
|
|
|
|
def _conditioning_get_any_value(conditioning, key, default=None):
|
|
for item in conditioning:
|
|
meta = item[1]
|
|
if key in meta and meta.get(key) is not None:
|
|
return meta.get(key)
|
|
return default
|
|
|
|
def _get_exact_crop_count_from_conditioning(conditioning):
|
|
value = _conditioning_get_any_value(conditioning, "nghtdrp_guide_crop_latent_frames", None)
|
|
if value is not None:
|
|
try:
|
|
return max(0, int(value))
|
|
except Exception:
|
|
return 0
|
|
keyframe_idxs = _conditioning_get_any_value(conditioning, "keyframe_idxs", None)
|
|
if keyframe_idxs is None:
|
|
return 0
|
|
try:
|
|
return int(torch.unique(keyframe_idxs[:, 0, :, 0]).shape[0])
|
|
except Exception:
|
|
return 0
|
|
|
|
def _get_noise_mask_for_crop(latent):
|
|
latent_image = latent["samples"]
|
|
noise_mask = latent.get("noise_mask", None)
|
|
if noise_mask is None:
|
|
batch, _, frames, _, _ = latent_image.shape
|
|
return torch.ones((batch, 1, frames, 1, 1), dtype=torch.float32, device=latent_image.device)
|
|
return noise_mask.clone()
|
|
|
|
|
|
class LTXDirectorCropGuides:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"positive": ("CONDITIONING",),
|
|
"negative": ("CONDITIONING",),
|
|
"latent": ("LATENT",),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("CONDITIONING", "CONDITIONING", "LATENT")
|
|
RETURN_NAMES = ("positive", "negative", "latent")
|
|
FUNCTION = "execute"
|
|
CATEGORY = "WhatDreamsCost CS"
|
|
|
|
def execute(self, positive, negative, latent):
|
|
latent_image = latent["samples"].clone()
|
|
noise_mask = _get_noise_mask_for_crop(latent)
|
|
|
|
crop_frames = _get_exact_crop_count_from_conditioning(positive)
|
|
if crop_frames <= 0:
|
|
return (positive, negative, {"samples": latent_image, "noise_mask": noise_mask})
|
|
|
|
crop_frames = min(crop_frames, max(0, latent_image.shape[2] - 1))
|
|
if crop_frames > 0:
|
|
latent_image = latent_image[:, :, :-crop_frames]
|
|
noise_mask = noise_mask[:, :, :-crop_frames]
|
|
|
|
clear_values = {
|
|
"keyframe_idxs": None,
|
|
"guide_attention_entries": None,
|
|
"nghtdrp_guide_crop_latent_frames": None,
|
|
}
|
|
positive = node_helpers.conditioning_set_values(positive, clear_values)
|
|
negative = node_helpers.conditioning_set_values(negative, clear_values)
|
|
return (positive, negative, {"samples": latent_image, "noise_mask": noise_mask})
|
|
|
|
NODE_CLASS_MAPPINGS = {
|
|
"LTXDirectorGuideCS": LTXDirectorGuide,
|
|
"LTXDirectorCropGuidesCS": LTXDirectorCropGuides,
|
|
}
|