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

refactor: split director execute helpers

Browse Source
This commit is contained in:
OpenClaw Agent
2026-07-09 13:53:10 +00:00
parent 51bd3b4968
commit 04ff3fc38a
1 changed files with 225 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

403
ltx_director.py
View File

@@ -1146,6 +1146,202 @@ def _encode_relay(model, clip, latent, global_prompt, local_prompts, segment_len
return patched, conditioning return patched, conditioning
def _load_character_slot_images(tdata: dict, image_cache: dict, input_dir: str):
char_images = []
char_slot_images = []
descriptions = ["", "", ""]
try:
characters = tdata.get("characters", [])
for idx, char_info in enumerate(characters[:3]):
descriptions[idx] = char_info.get("description", "")
for char_info in characters:
images_list = char_info.get("images", [])
legacy_b64 = char_info.get("imageB64", "")
if legacy_b64 and not images_list:
images_list = [{"b64": legacy_b64, "name": char_info.get("fileName", "")}]
slot_tensors = []
for img_info in images_list:
tensor = _load_image_source(
img_info.get("b64", ""),
img_info.get("name", ""),
cache=image_cache,
input_dir=input_dir,
)
char_images.append(tensor)
slot_tensors.append(tensor)
char_slot_images.append(slot_tensors)
except Exception as e:
log.warning("[LTXDirector] Could not process character slot inputs: %s", e)
return char_images, char_slot_images, descriptions
def _build_guide_data_from_timeline(
tdata: dict,
start_frame: int,
duration_frames: int,
frame_rate_f: float,
guide_strength: str,
custom_width: int,
custom_height: int,
resize_method: str,
divisible_by: int,
img_compression: int,
optional_latent,
input_dir: str,
image_cache: dict,
):
guide_data = {"images": [], "insert_frames": [], "strengths": [], "frame_rate": frame_rate_f}
derived_w, derived_h = custom_width, custom_height
try:
img_segs = [
s for s in tdata.get("segments", [])
if s.get("type", "image") in ("image", "video")
and (s.get("imageFile") or s.get("imageB64"))
and int(s.get("start", 0)) < start_frame + duration_frames
and int(s.get("start", 0)) + int(s.get("length", 1)) > start_frame
]
img_segs.sort(key=lambda s: s["start"])
strengths = []
if guide_strength.strip():
strengths = [float(x.strip()) for x in guide_strength.split(",") if x.strip()]
for idx, seg in enumerate(img_segs):
seg_start = int(seg.get("start", 0))
offset = max(0, start_frame - seg_start)
seg_length = int(seg.get("length", 1))
seg_for_load = seg
if seg.get("type") == "video":
if offset > 0:
seg_for_load = dict(seg)
seg_for_load["trimStart"] = float(seg.get("trimStart", 0)) + offset
seg_for_load["length"] = max(1, seg_length - offset)
tensor = _load_video_tensor(seg_for_load, frame_rate_f, input_dir=input_dir)
else:
tensor = _load_image_tensor(seg, cache=image_cache, input_dir=input_dir)
src_h, src_w = tensor.shape[1], tensor.shape[2]
tgt_w, tgt_h, resize_override = _target_resize_dims(
src_w, src_h, custom_width, custom_height, divisible_by
)
tensor = _resize_image(
tensor,
tgt_w,
tgt_h,
resize_override or resize_method,
divisible_by,
)
if img_compression > 0:
tensor = _compress_image(tensor, img_compression)
if idx == 0:
derived_h = tensor.shape[1]
derived_w = tensor.shape[2]
if seg.get("isEndFrame"):
insert_frame = max(0, seg_start + seg_length - 1 - start_frame)
else:
insert_frame = max(0, seg_start - start_frame)
strength = strengths[idx] if idx < len(strengths) else 1.0
guide_data["images"].append(tensor)
guide_data["insert_frames"].append(insert_frame)
guide_data["strengths"].append(float(strength))
if not guide_data["images"] and optional_latent is None:
src_w = derived_w if derived_w > 0 else 768
src_h = derived_h if derived_h > 0 else 512
found_dims = False
is_retake = tdata.get("retakeMode", False)
retake_vid = tdata.get("retakeVideo") or {}
retake_file = retake_vid.get("imageFile", "") if isinstance(retake_vid, dict) else ""
if is_retake and retake_file:
r_path = _resolve_input_path(retake_file, input_dir)
if r_path:
src_dims = _extract_video_dimensions(r_path)
if all(src_dims):
src_w, src_h = src_dims
found_dims = True
if not found_dims:
for mseg in tdata.get("motionSegments", []):
v_file = mseg.get("videoFile")
if not v_file:
continue
v_path = _resolve_input_path(v_file, input_dir)
if not v_path:
continue
src_dims = _extract_video_dimensions(v_path)
if all(src_dims):
src_w, src_h = src_dims
break
tensor = torch.zeros((1, src_h, src_w, 3), dtype=torch.float32)
tgt_w, tgt_h, resize_override = _target_resize_dims(
src_w, src_h, custom_width, custom_height, divisible_by
)
tensor = _resize_image(
tensor,
tgt_w,
tgt_h,
resize_override or resize_method,
divisible_by,
)
guide_data["images"].append(tensor)
guide_data["insert_frames"].append(0)
guide_data["strengths"].append(0.0)
derived_w = tensor.shape[2]
derived_h = tensor.shape[1]
except Exception as e:
log.warning("[PromptRelay] Could not build guide_data: %s", e)
return guide_data, derived_w, derived_h
def _build_motion_guide_data(tdata: dict, use_custom_motion: bool, start_frame: int,
duration_frames: int, frame_rate_f: float, resize_method: str):
motion_guide_data = {
"segments": [],
"frame_rate": frame_rate_f,
"duration_frames": int(duration_frames),
"resize_method": resize_method,
}
try:
motion_segments = tdata.get("motionSegments", []) if use_custom_motion else []
for seg in motion_segments:
seg_start = int(seg.get("start", 0))
length = int(seg.get("length", 1))
if seg_start >= start_frame + duration_frames or seg_start + length <= start_frame:
continue
if not seg.get("videoFile"):
continue
offset = max(0, start_frame - seg_start)
new_start = max(0, seg_start - start_frame)
clipped_len = min(length - offset, duration_frames - new_start)
if clipped_len <= 0:
continue
clean = dict(seg)
clean["start"] = new_start
clean["length"] = clipped_len
clean["trimStart"] = float(seg.get("trimStart", 0)) + offset
motion_guide_data["segments"].append(clean)
except Exception as e:
log.warning("[LTXDirector] Could not build motion_guide_data: %s", e)
return motion_guide_data
class LTXDirector(io.ComfyNode): class LTXDirector(io.ComfyNode):
"""WYSIWYG timeline variant — segments and lengths come from a visual editor in the node UI.""" """WYSIWYG timeline variant — segments and lengths come from a visual editor in the node UI."""
@@ -1365,36 +1561,10 @@ class LTXDirector(io.ComfyNode):
# One of: "Ghost Mask (End)", "Licon MSR (Prefix)", "OFF". # One of: "Ghost Mask (End)", "Licon MSR (Prefix)", "OFF".
reference_mode = tdata.get("reference_mode", "OFF") reference_mode = tdata.get("reference_mode", "OFF")
# --- Load character reference slots from the timeline JSON --- char_images, char_slot_images, char_descriptions = _load_character_slot_images(
# characters = [{ "images": [{"b64":..., "name":...}], "description": "..." }, ...] tdata, image_cache, input_dir
char_images = [] # flat list of every reference image tensor )
char_slot_images = [] # per-slot tensors, for @char tag filtering in MSR mode char1_val, char2_val, char3_val = char_descriptions
char1_val, char2_val, char3_val = "", "", ""
try:
characters = tdata.get("characters", [])
if len(characters) > 0:
char1_val = characters[0].get("description", "")
if len(characters) > 1:
char2_val = characters[1].get("description", "")
if len(characters) > 2:
char3_val = characters[2].get("description", "")
for char_info in characters:
images_list = char_info.get("images", [])
legacy_b64 = char_info.get("imageB64", "")
if legacy_b64 and not images_list:
images_list = [{"b64": legacy_b64, "name": char_info.get("fileName", "")}]
slot_tensors = []
for img_info in images_list:
image_b64 = img_info.get("b64", "")
file_name = img_info.get("name", "")
tensor = _load_image_source(image_b64, file_name, cache=image_cache, input_dir=input_dir)
char_images.append(tensor)
slot_tensors.append(tensor)
char_slot_images.append(slot_tensors)
except Exception as e:
log.warning("[LTXDirector] Could not process character slot inputs: %s", e)
# --- @charN substitution --- # --- @charN substitution ---
# Ghost Mask / OFF: swap @char tags for their VLM descriptions in the prompt text. # Ghost Mask / OFF: swap @char tags for their VLM descriptions in the prompt text.
@@ -1408,126 +1578,21 @@ class LTXDirector(io.ComfyNode):
) )
global_prompt, local_prompts = ref_global, ref_local global_prompt, local_prompts = ref_global, ref_local
# --- Build guide_data from image segments FIRST (to derive output dimensions) --- guide_data, derived_w, derived_h = _build_guide_data_from_timeline(
guide_data = {"images": [], "insert_frames": [], "strengths": [], "frame_rate": frame_rate} tdata=tdata,
derived_w, derived_h = custom_width, custom_height start_frame=start_frame,
try: duration_frames=duration_frames,
img_segs = [ frame_rate_f=frame_rate_f,
s for s in tdata.get("segments", []) guide_strength=guide_strength,
if s.get("type", "image") in ("image", "video") custom_width=custom_width,
and (s.get("imageFile") or s.get("imageB64")) custom_height=custom_height,
and int(s.get("start", 0)) < start_frame + duration_frames resize_method=resize_method,
and int(s.get("start", 0)) + int(s.get("length", 1)) > start_frame divisible_by=divisible_by,
] img_compression=img_compression,
img_segs.sort(key=lambda s: s["start"]) optional_latent=optional_latent,
input_dir=input_dir,
strengths = [] image_cache=image_cache,
if guide_strength.strip(): )
strengths = [float(x.strip()) for x in guide_strength.split(",") if x.strip()]
for idx, seg in enumerate(img_segs):
seg_start = int(seg.get("start", 0))
offset = max(0, start_frame - seg_start)
seg_length = int(seg.get("length", 1))
seg_for_load = seg
if seg.get("type") == "video":
if offset > 0:
seg_for_load = dict(seg)
seg_for_load["trimStart"] = float(seg.get("trimStart", 0)) + offset
seg_for_load["length"] = max(1, seg_length - offset)
tensor = _load_video_tensor(seg_for_load, frame_rate_f, input_dir=input_dir)
else:
tensor = _load_image_tensor(seg, cache=image_cache, input_dir=input_dir)
# Apply resize
src_h, src_w = tensor.shape[1], tensor.shape[2]
tgt_w, tgt_h, resize_override = _target_resize_dims(
src_w, src_h, custom_width, custom_height, divisible_by
)
tensor = _resize_image(
tensor,
tgt_w,
tgt_h,
resize_override or resize_method,
divisible_by,
)
# Apply compression
if img_compression > 0:
tensor = _compress_image(tensor, img_compression)
# Record dimensions of the first processed image for latent generation
if idx == 0:
derived_h = tensor.shape[1]
derived_w = tensor.shape[2]
if seg.get("isEndFrame"):
insert_frame = max(0, seg_start + seg_length - 1 - start_frame)
else:
insert_frame = max(0, seg_start - start_frame)
strength = strengths[idx] if idx < len(strengths) else 1.0
guide_data["images"].append(tensor)
guide_data["insert_frames"].append(insert_frame)
guide_data["strengths"].append(float(strength))
# If no images were loaded from the timeline, create a dummy image at strength 0
# to prevent artifacts in text-to-video mode.
if not guide_data["images"] and optional_latent is None:
src_w = derived_w if derived_w > 0 else 768
src_h = derived_h if derived_h > 0 else 512
# If there's an IC-LoRA video or retake base video on the timeline, extract its dimensions for accurate aspect ratio scaling
found_dims = False
# Check for retake base video first
is_retake = tdata.get("retakeMode", False)
retake_vid = tdata.get("retakeVideo") or {}
retake_file = retake_vid.get("imageFile", "") if isinstance(retake_vid, dict) else ""
if is_retake and retake_file:
r_path = _resolve_input_path(retake_file, input_dir)
if r_path:
src_dims = _extract_video_dimensions(r_path)
if all(src_dims):
src_w, src_h = src_dims
found_dims = True
# Fallback to normal motion segments
if not found_dims:
for mseg in tdata.get("motionSegments", []):
v_file = mseg.get("videoFile")
if v_file:
v_path = _resolve_input_path(v_file, input_dir)
if v_path:
src_dims = _extract_video_dimensions(v_path)
if all(src_dims):
src_w, src_h = src_dims
found_dims = True
break
# Create a dummy tensor of the exact source dimensions
tensor = torch.zeros((1, src_h, src_w, 3), dtype=torch.float32)
tgt_w, tgt_h, resize_override = _target_resize_dims(
src_w, src_h, custom_width, custom_height, divisible_by
)
tensor = _resize_image(
tensor,
tgt_w,
tgt_h,
resize_override or resize_method,
divisible_by,
)
guide_data["images"].append(tensor)
guide_data["insert_frames"].append(0)
guide_data["strengths"].append(0.0)
derived_w = tensor.shape[2]
derived_h = tensor.shape[1]
except Exception as e:
log.warning("[PromptRelay] Could not build guide_data: %s", e)
# --- Auto-generate LTXV latent if none was provided --- # --- Auto-generate LTXV latent if none was provided ---
# Apply the community 8n+1 rule directly to the timeline's duration_frames: # Apply the community 8n+1 rule directly to the timeline's duration_frames:
@@ -1844,36 +1909,14 @@ class LTXDirector(io.ComfyNode):
log.error("[PromptRelay] Could not generate empty audio latent: %s", e) log.error("[PromptRelay] Could not generate empty audio latent: %s", e)
raise e raise e
# --- Motion guide output from timeline video segments --- motion_guide_data = _build_motion_guide_data(
motion_guide_data = {"segments": [], "frame_rate": frame_rate_f, "duration_frames": int(duration_frames), "resize_method": resize_method} tdata=tdata,
try: use_custom_motion=use_custom_motion,
if use_custom_motion: start_frame=start_frame,
motion_segments = tdata.get("motionSegments", []) duration_frames=duration_frames,
else: frame_rate_f=frame_rate_f,
motion_segments = [] resize_method=resize_method,
for seg in motion_segments: )
seg_start = int(seg.get("start", 0))
length = int(seg.get("length", 1))
if seg_start >= start_frame + duration_frames or seg_start + length <= start_frame:
continue
if not seg.get("videoFile"):
continue
offset = max(0, start_frame - seg_start)
new_start = max(0, seg_start - start_frame)
# Trim length so it doesn't extend beyond duration_frames
clipped_len = min(length - offset, duration_frames - new_start)
if clipped_len <= 0:
continue
clean = dict(seg)
clean["start"] = new_start
clean["length"] = clipped_len
clean["trimStart"] = float(seg.get("trimStart", 0)) + offset
motion_guide_data["segments"].append(clean)
except Exception as e:
log.warning("[LTXDirector] Could not build motion_guide_data: %s", e)
# Inject raw timeline details for downstream masking in Retake Mode # Inject raw timeline details for downstream masking in Retake Mode
guide_data["timeline_data"] = timeline_data guide_data["timeline_data"] = timeline_data