Release standalone timeline editor 1.0.0

This commit is contained in:
Hopping Mad Games
2026-08-26 09:55:12 -06:00
commit 869537cefd
27 changed files with 7739 additions and 0 deletions
+6
View File
@@ -0,0 +1,6 @@
"""Timeline editor backend."""
from .editor import ETKLTXVTimelineImageEditor
__all__ = ["ETKLTXVTimelineImageEditor"]
+229
View File
@@ -0,0 +1,229 @@
"""Pure timeline schedule helpers for ETK LTXV timeline image editor nodes."""
import ast
import json
ETK_LTXV_TIMELINE_SCHEMA_VERSION = 2
def _etk_ltxv_timeline_payload_keyframes(payload):
schema_version = None
if isinstance(payload, dict):
raw_version = payload.get("schemaVersion", payload.get("schema_version", None))
if raw_version is not None:
try:
schema_version = int(raw_version)
except (TypeError, ValueError) as exc:
raise ValueError("timeline_json schemaVersion must be an integer") from exc
if schema_version < 1 or schema_version > ETK_LTXV_TIMELINE_SCHEMA_VERSION:
raise ValueError(f"unsupported timeline_json schemaVersion {schema_version}")
payload = payload.get("keyframes", [])
if not isinstance(payload, list):
raise ValueError("timeline_json must be a list or an object with a keyframes list")
return payload, schema_version
def _parse_etk_prompt_schedule(prompt_schedule):
if prompt_schedule is None:
return []
if isinstance(prompt_schedule, str):
text = prompt_schedule.strip()
if not text:
return []
try:
payload = json.loads(text)
except json.JSONDecodeError:
try:
payload = ast.literal_eval(text)
except (SyntaxError, ValueError) as exc:
raise ValueError(
"prompt_schedule must be a JSON/Python list of "
"(image_index, seconds, positive_prompt, latent_strength, prompt_strength) "
"or legacy (seconds, positive_prompt, latent_strength, prompt_strength) tuples"
) from exc
else:
payload = prompt_schedule
if isinstance(payload, dict):
payload = payload.get("keyframes", payload.get("schedule", []))
if not isinstance(payload, (list, tuple)):
raise ValueError("prompt_schedule must be a list")
schedule = []
for index, item in enumerate(payload):
image_index = None
if isinstance(item, dict):
image_index = item.get("image_index", item.get("imageIndex", item.get("image", None)))
seconds = item.get("seconds", item.get("time", item.get("t", item.get("position", item.get("keyframe_position_seconds")))))
prompt = item.get("positive_prompt", item.get("positivePrompt", item.get("prompt", "")))
latent_strength = item.get("latent_strength", item.get("latentStrength", item.get("strength", 1.0)))
prompt_strength = item.get("prompt_strength", item.get("promptStrength", 1.0))
elif isinstance(item, (list, tuple)) and len(item) >= 2:
has_image_index = False
if len(item) >= 3:
try:
float(item[1])
has_image_index = True
except (TypeError, ValueError):
has_image_index = False
if has_image_index:
image_index = item[0]
seconds = item[1]
prompt = item[2]
latent_strength = item[3] if len(item) >= 4 else 1.0
prompt_strength = item[4] if len(item) >= 5 else 1.0
else:
seconds = item[0]
prompt = item[1]
latent_strength = item[2] if len(item) >= 3 else 1.0
prompt_strength = item[3] if len(item) >= 4 else 1.0
else:
raise ValueError(f"prompt_schedule item {index} must be a tuple/list or dict")
try:
seconds = max(0.0, float(seconds))
if image_index is not None:
image_index = int(image_index)
if image_index < 0:
raise ValueError
latent_strength = max(0.0, min(1.0, float(latent_strength)))
prompt_strength = max(0.0, min(4.0, float(prompt_strength)))
except (TypeError, ValueError) as exc:
raise ValueError(f"prompt_schedule item {index} has invalid image index, seconds, or strength") from exc
prompt = str(prompt or "").strip()
if not prompt:
continue
schedule.append({
"image_index": image_index,
"seconds": seconds,
"positive_prompt": prompt,
"latent_strength": latent_strength,
"prompt_strength": prompt_strength,
})
return schedule
def _apply_etk_prompt_schedule_override(keyframes, prompt_schedule, fps, latent_frames, reset_existing=False):
schedule = _parse_etk_prompt_schedule(prompt_schedule)
if not schedule:
return keyframes, False
try:
fps = max(1e-6, float(fps))
except (TypeError, ValueError) as exc:
raise ValueError("prompt_schedule_fps must be a number") from exc
latent_frames = max(1, int(latent_frames))
merged = [] if reset_existing else [dict(keyframe) for keyframe in keyframes]
by_slot = {}
for keyframe in merged:
slot = max(0, min(latent_frames - 1, int(keyframe["slot"])))
keyframe["slot"] = slot
keyframe["frame"] = slot * 8
by_slot[slot] = keyframe
for entry in schedule:
frame = int(round(entry["seconds"] * fps))
slot = max(0, min(latent_frames - 1, int((frame / 8.0) + 0.5)))
keyframe = by_slot.get(slot)
if keyframe is None:
keyframe = {
"frame": slot * 8,
"slot": slot,
"fit_mode": "crop",
"pan_x": 0.0,
"pan_y": 0.0,
"zoom": 1.0,
"image": None,
"layers": [],
"input_image": False,
"positive_prompt": "",
"negative_prompt": "",
"prompt_strength": 1.0,
"latent_strength": None,
}
merged.append(keyframe)
by_slot[slot] = keyframe
keyframe["positive_prompt"] = entry["positive_prompt"]
keyframe["latent_strength"] = entry["latent_strength"]
keyframe["prompt_strength"] = entry["prompt_strength"]
keyframe.pop("time", None)
if entry.get("image_index") is None:
keyframe.pop("image_index", None)
else:
keyframe["image_index"] = entry["image_index"]
merged = [
keyframe
for keyframe in merged
if keyframe.get("image")
or keyframe.get("layers")
or keyframe.get("_input_image") is not None
or keyframe.get("input_image")
or str(keyframe.get("positive_prompt", "") or "").strip()
or str(keyframe.get("negative_prompt", "") or "").strip()
]
return sorted(merged, key=lambda item: item["slot"]), True
def _etk_timeline_keyframes_for_ui(keyframes, fps=25.0):
try:
fps = max(0.001, float(fps))
except (TypeError, ValueError):
fps = 25.0
ui_keyframes = []
for keyframe in sorted(keyframes, key=lambda item: item["slot"]):
ui_keyframe = {
"time": float(keyframe.get("time", int(keyframe["frame"]) / fps)),
"fitMode": keyframe.get("fit_mode", "crop"),
"panX": float(keyframe.get("pan_x", 0.0)),
"panY": float(keyframe.get("pan_y", 0.0)),
"zoom": float(keyframe.get("zoom", 1.0)),
"brightness": float(keyframe.get("brightness", 1.0)),
"contrast": float(keyframe.get("contrast", 1.0)),
"positivePrompt": str(keyframe.get("positive_prompt", "") or ""),
"negativePrompt": str(keyframe.get("negative_prompt", "") or ""),
"promptStrength": float(keyframe.get("prompt_strength", 1.0)),
}
if keyframe.get("latent_strength") is not None:
ui_keyframe["latentStrength"] = float(keyframe.get("latent_strength", 1.0))
if keyframe.get("image"):
ui_keyframe["image"] = keyframe["image"]
if keyframe.get("layers"):
ui_keyframe["layers"] = []
for layer in keyframe.get("layers", []):
ui_layer = {
"id": str(layer.get("id", "") or ""),
"fitMode": layer.get("fit_mode", "crop"),
"panX": float(layer.get("pan_x", 0.0)),
"panY": float(layer.get("pan_y", 0.0)),
"zoom": float(layer.get("zoom", 1.0)),
"brightness": float(layer.get("brightness", 1.0)),
"contrast": float(layer.get("contrast", 1.0)),
}
if layer.get("type") == "text":
ui_layer.update({
"type": "text",
"text": str(layer.get("text", "Text") or ""),
"fontFamily": str(layer.get("font_family", layer.get("fontFamily", "sans-serif")) or "sans-serif"),
"fontSize": float(layer.get("font_size", layer.get("fontSize", 72.0))),
"bold": bool(layer.get("bold", False)),
"italic": bool(layer.get("italic", False)),
"align": str(layer.get("align", layer.get("textAlign", "center")) or "center"),
"color": str(layer.get("color", "#ffffff") or "#ffffff"),
"outlineColor": str(layer.get("outline_color", layer.get("outlineColor", "#000000")) or "#000000"),
"outlineWidth": float(layer.get("outline_width", layer.get("outlineWidth", 0.0))),
"shadowColor": str(layer.get("shadow_color", layer.get("shadowColor", "#000000")) or "#000000"),
"shadowBlur": float(layer.get("shadow_blur", layer.get("shadowBlur", 0.0))),
"shadowOffsetX": float(layer.get("shadow_offset_x", layer.get("shadowOffsetX", 0.0))),
"shadowOffsetY": float(layer.get("shadow_offset_y", layer.get("shadowOffsetY", 0.0))),
})
elif layer.get("image"):
ui_layer["image"] = layer["image"]
else:
ui_layer["empty"] = True
ui_keyframe["layers"].append(ui_layer)
if ui_keyframe["layers"]:
ui_keyframe["selectedLayerId"] = ui_keyframe["layers"][0]["id"]
if keyframe.get("_input_image") is not None or keyframe.get("input_image"):
ui_keyframe["inputImage"] = True
ui_keyframes.append(ui_keyframe)
return ui_keyframes
+305
View File
@@ -0,0 +1,305 @@
"""Standalone LTXV timeline image-editor ComfyUI node."""
import math
from .images import (
_etk_scalar_input,
_etk_send_timeline_settings_to_ui,
_etk_timeline_editor_compat_inputs,
_etk_timeline_result_with_ui,
_etk_timeline_with_connected_images,
_etk_unwrap_single_input,
)
from .rendering import (
_etk_timeline_image_batch,
_etk_timeline_unedited_source_batch,
_fit_etk_timeline_keyframe_image,
_load_etk_timeline_keyframe_unedited_sources,
)
from .schema import _parse_etk_ltxv_timeline
def _etk_keyframe_latent_strength(keyframe, global_strength):
latent_strength = keyframe.get("latent_strength", None)
if latent_strength is None:
latent_strength = global_strength
try:
latent_strength = float(latent_strength)
except (TypeError, ValueError):
latent_strength = float(global_strength)
return max(0.0, min(1.0, latent_strength))
def _etk_timeline_list_input(name, value):
"""Unwrap one Comfy list-transport layer without collapsing singleton lists."""
if value is None:
return None
if not isinstance(value, (list, tuple)):
raise ValueError(f"{name} must be a LIST")
values = list(value)
if len(values) == 1 and isinstance(values[0], (list, tuple)):
values = list(values[0])
return values
def _etk_timeline_time_fields(value, index, fps, latent_frames):
try:
time_seconds = float(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"times[{index}] must be a number of seconds") from exc
if not math.isfinite(time_seconds) or time_seconds < 0.0:
raise ValueError(f"times[{index}] must be a finite, non-negative number of seconds")
frame = int(round(time_seconds * fps))
max_frame = max(0, (int(latent_frames) - 1) * 8)
if frame > max_frame:
raise ValueError(
f"times[{index}]={time_seconds} seconds maps to frame {frame}, "
f"outside video frame range 0..{max_frame}"
)
return time_seconds, frame, int((frame / 8.0) + 0.5)
def _etk_empty_timeline_keyframe(slot, fps):
frame = int(slot) * 8
return {
"frame": frame,
"slot": int(slot),
"time": frame / fps,
"fit_mode": "crop",
"pan_x": 0.0,
"pan_y": 0.0,
"zoom": 1.0,
"brightness": 1.0,
"contrast": 1.0,
"image": None,
"layers": [],
"selected_layer_id": "",
"input_image": False,
"positive_prompt": "",
"negative_prompt": "",
"prompt_strength": 1.0,
"latent_strength": None,
}
def _etk_apply_timeline_list_inputs(keyframes, prompts, strengths, times, fps, latent_frames):
supplied = {
"prompts": _etk_timeline_list_input("prompts", prompts),
"strengths": _etk_timeline_list_input("strengths", strengths),
"times": _etk_timeline_list_input("times", times),
}
provided = {name: values for name, values in supplied.items() if values is not None}
if not provided:
return keyframes, None
lengths = {name: len(values) for name, values in provided.items()}
if len(set(lengths.values())) != 1:
details = ", ".join(f"{name}={count}" for name, count in lengths.items())
raise ValueError(f"timeline list inputs must have equal lengths; got {details}")
item_count = next(iter(lengths.values()))
merged = [dict(keyframe) for keyframe in keyframes]
if len(merged) > item_count:
raise ValueError(
f"timeline list inputs contain {item_count} items, but timeline_json contains "
f"{len(merged)} keyframes"
)
occupied_slots = {int(keyframe["slot"]) for keyframe in merged}
cursor = max(occupied_slots, default=-1) + 1
while len(merged) < item_count:
index = len(merged)
if supplied["times"] is not None:
time_seconds, frame, slot = _etk_timeline_time_fields(
supplied["times"][index], index, fps, latent_frames
)
keyframe = _etk_empty_timeline_keyframe(slot, fps)
keyframe.update({"time": time_seconds, "frame": frame, "slot": slot})
else:
while cursor in occupied_slots:
cursor += 1
if cursor >= latent_frames:
raise ValueError(
f"timeline list inputs require {item_count} keyframes, but no free latent "
f"slot remains within 0..{latent_frames - 1}"
)
keyframe = _etk_empty_timeline_keyframe(cursor, fps)
cursor += 1
occupied_slots.add(int(keyframe["slot"]))
merged.append(keyframe)
for index, keyframe in enumerate(merged):
if supplied["prompts"] is not None:
prompt = supplied["prompts"][index]
if isinstance(prompt, (list, tuple, dict)):
raise ValueError(f"prompts[{index}] must be a string value")
keyframe["positive_prompt"] = str(prompt if prompt is not None else "")
if supplied["strengths"] is not None:
try:
strength = float(supplied["strengths"][index])
except (TypeError, ValueError) as exc:
raise ValueError(f"strengths[{index}] must be a number") from exc
if not math.isfinite(strength):
raise ValueError(f"strengths[{index}] must be finite")
keyframe["latent_strength"] = max(0.0, min(1.0, strength))
if supplied["times"] is not None:
time_seconds, frame, slot = _etk_timeline_time_fields(
supplied["times"][index], index, fps, latent_frames
)
keyframe.update({"time": time_seconds, "frame": frame, "slot": slot})
return sorted(merged, key=lambda item: (item["slot"], item["frame"])), item_count
class ETKLTXVTimelineImageEditor:
DISPLAY_NAME = "ETK LTXV Timeline Image Editor"
DESCRIPTION = (
"Standalone LTXV/LTX 2.x image timeline editor. It uses the same "
"timeline image-editing UI as the all-in-one PromptRelay node, then "
"emits composited preview images, matching unedited source images, "
"per-keyframe guide strengths, frame indexes, and normalized positive "
"prompts without loading a model, CLIP, or VAE. Optional prompt, "
"strength, and time lists replace keyframe metadata by index."
)
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"width": ("INT", {"default": 768, "min": 64, "max": 8192, "step": 32}),
"height": ("INT", {"default": 512, "min": 64, "max": 8192, "step": 32}),
"length": ("INT", {"default": 97, "min": 1, "max": 8192, "step": 8}),
"timeline_json": ("STRING", {"default": "[]", "multiline": True}),
"fps": ("FLOAT", {"default": 25.0, "min": 0.001, "max": 240.0, "step": 0.01}),
},
"optional": {
"images": ("IMAGE",),
"ui_instance_id": ("STRING", {"default": "", "multiline": False}),
"prompts": ("LIST", {
"tooltip": "Positive prompts aligned one-to-one with timeline keyframes/images.",
}),
"strengths": ("LIST", {
"tooltip": "Image guide strengths aligned one-to-one with timeline keyframes/images.",
}),
"times": ("LIST", {
"tooltip": "Keyframe times in seconds, aligned one-to-one with timeline keyframes/images.",
}),
},
"hidden": {
"unique_id": "UNIQUE_ID",
},
}
RETURN_TYPES = ("IMAGE", "IMAGE", "INT", "INT", "INT", "LIST", "LIST", "LIST", "LIST")
RETURN_NAMES = (
"edited_images",
"unedited_images",
"width",
"height",
"length",
"strengths",
"frame_indexes",
"position_seconds",
"prompts",
)
INPUT_IS_LIST = True
OUTPUT_IS_LIST = (False, False, False, False, False, False, False, False, False)
FUNCTION = "generate"
CATEGORY = "ETK/Image"
def generate(
self,
width,
height,
length,
timeline_json="[]",
fps=25.0,
strength=1.0,
images=None,
ui_instance_id="",
unique_id=None,
prompts=None,
strengths=None,
times=None,
):
width = _etk_scalar_input(width)
height = _etk_scalar_input(height)
length = _etk_scalar_input(length)
timeline_json = _etk_scalar_input(timeline_json)
fps = _etk_scalar_input(fps)
strength = _etk_scalar_input(strength)
images = _etk_unwrap_single_input(images)
ui_instance_id = _etk_scalar_input(ui_instance_id)
unique_id = _etk_scalar_input(unique_id)
width = max(64, int(round(int(width) / 32)) * 32)
height = max(64, int(round(int(height) / 32)) * 32)
length = max(1, int(length))
if (length - 1) % 8 != 0:
raise ValueError("LTXV video length must be 1 plus a multiple of 8")
try:
fps = max(0.001, float(fps))
except (TypeError, ValueError) as exc:
raise ValueError("timeline fps must be a number") from exc
timeline_json, strength = _etk_timeline_editor_compat_inputs(timeline_json, strength)
latent_frames = ((length - 1) // 8) + 1
keyframes, list_item_count = _etk_apply_timeline_list_inputs(
_parse_etk_ltxv_timeline(timeline_json, include_empty=True, fps=fps),
prompts,
strengths,
times,
fps,
latent_frames,
)
keyframes = _etk_timeline_with_connected_images(
keyframes,
images,
latent_frames,
)
if list_item_count is not None and len(keyframes) != list_item_count:
raise ValueError(
f"timeline list inputs contain {list_item_count} items, but timeline/images "
f"contain {len(keyframes)} keyframes"
)
fitted_images = []
unedited_images = []
output_strengths = []
frame_indexes = []
position_seconds = []
output_prompts = []
for keyframe in keyframes:
fitted = _fit_etk_timeline_keyframe_image(keyframe, width, height)
if fitted is not None:
fitted_images.append(fitted.copy())
for unedited in _load_etk_timeline_keyframe_unedited_sources(keyframe, width, height):
unedited_images.append(unedited.copy())
output_strengths.append(_etk_keyframe_latent_strength(keyframe, strength))
frame_index = int(keyframe["frame"])
frame_indexes.append(frame_index)
position_seconds.append(frame_index / fps)
output_prompts.append(str(keyframe.get("positive_prompt", "") or ""))
settings = {"width": width, "height": height, "length": length, "fps": fps}
_etk_send_timeline_settings_to_ui(unique_id, settings, ui_instance_id=ui_instance_id)
result = (
_etk_timeline_image_batch(fitted_images, width, height),
_etk_timeline_unedited_source_batch(unedited_images, width, height),
width,
height,
length,
output_strengths,
frame_indexes,
position_seconds,
output_prompts,
)
return _etk_timeline_result_with_ui(
result,
keyframes,
force_ui=images is not None or list_item_count is not None,
settings=settings,
ui_instance_id=ui_instance_id,
)
+406
View File
@@ -0,0 +1,406 @@
"""Timeline image resolution, conversion, connected-input, and UI payload helpers."""
import hashlib
import json
import os
import folder_paths
import numpy as np
import torch
from PIL import Image, ImageOps
from .core import (
ETK_LTXV_TIMELINE_SCHEMA_VERSION,
_etk_timeline_keyframes_for_ui,
)
def _resolve_etk_timeline_image_path(image_info):
if isinstance(image_info, str):
filename = image_info
subfolder = ""
folder_type = "input"
elif isinstance(image_info, dict):
filename = image_info.get("filename") or image_info.get("name")
subfolder = image_info.get("subfolder") or ""
folder_type = image_info.get("type") or "input"
else:
raise ValueError("timeline image entry must be a filename string or file info object")
if not filename:
raise ValueError("timeline image entry is missing filename")
base_dir = folder_paths.get_directory_by_type(folder_type)
if not base_dir:
raise ValueError(f"unknown ComfyUI folder type for timeline image: {folder_type}")
base_path = os.path.abspath(base_dir)
candidate = os.path.abspath(os.path.join(base_path, subfolder, filename))
if os.path.commonpath([base_path, candidate]) != base_path:
raise ValueError(f"timeline image path escapes ComfyUI {folder_type} directory")
if not os.path.isfile(candidate):
raise FileNotFoundError(candidate)
return candidate
def _fit_etk_timeline_image(image, width, height, fit_mode, pan_x=0.0, pan_y=0.0, zoom=1.0, transparent_pad=False):
image = ImageOps.exif_transpose(image)
has_alpha = "A" in image.getbands()
image = image.convert("RGBA")
pan_x = float(pan_x)
pan_y = float(pan_y)
zoom = max(0.01, min(20.0, float(zoom)))
if fit_mode == "pad":
scale = min(width / image.width, height / image.height) * zoom
else:
scale = max(width / image.width, height / image.height) * zoom
resized = image.resize(
(max(1, round(image.width * scale)), max(1, round(image.height * scale))),
Image.Resampling.LANCZOS,
)
pan_range_x = max(max(0, resized.width - width), width)
pan_range_y = max(max(0, resized.height - height), height)
paste_x = round((width - resized.width) / 2 + pan_x * pan_range_x / 2)
paste_y = round((height - resized.height) / 2 + pan_y * pan_range_y / 2)
source_left = max(0, -paste_x)
source_top = max(0, -paste_y)
source_right = min(resized.width, width - paste_x)
source_bottom = min(resized.height, height - paste_y)
canvas_alpha = 0 if (transparent_pad or has_alpha) else 255
canvas = Image.new("RGBA", (width, height), (0, 0, 0, canvas_alpha))
if source_right > source_left and source_bottom > source_top:
crop = resized.crop((source_left, source_top, source_right, source_bottom))
# Preserve RGB under transparent pixels so VAE encode can explicitly strip alpha later.
canvas.paste(crop, (max(0, paste_x), max(0, paste_y)))
return canvas
def _pil_to_comfy_image_tensor(image):
array = np.asarray(image, dtype=np.float32) / 255.0
return torch.from_numpy(array)
def _etk_rgb_with_alpha_bleed(image, bleed_px):
rgba = image.convert("RGBA")
array = np.asarray(rgba, dtype=np.float32) / 255.0
rgb = array[:, :, :3].copy()
alpha = array[:, :, 3]
opaque = alpha > (1.0 / 255.0)
if opaque.all() or not opaque.any():
return rgb
bleed_px = max(0, int(round(float(bleed_px))))
try:
from scipy import ndimage
distance, indices = ndimage.distance_transform_edt(
~opaque,
return_distances=True,
return_indices=True,
)
fill = (~opaque) if bleed_px == 0 else ((~opaque) & (distance <= bleed_px))
rgb[fill] = rgb[indices[0][fill], indices[1][fill]]
except Exception:
# Fallback for environments without scipy: one-pixel dilation per pass.
tensor = torch.from_numpy(rgb).permute(2, 0, 1).unsqueeze(0)
known = torch.from_numpy(opaque).view(1, 1, *opaque.shape)
passes = bleed_px if bleed_px > 0 else max(opaque.shape)
kernel = torch.ones((1, 1, 3, 3), dtype=torch.float32)
for _ in range(passes):
expanded = torch.nn.functional.conv2d(known.float(), kernel, padding=1) > 0
new_pixels = expanded & ~known
if not new_pixels.any():
break
neighbor_count = torch.nn.functional.conv2d(known.float(), kernel, padding=1).clamp_min(1.0)
summed = torch.cat([
torch.nn.functional.conv2d((tensor[:, c:c + 1] * known.float()), kernel, padding=1)
for c in range(3)
], dim=1)
averaged = summed / neighbor_count
tensor = torch.where(new_pixels.expand_as(tensor), averaged, tensor)
known = expanded
rgb = tensor.squeeze(0).permute(1, 2, 0).numpy()
return rgb
def _pil_to_ltxv_vae_image_tensor(image, alpha_rgb_mode="preserve_rgb", alpha_rgb_bleed_px=64):
mode = str(alpha_rgb_mode or "bleed_opaque")
if "A" not in image.getbands() or mode == "preserve_rgb":
return _pil_to_comfy_image_tensor(image.convert("RGB"))
if mode == "bleed_opaque":
return torch.from_numpy(_etk_rgb_with_alpha_bleed(image, alpha_rgb_bleed_px))
rgba = image.convert("RGBA")
array = np.asarray(rgba, dtype=np.float32) / 255.0
rgb = array[:, :, :3]
alpha = array[:, :, 3:4]
backgrounds = {
"composite_black": np.array([0.0, 0.0, 0.0], dtype=np.float32),
"composite_gray": np.array([0.5, 0.5, 0.5], dtype=np.float32),
"composite_white": np.array([1.0, 1.0, 1.0], dtype=np.float32),
}
background = backgrounds.get(mode)
if background is None:
background = backgrounds["composite_black"]
return torch.from_numpy((rgb * alpha) + (background * (1.0 - alpha)))
def _etk_unwrap_single_input(value):
while isinstance(value, (list, tuple)) and len(value) == 1:
value = value[0]
return value
def _etk_scalar_input(value):
value = _etk_unwrap_single_input(value)
while isinstance(value, (list, tuple)):
if not value:
return None
value = _etk_unwrap_single_input(value[0])
return value
def _comfy_image_tensor_batch_to_pil(images):
if images is None:
return []
if isinstance(images, (list, tuple)):
pil_images = []
for item in images:
pil_images.extend(_comfy_image_tensor_batch_to_pil(item))
return pil_images
if not torch.is_tensor(images):
raise ValueError("connected timeline images input must be an IMAGE tensor")
tensor = images.detach().cpu().float().clamp(0.0, 1.0)
if tensor.ndim == 3:
tensor = tensor.unsqueeze(0)
if tensor.ndim != 4:
raise ValueError(f"connected timeline images must have shape [B,H,W,C], got {tuple(tensor.shape)}")
if tensor.shape[-1] not in (1, 3, 4):
raise ValueError("connected timeline images must have 1, 3, or 4 channels")
pil_images = []
for image in tensor:
array = (image.numpy() * 255.0).round().astype(np.uint8)
channels = array.shape[-1]
if channels == 1:
pil_images.append(Image.fromarray(array[:, :, 0], mode="L").convert("RGBA"))
elif channels == 3:
pil_images.append(Image.fromarray(array, mode="RGB"))
else:
pil_images.append(Image.fromarray(array, mode="RGBA"))
return pil_images
def _etk_save_timeline_connected_image_preview(image, slot):
input_dir = folder_paths.get_input_directory()
subfolder = "ETKTimeline"
output_dir = os.path.join(input_dir, subfolder)
os.makedirs(output_dir, exist_ok=True)
preview = ImageOps.exif_transpose(image).convert("RGBA")
digest = hashlib.sha256(np.asarray(preview, dtype=np.uint8).tobytes()).hexdigest()[:16]
filename = f"etk_timeline_input_slot_{int(slot):04d}_{digest}.png"
preview.save(os.path.join(output_dir, filename), compress_level=4)
return {
"filename": filename,
"subfolder": subfolder,
"type": "input",
}
def _etk_attach_timeline_connected_image(keyframe, image):
keyframe["_input_image"] = image
image_info = _etk_save_timeline_connected_image_preview(image, keyframe["slot"])
keyframe["image"] = image_info
layers = [layer for layer in keyframe.get("layers", []) if isinstance(layer, dict)]
selected_layer_id = str(keyframe.get("selected_layer_id") or keyframe.get("selectedLayerId") or "")
layer_source = None
if selected_layer_id:
layer_source = next((layer for layer in layers if str(layer.get("id", "")) == selected_layer_id), None)
if layer_source is None and layers:
layer_source = layers[0]
layer_source = layer_source or {}
keyframe["layers"] = [{
"id": str(layer_source.get("id") or selected_layer_id or "input"),
"image": image_info,
"fit_mode": layer_source.get("fit_mode", keyframe.get("fit_mode", "crop")),
"pan_x": float(layer_source.get("pan_x", keyframe.get("pan_x", 0.0))),
"pan_y": float(layer_source.get("pan_y", keyframe.get("pan_y", 0.0))),
"zoom": float(layer_source.get("zoom", keyframe.get("zoom", 1.0))),
"brightness": float(layer_source.get("brightness", keyframe.get("brightness", 1.0))),
"contrast": float(layer_source.get("contrast", keyframe.get("contrast", 1.0))),
}]
return keyframe
def _etk_timeline_with_connected_images(
keyframes,
images,
latent_frames,
require_image_index=False,
):
connected_images = _comfy_image_tensor_batch_to_pil(images)
if not connected_images:
return keyframes
latent_frames = max(1, int(latent_frames))
merged = [dict(keyframe) for keyframe in keyframes]
used_image_indices = set()
for keyframe in merged:
explicit_index = keyframe.get("image_index", None)
if explicit_index is None:
continue
try:
explicit_index = int(explicit_index)
except (TypeError, ValueError) as exc:
raise ValueError(f"timeline keyframe at slot {keyframe.get('slot')} has invalid image_index") from exc
if explicit_index < 0 or explicit_index >= len(connected_images):
raise ValueError(
f"prompt_schedule image_index {explicit_index} is outside the connected image range "
f"0..{len(connected_images) - 1}"
)
_etk_attach_timeline_connected_image(keyframe, connected_images[explicit_index])
used_image_indices.add(explicit_index)
if require_image_index:
missing = [keyframe for keyframe in merged if keyframe.get("image_index", None) is None]
if missing:
raise ValueError(
"When prompt_schedule and images are connected, each schedule tuple must start "
"with a zero-based connected image index: "
"(image_index, seconds, positive_prompt, latent_strength, prompt_strength)."
)
return sorted(merged, key=lambda item: item["slot"])
remaining_images = [
image for index, image in enumerate(connected_images)
if index not in used_image_indices
]
remaining_index = 0
for keyframe in merged:
if remaining_index >= len(remaining_images):
break
if (keyframe.get("image") and not keyframe.get("input_image")) or keyframe.get("_input_image") is not None:
continue
_etk_attach_timeline_connected_image(keyframe, remaining_images[remaining_index])
remaining_index += 1
if remaining_index >= len(remaining_images):
return sorted(merged, key=lambda item: item["slot"])
occupied = {int(keyframe["slot"]) for keyframe in merged if 0 <= int(keyframe["slot"]) < latent_frames}
cursor = 0
if merged:
cursor = max(int(keyframe["slot"]) for keyframe in merged) + 1
for image in remaining_images[remaining_index:]:
slot = None
for candidate in range(cursor, latent_frames):
if candidate not in occupied:
slot = candidate
break
if slot is None:
raise ValueError(
f"connected timeline images provide {len(connected_images)} images, "
f"but there are only {max(0, latent_frames - remaining_index)} free latent slots "
"at or after the existing timeline keyframes"
)
occupied.add(slot)
cursor = slot + 1
merged.append(
_etk_attach_timeline_connected_image(
{
"frame": slot * 8,
"slot": slot,
"fit_mode": "crop",
"pan_x": 0.0,
"pan_y": 0.0,
"zoom": 1.0,
"image": None,
"layers": [],
"positive_prompt": "",
"negative_prompt": "",
},
image,
)
)
return sorted(merged, key=lambda item: item["slot"])
def _etk_timeline_result_with_ui(result, keyframes, force_ui=False, settings=None, ui_instance_id=None):
if (
settings is None
and not force_ui
and not any(keyframe.get("_input_image") is not None for keyframe in keyframes)
):
return result
instance_id = str(ui_instance_id or "")
keyframes_payload = {
"schemaVersion": ETK_LTXV_TIMELINE_SCHEMA_VERSION,
"ui_instance_id": instance_id,
"keyframes": _etk_timeline_keyframes_for_ui(keyframes, fps=(settings or {}).get("fps", 25.0)),
}
ui = {
"etk_ltxv_timeline_keyframes": [
json.dumps(keyframes_payload),
],
}
if settings is not None:
settings_payload = dict(settings)
settings_payload["ui_instance_id"] = instance_id
ui["etk_ltxv_timeline_settings"] = [json.dumps(settings_payload)]
return {
"ui": ui,
"result": result,
}
def _etk_send_timeline_settings_to_ui(unique_id, settings, ui_instance_id=None):
if unique_id is None or settings is None:
return
try:
from server import PromptServer
payload = dict(settings)
payload["ui_instance_id"] = str(ui_instance_id or "")
PromptServer.instance.send_sync(
"etk_ltxv_timeline_settings",
{
"node": str(unique_id),
"settings": payload,
},
getattr(PromptServer.instance, "client_id", None),
)
except Exception:
pass
def _etk_timeline_json_like(value):
if isinstance(value, (list, tuple, dict)):
return True
if not isinstance(value, str):
return False
stripped = value.strip()
return stripped == "" or stripped.startswith("[") or stripped.startswith("{")
def _etk_timeline_editor_compat_inputs(timeline_json, strength):
if _etk_timeline_json_like(strength):
if not _etk_timeline_json_like(timeline_json):
timeline_json, strength = strength, timeline_json
else:
if timeline_json in (None, "", "[]"):
timeline_json = strength
strength = 1.0
try:
strength = max(0.0, min(1.0, float(strength)))
except (TypeError, ValueError):
strength = 1.0
if timeline_json is None:
timeline_json = "[]"
return timeline_json, strength
+357
View File
@@ -0,0 +1,357 @@
"""Layer rendering, compositing, alpha conversion, and unedited-source output."""
import os
import torch
from PIL import Image, ImageColor, ImageDraw, ImageEnhance, ImageFilter, ImageFont, ImageOps
from .images import (
_comfy_image_tensor_batch_to_pil,
_fit_etk_timeline_image,
_pil_to_comfy_image_tensor,
_resolve_etk_timeline_image_path,
)
from .schema import (
_etk_normalize_timeline_text_align,
_etk_normalize_timeline_text_color,
)
def _etk_file_info_key(image_info):
if isinstance(image_info, str):
return (image_info, "", "input")
if isinstance(image_info, dict):
return (
image_info.get("filename") or image_info.get("name") or "",
image_info.get("subfolder") or "",
image_info.get("type") or "input",
)
return None
def _load_etk_timeline_layer_source(keyframe, layer):
if keyframe.get("_input_image") is not None and _etk_file_info_key(layer.get("image")) == _etk_file_info_key(keyframe.get("image")):
return ImageOps.exif_transpose(keyframe["_input_image"]).copy()
image_path = _resolve_etk_timeline_image_path(layer["image"])
with Image.open(image_path) as loaded:
return ImageOps.exif_transpose(loaded).copy()
def _apply_etk_timeline_filters(image, brightness=1.0, contrast=1.0):
brightness = max(0.0, min(2.0, float(brightness)))
contrast = max(0.0, min(2.0, float(contrast)))
if abs(brightness - 1.0) < 1e-6 and abs(contrast - 1.0) < 1e-6:
return image
has_alpha = image.mode in {"RGBA", "LA"} or "transparency" in image.info
if has_alpha:
rgba = image.convert("RGBA")
alpha = rgba.getchannel("A")
rgb = rgba.convert("RGB")
rgb = ImageEnhance.Brightness(rgb).enhance(brightness)
rgb = ImageEnhance.Contrast(rgb).enhance(contrast)
filtered = Image.merge("RGBA", (*rgb.split(), alpha))
return filtered
filtered = image.convert("RGB")
filtered = ImageEnhance.Brightness(filtered).enhance(brightness)
filtered = ImageEnhance.Contrast(filtered).enhance(contrast)
return filtered
def _etk_timeline_text_font_candidates(font_family, bold=False, italic=False):
requested = str(font_family or "").strip()
if requested and os.path.isfile(requested):
yield requested
aliases = {
"sans-serif": "dejavu sans",
"sans": "dejavu sans",
"serif": "dejavu serif",
"monospace": "dejavu sans mono",
"mono": "dejavu sans mono",
}
target = aliases.get(requested.lower(), requested.lower())
tokens = [token for token in target.replace("_", " ").replace("-", " ").split() if token]
font_roots = ["/usr/share/fonts", "/usr/local/share/fonts"]
scored = []
for root in font_roots:
if not os.path.isdir(root):
continue
for dirpath, _dirnames, filenames in os.walk(root):
for filename in filenames:
if not filename.lower().endswith((".ttf", ".otf")):
continue
lower = filename.lower().replace("_", " ").replace("-", " ")
if tokens and not all(token in lower for token in tokens):
continue
has_bold = "bold" in lower
has_italic = "italic" in lower or "oblique" in lower
score = 0
if bold == has_bold:
score += 10
if italic == has_italic:
score += 10
if not has_italic:
score += 1
scored.append((score, os.path.join(dirpath, filename)))
for _score, path in sorted(scored, reverse=True):
yield path
def _etk_timeline_text_font(font_family, size, bold=False, italic=False):
size = max(1, int(round(float(size))))
for candidate in _etk_timeline_text_font_candidates(font_family, bold, italic):
try:
return ImageFont.truetype(candidate, size)
except OSError:
continue
try:
return ImageFont.truetype(str(font_family or ""), size)
except OSError:
try:
return ImageFont.load_default(size=size)
except TypeError:
return ImageFont.load_default()
def _render_etk_timeline_text_layer(layer, width, height):
canvas = Image.new("RGBA", (int(width), int(height)), (0, 0, 0, 0))
text = str(layer.get("text", "") or "")
if not text:
return canvas
font = _etk_timeline_text_font(
layer.get("font_family", "sans-serif"),
layer.get("font_size", 72),
layer.get("bold", False),
layer.get("italic", False),
)
draw = ImageDraw.Draw(canvas)
color = ImageColor.getcolor(_etk_normalize_timeline_text_color(layer.get("color", "#ffffff")), "RGBA")
outline_width = max(0, int(round(float(layer.get("outline_width", 0) or 0))))
outline_color = ImageColor.getcolor(_etk_normalize_timeline_text_color(layer.get("outline_color", "#000000"), "#000000"), "RGBA")
shadow_color = ImageColor.getcolor(_etk_normalize_timeline_text_color(layer.get("shadow_color", "#000000"), "#000000"), "RGBA")
shadow_blur = max(0.0, float(layer.get("shadow_blur", 0) or 0))
shadow_offset_x = float(layer.get("shadow_offset_x", 0) or 0)
shadow_offset_y = float(layer.get("shadow_offset_y", 0) or 0)
align = _etk_normalize_timeline_text_align(layer.get("align", "center"))
lines = text.splitlines() or [text]
boxes = [draw.textbbox((0, 0), line if line else " ", font=font, stroke_width=outline_width) for line in lines]
heights = [max(1, bottom - top) for _left, top, _right, bottom in boxes]
line_step = max(1, int(round(max(heights) * 1.18)))
total_height = line_step * len(lines)
y = (height - total_height) / 2
placements = []
for line, box in zip(lines, boxes):
left, top, right, _bottom = box
text_width = right - left
if align == "left":
x = -left
elif align == "right":
x = width - text_width - left
else:
x = (width - text_width) / 2 - left
placements.append((line, x, y - top))
y += line_step
if shadow_color[3] > 0 and (shadow_blur > 0 or shadow_offset_x or shadow_offset_y):
shadow_layer = Image.new("RGBA", (int(width), int(height)), (0, 0, 0, 0))
shadow_draw = ImageDraw.Draw(shadow_layer)
for line, x, text_y in placements:
shadow_draw.text(
(x + shadow_offset_x, text_y + shadow_offset_y),
line,
font=font,
fill=shadow_color,
stroke_width=outline_width,
stroke_fill=shadow_color,
)
if shadow_blur > 0:
shadow_layer = shadow_layer.filter(ImageFilter.GaussianBlur(radius=shadow_blur))
canvas = Image.alpha_composite(canvas, shadow_layer)
draw = ImageDraw.Draw(canvas)
for line, x, text_y in placements:
draw.text(
(x, text_y),
line,
font=font,
fill=color,
stroke_width=outline_width,
stroke_fill=outline_color,
)
return canvas
def _fit_etk_timeline_keyframe_image(keyframe, width, height):
layers = [
layer for layer in keyframe.get("layers", [])
if layer.get("image") or layer.get("type") == "text"
]
if layers:
composite = Image.new("RGBA", (width, height), (0, 0, 0, 0))
used_layer = False
for layer in layers:
if layer.get("type") == "text":
source = _render_etk_timeline_text_layer(layer, width, height)
else:
source = _load_etk_timeline_layer_source(keyframe, layer)
source = _apply_etk_timeline_filters(
source,
layer.get("brightness", 1.0),
layer.get("contrast", 1.0),
)
fitted = _fit_etk_timeline_image(
source,
width,
height,
layer.get("fit_mode", "crop"),
layer.get("pan_x", 0.0),
layer.get("pan_y", 0.0),
layer.get("zoom", 1.0),
transparent_pad=True,
)
composite = Image.alpha_composite(composite, fitted.convert("RGBA"))
used_layer = True
return composite if used_layer else None
connected_image = keyframe.get("_input_image")
if connected_image is not None:
return _fit_etk_timeline_image(
_apply_etk_timeline_filters(connected_image, keyframe.get("brightness", 1.0), keyframe.get("contrast", 1.0)),
width,
height,
keyframe["fit_mode"],
keyframe["pan_x"],
keyframe["pan_y"],
keyframe["zoom"],
)
if not keyframe["image"]:
return None
image_path = _resolve_etk_timeline_image_path(keyframe["image"])
with Image.open(image_path) as loaded:
return _fit_etk_timeline_image(
_apply_etk_timeline_filters(loaded, keyframe.get("brightness", 1.0), keyframe.get("contrast", 1.0)),
width,
height,
keyframe["fit_mode"],
keyframe["pan_x"],
keyframe["pan_y"],
keyframe["zoom"],
)
def _etk_passthrough_connected_image_list(images):
if images is None:
return None
if not torch.is_tensor(images):
return None
tensor = images.detach().clone().float().clamp(0.0, 1.0)
if tensor.ndim == 3:
tensor = tensor.unsqueeze(0)
if tensor.ndim != 4:
raise ValueError(f"connected timeline images must have shape [B,H,W,C], got {tuple(tensor.shape)}")
if tensor.shape[-1] == 1:
tensor = tensor.expand(*tensor.shape[:-1], 3).clone()
if tensor.shape[-1] == 4:
tensor = tensor[..., :3].clone()
if tensor.shape[-1] != 3:
raise ValueError("connected timeline images must have 1, 3, or 4 channels")
return [image.unsqueeze(0).contiguous() for image in tensor]
def _etk_passthrough_connected_image_batch(images):
image_list = _etk_passthrough_connected_image_list(images)
if image_list is None:
return None
return torch.cat(image_list, dim=0).contiguous()
def _etk_unedited_source_image_tensor(image, width, height):
fitted = _fit_etk_timeline_image(
ImageOps.exif_transpose(image),
width,
height,
"pad",
)
return _pil_to_comfy_image_tensor(fitted.convert("RGBA")).unsqueeze(0).contiguous()
def _etk_connected_unedited_source_batch(images, width, height):
pil_images = _comfy_image_tensor_batch_to_pil(images)
if not pil_images:
return None
return torch.cat(
[_etk_unedited_source_image_tensor(image, width, height) for image in pil_images],
dim=0,
).contiguous()
def _load_etk_timeline_keyframe_unedited_sources(keyframe, width=None, height=None):
layers = [
layer for layer in keyframe.get("layers", [])
if layer.get("image") or layer.get("type") == "text"
]
if layers:
sources = []
for layer in layers:
if layer.get("type") == "text":
if width is None or height is None:
continue
sources.append(_render_etk_timeline_text_layer(layer, width, height))
else:
sources.append(_load_etk_timeline_layer_source(keyframe, layer))
return sources
if keyframe.get("_input_image") is not None:
return [ImageOps.exif_transpose(keyframe["_input_image"]).copy()]
if keyframe.get("image"):
image_path = _resolve_etk_timeline_image_path(keyframe["image"])
with Image.open(image_path) as loaded:
return [ImageOps.exif_transpose(loaded).copy()]
return []
def _load_etk_timeline_keyframe_unedited_source(keyframe):
sources = _load_etk_timeline_keyframe_unedited_sources(keyframe)
return sources[0] if sources else None
def _etk_timeline_unedited_source_list(images, width, height):
if not images:
return [torch.zeros((1, height, width, 4), dtype=torch.float32)]
return [
_etk_unedited_source_image_tensor(image, width, height)
for image in images
]
def _etk_timeline_unedited_source_batch(images, width, height):
return torch.cat(_etk_timeline_unedited_source_list(images, width, height), dim=0).contiguous()
def _etk_timeline_image_batch(images, width, height):
if not images:
return torch.zeros((1, height, width, 3), dtype=torch.float32)
return torch.stack([_pil_to_comfy_image_tensor(image.convert("RGBA")) for image in images], dim=0)
def _etk_timeline_alpha_tensor(image):
tensor = _pil_to_comfy_image_tensor(image)
if tensor.ndim == 3 and tensor.shape[-1] >= 4:
return tensor[:, :, 3].clamp(0.0, 1.0)
return torch.ones((image.height, image.width), dtype=torch.float32)
def _etk_alpha_noise_mask(image, strength, latent_height, latent_width, device, dtype):
alpha = _etk_timeline_alpha_tensor(image)
if torch.all(alpha >= 1.0):
return None
alpha = alpha.to(device=device, dtype=dtype).view(1, 1, 1, image.height, image.width)
alpha = torch.nn.functional.interpolate(
alpha.view(1, 1, image.height, image.width),
size=(latent_height, latent_width),
mode="area",
).view(1, 1, 1, latent_height, latent_width)
return (1.0 - (float(strength) * alpha)).clamp(0.0, 1.0)
+257
View File
@@ -0,0 +1,257 @@
"""Timeline JSON parsing and normalized layer/keyframe schema."""
import json
from PIL import ImageColor
from .core import (
ETK_LTXV_TIMELINE_SCHEMA_VERSION,
_etk_ltxv_timeline_payload_keyframes,
)
def _etk_bool(value, fallback=False):
if isinstance(value, bool):
return value
if isinstance(value, str):
lowered = value.strip().lower()
if lowered in {"true", "1", "yes", "bold"}:
return True
if lowered in {"false", "0", "no", "normal"}:
return False
if isinstance(value, (int, float)):
return value != 0
return bool(fallback)
ETK_TIMELINE_TEXT_LAYER_FIELDS = frozenset({
"text",
"fontFamily",
"font_family",
"fontSize",
"font_size",
"bold",
"italic",
"align",
"textAlign",
"text_align",
"color",
"outlineColor",
"outline_color",
"outlineWidth",
"outline_width",
"shadowColor",
"shadow_color",
"shadowBlur",
"shadow_blur",
"shadowOffsetX",
"shadow_offset_x",
"shadowOffsetY",
"shadow_offset_y",
})
def _etk_timeline_layer_is_text(raw, fallback=None):
fallback = fallback or {}
if not isinstance(raw, dict):
return False
layer_type = str(
raw.get("type", raw.get("layerType", raw.get("layer_type", fallback.get("type", ""))))
or ""
).lower()
return layer_type == "text" or any(field in raw for field in ETK_TIMELINE_TEXT_LAYER_FIELDS)
def _etk_normalize_timeline_text_color(value, fallback="#ffffff"):
text = str(value if value is not None else fallback).strip()
if not text:
text = fallback
try:
ImageColor.getcolor(text, "RGBA")
except ValueError:
text = fallback
return text
def _etk_float(value, fallback=0.0, minimum=None, maximum=None):
try:
number = float(value)
except (TypeError, ValueError):
number = float(fallback)
if minimum is not None:
number = max(float(minimum), number)
if maximum is not None:
number = min(float(maximum), number)
return number
def _etk_normalize_timeline_text_align(value, fallback="center"):
text = str(value if value is not None else fallback).strip().lower()
return text if text in {"left", "center", "right"} else fallback
def _etk_parse_timeline_layer(raw, fallback=None):
fallback = fallback or {}
if not isinstance(raw, dict):
return None
is_text_layer = _etk_timeline_layer_is_text(raw, fallback)
image_info = None if is_text_layer else raw.get("image") or raw.get("file") or raw.get("filename") or fallback.get("image")
has_layer_slot = bool(
image_info
or is_text_layer
or raw.get("id")
or raw.get("empty")
or "image" in raw
or "file" in raw
or "filename" in raw
or "type" in raw
or "layerType" in raw
or "layer_type" in raw
or any(field in raw for field in ETK_TIMELINE_TEXT_LAYER_FIELDS)
)
if not has_layer_slot:
return None
fit_mode = str(raw.get("fitMode", raw.get("fit_mode", fallback.get("fit_mode", "crop")))).lower()
if fit_mode not in {"crop", "pad"}:
fit_mode = "crop"
try:
pan_x = float(raw.get("panX", raw.get("pan_x", fallback.get("pan_x", 0.0))))
pan_y = float(raw.get("panY", raw.get("pan_y", fallback.get("pan_y", 0.0))))
zoom = max(0.01, min(20.0, float(raw.get("zoom", fallback.get("zoom", 1.0)))))
brightness = max(0.0, min(2.0, float(raw.get("brightness", fallback.get("brightness", 1.0)))))
contrast = max(0.0, min(2.0, float(raw.get("contrast", fallback.get("contrast", 1.0)))))
except (TypeError, ValueError) as exc:
raise ValueError("timeline layer has invalid pan/zoom/filter values") from exc
parsed = {
"id": str(raw.get("id", fallback.get("id", "")) or ""),
"type": "text" if is_text_layer else "image",
"image": image_info,
"empty": not bool(image_info) and not is_text_layer,
"fit_mode": fit_mode,
"pan_x": pan_x,
"pan_y": pan_y,
"zoom": zoom,
"brightness": brightness,
"contrast": contrast,
}
if is_text_layer:
try:
font_size = max(1.0, min(512.0, float(raw.get("fontSize", raw.get("font_size", fallback.get("font_size", 72))))))
except (TypeError, ValueError) as exc:
raise ValueError("timeline text layer has invalid font size") from exc
parsed.update({
"text": str(raw.get("text", fallback.get("text", "Text")) or ""),
"font_family": str(raw.get("fontFamily", raw.get("font_family", fallback.get("font_family", "sans-serif"))) or "sans-serif"),
"font_size": font_size,
"bold": _etk_bool(raw.get("bold", fallback.get("bold", False))),
"italic": _etk_bool(raw.get("italic", fallback.get("italic", False))),
"align": _etk_normalize_timeline_text_align(raw.get("align", raw.get("textAlign", raw.get("text_align", fallback.get("align", "center"))))),
"color": _etk_normalize_timeline_text_color(raw.get("color", fallback.get("color", "#ffffff"))),
"outline_color": _etk_normalize_timeline_text_color(raw.get("outlineColor", raw.get("outline_color", fallback.get("outline_color", "#000000"))), "#000000"),
"outline_width": _etk_float(raw.get("outlineWidth", raw.get("outline_width", fallback.get("outline_width", 0.0))), 0.0, 0.0, 64.0),
"shadow_color": _etk_normalize_timeline_text_color(raw.get("shadowColor", raw.get("shadow_color", fallback.get("shadow_color", "#000000"))), "#000000"),
"shadow_blur": _etk_float(raw.get("shadowBlur", raw.get("shadow_blur", fallback.get("shadow_blur", 0.0))), 0.0, 0.0, 128.0),
"shadow_offset_x": _etk_float(raw.get("shadowOffsetX", raw.get("shadow_offset_x", fallback.get("shadow_offset_x", 0.0))), 0.0, -512.0, 512.0),
"shadow_offset_y": _etk_float(raw.get("shadowOffsetY", raw.get("shadow_offset_y", fallback.get("shadow_offset_y", 0.0))), 0.0, -512.0, 512.0),
})
return parsed
def _parse_etk_ltxv_timeline(timeline_json, include_empty=False, fps=25.0):
if not timeline_json:
return []
try:
fps = max(0.001, float(fps))
except (TypeError, ValueError) as exc:
raise ValueError("timeline fps must be a number") from exc
try:
payload = json.loads(timeline_json)
except json.JSONDecodeError as exc:
raise ValueError(f"timeline_json is not valid JSON: {exc}") from exc
payload, schema_version = _etk_ltxv_timeline_payload_keyframes(payload)
keyframes = []
for raw in payload:
if not isinstance(raw, dict):
continue
image_info = raw.get("image") or raw.get("file") or raw.get("filename")
positive_prompt = str(raw.get("positivePrompt", raw.get("positive_prompt", "")) or "")
negative_prompt = str(raw.get("negativePrompt", raw.get("negative_prompt", "")) or "")
if schema_version == ETK_LTXV_TIMELINE_SCHEMA_VERSION and raw.get("time") is None:
raise ValueError("timeline_json schemaVersion 2 keyframes must include time")
try:
if raw.get("time") is not None:
time_seconds = float(raw.get("time", 0.0))
elif raw.get("slot") is not None:
time_seconds = float(raw.get("slot", 0.0)) * 8.0 / fps
elif raw.get("frame") is not None:
time_seconds = float(raw.get("frame", 0.0)) / fps
else:
time_seconds = 0.0
frame = int(round(time_seconds * fps))
slot = int((frame / 8.0) + 0.5)
except (TypeError, ValueError) as exc:
raise ValueError(f"timeline keyframe has invalid time value: {raw.get('time')}") from exc
if time_seconds < 0 or slot < 0:
raise ValueError(f"timeline keyframe time {time_seconds} maps outside video latent range")
fit_mode = str(raw.get("fitMode", raw.get("fit_mode", "crop"))).lower()
if fit_mode not in {"crop", "pad"}:
fit_mode = "crop"
try:
pan_x = float(raw.get("panX", raw.get("pan_x", 0.0)))
pan_y = float(raw.get("panY", raw.get("pan_y", 0.0)))
zoom = max(0.01, min(20.0, float(raw.get("zoom", 1.0))))
brightness = max(0.0, min(2.0, float(raw.get("brightness", 1.0))))
contrast = max(0.0, min(2.0, float(raw.get("contrast", 1.0))))
prompt_strength = max(0.0, min(4.0, float(raw.get("promptStrength", raw.get("prompt_strength", 1.0)))))
latent_strength_raw = raw.get("latentStrength", raw.get("latent_strength", raw.get("strength", None)))
latent_strength = None if latent_strength_raw is None else max(0.0, min(1.0, float(latent_strength_raw)))
except (TypeError, ValueError) as exc:
raise ValueError("timeline keyframe has invalid pan/zoom/filter values") from exc
fallback_layer = {
"image": image_info,
"fit_mode": fit_mode,
"pan_x": pan_x,
"pan_y": pan_y,
"zoom": zoom,
"brightness": brightness,
"contrast": contrast,
}
layers = []
if isinstance(raw.get("layers"), list):
for layer in raw.get("layers", []):
parsed_layer = _etk_parse_timeline_layer(layer, {**fallback_layer, "image": None})
if parsed_layer is not None:
layers.append(parsed_layer)
if not layers and image_info:
parsed_layer = _etk_parse_timeline_layer({"image": image_info}, fallback_layer)
if parsed_layer is not None:
layers.append(parsed_layer)
if layers:
image_info = layers[0]["image"]
if not include_empty and not layers and not positive_prompt and not negative_prompt:
continue
keyframes.append(
{
"frame": frame,
"slot": slot,
"time": time_seconds,
"fit_mode": fit_mode,
"pan_x": pan_x,
"pan_y": pan_y,
"zoom": zoom,
"brightness": brightness,
"contrast": contrast,
"image": image_info,
"layers": layers,
"selected_layer_id": str(raw.get("selectedLayerId", raw.get("selected_layer_id", "")) or ""),
"input_image": bool(raw.get("inputImage", raw.get("input_image", False))),
"positive_prompt": positive_prompt,
"negative_prompt": negative_prompt,
"prompt_strength": prompt_strength,
"latent_strength": latent_strength,
}
)
return sorted(keyframes, key=lambda item: item["slot"])