chris.dumas/WhatDreamsCost-Dumas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf: refactor director guide execution paths

Browse Source
This commit is contained in:
OpenClaw Agent
2026-07-09 15:30:50 +00:00
parent 7099b3c6f1
commit a536ac5a14
1 changed files with 595 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files

893
ltx_director_guide.py
View File

@@ -9,11 +9,10 @@ import comfy.sd
import comfy.utils
import folder_paths
import node_helpers
from comfy_extras import nodes_lt
from comfy_api.latest import io
from .ltx_director import GuideData, MotionGuideData, _resize_image, _execute_comfy_node, _unpack
from comfy_extras import nodes_lt
from .ltx_director import _resize_image, _execute_comfy_node, _unpack
log = logging.getLogger(__name__)
log = logging.getLogger(__name__)
# --- Helper Functions from Nghtdrp ---
@@ -47,15 +46,27 @@ def _append_guide_attention_entry(
)
return _set_guide_attention_entries(conditioning, entries)
def _clone_noise_mask(latent, latent_image):
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,
)
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
@@ -74,9 +85,68 @@ def _resize_latent_spatial(latent_image, noise_mask, width, height, method):
return latent_image, noise_mask
def _ceil_to_multiple(value, multiple):
multiple = max(1, int(multiple))
return int(math.ceil(value / multiple) * multiple)
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))))
@@ -272,7 +342,7 @@ def _load_motion_video_frames(video_file, trim_start_frames, length_frames, dire
# --- Main Class ---
class LTXDirectorGuide:
class LTXDirectorGuide:
@classmethod
def INPUT_TYPES(cls):
loras = folder_paths.get_filename_list("loras")
@@ -304,19 +374,421 @@ class LTXDirectorGuide:
}
RETURN_TYPES = ("CONDITIONING", "CONDITIONING", "LATENT", "MODEL", "FLOAT")
RETURN_NAMES = ("positive", "negative", "latent", "model", "latent_downscale_factor")
FUNCTION = "execute"
@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}")
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:
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
@@ -336,26 +808,17 @@ class LTXDirectorGuide:
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:
gb, gc, gf, gh, gw = latent_image.shape
latent_image = torch.cat(
[latent_image, torch.zeros((gb, gc, _ghost_pre_extend, gh, gw), dtype=latent_image.dtype, device=latent_image.device)],
dim=2,
)
mb, mc, mf, mh, mw = noise_mask.shape
noise_mask = torch.cat(
[noise_mask, torch.ones((mb, mc, _ghost_pre_extend, mh, mw), dtype=noise_mask.dtype, device=noise_mask.device)],
dim=2,
)
_, _, latent_length, latent_height, latent_width = latent_image.shape
initial_latent_length = int(latent_length)
# 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
@@ -364,16 +827,10 @@ class LTXDirectorGuide:
if msr is not None:
return cls._inject_msr(positive, negative, vae, latent_image, noise_mask, msr, model, latent_downscale_factor, msr_strength)
# Parse timeline JSON to see if retake mode is active in UI
import json
timeline_data_str = guide_data.get("timeline_data", "{}") if guide_data else "{}"
try:
tdata = json.loads(timeline_data_str)
except Exception:
tdata = {}
is_retake_active = bool(retake_mode) or tdata.get("retakeMode", False)
is_empty_latent = (latent_image.abs().max().item() < 1e-5)
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 []
@@ -391,112 +848,31 @@ class LTXDirectorGuide:
# Retake Mode Branch:
# Load single base video, encode continuously, apply temporal mask.
# -----------------------------------------------------------------------
if is_retake_active:
print(f"[LTXDirectorGuide] Retake Mode active. Preserving base latent, masking selected regions. is_empty_latent: {is_empty_latent}")
target_width = latent_width * 32
target_height = latent_height * 32
# Calculate retake region latent indices first so we know what to copy/paste
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 = relative_start // time_scale_factor
l_end = int(math.ceil((relative_start + retake_len) / time_scale_factor))
l_start = min(l_start, latent_length)
l_end = min(l_end, latent_length)
# Stage 2 optimization: If the retake region covers the entire generation area,
# there are no preserved regions to copy over in Stage 2. We can bypass video loading/encoding.
need_base_video = True
if not is_empty_latent and l_start == 0 and l_end >= latent_length:
need_base_video = False
print("[LTXDirectorGuide] Stage 2: Retake region covers the entire generated range. Skipping base video loading and VAE encoding.")
# 1. Try to load and encode base video from timeline data
retake_vid_info = tdata.get("retakeVideo") or {}
video_file = retake_vid_info.get("imageFile", "") if isinstance(retake_vid_info, dict) else ""
# Fallback to first segment only if it exists and we have no retake video info at all (old workflows)
if not video_file and not retake_vid_info and len(segments) > 0:
video_file = segments[0].get("videoFile", "")
if need_base_video:
if 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."
)
else:
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 video_file and need_base_video:
try:
print(f"[LTXDirectorGuide] Loading and encoding base video file: {video_file} starting at frame {start_frame} for length {ltxv_length} at resolution {target_width}x{target_height}")
video_frames = _load_motion_video_frames(
video_file, trim_start_frames=start_frame, length_frames=ltxv_length, director_fps=director_fps, resample_mode="nearest"
)
# Retake base video must match the exact target latent shape.
# "maintain aspect ratio" doesn't pad, which causes shape mismatch in VAE encode.
# So we fallback "maintain aspect ratio" to "pad" for retake base video.
retake_resize_method = active_resize_method
if retake_resize_method == "maintain aspect ratio":
retake_resize_method = "pad"
pixels = _resize_image(video_frames, target_width, target_height, retake_resize_method, divisible_by=1)
num_clip_frames = pixels.shape[0]
num_frames_to_keep = ((num_clip_frames - 1) // time_scale_factor) * time_scale_factor + 1
encode_src = pixels[:num_frames_to_keep, :, :, :3]
if use_tiled_encode:
base_latent = vae.encode_tiled(encode_src, tile_x=tile_size, tile_y=tile_size, overlap=tile_overlap)
else:
base_latent = vae.encode(encode_src)
base_latent = base_latent.to(device=latent_image.device, dtype=latent_image.dtype)
# Copy to latent_image
paste_len = min(base_latent.shape[2], latent_length)
if is_empty_latent:
# Stage 1: Overwrite entire latent with base video VAE encode
latent_image[:, :, :paste_len] = base_latent[:, :, :paste_len]
else:
# Stage 2: Overwrite only preserved regions (before l_start and after l_end)
# leaving the generated retake region from Stage 1 untouched
print(f"[LTXDirectorGuide] Stage 2: Copying high-resolution base latent for preserved 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.")
# 2. Build the temporal noise mask (0.0 = frozen, 1.0 = regenerate)
noise_mask = torch.zeros_like(noise_mask) # Initialize fully frozen
l_start = min(l_start, latent_length)
l_end = min(l_end, latent_length)
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()}")
# In retake mode, skip normal mode processing entirely and return immediately!
exact_crop_frames = max(0, int(latent_image.shape[2]) - initial_latent_length)
positive = node_helpers.conditioning_set_values(positive, {"nghtdrp_guide_crop_latent_frames": exact_crop_frames})
negative = node_helpers.conditioning_set_values(negative, {"nghtdrp_guide_crop_latent_frames": exact_crop_frames})
return (positive, negative, {"samples": latent_image, "noise_mask": noise_mask}, model, float(latent_downscale_factor))
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:
@@ -504,126 +880,56 @@ class LTXDirectorGuide:
# 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}")
# A. Process Image Guides
for idx, img_tensor in enumerate(images):
f_idx = 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
B_img, H_img, W_img, C_img = img_tensor.shape
target_pix_w = int(latent_width * 32)
target_pix_h = int(latent_height * 32)
if target_pix_w != W_img or target_pix_h != H_img:
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)
image_pixels, guide_latent = nodes_lt.LTXVAddGuide.encode(vae, latent_width, latent_height, img_tensor, scale_factors)
frame_idx, latent_idx = nodes_lt.LTXVAddGuide.get_latent_index(positive, latent_length, len(image_pixels), int(f_idx), 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
)
if is_lora_active:
positive = _append_guide_attention_entry(positive, tokens_added, guide_orig_shape, attention_strength=image_attention_strength)
negative = _append_guide_attention_entry(negative, tokens_added, guide_orig_shape, attention_strength=image_attention_strength)
# B. Process Motion Video Segments
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
start_frame_aligned = start_frame
video_frames = _load_motion_video_frames(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 = int(start_frame_aligned) == 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_aligned
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:])
B_g, C_g, F_g, H_g, W_g = guide_latent.shape
guide_mask = torch.ones((B_g, 1, F_g, H_g, W_g), device=guide_latent.device, dtype=guide_latent.dtype)
if start_frame > 0:
ramp_steps = [0.25, 0.65]
for i, s in enumerate(ramp_steps):
if i < F_g:
guide_mask_val = 1.0 + video_strength * (1.0 - s)
guide_mask[:, :, i, :, :] = guide_mask_val
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_mask = dilated["noise_mask"]
guide_latent = dilated["samples"]
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
)
if is_lora_active:
positive = _append_guide_attention_entry(positive, tokens_added, guide_orig_shape, attention_strength=video_attention_strength)
negative = _append_guide_attention_entry(negative, tokens_added, guide_orig_shape, attention_strength=video_attention_strength)
except Exception as e:
raise RuntimeError(f"LTX Director Guide motion segment failed for {seg}: {e}") from e
else:
print("[LTXDirectorGuide] No timeline guides present. Passing through.")
exact_crop_frames = max(0, int(latent_image.shape[2]) - initial_latent_length)
positive = node_helpers.conditioning_set_values(positive, {"nghtdrp_guide_crop_latent_frames": exact_crop_frames})
negative = node_helpers.conditioning_set_values(negative, {"nghtdrp_guide_crop_latent_frames": exact_crop_frames})
return (positive, negative, {"samples": latent_image, "noise_mask": noise_mask}, model, float(latent_downscale_factor))
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):
@@ -651,20 +957,11 @@ class LTXDirectorGuide:
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:
B, C, F, H, W = latent_image.shape
latent_image = torch.cat(
[latent_image, torch.zeros((B, C, pad_latents, H, W), dtype=latent_image.dtype, device=latent_image.device)],
dim=2,
)
mb, mc, mf, mh, mw = noise_mask.shape
noise_mask = torch.cat(
[noise_mask, torch.ones((mb, mc, pad_latents, mh, mw), dtype=noise_mask.dtype, device=noise_mask.device)],
dim=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}
@@ -764,4 +1061,4 @@ class LTXDirectorCropGuides:
NODE_CLASS_MAPPINGS = {
"LTXDirectorGuideCS": LTXDirectorGuide,
"LTXDirectorCropGuidesCS": LTXDirectorCropGuides,
}
}