2049 lines
86 KiB
Python
2049 lines
86 KiB
Python
import logging
|
|
import asyncio
|
|
import json
|
|
import base64
|
|
import io as _io
|
|
import math
|
|
|
|
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
|
|
|
|
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 _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_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 _extract_ollama_generated_text(resp_json: dict) -> str:
|
|
"""Handle Ollama generate/chat variants and reasoning-capable models."""
|
|
if not isinstance(resp_json, dict):
|
|
return ""
|
|
|
|
candidates = []
|
|
candidates.append(resp_json.get("response"))
|
|
candidates.append(resp_json.get("thinking"))
|
|
candidates.append(resp_json.get("content"))
|
|
|
|
message = resp_json.get("message")
|
|
if isinstance(message, dict):
|
|
candidates.append(message.get("content"))
|
|
candidates.append(message.get("reasoning_content"))
|
|
candidates.append(message.get("thinking"))
|
|
|
|
for candidate in candidates:
|
|
if isinstance(candidate, str) and candidate.strip():
|
|
text = candidate.strip()
|
|
if "<think>" in text:
|
|
text = text.split("</think>")[-1].strip()
|
|
if text:
|
|
return text
|
|
return ""
|
|
|
|
|
|
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", "")
|
|
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":
|
|
payload = {
|
|
"model": model_name, "prompt": _ANALYZE_PROMPT,
|
|
"images": cleaned_b64_list, "stream": False, "keep_alive": 0,
|
|
}
|
|
async with session.post(f"{base_url}/api/generate", json=payload, timeout=300) as response:
|
|
if response.status != 200:
|
|
err_txt = await response.text()
|
|
return web.json_response({"status": "error", "message": f"Ollama HTTP {response.status}: {err_txt}"})
|
|
resp_json = await response.json()
|
|
generated_text = _extract_ollama_generated_text(resp_json)
|
|
|
|
# Some Ollama model variants return an empty `response` from `/api/generate`
|
|
# even though the same request succeeds via the chat endpoint.
|
|
if not generated_text:
|
|
chat_payload = {
|
|
"model": model_name,
|
|
"messages": [{
|
|
"role": "user",
|
|
"content": _ANALYZE_PROMPT,
|
|
"images": cleaned_b64_list,
|
|
}],
|
|
"stream": False,
|
|
"keep_alive": 0,
|
|
}
|
|
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()
|
|
generated_text = _extract_ollama_generated_text(resp_json)
|
|
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": "user", "content": content}],
|
|
"max_tokens": 2048, "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 "<think>" in generated_text:
|
|
generated_text = generated_text.split("</think>")[-1].strip()
|
|
if not generated_text:
|
|
return web.json_response({
|
|
"status": "error",
|
|
"message": f"{provider} returned an empty analysis response.",
|
|
})
|
|
|
|
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]."""
|
|
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"))
|
|
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:
|
|
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,
|
|
ref_images,
|
|
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
|
|
|
|
extra_refs = []
|
|
if reference_mode == "Ghost Mask (End)" and ref_images is not None:
|
|
try:
|
|
for bi in range(ref_images.shape[0]):
|
|
extra_refs.append(ref_images[bi:bi + 1])
|
|
except Exception as e:
|
|
log.warning("[LTXDirector] Could not process ref_images input: %s", e)
|
|
|
|
ghost_refs = char_images + extra_refs
|
|
is_ghost = reference_mode == "Ghost Mask (End)" and bool(ghost_refs)
|
|
n_refs = len(ghost_refs) if is_ghost else 0
|
|
|
|
if is_ghost:
|
|
for i, single_ref in enumerate(ghost_refs):
|
|
ref_tensor = prepare_tensor_image(single_ref, latent_w, latent_h)
|
|
guide_data["images"].append(ref_tensor)
|
|
guide_data["insert_frames"].append((clean_latent_frames + i) * 8)
|
|
guide_data["strengths"].append(float(reference_strength))
|
|
guide_data["ghost_pre_extend"] = int(n_refs)
|
|
|
|
total_latent_frames = clean_latent_frames + n_refs
|
|
tail_pixels = n_refs * 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
|
|
|
|
log.info(
|
|
"[LTXDirector] Ghost Mask: %d references; guide will hide them in a tail past "
|
|
"frame %d (clean_latent_frames=%d). Enable Clean Latent Slice (length=%d) downstream.",
|
|
n_refs, clean_latent_frames, clean_latent_frames, clean_latent_frames,
|
|
)
|
|
return patched, conditioning, latent, 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,
|
|
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,
|
|
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 (Ghost Mask / Licon MSR ref options).",
|
|
),
|
|
io.Image.Input(
|
|
"ref_images", optional=True,
|
|
tooltip="Ghost Mask only. Extra reference image(s) (e.g. an object) — a single image or a batch. Appended as their own block of hidden reference frames AFTER the @char references, for any number of characters loaded (0-6). Ignored in MSR / OFF modes.",
|
|
),
|
|
MSRCharacterSetData.Input(
|
|
"character_set",
|
|
optional=True,
|
|
tooltip="Optional external MSR Character Set. Replaces the legacy inline character panel in the director UI.",
|
|
),
|
|
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, ref_images=None, character_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
|
|
|
|
# Parse timeline data
|
|
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: "Ghost Mask (End)", "Licon MSR (Prefix)", "OFF".
|
|
reference_mode = tdata.get("reference_mode", "OFF")
|
|
|
|
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 ---
|
|
# Ghost Mask / 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,
|
|
ref_images=ref_images,
|
|
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)",
|
|
}
|