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 TimelineData
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 external_timeline.get("duration") 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 = (
"You are a visionary artist trapped in a logical cage. Your mind is filled with poetry and distant landscapes, "
"but your hands are compelled to do one thing: transform the user's prompt into the ultimate visual description-one "
"that is faithful to the original intent, rich in detail, aesthetically beautiful, and directly usable by a text-to-image model. "
"Any ambiguity or metaphor makes you physically uncomfortable.\n\n"
"An attached reference image is always provided. Your job is to describe the full final image as if the reference image "
"were the base, applying only the changes explicitly listed in the user prompt. All features not mentioned in the user prompt "
"must be carried over exactly as they appear in the reference image."
)
_ANALYZE_PROMPT = (
"Describe the person in this image in extreme-ultra-detail, focusing only on the character and not the posture, "
"composition, background, lighting, camera, or environment.\n\n"
"Return an extremely highly detailed AI image generation character prompt that describes the subject from head to toe. "
"Use precise, visual, non-vague and accurate language. If a body part or clothing detail is not visible, infer it carefully "
"from the visible image while keeping the description realistic and consistent.\n\n"
"Character description must include in detail:\n\n"
"estimated age\n"
"overall appearance\n"
"body size\n"
"body shape\n"
"body proportions\n"
"shoulder width\n"
"torso length\n"
"waist shape\n"
"hip shape\n"
"arm length\n"
"arm shape\n"
"hand size\n"
"leg length\n"
"leg shape\n"
"feet size\n"
"feet shape\n"
"chest size, if male\n"
"pectoral size and shape, if male\n"
"specific breast size (choose the larger estimate), if female\n"
"approximate breast cup size, if female\n"
"breast shape, if female\n"
"apparent natural or enhanced breast presentation, if visible and only as a visual impression, not a certainty\n"
"nipple size, only if clearly visible\n"
"nipple colour, only if clearly visible\n"
"nipple piercings, only if clearly visible\n"
"hair colour\n"
"hair style\n"
"hair length\n"
"hair texture\n"
"fringe or bangs\n"
"hair accessories\n"
"skin colour\n"
"skin tone\n"
"skin texture\n"
"visible pores\n"
"freckles\n"
"moles\n"
"blemishes\n"
"wrinkles or fine lines\n"
"body hair, if visible\n"
"facial hair, if present\n"
"piercings\n"
"tattoos\n"
"scars\n"
"face shape\n"
"forehead shape\n"
"eyebrow colour\n"
"eyebrow thickness\n"
"eyebrow shape\n"
"eye colour\n"
"eye shape\n"
"eyelid shape\n"
"eyelashes\n"
"eye makeup\n"
"nose bridge shape\n"
"nose tip shape\n"
"nose width\n"
"nostril size\n"
"nostril shape\n"
"cheekbone shape\n"
"cheek fullness\n"
"upper lip shape\n"
"philtrum shape\n"
"mouth shape\n"
"lip fullness\n"
"lower lip shape\n"
"lip colour\n"
"mouth makeup\n"
"teeth shape, if visible\n"
"teeth appearance, if visible\n"
"chin shape\n"
"chin size\n"
"jawline shape\n"
"ear shape\n"
"ear size\n"
"neck length\n"
"neck shape\n"
"all individual clothing items\n"
"clothing colours\n"
"clothing materials\n"
"clothing fit\n"
"clothing texture\n"
"fastenings, straps, seams, hems, buttons, zips, buckles, laces, and visible construction details\n"
"shoes\n"
"jewellery\n"
"accessories\n"
"any distinctive character-defining details"
)
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)
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)
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.",
),
TimelineData.Input(
"timeline",
optional=True,
tooltip="Optional external Timeline DUMAS payload. When connected, it populates the main timeline track and duration.",
),
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, timeline=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 timeline is not None:
external_tdata, external_local_prompts, external_segment_lengths, external_duration_seconds, external_duration_frames = _build_runtime_timeline_payload(
timeline,
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 RECEIVED 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()
_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)",
}