import logging
import asyncio
import json
import base64
import io as _io
import math
import re
import numpy as np
import torch
import torch.nn.functional as F
import av
from PIL import Image
import os
import platform
import subprocess
from urllib.parse import parse_qs, urlparse
import folder_paths
import comfy.model_management
from server import PromptServer
from aiohttp import web
from comfy_api.latest import io
from .prompt_relay import (
get_raw_tokenizer,
map_token_indices,
build_segments,
create_mask_fn,
distribute_segment_lengths,
)
from .patches import detect_model_type, apply_patches
from .msr_character import MSRCharacterSetData
from .timeline_nodes import TimelineSectionSetData
log = logging.getLogger(__name__)
# Setup global event loop exception handler to silence ConnectionResetError (WinError 10054/10053) on Windows
try:
loop = None
try:
loop = asyncio.get_event_loop()
except RuntimeError:
try:
loop = asyncio.get_event_loop_policy().get_event_loop()
except Exception:
pass
if loop is not None:
old_handler = loop.get_exception_handler()
def silence_connection_reset_handler(loop, context):
exception = context.get('exception')
if (isinstance(exception, (ConnectionResetError, ConnectionAbortedError)) or
(isinstance(exception, OSError) and getattr(exception, 'winerror', None) in (10054, 10053))):
# Suppress WinError 10054 and WinError 10053 tracebacks in logging
return
if old_handler:
old_handler(loop, context)
else:
loop.default_exception_handler(context)
loop.set_exception_handler(silence_connection_reset_handler)
except Exception:
pass
# Custom socket type shared with LTXSequencer
GuideData = io.Custom("GUIDE_DATA")
MotionGuideData = io.Custom("MOTION_GUIDE_DATA")
# --- Licon MSR engine fixed parameters (were node widgets; hardcoded for a clean UI) ---
# Slideshow length. 41 is the Licon training default (valid: 17 / 25 / 33 / 41).
MSR_PREFIX_FRAMES = 41
# latent_downscale_factor for the IC-LoRA reference frames. Tied to how the MSR LoRA was
# trained; Licon MSR V1 uses full-resolution references (1.0). This is INDEPENDENT of any
# multi-stage scale_by / x2 upscaling, which LTXDirectorGuide handles downstream.
MSR_LATENT_DOWNSCALE = 1.0
EMPTY_IMAGE_SHAPE = (1, 512, 512, 3)
PEAK_BUCKETS = 200
def _empty_image_tensor() -> torch.Tensor:
return torch.zeros(EMPTY_IMAGE_SHAPE, dtype=torch.float32)
def _normalise_b64_payload(payload: str) -> str:
if payload and "," in payload:
return payload.split(",", 1)[1]
return payload or ""
def _image_to_tensor(img: Image.Image) -> torch.Tensor:
arr = np.asarray(img.convert("RGB"), dtype=np.float32) / 255.0
return torch.from_numpy(arr).unsqueeze(0)
def _resolve_input_path(path: str, input_dir: str = None) -> str | None:
if not path:
return None
input_dir = input_dir or folder_paths.get_input_directory()
clean_path = path.replace("\\", "/")
candidates = [
os.path.join(input_dir, clean_path),
os.path.join(input_dir, "whatdreamscost", os.path.basename(clean_path)),
os.path.join(input_dir, os.path.basename(clean_path)),
]
for candidate in candidates:
if os.path.exists(candidate):
return candidate
return None
def _load_image_reference(b64_or_url: str = "", filename: str = None, cache: dict | None = None,
input_dir: str = None) -> torch.Tensor:
"""Load a reference image from a base64 string, a Comfy /view URL, or an input filename."""
if not b64_or_url and not filename:
return _empty_image_tensor()
input_dir = input_dir or folder_paths.get_input_directory()
cache = cache if cache is not None else {}
file_path = None
if b64_or_url and "view?" in b64_or_url:
try:
parsed = urlparse(b64_or_url)
q = parse_qs(parsed.query)
fname = q.get("filename", [None])[0]
subfolder = q.get("subfolder", [""])[0]
if fname:
file_path = _resolve_input_path(os.path.join(subfolder, fname), input_dir)
except Exception as e:
log.debug(f"[PromptRelay] URL parsing failed for {b64_or_url}: {e}")
if file_path is None and filename:
file_path = _resolve_input_path(filename, input_dir)
if file_path:
cache_key = ("file", file_path)
cached = cache.get(cache_key)
if cached is not None:
return cached
try:
tensor = _image_to_tensor(Image.open(file_path))
cache[cache_key] = tensor
return tensor
except Exception as e:
log.debug(f"[PromptRelay] File image loading failed for {file_path}: {e}")
if b64_or_url:
try:
b64_str = _normalise_b64_payload(b64_or_url)
cache_key = ("b64", b64_str)
cached = cache.get(cache_key)
if cached is not None:
return cached
img_bytes = base64.b64decode(b64_str)
tensor = _image_to_tensor(Image.open(_io.BytesIO(img_bytes)))
cache[cache_key] = tensor
return tensor
except Exception as e:
log.debug(f"[PromptRelay] Base64 decoding failed: {e}")
return _empty_image_tensor()
def _build_runtime_timeline_payload(external_timeline, frame_rate: float) -> tuple[dict, str, str, float, int]:
tdata = {
"segments": [],
"motionSegments": [],
"audioSegments": [],
}
if not isinstance(external_timeline, dict):
return tdata, "", "", 0.0, 0
sections = external_timeline.get("sections") or []
current_start = 0
local_prompts = []
segment_lengths = []
total_duration_seconds = 0.0
for index, section in enumerate(sections):
if not isinstance(section, dict):
continue
prompt = (section.get("text") or "").strip()
if not prompt:
continue
section_seconds = max(0.1, float(section.get("duration") or 0.0))
section_frames = max(1, int(round(section_seconds * frame_rate)))
total_duration_seconds += section_seconds
seg = {
"id": f"external_{index}",
"start": current_start,
"length": section_frames,
"prompt": prompt,
"type": "image" if section.get("image") is not None else "text",
}
if section.get("image") is not None:
seg["runtime_image"] = section.get("image")
tdata["segments"].append(seg)
local_prompts.append(prompt)
segment_lengths.append(str(section_frames))
current_start += section_frames
duration_seconds = max(0.1, float(total_duration_seconds or 0.0))
duration_frames = max(1, int(round(duration_seconds * frame_rate)))
return tdata, " | ".join(local_prompts), ",".join(segment_lengths), duration_seconds, duration_frames
def _make_json_safe(value):
"""Return a JSON-serializable copy of runtime data by stripping live tensor payloads."""
if isinstance(value, torch.Tensor):
return None
if isinstance(value, dict):
cleaned = {}
for key, item in value.items():
safe_item = _make_json_safe(item)
if safe_item is not None:
cleaned[key] = safe_item
return cleaned
if isinstance(value, (list, tuple)):
cleaned = []
for item in value:
safe_item = _make_json_safe(item)
if safe_item is not None:
cleaned.append(safe_item)
return cleaned
return value
def _compute_signal_peaks(samples: np.ndarray, num_peaks: int = PEAK_BUCKETS) -> list[float]:
if samples.size == 0:
return [0.0] * num_peaks
step = max(1, int(math.ceil(samples.size / num_peaks)))
padded = np.pad(samples, (0, step * num_peaks - samples.size), mode="constant")
chunks = padded.reshape(num_peaks, step)
return (np.max(np.abs(chunks), axis=1) / 32767.0).astype(np.float32).tolist()
def _safe_json_loads(raw_value, default=None):
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 _read_audio_bytes(container, stream, resampler) -> bytes:
audio_bytes = bytearray()
for frame in container.decode(stream):
for resampled_frame in resampler.resample(frame):
audio_bytes.extend(resampled_frame.to_ndarray().tobytes())
for resampled_frame in resampler.resample(None):
audio_bytes.extend(resampled_frame.to_ndarray().tobytes())
return bytes(audio_bytes)
def _target_resize_dims(src_w: int, src_h: int, custom_width: int, custom_height: int, divisible_by: int) -> tuple[int, int, str]:
def snap(val, div):
return max(div, (val // div) * div)
if custom_width > 0 and custom_height > 0:
return custom_width, custom_height, None
if custom_width > 0:
tgt_w = snap(custom_width, divisible_by)
tgt_h = snap(int(src_h * tgt_w / src_w), divisible_by)
return tgt_w, tgt_h, "stretch to fit"
if custom_height > 0:
tgt_h = snap(custom_height, divisible_by)
tgt_w = snap(int(src_w * tgt_h / src_h), divisible_by)
return tgt_w, tgt_h, "stretch to fit"
return src_w, src_h, "maintain aspect ratio"
def _extract_video_dimensions(file_path: str) -> tuple[int | None, int | None]:
try:
with av.open(file_path) as container:
stream = container.streams.video[0]
return stream.width or stream.codec_context.width, stream.height or stream.codec_context.height
except Exception:
return None, None
def _get_audio_segment_source(seg: dict, override_audio: bool = False) -> tuple[str | None, str | _io.BytesIO | None]:
file_key = "videoFile" if override_audio else "audioFile"
source_path = seg.get(file_key)
if source_path:
file_path = _resolve_input_path(source_path)
if file_path:
return f"file:{file_path}", file_path
if not override_audio and seg.get("audioB64"):
b64 = seg.get("audioB64")
if "," in b64:
b64 = b64.split(",", 1)[1]
try:
audio_bytes = base64.b64decode(b64)
cache_key = f"b64:{len(b64)}:{b64[:64]}"
return cache_key, _io.BytesIO(audio_bytes)
except Exception:
return None, None
return None, None
def _decode_audio_waveform(buffer, target_sr: int) -> torch.Tensor | None:
clip_arrays = []
with av.open(buffer) as container:
if not container.streams.audio:
return None
stream = container.streams.audio[0]
resampler = av.AudioResampler(
format="fltp",
layout="stereo",
rate=target_sr,
)
for frame in container.decode(stream):
for resampled_frame in resampler.resample(frame):
clip_arrays.append(resampled_frame.to_ndarray())
for resampled_frame in resampler.resample(None):
clip_arrays.append(resampled_frame.to_ndarray())
if not clip_arrays:
return None
waveform_np = clip_arrays[0] if len(clip_arrays) == 1 else np.concatenate(clip_arrays, axis=1)
return torch.from_numpy(waveform_np)
def _time_range_to_latent_indices(rel_start_frames: float, rel_length_frames: float, start_frame: int,
total_frames: int, frame_rate: float, latent_frames: int) -> tuple[int, int]:
total_sec = total_frames / frame_rate
start_sec = max(0.0, rel_start_frames - start_frame) / frame_rate
end_sec = max(start_sec, start_sec + (rel_length_frames / frame_rate))
start_idx = int((start_sec / total_sec) * latent_frames)
end_idx = int((end_sec / total_sec) * latent_frames)
start_idx = max(0, min(latent_frames, start_idx))
end_idx = max(0, min(latent_frames, end_idx))
return start_idx, end_idx
def _normalize_character_alias(alias: str) -> str:
alias = (alias or "").strip()
if alias.startswith("@"):
alias = alias[1:]
return alias.strip()
def _build_character_tag_groups(characters: list[dict]) -> list[tuple[str, ...]]:
groups: list[tuple[str, ...]] = []
for idx, character in enumerate(characters or []):
tags = [f"@character{idx + 1}", f"@char{idx + 1}"]
alias = _normalize_character_alias(character.get("alias", ""))
if alias:
tags.append(f"@{alias}")
groups.append(tuple(tags))
return groups
def _character_prompt_replacements(characters: list[dict]) -> dict[str, str]:
replacements: dict[str, str] = {}
for character, tags in zip(characters or [], _build_character_tag_groups(characters)):
replacement = character.get("description", "") or ""
for tag in tags:
replacements[tag] = replacement
return replacements
def _preprocess_prompts_with_characters(global_prompt, local_prompts, characters: list[dict] | None = None):
"""Invisibly swaps out @characterN/@charN/@alias tags with their character descriptions."""
if "@" not in (global_prompt or "") and "@" not in (local_prompts or ""):
return global_prompt or "", local_prompts or ""
replacements = _character_prompt_replacements(characters or [])
def apply_replacements(text: str) -> str:
updated = text or ""
for tag, replacement in replacements.items():
if tag in updated:
updated = updated.replace(tag, replacement)
return updated
gp = apply_replacements(global_prompt or "")
if not local_prompts:
return gp, ""
processed_locals = [apply_replacements(part.strip()) for part in local_prompts.split("|")]
return gp, " | ".join(processed_locals)
def _load_image_source(b64_or_url: str, filename: str = None, cache: dict | None = None,
input_dir: str = None) -> torch.Tensor:
return _load_image_reference(b64_or_url=b64_or_url, filename=filename, cache=cache, input_dir=input_dir)
def _execute_comfy_node(node_class, **kwargs):
"""Invoke a ComfyUI node's main entrypoint, whether it is a comfy_api io.ComfyNode
(classmethod 'execute') or a legacy node (instance method named by FUNCTION)."""
if hasattr(node_class, "execute"):
return node_class.execute(**kwargs)
fn_name = getattr(node_class, "FUNCTION", None)
instance = node_class()
if fn_name and hasattr(instance, fn_name):
return getattr(instance, fn_name)(**kwargs)
raise RuntimeError(f"Could not determine how to execute node {node_class!r}")
def _unpack(out):
"""Normalise a node return (io.NodeOutput, tuple, list or dict) into a tuple of outputs."""
if out is None:
return ()
for attr in ("result", "args", "values", "outputs"):
if hasattr(out, attr):
val = getattr(out, attr)
if callable(val):
try:
val = val()
except Exception:
continue
if isinstance(val, (tuple, list)):
return tuple(val)
if isinstance(out, (tuple, list)):
return tuple(out)
if isinstance(out, dict) and isinstance(out.get("result"), (tuple, list)):
return tuple(out["result"])
return (out,)
# --- File Check Endpoint for Deduplication ---
@PromptServer.instance.routes.get("/ltx_director_check_file")
async def ltx_director_check_file(request):
filename = request.query.get("filename", "")
file_size = request.query.get("size", "")
if not filename:
return web.json_response({"exists": False})
upload_dir = folder_paths.get_input_directory()
temp_dir = os.path.join(upload_dir, "whatdreamscost")
# 1. Check if the exact filename exists in whatdreamscost or root input dir
possible_paths = [
os.path.join(temp_dir, filename),
os.path.join(upload_dir, filename)
]
found_path = None
for p in possible_paths:
if os.path.exists(p) and os.path.isfile(p):
if file_size:
try:
if os.path.getsize(p) == int(file_size):
found_path = p
break
except ValueError:
found_path = p
break
else:
found_path = p
break
if found_path:
rel_name = os.path.relpath(found_path, upload_dir).replace('\\', '/')
return web.json_response({"exists": True, "name": rel_name})
# 2. Suffix search if exact match not found
base_name = os.path.basename(filename)
suffix = f"_{base_name}"
try:
for search_dir in [temp_dir, upload_dir]:
if os.path.exists(search_dir):
for f_name in os.listdir(search_dir):
if f_name.endswith(suffix) or f_name == base_name:
pot_path = os.path.join(search_dir, f_name)
if os.path.isfile(pot_path):
if file_size:
try:
if os.path.getsize(pot_path) == int(file_size):
rel_name = os.path.relpath(pot_path, upload_dir).replace('\\', '/')
return web.json_response({"exists": True, "name": rel_name})
except ValueError:
pass
else:
rel_name = os.path.relpath(pot_path, upload_dir).replace('\\', '/')
return web.json_response({"exists": True, "name": rel_name})
except Exception as e:
log.warning(f"[LTXDirector] Error listing input directory: {e}")
return web.json_response({"exists": False})
# --- Provider defaults shared by the analyze + unload endpoints ---
_PROVIDER_DEFAULTS = {
"ollama": {"url": "http://127.0.0.1:11434", "model": "huihui_ai/qwen3.5-abliterated:2b"},
"lmstudio": {"url": "http://127.0.0.1:1234", "model": ""},
"custom": {"url": "", "model": ""},
}
_ANALYZE_SYSTEM_PROMPT = ""
_ANALYZE_PROMPT = (
"Describe the character's physical appearance in two concise sentences. "
"Specify their hair color/style, face details, and their clothing type/color. "
"Keep the entire response very brief."
)
def _resolve_analyze_prompt(data: dict) -> str:
prompt = (data.get("prompt") or "").strip()
return prompt or _ANALYZE_PROMPT
def _extract_ollama_generated_text(resp_json: dict) -> str:
if not isinstance(resp_json, dict):
return ""
candidates = [resp_json.get("response"), resp_json.get("content"), resp_json.get("thinking")]
message = resp_json.get("message")
if isinstance(message, dict):
candidates.extend([
message.get("content"),
message.get("reasoning_content"),
message.get("thinking"),
])
cleaned_candidates = []
for candidate in candidates:
if isinstance(candidate, str):
text = candidate.strip()
if "" in text:
text = text.split("")[-1].strip()
if text:
cleaned_candidates.append(text)
if not cleaned_candidates:
return ""
usable = [text for text in cleaned_candidates if _analysis_text_is_usable(text)]
if usable:
return max(usable, key=len)
return max(cleaned_candidates, key=len)
def _collect_ollama_text_candidates(resp_json: dict) -> list[str]:
if not isinstance(resp_json, dict):
return []
candidates = [resp_json.get("response"), resp_json.get("content"), resp_json.get("thinking")]
message = resp_json.get("message")
if isinstance(message, dict):
candidates.extend([
message.get("content"),
message.get("reasoning_content"),
message.get("thinking"),
])
cleaned = []
for candidate in candidates:
if isinstance(candidate, str):
text = candidate.strip()
if "" in text:
text = text.split("")[-1].strip()
if text:
cleaned.append(text)
return cleaned
def _analysis_text_is_usable(text: str) -> bool:
text = (text or "").strip()
if not text:
return False
if len(text) < 24:
return False
if len(text.split()) < 6:
return False
return True
def _compress_analysis_image_b64(b64_payload: str, max_dim: int = 768, quality: int = 82) -> str:
"""Shrink analysis images so multimodal providers do not burn their full context on pixels."""
try:
raw = base64.b64decode(_normalise_b64_payload(b64_payload))
with Image.open(_io.BytesIO(raw)) as img:
img = img.convert("RGB")
w, h = img.size
if max(w, h) > max_dim:
scale = max_dim / float(max(w, h))
img = img.resize((max(1, int(round(w * scale))), max(1, int(round(h * scale)))), Image.LANCZOS)
out = _io.BytesIO()
img.save(out, format="JPEG", quality=quality, optimize=True)
return base64.b64encode(out.getvalue()).decode("ascii")
except Exception:
return _normalise_b64_payload(b64_payload)
def _prepare_analysis_images(cleaned_b64_list: list[str], max_dim: int = 768, quality: int = 82) -> list[str]:
return [
_compress_analysis_image_b64(b64, max_dim=max_dim, quality=quality)
for b64 in cleaned_b64_list
]
def _resolve_provider(data):
provider = (data.get("provider") or "ollama").lower()
defs = _PROVIDER_DEFAULTS.get(provider, _PROVIDER_DEFAULTS["ollama"])
base_url = (data.get("base_url") or defs["url"]).rstrip("/")
model = data.get("model") or defs["model"]
return provider, base_url, model
# --- Character reference analysis endpoint (Ollama / LM Studio / Custom OpenAI-compatible) ---
@PromptServer.instance.routes.post("/ltx_director/analyze_character")
async def analyze_character_endpoint(request):
try:
import aiohttp
data = await request.json()
image_b64 = data.get("image_b64", "")
image_debug = data.get("image_debug") or []
char_index = int(data.get("char_index", 0))
provider, base_url, model_name = _resolve_provider(data)
analyze_prompt = _resolve_analyze_prompt(data)
if provider == "off":
return web.json_response({"status": "error", "message": "Analyze is set to Off / Manual."})
if not image_b64:
return web.json_response({"status": "error", "message": "No image provided for analysis."})
b64_list = image_b64 if isinstance(image_b64, list) else [image_b64]
cleaned_b64_list = []
for b64 in b64_list:
if "," in b64:
b64 = b64.split(",", 1)[1]
cleaned_b64_list.append(b64)
if not cleaned_b64_list:
return web.json_response({"status": "error", "message": "No valid base64 images decoded."})
if provider in ("lmstudio", "custom") and not model_name:
return web.json_response({
"status": "error",
"message": f"No model name set for {provider}. Open the gear menu and enter your loaded model's name.",
})
log.info("[LTXDirector] Analyzing Character %d via %s (%s, model '%s')...",
char_index + 1, provider, base_url, model_name)
try:
async with aiohttp.ClientSession() as session:
if provider == "ollama":
debug_candidates = []
analysis_images = cleaned_b64_list
payload = {
"model": model_name,
"system": _ANALYZE_SYSTEM_PROMPT,
"prompt": analyze_prompt,
"images": analysis_images,
"stream": False,
"keep_alive": 0,
"options": {
"temperature": 0.2,
"num_predict": 768,
},
}
async with session.post(f"{base_url}/api/generate", json=payload, timeout=300) as response:
if response.status != 200:
err_txt = await response.text()
if response.status == 400 and "exceeds the available context size" in err_txt:
analysis_images = _prepare_analysis_images(cleaned_b64_list, max_dim=512, quality=70)
payload["images"] = analysis_images
async with session.post(f"{base_url}/api/generate", json=payload, timeout=300) as retry_response:
if retry_response.status != 200:
retry_err = await retry_response.text()
return web.json_response({"status": "error", "message": f"Ollama HTTP {retry_response.status}: {retry_err}"})
resp_json = await retry_response.json()
debug_candidates.extend(_collect_ollama_text_candidates(resp_json))
generated_text = _extract_ollama_generated_text(resp_json)
else:
return web.json_response({"status": "error", "message": f"Ollama HTTP {response.status}: {err_txt}"})
else:
resp_json = await response.json()
debug_candidates.extend(_collect_ollama_text_candidates(resp_json))
generated_text = _extract_ollama_generated_text(resp_json)
if not _analysis_text_is_usable(generated_text):
chat_payload = {
"model": model_name,
"messages": [
{
"role": "system",
"content": _ANALYZE_SYSTEM_PROMPT,
},
{
"role": "user",
"content": analyze_prompt,
"images": analysis_images,
},
],
"stream": False,
"keep_alive": 0,
"options": {
"temperature": 0.2,
"num_predict": 768,
},
}
async with session.post(f"{base_url}/api/chat", json=chat_payload, timeout=300) as response:
if response.status == 200:
resp_json = await response.json()
debug_candidates.extend(_collect_ollama_text_candidates(resp_json))
chat_text = _extract_ollama_generated_text(resp_json)
if _analysis_text_is_usable(chat_text):
generated_text = chat_text
if not _analysis_text_is_usable(generated_text):
return web.json_response({
"status": "error",
"message": "Ollama returned only a truncated analysis response.",
"debug_candidates": debug_candidates,
"image_debug": image_debug,
"image_lengths": [len(b64) for b64 in cleaned_b64_list],
"description": generated_text,
})
else:
# OpenAI-compatible vision chat (LM Studio / Custom).
content = [{"type": "text", "text": analyze_prompt}]
for b64 in cleaned_b64_list:
content.append({"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}})
payload = {
"model": model_name,
"messages": [
{"role": "system", "content": _ANALYZE_SYSTEM_PROMPT},
{"role": "user", "content": content},
],
"max_tokens": 4096, "stream": False,
}
async with session.post(f"{base_url}/v1/chat/completions", json=payload, timeout=120) as response:
if response.status != 200:
err_txt = await response.text()
return web.json_response({"status": "error", "message": f"{provider} HTTP {response.status}: {err_txt}"})
resp_json = await response.json()
try:
msg = resp_json["choices"][0]["message"]
generated_text = (msg.get("content") or "").strip()
# Reasoning/"thinking" models (Gemma, Qwen-thinking, etc.) may leave
# content empty and put their output in reasoning_content instead.
if not generated_text:
generated_text = (msg.get("reasoning_content") or "").strip()
except (KeyError, IndexError, TypeError):
return web.json_response({"status": "error", "message": f"Unexpected response shape from {provider}."})
except aiohttp.ClientConnectorError:
return web.json_response({
"status": "error",
"message": f"Could not connect to {provider} at {base_url}. Make sure the server is running and reachable.",
})
if "" in generated_text:
generated_text = generated_text.split("")[-1].strip()
log.info("[LTXDirector] Analysis complete: %s", generated_text)
return web.json_response({"status": "success", "description": generated_text})
except Exception as e:
log.error(f"[LTXDirector] Failed to analyze character: {e}")
return web.json_response({"status": "error", "message": str(e)}, status=500)
@PromptServer.instance.routes.post("/ltx_director/unload_ollama")
async def unload_ollama_endpoint(request):
"""Evict the analysis model from VRAM right before a generation run, so the VLM doesn't
compete with LTX for VRAM (the cause of intermittent CUDA offload crashes).
Ollama supports a clean instant unload (keep_alive=0). LM Studio / Custom have no reliable
cross-version HTTP unload, so for those this is a graceful no-op — users should set a short
JIT / auto-unload TTL in their server instead. Fully tolerant: never raises into the run.
"""
try:
import aiohttp
try:
data = await request.json()
except Exception:
data = {}
provider, base_url, model_name = _resolve_provider(data)
if provider == "ollama":
payload = {"model": model_name, "keep_alive": 0}
try:
async with aiohttp.ClientSession() as session:
async with session.post(f"{base_url}/api/generate", json=payload, timeout=8) as response:
await response.text()
log.info("[LTXDirector] Asked Ollama to release '%s' from VRAM before the run.", model_name)
except Exception:
pass # Ollama not running / unreachable -> nothing to free.
return web.json_response({"status": "ok", "provider": provider})
# LM Studio / Custom: no reliable HTTP unload across versions -> graceful no-op.
return web.json_response({"status": "ok", "provider": provider, "note": "no-op (set a JIT/TTL unload in your server)"})
except Exception as e:
return web.json_response({"status": "error", "message": str(e)})
def read_wav_peaks(wav_path):
import wave
with wave.open(wav_path, 'rb') as w:
n_frames = w.getnframes()
if n_frames <= 0:
return [0.0] * PEAK_BUCKETS
frames_bytes = w.readframes(n_frames)
samples = np.frombuffer(frames_bytes, dtype=np.int16)
return _compute_signal_peaks(samples)
def extract_audio_from_video(video_path):
import wave
try:
base, _ = os.path.splitext(video_path)
output_wav = base + "_extracted_audio.wav"
# Check if already exists, is not empty, and has the correct 44100Hz sample rate
if os.path.exists(output_wav) and os.path.getsize(output_wav) > 44:
try:
with wave.open(output_wav, 'rb') as w_check:
if w_check.getframerate() == 44100:
peaks = read_wav_peaks(output_wav)
input_dir = folder_paths.get_input_directory()
rel_output = os.path.relpath(output_wav, input_dir).replace('\\', '/')
return rel_output, peaks
except Exception:
pass
# Decode the video using PyAV
with av.open(video_path) as container:
if not container.streams.audio:
return None, None
stream = container.streams.audio[0]
# Setup resampler to 44100Hz, Mono, signed 16-bit integer (s16)
resampler = av.AudioResampler(
format='s16',
layout='mono',
rate=44100,
)
audio_bytes = _read_audio_bytes(container, stream, resampler)
if not audio_bytes:
return None, None
# Write WAV file
with wave.open(output_wav, 'wb') as w:
w.setnchannels(1)
w.setsampwidth(2) # 16-bit
w.setframerate(44100)
w.writeframes(audio_bytes)
# Calculate peaks
samples = np.frombuffer(audio_bytes, dtype=np.int16)
peaks = _compute_signal_peaks(samples)
input_dir = folder_paths.get_input_directory()
rel_output = os.path.relpath(output_wav, input_dir).replace('\\', '/')
return rel_output, peaks
except Exception as e:
print(f"[LTXDirector] Server audio extraction failed: {e}")
return None, None
def get_audio_peaks(audio_path):
import wave
# If it is already a WAV file, read peaks directly
_, ext = os.path.splitext(audio_path)
if ext.lower() == ".wav":
try:
return read_wav_peaks(audio_path)
except Exception:
pass # fallback to PyAV
# Use PyAV to decode and resample the audio file
try:
with av.open(audio_path) as container:
if not container.streams.audio:
return None
stream = container.streams.audio[0]
resampler = av.AudioResampler(
format='s16',
layout='mono',
rate=8000,
)
audio_bytes = _read_audio_bytes(container, stream, resampler)
if not audio_bytes:
return None
samples = np.frombuffer(audio_bytes, dtype=np.int16)
return _compute_signal_peaks(samples)
except Exception as e:
print(f"[LTXDirector] Failed to get audio peaks via PyAV: {e}")
return None
@PromptServer.instance.routes.get("/ltx_director_get_audio")
async def ltx_director_get_audio(request):
filename = request.query.get("filename")
if not filename:
return web.json_response({"error": "Missing filename"}, status=400)
upload_dir = folder_paths.get_input_directory()
clean_filename = filename.replace('\\', '/')
file_path = os.path.join(upload_dir, clean_filename)
if not os.path.exists(file_path):
basename = os.path.basename(clean_filename)
temp_path = os.path.join(upload_dir, "whatdreamscost", basename)
if os.path.exists(temp_path):
file_path = temp_path
else:
file_path = os.path.join(upload_dir, basename)
if not os.path.exists(file_path) or not os.path.isfile(file_path):
return web.json_response({"error": "File not found"}, status=404)
_, ext = os.path.splitext(file_path)
is_audio = ext.lower() in [".wav", ".mp3", ".ogg", ".flac", ".m4a"]
if is_audio:
peaks = None
try:
peaks = get_audio_peaks(file_path)
except Exception as e:
print(f"[LTXDirector] Failed to get audio peaks for audio file: {e}")
rel_path = os.path.relpath(file_path, upload_dir).replace('\\', '/')
return web.json_response({
"audio_file": rel_path,
"peaks": peaks
})
audio_file, peaks = None, None
try:
loop = asyncio.get_event_loop()
audio_file, peaks = await loop.run_in_executor(None, extract_audio_from_video, file_path)
except Exception as e:
print(f"[LTXDirector] Error extracting audio: {e}")
return web.json_response({
"audio_file": audio_file,
"peaks": peaks
})
@PromptServer.instance.routes.get("/ltx_director_open_folder")
async def ltx_director_open_folder(request):
upload_dir = os.path.join(folder_paths.get_input_directory(), "whatdreamscost")
os.makedirs(upload_dir, exist_ok=True)
try:
current_os = platform.system()
if current_os == "Windows":
subprocess.Popen(["explorer", os.path.normpath(upload_dir)])
elif current_os == "Darwin":
subprocess.Popen(["open", upload_dir])
else:
subprocess.Popen(["xdg-open", upload_dir])
return web.json_response({"success": True})
except Exception as e:
print(f"[LTXDirector] Failed to open workspace folder: {e}")
return web.json_response({"success": False, "error": str(e)}, status=500)
def _read_and_write_file_chunk(file, file_path, mode):
chunk_bytes = file.file.read()
with open(file_path, mode) as f:
f.write(chunk_bytes)
# --- LTX Director Chunked Video Upload Endpoint ---
# Bypasses the 413 Payload Too Large error for large video files.
# This endpoint is self-contained and independent of any other node.
@PromptServer.instance.routes.post("/ltx_director_upload_chunk")
async def ltx_director_upload_chunk(request):
post = await request.post()
file = post.get("file")
filename = post.get("filename")
chunk_index = int(post.get("chunk_index"))
total_chunks = int(post.get("total_chunks"))
upload_dir = os.path.join(folder_paths.get_input_directory(), "whatdreamscost")
os.makedirs(upload_dir, exist_ok=True)
# Sanitize filename to prevent path traversal attacks (e.g. ../../etc/passwd)
filename = os.path.basename(filename)
file_path = os.path.join(upload_dir, filename)
# Belt-and-suspenders: confirm the resolved path is still inside the upload directory
if not os.path.realpath(file_path).startswith(os.path.realpath(upload_dir)):
return web.json_response({"error": "Invalid filename"}, status=400)
# Append chunk to file (write fresh on first chunk, append on subsequent)
mode = "ab" if chunk_index > 0 else "wb"
# Offload the blocking read/write disk I/O to a thread executor
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, _read_and_write_file_chunk, file, file_path, mode)
if chunk_index == total_chunks - 1:
audio_file, peaks = None, None
try:
audio_file, peaks = await loop.run_in_executor(None, extract_audio_from_video, file_path)
except Exception as e:
print(f"[LTXDirector] Error in final chunk audio extraction: {e}")
return web.json_response({
"name": f"whatdreamscost/{filename}",
"audio_file": audio_file,
"peaks": peaks
})
return web.json_response({"status": "ok"})
def _load_image_tensor(seg: dict, cache: dict | None = None, input_dir: str = None) -> torch.Tensor:
"""Decode an image from the ComfyUI input folder (if imageFile provided) or fallback to base64
to a ComfyUI-style image tensor of shape [1, H, W, 3], float32 in [0, 1]."""
if seg.get("runtime_image") is not None:
return seg.get("runtime_image")
return _load_image_reference(
b64_or_url=seg.get("imageB64", ""),
filename=seg.get("imageFile"),
cache=cache,
input_dir=input_dir,
)
def _load_video_tensor(seg: dict, frame_rate: float, input_dir: str = None) -> torch.Tensor:
"""Extracts a sequence of frames from a video file based on the segment's trim parameters,
and returns them as an [N, H, W, 3] float32 tensor."""
file_path = _resolve_input_path(seg.get("imageFile", ""), input_dir)
if not file_path:
return _empty_image_tensor()
trim_start_frames = float(seg.get("trimStart", 0))
length_frames = float(seg.get("length", 1))
start_sec = trim_start_frames / frame_rate
frames = []
try:
with av.open(file_path) as container:
stream = container.streams.video[0]
stream.thread_type = "AUTO"
# Seek slightly before target to hit a keyframe
if stream.time_base:
seek_pts = int((max(0, start_sec - 0.5)) / float(stream.time_base))
else:
seek_pts = int((max(0, start_sec - 0.5)) * av.time_base)
container.seek(seek_pts, stream=stream, backward=True)
for frame in container.decode(stream):
frame_time = frame.time
if frame_time is None and frame.pts is not None and stream.time_base:
frame_time = float(frame.pts * stream.time_base)
if frame_time is None:
frame_time = 0.0
if frame_time < start_sec - 0.01:
continue
frames.append(frame.to_ndarray(format='rgb24'))
if len(frames) >= int(length_frames):
break
except Exception as e:
log.warning(f"[PromptRelay] Video extract error: {e}")
if not frames:
return _empty_image_tensor()
frames_np = np.array(frames, dtype=np.float32) / 255.0
return torch.from_numpy(frames_np)
def _resize_image(tensor: torch.Tensor, target_w: int, target_h: int, method: str, divisible_by: int) -> torch.Tensor:
"""Resize an [N, H, W, 3] float32 tensor to target dimensions using the given method,
then snap the final dimensions to be divisible by `divisible_by`."""
def snap(val, div):
return max(div, (val // div) * div)
tw = snap(target_w, divisible_by)
th = snap(target_h, divisible_by)
N, H, W, C = tensor.shape
if H == th and W == tw:
return tensor
t_nchw = tensor.permute(0, 3, 1, 2)
if method == "stretch to fit":
resized = F.interpolate(t_nchw, size=(th, tw), mode="bilinear", align_corners=False)
elif method == "maintain aspect ratio":
ratio = min(tw / W, th / H)
new_w = snap(int(W * ratio), divisible_by)
new_h = snap(int(H * ratio), divisible_by)
resized = F.interpolate(t_nchw, size=(new_h, new_w), mode="bilinear", align_corners=False)
elif method == "pad" or method == "pad green":
ratio = min(tw / W, th / H)
new_w = snap(int(W * ratio), divisible_by)
new_h = snap(int(H * ratio), divisible_by)
inner = F.interpolate(t_nchw, size=(new_h, new_w), mode="bilinear", align_corners=False)
pad_l = (tw - new_w) // 2
pad_t = (th - new_h) // 2
if method == "pad green":
resized = torch.zeros((N, C, th, tw), dtype=t_nchw.dtype, device=t_nchw.device)
# #66FF00 is roughly R: 102/255, G: 255/255, B: 0
resized[:, 0, :, :] = 102 / 255.0
resized[:, 1, :, :] = 1.0
resized[:, 2, :, :] = 0.0
resized[:, :, pad_t:pad_t+new_h, pad_l:pad_l+new_w] = inner
else:
resized = F.pad(inner, (pad_l, tw - new_w - pad_l, pad_t, th - new_h - pad_t), mode="constant", value=0)
elif method == "crop":
ratio = max(tw / W, th / H)
new_w = int(W * ratio)
new_h = int(H * ratio)
inner = F.interpolate(t_nchw, size=(new_h, new_w), mode="bilinear", align_corners=False)
left = (new_w - tw) // 2
top = (new_h - th) // 2
resized = inner[:, :, top:top+th, left:left+tw]
else:
resized = F.interpolate(t_nchw, size=(th, tw), mode="bilinear", align_corners=False)
return resized.permute(0, 2, 3, 1)
def _compress_image(tensor: torch.Tensor, crf: int) -> torch.Tensor:
"""Apply H.264 compression artefacts to an [N, H, W, 3] float32 tensor (ComfyUI image format).
crf=0 means no compression. Uses PyAV to encode/decode frames in-memory."""
if crf == 0:
return tensor
N, H, W, C = tensor.shape
# Dimensions must be even for H.264
h = (H // 2) * 2
w = (W // 2) * 2
# uint8 [N, H, W, 3]
tensor_bytes = (tensor[:, :h, :w, :] * 255.0).byte().cpu().numpy()
try:
buf = _io.BytesIO()
container = av.open(buf, mode="w", format="mp4")
stream = container.add_stream("libx264", rate=24)
stream.width = w
stream.height = h
stream.pix_fmt = "yuv420p"
stream.options = {"crf": str(crf), "preset": "ultrafast"}
for i in range(N):
frame = av.VideoFrame.from_ndarray(tensor_bytes[i], format="rgb24")
for pkt in stream.encode(frame):
container.mux(pkt)
for pkt in stream.encode(None):
container.mux(pkt)
container.close()
buf.seek(0)
container_r = av.open(buf, mode="r")
decoded = [frame_r.to_ndarray(format="rgb24") for frame_r in container_r.decode(video=0)]
container_r.close()
if not decoded:
return tensor
decoded_np = np.stack(decoded).astype(np.float32) / 255.0
# Re-embed into original tensor shape (may have been cropped by even-rounding)
out = tensor.clone()
dec_N = min(N, len(decoded))
out[:dec_N, :h, :w] = torch.from_numpy(decoded_np[:dec_N]).to(tensor.device, tensor.dtype)
return out
except Exception as e:
log.warning("[PromptRelay] img_compression encode/decode failed: %s", e)
return tensor
def _build_combined_audio(timeline_data_value, start_frame: int, duration_frames: int, frame_rate: float, override_audio: bool = False) -> dict:
"""Parses timeline JSON, loads/trims audio directly from memory using PyAV,
and aligns to a global timeline yielding ComfyUI's format.
Output length explicitly mimics the timeline's duration_frames length."""
target_sr = 44100
total_samples = max(1, int(math.ceil(duration_frames / frame_rate * target_sr)))
empty_audio = {"waveform": torch.zeros((1, 2, total_samples), dtype=torch.float32), "sample_rate": target_sr}
if not timeline_data_value:
return empty_audio
try:
data = _safe_json_loads(timeline_data_value)
is_retake = data.get("retakeMode", False)
if is_retake and data.get("retakeVideo"):
retake_vid = data.get("retakeVideo")
audio_segs = [{
"videoFile": retake_vid.get("imageFile") or retake_vid.get("fileName"),
"audioFile": retake_vid.get("imageFile") or retake_vid.get("fileName"),
"start": 0,
"length": retake_vid.get("videoDurationFrames", duration_frames),
"trimStart": 0
}]
override_audio = True
elif override_audio:
audio_segs = data.get("motionSegments", [])
else:
audio_segs = data.get("audioSegments", [])
except Exception:
return empty_audio
if not audio_segs:
return empty_audio
out_waveform = torch.zeros((2, total_samples), dtype=torch.float32)
decoded_waveform_cache = {}
for seg in audio_segs:
cache_key, buffer = _get_audio_segment_source(seg, override_audio=override_audio)
if not buffer:
continue
try:
waveform = decoded_waveform_cache.get(cache_key)
if waveform is None:
waveform = _decode_audio_waveform(buffer, target_sr)
if waveform is None:
continue
if cache_key:
decoded_waveform_cache[cache_key] = waveform
if waveform.shape[1] <= 0:
continue
# Calculate interactive trim boundaries
trim_start_frames = float(seg.get("trimStart", 0))
length_frames = float(seg.get("length", 1))
start_frames = float(seg.get("start", 0))
if start_frames + length_frames <= start_frame:
continue
offset = max(0, start_frame - start_frames)
trim_start_frames += offset
length_frames = max(1, length_frames - offset)
start_frames = max(0, start_frames - start_frame)
start_sample_src = int(trim_start_frames / frame_rate * target_sr)
length_samples = int(length_frames / frame_rate * target_sr)
end_sample_src = start_sample_src + length_samples
if start_sample_src < 0: start_sample_src = 0
if end_sample_src > waveform.shape[1]:
end_sample_src = waveform.shape[1]
actual_length = end_sample_src - start_sample_src
if actual_length <= 0: continue
# Extract the correct segment of the audio
clip_waveform = waveform[:, start_sample_src:end_sample_src]
# Position onto the timeline
start_sample_dst = int(start_frames / frame_rate * target_sr)
if start_sample_dst >= out_waveform.shape[1]:
continue
end_sample_dst = start_sample_dst + actual_length
# Clip any trailing overflow so we don't index past the timeline bounds
if end_sample_dst > out_waveform.shape[1]:
actual_length = out_waveform.shape[1] - start_sample_dst
clip_waveform = clip_waveform[:, :actual_length]
end_sample_dst = start_sample_dst + actual_length
if actual_length <= 0:
continue
# Additive composite (allows clips overlapping to sum together naturally)
out_waveform[:, start_sample_dst:end_sample_dst] += clip_waveform
except Exception as e:
log.warning("[PromptRelay] Audio process error for segment %s: %s", seg.get("fileName"), e)
continue
return {"waveform": out_waveform.unsqueeze(0), "sample_rate": target_sr}
def _convert_to_latent_lengths(pixel_lengths, temporal_stride, latent_frames):
"""Convert pixel-space segment lengths to integer latent-space lengths using the
largest-remainder method. Targets the full `latent_frames` when the pixel sum looks
like full coverage (within one stride of latent_frames * stride). Otherwise targets
round(total_pixel / temporal_stride) so partial-coverage timelines stay partial.
"""
if not pixel_lengths:
return []
total_pixel = sum(pixel_lengths)
if total_pixel <= 0:
return [1] * len(pixel_lengths)
naive_total = max(1, round(total_pixel / temporal_stride))
target_total = min(latent_frames, naive_total)
# Within one frame of full → user clearly intended full coverage; pin to latent_frames.
if target_total >= latent_frames - 1:
target_total = latent_frames
exact = [p * target_total / total_pixel for p in pixel_lengths]
result = [int(e) for e in exact]
diff = target_total - sum(result)
if diff > 0:
order = sorted(range(len(exact)), key=lambda i: -(exact[i] - int(exact[i])))
for k in range(diff):
result[order[k % len(order)]] += 1
# Ensure every segment has ≥ 1 latent frame (steal from the largest if needed).
for i in range(len(result)):
if result[i] < 1:
max_idx = max(range(len(result)), key=lambda j: result[j])
if result[max_idx] > 1:
result[max_idx] -= 1
result[i] = 1
return result
def _encode_relay(model, clip, latent, global_prompt, local_prompts, segment_lengths, epsilon):
for name, val in (("global_prompt", global_prompt),
("local_prompts", local_prompts),
("segment_lengths", segment_lengths)):
if val is None:
raise ValueError(
f"PromptRelay: '{name}' arrived as None. "
"Likely causes: a stale workflow JSON saved with null, the timeline "
"editor's web extension failing to load, or an upstream node returning None. "
"Set the field to an empty string or fix the upstream connection."
)
# Split prompts but do NOT filter out empty ones yet, so we can detect them
locals_list = [p.strip() for p in local_prompts.split("|")]
# If there are no visual segments on the timeline (e.g., only using IC-LoRA motion track),
# bypass the local prompt chunking entirely and just use the global prompt.
if not locals_list or (len(locals_list) == 1 and not locals_list[0]):
log.info("[PromptRelay] No local segments found. Using global prompt exclusively.")
conditioning = clip.encode_from_tokens_scheduled(clip.tokenize(global_prompt))
return model.clone(), conditioning
# Check if any specific segment is empty and apply fallbacks
for i, p in enumerate(locals_list):
if not p:
fallback = global_prompt.strip() if global_prompt else "video"
if not fallback:
fallback = "video"
locals_list[i] = fallback
arch, patch_size, temporal_stride = detect_model_type(model)
samples = latent["samples"]
latent_frames = samples.shape[2]
tokens_per_frame = (samples.shape[3] // patch_size[1]) * (samples.shape[4] // patch_size[2])
parsed_lengths = None
if segment_lengths.strip():
pixel_lengths = [int(float(x.strip())) for x in segment_lengths.split(",") if x.strip()]
parsed_lengths = _convert_to_latent_lengths(pixel_lengths, temporal_stride, latent_frames)
raw_tokenizer = get_raw_tokenizer(clip)
full_prompt, token_ranges = map_token_indices(raw_tokenizer, global_prompt, locals_list)
log.info("[PromptRelay] Global: tokens [0:%d] (%d tokens)", token_ranges[0][0], token_ranges[0][0])
for i, (s, e) in enumerate(token_ranges):
log.info("[PromptRelay] Segment %d: tokens [%d:%d] (%d tokens)", i, s, e, e - s)
conditioning = clip.encode_from_tokens_scheduled(clip.tokenize(full_prompt))
effective_lengths = distribute_segment_lengths(len(locals_list), latent_frames, parsed_lengths)
log.info(
"[PromptRelay] Latent: %d frames, %d tokens/frame, segments: %s",
latent_frames, tokens_per_frame, effective_lengths,
)
q_token_idx = build_segments(token_ranges, effective_lengths, epsilon, None)
mask_fn = create_mask_fn(q_token_idx, tokens_per_frame, latent_frames)
patched = model.clone()
apply_patches(patched, arch, mask_fn)
return patched, conditioning
def _extract_runtime_character_entries(tdata: dict, character_set, image_cache: dict, input_dir: str) -> list[dict]:
entries: list[dict] = []
if isinstance(character_set, dict) and character_set.get("characters"):
try:
for item in character_set.get("characters", []):
images = [img for img in (item.get("images") or []) if img is not None]
entries.append(
{
"images": images,
"description": item.get("description", "") or "",
"alias": _normalize_character_alias(item.get("alias", "")),
}
)
except Exception as e:
log.warning("[LTXDirector] Could not process external character_set input: %s", e)
if entries:
return entries
try:
for char_info in tdata.get("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", "")}]
loaded_images = []
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,
)
loaded_images.append(tensor)
entries.append(
{
"images": loaded_images,
"description": char_info.get("description", "") or "",
"alias": _normalize_character_alias(char_info.get("alias", "")),
}
)
except Exception as e:
log.warning("[LTXDirector] Could not process legacy timeline character slots: %s", e)
return entries
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
processed_still_cache = {}
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") or s.get("runtime_image") is not None)
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)
cache_key = None
else:
if seg.get("runtime_image") is not None:
cache_key = None
cached_tensor = None
tensor = _load_image_tensor(seg, cache=image_cache, input_dir=input_dir)
else:
cache_key = (
seg.get("imageFile") or "",
seg.get("imageB64") or "",
custom_width,
custom_height,
resize_method,
divisible_by,
img_compression,
)
cached_tensor = processed_still_cache.get(cache_key)
if cached_tensor is not None:
tensor = cached_tensor
else:
tensor = _load_image_tensor(seg, cache=image_cache, input_dir=input_dir)
if cache_key is None or cached_tensor is None:
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 cache_key is not None:
processed_still_cache[cache_key] = tensor
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
def _build_reference_mode_outputs(
reference_mode: str,
vae,
global_prompt: str,
local_prompts: str,
segment_lengths: str,
duration_frames: int,
epsilon: float,
characters: list[dict],
char_images: list,
char_slot_images: list,
guide_data: dict,
latent_w: int,
latent_h: int,
latent_grid_h: int,
latent_grid_w: int,
clean_latent_frames: int,
reference_strength: float,
optional_latent,
resize_method: str,
divisible_by: int,
prepare_tensor_image,
model,
clip,
conditioning_neg,
_dev,
):
if reference_mode == "Licon MSR (Prefix)":
if vae is None:
raise ValueError("Licon MSR (Prefix) ref option requires connecting the VAE to LTX Director!")
prompt_text = (global_prompt or "") + " " + (local_prompts or "")
tag_groups = _build_character_tag_groups(characters)
referenced_slots = [i for i, tags in enumerate(tag_groups) if any(t in prompt_text for t in tags)]
selected = []
for slot in referenced_slots:
if slot < len(char_slot_images):
selected.extend(char_slot_images[slot])
if not selected:
selected = char_images
log.info("[LTXDirector] MSR slideshow subjects: referenced slots=%s, images=%d", referenced_slots, len(selected))
identity_images = [prepare_tensor_image(c, latent_w, latent_h) for c in selected]
scene_images = list(guide_data["images"])
if scene_images:
bg_slide = scene_images[0]
if bg_slide.shape[1] != latent_h or bg_slide.shape[2] != latent_w:
bg_slide = _resize_image(bg_slide, latent_w, latent_h, resize_method, divisible_by)
else:
bg_slide = torch.zeros((1, latent_h, latent_w, 3), dtype=torch.float32)
slideshow_sources = identity_images + [bg_slide]
base_count = MSR_PREFIX_FRAMES // len(slideshow_sources)
remainder = MSR_PREFIX_FRAMES % len(slideshow_sources)
slideshow_tensors = []
for index, src_img in enumerate(slideshow_sources):
repeats = base_count + (1 if index < remainder else 0)
slideshow_tensors.extend([src_img[0]] * repeats)
slideshow_video = torch.stack(slideshow_tensors)
keyframe_images = slideshow_sources
prefix_latents = ((MSR_PREFIX_FRAMES - 1) // 8) + 1
tail_latents = prefix_latents
total_latent_frames = clean_latent_frames + tail_latents
tail_pixels = tail_latents * 8
local_part = local_prompts.strip() if local_prompts.strip() else global_prompt
clean_lengths = segment_lengths.strip() if segment_lengths.strip() else str(duration_frames)
injected_local = f"{local_part} | {global_prompt}"
injected_lengths = f"{clean_lengths},{tail_pixels}"
dummy_full = {"samples": torch.zeros([1, 128, total_latent_frames, latent_grid_h, latent_grid_w], device=_dev)}
patched, conditioning = _encode_relay(
model, clip, dummy_full, global_prompt, injected_local, injected_lengths, epsilon,
)
if optional_latent is None:
latent = {
"samples": torch.zeros([1, 128, clean_latent_frames, latent_grid_h, latent_grid_w], device=_dev),
"noise_mask": torch.ones((1, 1, clean_latent_frames, latent_grid_h, latent_grid_w), dtype=torch.float32, device=_dev),
}
else:
latent = optional_latent
out_guide_data = {
"images": [],
"insert_frames": [],
"strengths": [],
"frame_rate": guide_data.get("frame_rate"),
"msr": {
"slideshow": slideshow_video,
"keyframes": keyframe_images,
"prefix_latents": int(prefix_latents),
"strength": float(reference_strength),
"downscale": float(MSR_LATENT_DOWNSCALE),
"clean_latent_frames": int(clean_latent_frames),
"negative": conditioning_neg,
},
}
log.info(
"[LTXDirector] Licon MSR Engine: clean=%d, tail=%d (prefix), %d keyframes for the guide node.",
clean_latent_frames, tail_latents, len(keyframe_images),
)
return patched, conditioning, latent, out_guide_data
if optional_latent is None:
latent = {"samples": torch.zeros([1, 128, clean_latent_frames, latent_grid_h, latent_grid_w], device=_dev)}
log.info(
"[PromptRelay] Auto-generated LTXV latent: %dx%d, %d pixel frames (%d latent frames)",
latent_w, latent_h, ((clean_latent_frames - 1) * 8) + 1, clean_latent_frames,
)
else:
latent = optional_latent
patched, conditioning = _encode_relay(
model, clip, latent, global_prompt, local_prompts, segment_lengths, epsilon,
)
return patched, conditioning, latent, guide_data
def _build_audio_latent(
audio_vae,
audio_out,
use_custom_audio: bool,
override_audio: bool,
is_retake_active: bool,
inpaint_audio: bool,
tdata: dict,
start_frame: int,
ltxv_length: int,
frame_rate_f: float,
):
if audio_vae is None:
return {}
def get_empty_latent():
inner = getattr(audio_vae, "first_stage_model", audio_vae)
z_channels = audio_vae.latent_channels
audio_freq = inner.latent_frequency_bins
num_audio_latents = inner.num_of_latents_from_frames(ltxv_length, frame_rate_f)
audio_latents = torch.zeros(
(1, z_channels, num_audio_latents, audio_freq),
device=comfy.model_management.intermediate_device(),
)
return {"samples": audio_latents, "type": "audio"}
if not (use_custom_audio or override_audio or is_retake_active):
audio_latent = get_empty_latent()
log.info("[PromptRelay] Auto-generated empty audio latent.")
return audio_latent
if audio_out is None:
raise ValueError("No audio waveform to encode.")
waveform = audio_out["waveform"]
if waveform.ndim == 2:
waveform = waveform.unsqueeze(0)
if waveform.ndim != 3:
raise ValueError(f"Expected custom audio waveform with 2 or 3 dims, got shape {tuple(waveform.shape)}")
if hasattr(audio_vae, "first_stage_model"):
latent_samples = audio_vae.encode(waveform.movedim(1, -1))
else:
latent_samples = audio_vae.encode({
"waveform": waveform,
"sample_rate": audio_out["sample_rate"],
})
if latent_samples.numel() == 0:
raise ValueError("Encoded audio latent is empty (0 elements).")
B, _C, F_len, H_len = latent_samples.shape
if is_retake_active:
gap_mask = torch.zeros((B, F_len, H_len), dtype=torch.float32, device=latent_samples.device)
retake_start = float(tdata.get("retakeStart", 0))
retake_len = float(tdata.get("retakeLength", 0))
overlap_start = max(start_frame, retake_start)
overlap_end = min(start_frame + ltxv_length, retake_start + retake_len)
if overlap_end > overlap_start:
rel_start = overlap_start - start_frame
rel_len = overlap_end - overlap_start
start_idx, end_idx = _time_range_to_latent_indices(rel_start, rel_len, 0, ltxv_length, frame_rate_f, F_len)
gap_mask[:, start_idx:end_idx, :] = 1.0
else:
gap_mask = torch.ones((B, F_len, H_len), dtype=torch.float32, device=latent_samples.device)
audio_segs_key = "motionSegments" if override_audio else "audioSegments"
file_key = "videoFile" if override_audio else "audioFile"
for seg in tdata.get(audio_segs_key, []):
if not seg.get(file_key):
continue
seg_start = float(seg.get("start", 0))
seg_len = float(seg.get("length", 1))
if seg_start + seg_len <= start_frame or seg_start >= start_frame + ltxv_length:
continue
offset = max(0, start_frame - seg_start)
seg_len = max(1.0, seg_len - offset)
seg_start = max(0, seg_start - start_frame)
start_idx, end_idx = _time_range_to_latent_indices(seg_start, seg_len, 0, ltxv_length, frame_rate_f, F_len)
gap_mask[:, start_idx:end_idx, :] = 0.0
if inpaint_audio:
mask = gap_mask
else:
mask = torch.zeros((B, F_len, H_len), dtype=torch.float32, device=latent_samples.device)
audio_latent = {"samples": latent_samples, "type": "audio", "noise_mask": mask}
log.info("[PromptRelay] Generated custom audio latent with dynamic noise mask.")
return audio_latent
class LTXDirector(io.ComfyNode):
"""WYSIWYG timeline variant — segments and lengths come from a visual editor in the node UI."""
@classmethod
def define_schema(cls):
return io.Schema(
node_id="LTXDirectorCS",
display_name="LTX Director DUMAS",
category="WhatDreamsCost DUMAS",
description=(
"Same as Prompt Relay Encode, but local prompts and segment lengths are edited "
"visually as draggable blocks on a timeline. The duration_frames input only sets the "
"timeline scale (pixel space) — actual frame count is still read from the latent."
),
inputs=[
io.Model.Input("model"),
io.Clip.Input("clip"),
io.Vae.Input("audio_vae", optional=True, tooltip="Optional. Connect an Audio VAE to generate audio latents."),
io.Latent.Input("optional_latent", optional=True, tooltip="Optional. Connect a latent to override the auto-generated one."),
io.String.Input(
"global_prompt", multiline=True, default="", force_input=True, optional=True,
tooltip="Conditions the entire video. Anchors persistent characters, objects, and scene context.",
),
io.Float.Input(
"start_second", default=0.0, min=0.0, max=1000.0, step=0.01,
tooltip="Start time in seconds of the timeline generation.",
),
io.Float.Input(
"end_second", default=5.0, min=0.0, max=1000.0, step=0.01,
tooltip="End time in seconds of the timeline generation.",
),
io.Float.Input(
"duration_seconds", default=5.0, min=0.1, max=1000.0, step=0.01,
tooltip="Total timeline duration in seconds (computed/synced from frames).",
),
io.Int.Input(
"start_frame", default=0, min=0, max=10000, step=1,
tooltip="Start frame of the timeline generation.",
),
io.Int.Input(
"end_frame", default=120, min=1, max=10000, step=1,
tooltip="End frame of the timeline generation.",
),
io.Int.Input(
"duration_frames", default=120, min=1, max=10000, step=1,
tooltip="Total timeline length in pixel-space frames. Used by the editor for visual scale only.",
),
io.String.Input(
"timeline_data", default="",
tooltip="JSON state of the timeline editor (auto-managed; do not edit by hand).",
),
io.Boolean.Input(
"use_custom_audio", default=False, optional=True,
tooltip="Toggle between using timeline audio (ON) and generating audio from scratch (OFF).",
),
io.Boolean.Input(
"use_custom_motion", default=True, optional=True,
tooltip="Toggle between using timeline motion guidance (ON) and ignoring motion video segments (OFF).",
),
io.Boolean.Input(
"inpaint_audio", default=True, optional=True,
tooltip="Toggle whether empty gaps in the audio track are inpainted with generated audio.",
),
io.String.Input(
"local_prompts", multiline=True, default="",
tooltip="Auto-populated from the timeline editor.",
),
io.String.Input(
"segment_lengths", default="",
tooltip="Auto-populated from the timeline editor (pixel-space frame counts).",
),
io.Float.Input(
"epsilon", default=0.001, min=0.0001, max=0.99, step=0.0001,
tooltip="Penalty decay parameter. Values below ~0.1 all produce sharp boundaries (paper default 0.001). For softer transitions, try 0.5 or higher.",
),
io.Float.Input(
"frame_rate", default=24, min=1, max=240, step=1, optional=True,
tooltip="Frames per second — only affects how time is displayed in the timeline editor when time_units is set to 'seconds'.",
),
io.Combo.Input(
"display_mode", options=["frames", "seconds"], default="seconds", optional=True,
tooltip="Display the ruler, segment ranges, length input, and total in frames or seconds. Internal storage is always pixel-space frames.",
),
io.String.Input(
"guide_strength", default="",
tooltip="Auto-populated from the timeline editor (comma-separated guide strengths for image segments).",
),
io.Int.Input(
"custom_width", default=0, min=0, max=8192, step=1, optional=True, force_input=True,
tooltip="Target output width for all image segments. Set to 0 to use the original image width.",
),
io.Int.Input(
"custom_height", default=0, min=0, max=8192, step=1, optional=True, force_input=True,
tooltip="Target output height for all image segments. Set to 0 to use the original image height.",
),
io.Combo.Input(
"resize_method",
options=["maintain aspect ratio", "stretch to fit", "pad", "pad green", "crop"],
default="maintain aspect ratio",
optional=True,
tooltip="How to resize image segments to fit the target dimensions.",
),
io.Int.Input(
"divisible_by", default=32, min=1, max=256, step=1, optional=True,
tooltip="Snap the final output image dimensions to be divisible by this number (e.g. 32 for LTX).",
),
io.Int.Input(
"img_compression", default=18, min=0, max=100, step=1, optional=True,
tooltip="H.264 CRF compression to apply to each guide image. 0 = no compression, higher = more artefacts.",
),
io.Boolean.Input(
"override_audio", default=False, optional=True,
tooltip="Use the audio from the IC-LoRA video instead of using the audio track.",
),
io.Vae.Input(
"vae", optional=True,
tooltip="Optional. Connect the LTX Autoencoder/VAE here to natively encode the MSR visual reference slideshow into the latent prefix. Required for the 'Licon MSR' ref option.",
),
io.Float.Input(
"reference_strength", default=1.0, min=0.0, max=5.0, step=0.05, optional=True,
tooltip="Guide strength applied to the character reference images when using Licon MSR.",
),
MSRCharacterSetData.Input(
"character_set",
optional=True,
tooltip="Optional external MSR Character Set. Replaces the legacy inline character panel in the director UI.",
),
TimelineSectionSetData.Input(
"section_set",
optional=True,
tooltip="Optional external Timeline Section Set DUMAS payload. When connected, it populates the main timeline track and derives the duration from the sections.",
),
io.Float.Input(
"start", force_input=True, optional=True, default=0.0,
tooltip="Automation (connection-only). Start time in SECONDS. Overrides the panel Start when connected.",
),
io.Float.Input(
"end", force_input=True, optional=True, default=0.0,
tooltip="Automation (connection-only). End time in SECONDS. When connected (and duration is not), the render length is derived from start..end.",
),
io.Float.Input(
"duration", force_input=True, optional=True, default=0.0,
tooltip="Automation (connection-only). Duration in SECONDS. Overrides the panel Duration and sets the render length when connected.",
),
],
outputs=[
io.Model.Output(display_name="model"),
io.Conditioning.Output(display_name="positive"),
io.Conditioning.Output(display_name="negative", tooltip="Negative conditioning emitted by the Director (neutral/empty). Wire to LTX Director Guide's 'negative' input."),
io.Latent.Output(display_name="video_latent", tooltip="Auto-generated LTXV empty latent (only populated when no latent is connected)."),
io.Latent.Output(display_name="audio_latent", tooltip="Auto-generated audio latent (uses custom audio if enabled)."),
GuideData.Output(display_name="guide_data"),
MotionGuideData.Output(display_name="motion_guide_data"),
io.Float.Output(display_name="frame_rate", tooltip="The frame rate used for the timeline."),
io.Audio.Output(display_name="combined_audio", tooltip="Combined timeline audio layout."),
io.Int.Output(display_name="clean_latent_frames", tooltip="Number of clean (visible) latent frames. Plug into Clean Latent Slice 'length'."),
io.Int.Output(display_name="clean_pixel_frames", tooltip="Number of clean (visible) pixel frames = (clean_latent_frames - 1) * 8 + 1."),
],
)
@classmethod
def execute(cls, model, clip, start_second, end_second, duration_seconds, start_frame, end_frame, duration_frames,
timeline_data, local_prompts, segment_lengths, global_prompt="", guide_strength="", epsilon=1e-3,
frame_rate=24, display_mode="seconds",
custom_width=768, custom_height=512, resize_method="maintain aspect ratio",
divisible_by=32, img_compression=0, audio_vae=None, optional_latent=None,
use_custom_audio=False, inpaint_audio=True, use_custom_motion=True, override_audio=False,
vae=None, reference_strength=1.0, character_set=None, section_set=None,
start=None, end=None, duration=None) -> io.NodeOutput:
input_dir = folder_paths.get_input_directory()
image_cache = {}
processed_image_cache = {}
def prepare_tensor_image(tensor: torch.Tensor, width: int, height: int) -> torch.Tensor:
cache_key = (id(tensor), width, height, resize_method, divisible_by, img_compression)
cached = processed_image_cache.get(cache_key)
if cached is not None:
return cached
processed = _resize_image(tensor, width, height, resize_method, divisible_by)
if img_compression > 0:
processed = _compress_image(processed, img_compression)
processed_image_cache[cache_key] = processed
return processed
if section_set is not None:
external_tdata, external_local_prompts, external_segment_lengths, external_duration_seconds, external_duration_frames = _build_runtime_timeline_payload(
{"sections": list((section_set or {}).get("sections") or [])},
float(frame_rate) if frame_rate else 24.0,
)
try:
existing_tdata = _safe_json_loads(timeline_data)
except Exception:
existing_tdata = {}
merged_tdata = dict(existing_tdata or {})
merged_tdata["segments"] = external_tdata["segments"]
merged_tdata.setdefault("motionSegments", existing_tdata.get("motionSegments", []))
merged_tdata.setdefault("audioSegments", existing_tdata.get("audioSegments", []))
timeline_data = json.dumps(_make_json_safe(merged_tdata))
local_prompts = external_local_prompts
segment_lengths = external_segment_lengths
if duration is None:
duration_seconds = external_duration_seconds
duration_frames = external_duration_frames
else:
merged_tdata = None
# Parse timeline data
if merged_tdata is not None:
tdata = merged_tdata
else:
try:
tdata = _safe_json_loads(timeline_data)
except Exception as e:
log.error(f"[LTXDirector] execute timeline_data parse error: {e}")
tdata = {}
# --- Automation overrides (connection-only inputs, in SECONDS) ---
# When wired, these take precedence over the panel/timeline values.
_auto_fps = float(frame_rate) if frame_rate else 24.0
if start is not None:
start_second = float(start)
start_frame = int(round(start_second * _auto_fps))
if end is not None:
end_second = float(end)
end_frame = int(round(end_second * _auto_fps))
if duration is not None:
duration_seconds = float(duration)
duration_frames = max(1, int(round(duration_seconds * _auto_fps)))
elif end is not None:
# Derive the window length from start..end when duration is not explicitly wired.
duration_frames = max(1, end_frame - start_frame)
duration_seconds = duration_frames / _auto_fps
frame_rate_f = float(frame_rate)
is_retake_mode = tdata.get("retakeMode", False)
is_retake_active = is_retake_mode and tdata.get("retakeVideo") is not None
# Extract global_prompt from timeline_data if not connected/empty
if not global_prompt:
if is_retake_mode:
global_prompt = tdata.get("retake_global_prompt", "")
else:
global_prompt = tdata.get("global_prompt", "")
log.info(f"[LTXDirector] execute RAW global_prompt: {repr(global_prompt)}")
# --- Reference option (set by the toolbar "Ref Option" dropdown, stored in timeline JSON) ---
# One of: "Licon MSR (Prefix)", "OFF".
reference_mode = tdata.get("reference_mode", "OFF")
if reference_mode not in {"Licon MSR (Prefix)", "OFF"}:
reference_mode = "OFF"
tdata["reference_mode"] = reference_mode
characters = _extract_runtime_character_entries(
tdata=tdata,
character_set=character_set,
image_cache=image_cache,
input_dir=input_dir,
)
char_slot_images = [list(character.get("images") or []) for character in characters]
char_images = [img for slot_images in char_slot_images for img in slot_images]
# --- @charN substitution ---
# OFF: swap @char tags for their VLM descriptions in the prompt text.
# Licon MSR: leave the tags raw — there the reference IMAGE drives identity, and the
# tags are used only to pick which character slots feed the slideshow.
if reference_mode == "Licon MSR (Prefix)":
ref_global, ref_local = global_prompt, local_prompts
else:
ref_global, ref_local = _preprocess_prompts_with_characters(
global_prompt, local_prompts, characters
)
global_prompt, local_prompts = ref_global, ref_local
guide_data, derived_w, derived_h = _build_guide_data_from_timeline(
tdata=tdata,
start_frame=start_frame,
duration_frames=duration_frames,
frame_rate_f=frame_rate_f,
guide_strength=guide_strength,
custom_width=custom_width,
custom_height=custom_height,
resize_method=resize_method,
divisible_by=divisible_by,
img_compression=img_compression,
optional_latent=optional_latent,
input_dir=input_dir,
image_cache=image_cache,
)
# --- Auto-generate LTXV latent if none was provided ---
# Apply the community 8n+1 rule directly to the timeline's duration_frames:
# int(ceil(((duration_frames) - 1) / 8) * 8) + 1
# This ensures we get AT LEAST the requested frames, snapped to LTXV's requirements.
ltxv_length = int(math.ceil((duration_frames - 1) / 8.0) * 8) + 1
latent_w = max(32, (derived_w // 32) * 32)
latent_h = max(32, (derived_h // 32) * 32)
latent_grid_h = latent_h // 32
latent_grid_w = latent_w // 32
# Clean (visible) region: latent frames and the matching pixel frames.
clean_latent_frames = ((ltxv_length - 1) // 8) + 1
clean_pixel_frames = int(ltxv_length)
# Negative conditioning emitted on the "negative" output (slot 2, right under positive).
# The Director no longer takes a negative input; it emits a neutral EMPTY negative so the
# downstream guide always has a valid required input. Wire your own Negative Prompt node
# straight into the guide if you want custom negative text.
neg_tokens = clip.tokenize("")
conditioning_neg = clip.encode_from_tokens_scheduled(neg_tokens)
# Fall back to the first local prompt as the global anchor if no global prompt was given.
if not (global_prompt or "").strip() and local_prompts:
global_prompt = local_prompts.split("|")[0].strip()
log.info(f"[LTXDirector] execute EFFECTIVE global_prompt: {repr(global_prompt)}")
_dev = comfy.model_management.intermediate_device()
patched, conditioning, latent, guide_data = _build_reference_mode_outputs(
reference_mode=reference_mode,
vae=vae,
global_prompt=global_prompt,
local_prompts=local_prompts,
segment_lengths=segment_lengths,
duration_frames=duration_frames,
epsilon=epsilon,
characters=characters,
char_images=char_images,
char_slot_images=char_slot_images,
guide_data=guide_data,
latent_w=latent_w,
latent_h=latent_h,
latent_grid_h=latent_grid_h,
latent_grid_w=latent_grid_w,
clean_latent_frames=clean_latent_frames,
reference_strength=reference_strength,
optional_latent=optional_latent,
resize_method=resize_method,
divisible_by=divisible_by,
prepare_tensor_image=prepare_tensor_image,
model=model,
clip=clip,
conditioning_neg=conditioning_neg,
_dev=_dev,
)
# --- Build Audio Output ---
audio_out = _build_combined_audio(tdata, start_frame, ltxv_length, frame_rate_f, override_audio=override_audio)
try:
audio_latent = _build_audio_latent(
audio_vae=audio_vae,
audio_out=audio_out,
use_custom_audio=use_custom_audio,
override_audio=override_audio,
is_retake_active=is_retake_active,
inpaint_audio=inpaint_audio,
tdata=tdata,
start_frame=start_frame,
ltxv_length=ltxv_length,
frame_rate_f=frame_rate_f,
)
except Exception as e:
log.error("[PromptRelay] Failed to generate custom audio latent: %s", e)
raise e
motion_guide_data = _build_motion_guide_data(
tdata=tdata,
use_custom_motion=use_custom_motion,
start_frame=start_frame,
duration_frames=duration_frames,
frame_rate_f=frame_rate_f,
resize_method=resize_method,
)
# Inject raw timeline details for downstream masking in Retake Mode
guide_data["timeline_data"] = timeline_data
guide_data["start_frame"] = start_frame
guide_data["duration_frames"] = duration_frames
guide_data["resize_method"] = resize_method
return io.NodeOutput(patched, conditioning, conditioning_neg, latent, audio_latent, guide_data, motion_guide_data, frame_rate_f, audio_out, int(clean_latent_frames), int(clean_pixel_frames))
NODE_CLASS_MAPPINGS = {
"LTXDirectorCS": LTXDirector,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"PromptRelayEncodeTimeline": "Prompt Relay Encode (Timeline)",
}