Updated IAMCCS-nodes to version 1.4.6

This commit is contained in:
IAMCCS
2026-05-21 00:20:33 +02:00
parent 9efc7ba8e3
commit 246f2f1966
16 changed files with 12258 additions and 859 deletions
+2
View File
@@ -1,5 +1,7 @@
# IAMCCS Nodes - Changelog
## 🆕 2026-05-20 - version 1.4.6 — Shotboard planner v2 and v3 added
## 🆕 2026-05-12 - version 1.4.5 — Cine nodes added
## 🆕 2026-05-04 - version 1.4.4 — Supernodes v.2 and bug fixed pplus utilities added
+2
View File
@@ -9,6 +9,8 @@
### Category: ComfyUI Custom Nodes
### Main Feature: Fix for LoRA loading in native WANAnimate workflows + general nodes 4 ComfyUI
## Version: 1.4.6 (Shotboard planner v2 and v3 added)
Version: 1.4.5 (Cine nodes added)
Version: 1.4.4 (Supernodes and wan 2.2 + LTX 2.2 utilities added)
+279 -2
View File
@@ -1,4 +1,4 @@
# ==========================================================
# ==========================================================
# __init__.py — Registro nodi IAMCCS
# ==========================================================
@@ -108,14 +108,24 @@ from .iamccs_cine_nodes import (
IAMCCS_CineShotboardTimelinePro,
IAMCCS_CineShotboardPlannerPro,
IAMCCS_CineShotboardPlannerProV2,
IAMCCS_CineShotboardPlannerV3,
IAMCCS_CineShotboardLite,
IAMCCS_CineShotboardPlannerProLegacy,
IAMCCS_CineInfo,
IAMCCS_CineInfoV2,
IAMCCS_CineFLFProductor,
IAMCCS_CineFilmmaker,
IAMCCS_CineFilmmakerBackend,
IAMCCS_CineShotboardBackendPro,
IAMCCS_CineFilmmakerGuide,
IAMCCS_CineFilmmakerGuide1to1,
IAMCCS_CineSwitch,
IAMCCS_CinePromptRelayLatentShapeSync,
IAMCCS_CineFLFLengthCompensator,
IAMCCS_CinePromptRelaySafeEncode,
IAMCCS_CineRelayOrBypass,
IAMCCS_CinePromptArchitect,
IAMCCS_BoardMaker,
IAMCCS_CineMusicVideoPlanner,
IAMCCS_CineShotPlanner,
IAMCCS_CineRefLatentControl,
@@ -131,6 +141,12 @@ from .iamccs_cine_nodes import (
IAMCCS_CineV2VAssetSelector,
IAMCCS_CineWorkflowInspector,
)
from .iamccs_cine_resolution_parity import IAMCCS_CineResolutionParityTranslator
from .iamccs_cine_stage_switch import IAMCCS_CineStage2BypassSwitch
from .iamccs_cine_stage2_preview_toggle import IAMCCS_CineStage2PreviewToggle
from .iamccs_cine_flf_productor_dyno import IAMCCS_CineFLFProductorDyno
from .iamccs_cine_flf_engine_simple_dyno import IAMCCS_CineFLFEngineSimpleDyno
from .iamccs_cine_duration_lock import IAMCCS_CineBoardDurationLock, IAMCCS_CineLatentDurationCrop
from .iamccs_ltx2_temporal_overlap_samplers import (
IAMCCS_LTX2_ConditionNextLatentWithPrevOverlap,
@@ -212,6 +228,12 @@ from .iamccs_hw_probe_node import (
IAMCCS_HWProbeRecommendations,
)
from .iamccs_detail_atelier import (
IAMCCS_DetailAtelier,
IAMCCS_DetailAtelierAdvanced,
IAMCCS_DetailAtelierSampler,
)
from .iamccs_qwen_vl_flf import (
IAMCCS_QWEN_VL_FLF,
IAMCCS_QWEN_VL_FLF_Advanced,
@@ -374,14 +396,30 @@ NODE_CLASS_MAPPINGS = {
"IAMCCS_CineShotboardTimelinePro": IAMCCS_CineShotboardTimelinePro,
"IAMCCS_CineShotboardPlannerPro": IAMCCS_CineShotboardPlannerPro,
"IAMCCS_CineShotboardPlannerProV2": IAMCCS_CineShotboardPlannerProV2,
"IAMCCS_CineShotboardPlannerV3": IAMCCS_CineShotboardPlannerV3,
"IAMCCS_CineShotboardLite": IAMCCS_CineShotboardLite,
"IAMCCS_CineShotboardPlannerProLegacy": IAMCCS_CineShotboardPlannerProLegacy,
"IAMCCS_CineResolutionParityTranslator": IAMCCS_CineResolutionParityTranslator,
"IAMCCS_CineStage2BypassSwitch": IAMCCS_CineStage2BypassSwitch,
"IAMCCS_CineStage2PreviewToggle": IAMCCS_CineStage2PreviewToggle,
"IAMCCS_CineInfo": IAMCCS_CineInfo,
"IAMCCS_CineInfoV2": IAMCCS_CineInfoV2,
"IAMCCS_CineFLFProductor": IAMCCS_CineFLFProductor,
"IAMCCS_CineFLFProductorDyno": IAMCCS_CineFLFProductorDyno,
"IAMCCS_CineFilmmaker": IAMCCS_CineFilmmaker,
"IAMCCS_CineFilmmakerBackend": IAMCCS_CineFilmmakerBackend,
"IAMCCS_CineShotboardBackendPro": IAMCCS_CineShotboardBackendPro,
"IAMCCS_CineFilmmakerGuide": IAMCCS_CineFilmmakerGuide,
"IAMCCS_CineFilmmakerGuide1to1": IAMCCS_CineFilmmakerGuide1to1,
"IAMCCS_CineSwitch": IAMCCS_CineSwitch,
"IAMCCS_CinePromptRelayLatentShapeSync": IAMCCS_CinePromptRelayLatentShapeSync,
"IAMCCS_CineFLFLengthCompensator": IAMCCS_CineFLFLengthCompensator,
"IAMCCS_CineBoardDurationLock": IAMCCS_CineBoardDurationLock,
"IAMCCS_CineLatentDurationCrop": IAMCCS_CineLatentDurationCrop,
"IAMCCS_CinePromptRelaySafeEncode": IAMCCS_CinePromptRelaySafeEncode,
"IAMCCS_CineRelayOrBypass": IAMCCS_CineRelayOrBypass,
"IAMCCS_CinePromptArchitect": IAMCCS_CinePromptArchitect,
"IAMCCS_BoardMaker": IAMCCS_BoardMaker,
"IAMCCS_CineMusicVideoPlanner": IAMCCS_CineMusicVideoPlanner,
"IAMCCS_CineShotPlanner": IAMCCS_CineShotPlanner,
"IAMCCS_CineRefLatentControl": IAMCCS_CineRefLatentControl,
@@ -401,6 +439,7 @@ NODE_CLASS_MAPPINGS = {
"IAMCCS_WDC_LTXSequencer": IAMCCS_WDC_LTXSequencer,
"IAMCCS_CineLTXSequencerExact": IAMCCS_CineLTXSequencerExact,
"IAMCCS_CineFLFEngineSimple": IAMCCS_CineFLFEngineSimple,
"IAMCCS_CineFLFEngineSimpleDyno": IAMCCS_CineFLFEngineSimpleDyno,
"IAMCCS_WDC_LTXSequencerFixed5": IAMCCS_WDC_LTXSequencerFixed5,
"IAMCCS_LTX2_InitLatentSampler": IAMCCS_LTX2_InitLatentSampler,
"IAMCCS_LTX2_LoopingSampler": IAMCCS_LTX2_LoopingSampler,
@@ -457,6 +496,9 @@ NODE_CLASS_MAPPINGS = {
"IAMCCS_VAEDecodeTiledSafe": IAMCCS_VAEDecodeTiledSafe,
"IAMCCS_VAEDecodeToDisk": IAMCCS_VAEDecodeToDisk,
"IAMCCS_HWProbeRecommendations": IAMCCS_HWProbeRecommendations,
"IAMCCS_DetailAtelier": IAMCCS_DetailAtelier,
"IAMCCS_DetailAtelierAdvanced": IAMCCS_DetailAtelierAdvanced,
"IAMCCS_DetailAtelierSampler": IAMCCS_DetailAtelierSampler,
"IAMCCS_MoveAhead": IAMCCS_MoveAhead,
"IAMCCS_MoveAheadEnforcer": IAMCCS_MoveAheadEnforcer,
@@ -586,14 +628,30 @@ NODE_DISPLAY_NAME_MAPPINGS = {
"IAMCCS_CineShotboardTimelinePro": "IAMCCS Cine Shotboard Timeline Pro",
"IAMCCS_CineShotboardPlannerPro": "IAMCCS Cine Shotboard Planner Pro",
"IAMCCS_CineShotboardPlannerProV2": "IAMCCS Cine Shotboard Planner Pro V2",
"IAMCCS_CineShotboardPlannerV3": "IAMCCS Cine Shotboard Planner V3",
"IAMCCS_CineShotboardLite": "IAMCCS Cine Shotboard Lite",
"IAMCCS_CineShotboardPlannerProLegacy": "IAMCCS Cine Shotboard Planner Pro Legacy Outputs",
"IAMCCS_CineResolutionParityTranslator": "IAMCCS Cine Resolution Parity Translator",
"IAMCCS_CineStage2BypassSwitch": "IAMCCS Cine Stage 2 Bypass Switch",
"IAMCCS_CineStage2PreviewToggle": "IAMCCS Cine Stage 2 Preview Toggle",
"IAMCCS_CineInfo": "IAMCCS CineInfo",
"IAMCCS_CineInfoV2": "IAMCCS CineInfo V2",
"IAMCCS_CineFLFProductor": "IAMCCS Cine FLF Productor",
"IAMCCS_CineFLFProductorDyno": "IAMCCS Cine FLF Productor Dyno",
"IAMCCS_CineFilmmaker": "IAMCCS Cine Filmmaker",
"IAMCCS_CineFilmmakerBackend": "IAMCCS Cine Filmmaker Backend",
"IAMCCS_CineShotboardBackendPro": "IAMCCS Cine Shotboard Backend Pro",
"IAMCCS_CineFilmmakerGuide": "IAMCCS Cine Filmmaker Guide",
"IAMCCS_CineFilmmakerGuide1to1": "IAMCCS Cine Filmmaker Guide 1:1",
"IAMCCS_CineSwitch": "IAMCCS CineSwitch Lazy FLF/PromptRelay",
"IAMCCS_CinePromptRelayLatentShapeSync": "IAMCCS Cine PromptRelay Latent Shape Sync",
"IAMCCS_CineFLFLengthCompensator": "IAMCCS Cine FLF Length Compensator",
"IAMCCS_CineBoardDurationLock": "IAMCCS Cine Board Duration Lock",
"IAMCCS_CineLatentDurationCrop": "IAMCCS Cine Latent Duration Crop",
"IAMCCS_CinePromptRelaySafeEncode": "IAMCCS Cine PromptRelay Safe Encode",
"IAMCCS_CineRelayOrBypass": "IAMCCS Cine Relay Or Bypass",
"IAMCCS_CinePromptArchitect": "IAMCCS CinePrompt Architect",
"IAMCCS_BoardMaker": "IAMCCS_BoardMaker",
"IAMCCS_CineMusicVideoPlanner": "IAMCCS Cine Videoclip Maker Planner",
"IAMCCS_CineShotPlanner": "IAMCCS Cine Shot Planner",
"IAMCCS_CineRefLatentControl": "IAMCCS Cine Reference Latent Control",
@@ -613,6 +671,7 @@ NODE_DISPLAY_NAME_MAPPINGS = {
"IAMCCS_WDC_LTXSequencer": "IAMCCS Cine LTX Sequencer (legacy alias)",
"IAMCCS_CineLTXSequencerExact": "IAMCCS Cine LTX Sequencer Exact",
"IAMCCS_CineFLFEngineSimple": "IAMCCS Cine FLF Engine Simple",
"IAMCCS_CineFLFEngineSimpleDyno": "IAMCCS Cine FLF Engine Simple Dyno",
"IAMCCS_WDC_LTXSequencerFixed5": "IAMCCS Cine LTX Sequencer Fixed 5 (legacy alias)",
"IAMCCS_LTX2_InitLatentSampler": "LTX-2 Init Latent Sampler 🧱",
"IAMCCS_LTX2_LoopingSampler": "LTX-2 Looping Sampler (temporal overlap) 🧷",
@@ -700,6 +759,9 @@ NODE_DISPLAY_NAME_MAPPINGS = {
"IAMCCS_VAEDecodeTiledSafe": "VAE Decode Tiled (safe, optional cleanup)",
"IAMCCS_VAEDecodeToDisk": "VAE Decode → Disk (frames, low RAM)",
"IAMCCS_HWProbeRecommendations": "HW Probe Recommendations (JSON)",
"IAMCCS_DetailAtelier": "IAMCCS Detail Atelier",
"IAMCCS_DetailAtelierAdvanced": "IAMCCS Detail Atelier Advanced",
"IAMCCS_DetailAtelierSampler": "IAMCCS Detail Atelier",
"IAMCCS_MoveAhead": "MoveAhead (FreeLong spectral blend) 🎬",
"IAMCCS_MoveAheadEnforcer": "MoveAhead Enforcer (3-tier motion lock) 🎬",
@@ -855,7 +917,7 @@ def setup_api_routes() -> None:
im = ImageOps.exif_transpose(im).convert("RGB")
if abs(rotation) > 0.001:
fill = tuple(int(v) for v in im.resize((1, 1), Image.Resampling.BILINEAR).getpixel((0, 0)))
im = im.rotate(rotation, resample=Image.Resampling.BICUBIC, expand=True, fillcolor=fill)
im = im.rotate(-rotation, resample=Image.Resampling.BICUBIC, expand=True, fillcolor=fill)
src_w, src_h = im.size
crop_box = None
preview_crop_box = None
@@ -940,6 +1002,221 @@ def setup_api_routes() -> None:
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@routes.post("/api/iamccs/cine/save_shotboard_package")
async def iamccs_cine_save_shotboard_package(request):
try:
import base64
import copy
import json
import re
import shutil
import time
import folder_paths
def _sanitize(value, fallback="cine_shotboard_package"):
clean = re.sub(r'[<>:"/\\|?*\x00-\x1F]+', "_", str(value or fallback).strip())
clean = re.sub(r"\s+", "_", clean).strip("._")
return (clean[:90] or fallback)
def _split_paths(value):
if isinstance(value, list):
return [str(item).strip() for item in value if str(item or "").strip()]
raw = str(value or "").strip()
if not raw:
return []
try:
parsed = json.loads(raw)
if isinstance(parsed, list):
return [str(item).strip() for item in parsed if str(item or "").strip()]
except Exception:
pass
if "\n" in raw or "\r" in raw:
return [item.strip() for item in raw.splitlines() if item.strip()]
return [item.strip() for item in raw.split(",") if item.strip()] if "," in raw else [raw]
def _add_path(paths, seen, value):
clean = str(value or "").strip()
if clean and clean not in seen:
seen.add(clean)
paths.append(clean)
def _collect_paths(board):
paths = []
seen = set()
for path in _split_paths(board.get("image_paths")):
_add_path(paths, seen, path)
for item in board.get("images") or []:
if isinstance(item, dict):
_add_path(paths, seen, item.get("path") or item.get("original_path") or item.get("filename") or item.get("name"))
for seg in (board.get("segments") or []):
if isinstance(seg, dict):
_add_path(paths, seen, seg.get("imageFile") or seg.get("image_file") or seg.get("path"))
timeline = board.get("timeline")
if isinstance(timeline, dict):
for seg in (timeline.get("segments") or []):
if isinstance(seg, dict):
_add_path(paths, seen, seg.get("imageFile") or seg.get("image_file") or seg.get("path"))
timeline_data = board.get("timeline_data")
if isinstance(timeline_data, str) and timeline_data.strip():
try:
parsed = json.loads(timeline_data)
for path in _split_paths(parsed.get("image_paths") if isinstance(parsed, dict) else None):
_add_path(paths, seen, path)
for seg in (parsed.get("segments") if isinstance(parsed, dict) else []) or []:
if isinstance(seg, dict):
_add_path(paths, seen, seg.get("imageFile") or seg.get("image_file") or seg.get("path"))
except Exception:
pass
return paths
def _resolve_source(path, input_dir):
clean = str(path or "").strip()
if not clean or clean.startswith("data:"):
return None
if os.path.isabs(clean):
return os.path.abspath(os.path.expanduser(clean))
return os.path.abspath(os.path.join(input_dir, clean.replace("/", os.sep)))
def _rewrite_segments(segments, path_map):
if not isinstance(segments, list):
return
for seg in segments:
if not isinstance(seg, dict):
continue
for key in ("imageFile", "image_file", "path"):
value = str(seg.get(key) or "").strip()
if value in path_map:
seg[key] = path_map[value]
def _rewrite_board_paths(board, ordered_paths, path_map):
board["image_paths"] = [path_map.get(path, path) for path in ordered_paths if path_map.get(path, path)]
if isinstance(board.get("images"), list):
for item in board["images"]:
if not isinstance(item, dict):
continue
value = str(item.get("path") or item.get("original_path") or item.get("filename") or item.get("name") or "").strip()
if value in path_map:
item["original_path"] = value
item["path"] = path_map[value]
_rewrite_segments(board.get("segments"), path_map)
if isinstance(board.get("timeline"), dict):
board["timeline"]["image_paths"] = [path_map.get(path, path) for path in ordered_paths if path_map.get(path, path)]
_rewrite_segments(board["timeline"].get("segments"), path_map)
if isinstance(board.get("timeline_data"), str) and board["timeline_data"].strip():
try:
parsed = json.loads(board["timeline_data"])
if isinstance(parsed, dict):
parsed["image_paths"] = [path_map.get(path, path) for path in ordered_paths if path_map.get(path, path)]
_rewrite_segments(parsed.get("segments"), path_map)
board["timeline_data"] = json.dumps(parsed, indent=2)
except Exception:
pass
def _write_data_url(value, out_path):
header, _, payload = str(value or "").partition(",")
if not header.startswith("data:image") or not payload:
return False
with open(out_path, "wb") as fh:
fh.write(base64.b64decode(payload))
return True
data = await request.json()
board = data.get("board")
if not isinstance(board, dict):
return web.json_response({"error": "Missing board object"}, status=400)
input_dir = folder_paths.get_input_directory()
package_root = os.path.join(input_dir, "IAMCCS_shotboard_packages")
package_name = _sanitize(data.get("package_name") or data.get("label") or f"cine_shotboard_{int(time.time())}")
package_dir = os.path.join(package_root, package_name)
images_dir = os.path.join(package_dir, "images")
os.makedirs(images_dir, exist_ok=True)
original_board = copy.deepcopy(board)
ordered_paths = _collect_paths(original_board)
path_map = {}
manifest_images = []
allowed_ext = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tif", ".tiff", ".avif"}
for index, original_path in enumerate(ordered_paths, start=1):
source_path = _resolve_source(original_path, input_dir)
source_name = os.path.basename(source_path or original_path) or f"ref_{index:03d}.png"
ext = os.path.splitext(source_name)[1].lower()
if ext not in allowed_ext:
ext = ".png"
clean_stem = _sanitize(os.path.splitext(source_name)[0], f"ref_{index:03d}")[:52]
filename = f"ref_{index:03d}_{clean_stem}{ext}"
target_path = os.path.join(images_dir, filename)
rel_path = "/".join(["IAMCCS_shotboard_packages", package_name, "images", filename])
entry = {
"ref": index,
"original_path": original_path,
"package_path": rel_path,
"filename": filename,
}
try:
if str(original_path).startswith("data:image"):
if not _write_data_url(original_path, target_path):
raise ValueError("Unsupported data URL")
else:
if not source_path or not os.path.isfile(source_path):
raise FileNotFoundError(f"Image not found: {original_path}")
shutil.copy2(source_path, target_path)
entry["source_path"] = source_path
entry["bytes"] = os.path.getsize(target_path)
path_map[original_path] = rel_path
except Exception as err:
entry["error"] = str(err)
manifest_images.append(entry)
packaged_board = copy.deepcopy(original_board)
_rewrite_board_paths(packaged_board, ordered_paths, path_map)
saved_at = time.strftime("%Y-%m-%dT%H:%M:%S%z")
packaged_board["metadata"] = {
**(packaged_board.get("metadata") or {}),
"packaged_at": saved_at,
"package_schema": "iamccs.cine.shotboard.package",
}
packaged_board["package"] = {
"name": package_name,
"root": package_dir,
"images_dir": "images",
"images": manifest_images,
}
manifest = {
"metadata": {
"schema": "iamccs.cine.shotboard.package",
"schema_version": 2,
"saved_at": saved_at,
"package_name": package_name,
"package_root": package_dir,
},
"board_file": "board.json",
"images_dir": "images",
"image_count": len(ordered_paths),
"images": manifest_images,
}
with open(os.path.join(package_dir, "board.json"), "w", encoding="utf-8") as fh:
json.dump(packaged_board, fh, indent=2, ensure_ascii=False)
with open(os.path.join(package_dir, "manifest.json"), "w", encoding="utf-8") as fh:
json.dump(manifest, fh, indent=2, ensure_ascii=False)
failed = sum(1 for item in manifest_images if item.get("error"))
return web.json_response({
"ok": True,
"package_name": package_name,
"package_dir": package_dir,
"board_file": os.path.join(package_dir, "board.json"),
"manifest_file": os.path.join(package_dir, "manifest.json"),
"image_count": len(ordered_paths),
"failed_images": failed,
"images": manifest_images,
})
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
@routes.get("/api/iamccs/hw_probe")
async def iamccs_hw_probe_endpoint(request):
try:
+234
View File
@@ -0,0 +1,234 @@
import json
import math
from typing import Any, Dict, Optional
try:
import torch
except Exception: # pragma: no cover - ComfyUI normally provides torch
torch = None
SUPERNODE_LINX_TYPE = "IAMCCS_SUPERNODE_LINX"
def _safe_float(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except Exception:
return float(default)
def _safe_int(value: Any, default: int = 0) -> int:
try:
return int(round(float(value)))
except Exception:
return int(default)
def _safe_json_loads(value: Any) -> Any:
try:
return json.loads(str(value or ""))
except Exception:
return None
def _round_ltx_frames(frames: int, mode: str) -> int:
value = max(1, int(frames))
if str(mode) == "none":
return value
rounded = int(round((value - 1) / 8.0) * 8 + 1)
if str(mode) == "nearest_8n_plus_1":
return max(1, rounded)
return max(1, int(math.ceil(max(0, value - 1) / 8.0) * 8 + 1))
def _cine_linx_resources(cine_linx: Any) -> Dict[str, Any]:
if not isinstance(cine_linx, dict):
return {}
resources = cine_linx.get("resources")
return resources if isinstance(resources, dict) else {}
def _cine_linx_outputs(cine_linx: Any) -> Dict[str, Any]:
if not isinstance(cine_linx, dict):
return {}
outputs = cine_linx.get("outputs")
return outputs if isinstance(outputs, dict) else {}
def _duration_from_cine_linx(cine_linx: Any) -> Optional[float]:
resources = _cine_linx_resources(cine_linx)
outputs = _cine_linx_outputs(cine_linx)
payload = resources.get("cine_payload") if isinstance(resources.get("cine_payload"), dict) else {}
for source in (resources, outputs, payload):
for key in ("cine_duration_seconds", "duration_seconds", "duration", "duration_sec"):
if isinstance(source, dict) and key in source:
value = _safe_float(source.get(key), 0.0)
if value > 0:
return value
return None
def _frame_rate_from_cine_linx(cine_linx: Any) -> Optional[int]:
resources = _cine_linx_resources(cine_linx)
outputs = _cine_linx_outputs(cine_linx)
payload = resources.get("cine_payload") if isinstance(resources.get("cine_payload"), dict) else {}
for source in (resources, outputs, payload):
for key in ("cine_frame_rate", "frame_rate", "fps"):
if isinstance(source, dict) and key in source:
value = _safe_int(source.get(key), 0)
if value > 0:
return value
return None
def _duration_from_timeline_data(timeline_data: Any) -> Optional[float]:
data = _safe_json_loads(timeline_data)
if not isinstance(data, dict):
return None
for key in ("duration_seconds", "duration", "duration_sec"):
value = _safe_float(data.get(key), 0.0)
if value > 0:
return value
payload = data.get("payload") if isinstance(data.get("payload"), dict) else {}
for key in ("duration_seconds", "duration", "duration_sec"):
value = _safe_float(payload.get(key), 0.0)
if value > 0:
return value
return None
def _guide_count_from_timeline_data(timeline_data: Any) -> int:
data = _safe_json_loads(timeline_data)
if isinstance(data, dict):
rows = data.get("guides") or data.get("keyframes") or data.get("rows") or data.get("segments") or []
elif isinstance(data, list):
rows = data
else:
rows = []
count = 0
for row in rows:
if not isinstance(row, dict):
continue
if row.get("use_guide") is False:
continue
ref = _safe_int(row.get("ref", row.get("reference_index", row.get("image_ref", 0))), 0)
strength = _safe_float(row.get("strength", row.get("guide_strength", row.get("force", 1.0))), 1.0)
if ref > 0 and strength > 0:
count += 1
return count
class IAMCCS_CineBoardDurationLock:
"""Lock production latent length to the shotboard duration, without guide-tail padding."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"timeline_data": ("STRING", {"default": "", "multiline": True}),
"duration_seconds": ("INT", {"default": 8, "min": 1, "max": 36000, "step": 1}),
"frame_rate": ("INT", {"default": 24, "min": 1, "max": 120, "step": 1}),
"ltx_round_mode": (["up_8n_plus_1", "nearest_8n_plus_1", "none"], {"default": "up_8n_plus_1"}),
},
"optional": {
"cine_linx": (SUPERNODE_LINX_TYPE,),
},
}
RETURN_TYPES = ("INT", "INT", "FLOAT", "INT", "STRING")
RETURN_NAMES = ("model_length_frames", "target_frames", "duration_seconds_exact", "frame_rate_int", "report")
FUNCTION = "compute"
CATEGORY = "IAMCCS/Cine/00 Utilities"
def compute(self, timeline_data, duration_seconds, frame_rate, ltx_round_mode, cine_linx=None):
fps = max(1, _frame_rate_from_cine_linx(cine_linx) or _safe_int(frame_rate, 24))
duration = (
_duration_from_cine_linx(cine_linx)
or _duration_from_timeline_data(timeline_data)
or max(0.1, _safe_float(duration_seconds, 8.0))
)
raw_frames = max(1, int(round(float(duration) * fps)))
target_frames = _round_ltx_frames(raw_frames, str(ltx_round_mode))
guide_count = _guide_count_from_timeline_data(timeline_data)
report = json.dumps(
{
"node": "IAMCCS_CineBoardDurationLock",
"duration_seconds_exact": float(duration),
"frame_rate": int(fps),
"raw_frames": int(raw_frames),
"target_frames": int(target_frames),
"model_length_frames": int(target_frames),
"guide_count_observed": int(guide_count),
"guide_tail_padding_frames": 0,
"ltx_round_mode": str(ltx_round_mode),
"truth": "The production latent length is locked to the board duration. Guides do not extend final narrative duration.",
},
ensure_ascii=False,
indent=2,
)
return int(target_frames), int(target_frames), float(duration), int(fps), report
class IAMCCS_CineLatentDurationCrop:
"""Safety crop for video latents when a workflow branch still carries padded tail frames."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"latent": ("LATENT",),
"target_frames": ("INT", {"default": 81, "min": 1, "max": 36000, "step": 1}),
"ltx_time_factor": ("INT", {"default": 8, "min": 1, "max": 32, "step": 1}),
}
}
RETURN_TYPES = ("LATENT", "STRING")
RETURN_NAMES = ("latent", "report")
FUNCTION = "crop"
CATEGORY = "IAMCCS/Cine/00 Utilities"
def crop(self, latent, target_frames, ltx_time_factor):
if not isinstance(latent, dict) or torch is None:
return latent, json.dumps({"node": "IAMCCS_CineLatentDurationCrop", "changed": False, "reason": "invalid latent"})
samples = latent.get("samples")
if not torch.is_tensor(samples) or samples.ndim < 3:
return latent, json.dumps({"node": "IAMCCS_CineLatentDurationCrop", "changed": False, "reason": "missing samples tensor"})
target = max(1, _safe_int(target_frames, 1))
time_factor = max(1, _safe_int(ltx_time_factor, 8))
target_latent_frames = max(1, int(math.ceil(max(1, target - 1) / float(time_factor))) + 1)
current_latent_frames = int(samples.shape[2]) if samples.ndim == 5 else int(samples.shape[0])
next_latent = dict(latent)
if samples.ndim == 5 and current_latent_frames > target_latent_frames:
next_latent["samples"] = samples[:, :, :target_latent_frames, :, :].clone()
elif samples.ndim != 5 and current_latent_frames > target_latent_frames:
next_latent["samples"] = samples[:target_latent_frames].clone()
if "noise_mask" in latent and torch.is_tensor(latent["noise_mask"]):
mask = latent["noise_mask"]
if mask.ndim == 5 and int(mask.shape[2]) > target_latent_frames:
next_latent["noise_mask"] = mask[:, :, :target_latent_frames, :, :].clone()
elif mask.ndim != 5 and int(mask.shape[0]) > target_latent_frames:
next_latent["noise_mask"] = mask[:target_latent_frames].clone()
report = json.dumps(
{
"node": "IAMCCS_CineLatentDurationCrop",
"target_pixel_frames": int(target),
"target_latent_frames": int(target_latent_frames),
"current_latent_frames": int(current_latent_frames),
"changed": bool(current_latent_frames > target_latent_frames),
},
ensure_ascii=False,
indent=2,
)
return next_latent, report
NODE_CLASS_MAPPINGS = {
"IAMCCS_CineBoardDurationLock": IAMCCS_CineBoardDurationLock,
"IAMCCS_CineLatentDurationCrop": IAMCCS_CineLatentDurationCrop,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"IAMCCS_CineBoardDurationLock": "IAMCCS Cine Board Duration Lock",
"IAMCCS_CineLatentDurationCrop": "IAMCCS Cine Latent Duration Crop",
}
+170
View File
@@ -0,0 +1,170 @@
from __future__ import annotations
import torch
from comfy_api.latest import io as comfy_io
try:
from comfy_extras.nodes_lt import _append_guide_attention_entry
except Exception: # pragma: no cover - depends on ComfyUI/KJ/LTX version
_append_guide_attention_entry = None
from .iamccs_wdc_ltx_port import IAMCCS_CineFLFEngineSimple
class IAMCCS_CineFLFEngineSimpleDyno(IAMCCS_CineFLFEngineSimple):
"""Timeline/manual FLF engine with locked guide frames plus dynamic attention."""
@classmethod
def define_schema(cls):
inputs = [
comfy_io.Conditioning.Input("positive", tooltip="Positive conditioning to which guide keyframe info will be added"),
comfy_io.Conditioning.Input("negative", tooltip="Negative conditioning to which guide keyframe info will be added"),
comfy_io.Vae.Input("vae", tooltip="Video VAE used to encode the guide images"),
comfy_io.Latent.Input("latent", tooltip="Video latent, guides are added to this latent"),
comfy_io.Image.Input("multi_input", tooltip="Batched images from the Cine Shotboard or Reference Board"),
]
inputs.append(comfy_io.Int.Input(
"num_images",
default=0,
min=0,
max=50,
step=1,
display_name="images_loaded",
tooltip="How many reference guide slots to read.",
))
inputs.append(comfy_io.Combo.Input(
"insert_mode",
options=["frames", "seconds"],
default="frames",
tooltip="How manual_keyframes are interpreted.",
))
inputs.append(comfy_io.Int.Input(
"frame_rate",
default=24,
min=1,
max=120,
step=1,
tooltip="FPS used when manual_keyframes/timeline_data use seconds.",
))
inputs.append(comfy_io.String.Input(
"timeline_data",
default="",
multiline=True,
tooltip="Optional synced Cine Shotboard FLF timeline. When connected, this overrides manual keyframes.",
))
inputs.append(comfy_io.String.Input(
"manual_keyframes",
default="0 | 1 | 1.00\n96 | 2 | 0.92\n184 | 3 | 0.88\n288 | 4 | 0.84\n-1 | 5 | 0.88",
multiline=True,
tooltip="Compact fallback. Frames mode: frame | ref | strength. Seconds mode: second | ref | strength.",
))
return comfy_io.Schema(
node_id="IAMCCS_CineFLFEngineSimpleDyno",
display_name="IAMCCS Cine FLF Engine Simple Dyno",
category="IAMCCS/Cine/02 Single Generation",
description=(
"Compact IAMCCS technical FLF engine. Dyno keeps append_keyframe image locking "
"and also adds legacy guide attention entries for motion-heavy shots."
),
inputs=inputs,
outputs=[
comfy_io.Conditioning.Output(display_name="positive"),
comfy_io.Conditioning.Output(display_name="negative"),
comfy_io.Latent.Output(display_name="latent", tooltip="Video latent with added FLF guides"),
],
)
@classmethod
def _append_dyno_attention(cls, positive, negative, encoded, strength):
if _append_guide_attention_entry is None:
return positive, negative, False
pre_filter_count = int(encoded.shape[2] * encoded.shape[3] * encoded.shape[4])
guide_latent_shape = [int(v) for v in encoded.shape[2:]]
positive, negative = _append_guide_attention_entry(
positive,
negative,
pre_filter_count,
guide_latent_shape,
strength=float(strength),
)
return positive, negative, True
@classmethod
def execute(cls, positive, negative, vae, latent, multi_input, num_images, **kwargs) -> comfy_io.NodeOutput:
scale_factors = vae.downscale_index_formula
latent_image = latent["samples"].clone()
if "noise_mask" in latent:
noise_mask = latent["noise_mask"].clone()
else:
batch, _, latent_frames, _, _ = latent_image.shape
noise_mask = torch.ones(
(batch, 1, latent_frames, 1, 1),
dtype=torch.float32,
device=latent_image.device,
)
_, _, latent_length, latent_height, latent_width = latent_image.shape
batch_size = multi_input.shape[0] if multi_input is not None else 0
insert_mode = kwargs.get("insert_mode", "frames")
frame_rate = kwargs.get("frame_rate", 24)
try:
image_limit = max(0, min(50, int(num_images)))
except Exception:
image_limit = 0
if image_limit <= 0 or batch_size <= 0:
return comfy_io.NodeOutput(positive, negative, {"samples": latent_image, "noise_mask": noise_mask})
synced_keyframes = cls._parse_synced_timeline(kwargs.get("timeline_data", ""), frame_rate)
if synced_keyframes:
guide_items = synced_keyframes
else:
guide_items = cls._parse_manual_keyframes(
kwargs.get("manual_keyframes", ""),
insert_mode,
frame_rate,
image_limit,
)
max_ref = min(batch_size, image_limit)
applied = 0
attention_entries = 0
for guide in guide_items:
i = int(guide.get("reference_index", 1))
if i > max_ref:
continue
img = multi_input[i - 1:i]
if img is None:
continue
f_idx = int(guide.get("frame", 0))
strength = cls._clamp_float(guide.get("strength", 1.0), 0.0, 1.0, 1.0)
image_1, encoded = cls.encode(vae, latent_width, latent_height, img, scale_factors)
frame_idx, latent_idx = cls.get_latent_index(positive, latent_length, len(image_1), f_idx, scale_factors)
assert latent_idx + encoded.shape[2] <= latent_length, "Conditioning frames exceed the length of the latent sequence."
positive, negative, latent_image, noise_mask = cls.append_keyframe(
positive,
negative,
frame_idx,
latent_image,
noise_mask,
encoded,
strength,
scale_factors,
)
positive, negative, did_attention = cls._append_dyno_attention(positive, negative, encoded, strength)
applied += 1
if did_attention:
attention_entries += 1
print(
"[IAMCCS CineFLFEngineSimpleDyno] "
f"guides={applied} guide_attention_available={_append_guide_attention_entry is not None} "
f"guide_attention_entries={attention_entries}"
)
return comfy_io.NodeOutput(positive, negative, {"samples": latent_image, "noise_mask": noise_mask})
+287
View File
@@ -0,0 +1,287 @@
from __future__ import annotations
import json
from copy import deepcopy
from typing import Any, Dict
import torch
from comfy_extras.nodes_lt import LTXVAddGuide
try:
from comfy_extras.nodes_lt import _append_guide_attention_entry
except Exception: # pragma: no cover - depends on ComfyUI/KJ/LTX version
_append_guide_attention_entry = None
from .iamccs_cine_nodes import (
IAMCCS_CineFLFProductor,
_clamp,
_json_report,
_safe_bool,
_safe_float,
_safe_int,
)
class IAMCCS_CineFLFProductorDyno(IAMCCS_CineFLFProductor):
"""Dynamic guide Productor variant.
The normal Productor keeps the Filmmaker V3 clean guide path for maximum
image lock. Dyno keeps the same inputs/outputs and guide plan contract, but
also appends the legacy guide attention entry used by the older dynamic
FLF path. This makes it safe to A/B test motion-heavy shots without
changing existing workflows.
"""
CATEGORY = "IAMCCS/Cine/02 Single Generation"
@classmethod
def INPUT_TYPES(cls):
schema = deepcopy(IAMCCS_CineFLFProductor.INPUT_TYPES())
schema.setdefault("optional", {})
schema["optional"]["guide_attention_mode"] = (
["skip_step_transitions", "all_guides", "off"],
{
"default": "skip_step_transitions",
"tooltip": (
"Controls Dyno guide_attention_entries. "
"skip_step_transitions keeps step-transition pairs on the clean Production guide path."
),
},
)
return schema
@staticmethod
def _normalise_attention_mode(value: Any) -> str:
mode = str(value or "skip_step_transitions").strip().lower()
if mode in {"all", "all_guides", "on", "true"}:
return "all_guides"
if mode in {"off", "none", "false", "0"}:
return "off"
return "skip_step_transitions"
@staticmethod
def _apply_attention_entry(positive, negative, encoded, strength: float):
if _append_guide_attention_entry is None:
return positive, negative, False, "guide attention helper unavailable"
pre_filter_count = int(encoded.shape[2] * encoded.shape[3] * encoded.shape[4])
guide_latent_shape = [int(v) for v in encoded.shape[2:]]
positive, negative = _append_guide_attention_entry(
positive,
negative,
pre_filter_count,
guide_latent_shape,
strength=float(strength),
)
return positive, negative, True, ""
@staticmethod
def _execute_guide_data(
positive,
negative,
vae,
latent,
guide_data: Dict[str, Any],
strength_scale: float,
tail_safety_frames: int,
guide_attention_mode: str = "skip_step_transitions",
):
latent_samples = latent.get("samples") if isinstance(latent, dict) else None
if not torch.is_tensor(latent_samples) or latent_samples.ndim != 5:
return positive, negative, latent, {
"applied_guides": [],
"skipped_guides": [{"reason": "invalid latent samples"}],
"latent_pixel_frames": 0,
"guide_attention_available": _append_guide_attention_entry is not None,
"guide_attention_entries": False,
"guide_attention_entry_count": 0,
}
scale_factors = vae.downscale_index_formula
latent_image = latent_samples.clone()
if "noise_mask" in latent:
noise_mask = latent["noise_mask"].clone()
else:
batch, _, latent_frames, _, _ = latent_image.shape
noise_mask = torch.ones((batch, 1, latent_frames, 1, 1), dtype=torch.float32, device=latent_image.device)
_, _, latent_length, latent_height, latent_width = latent_image.shape
time_scale_factor = int(scale_factors[0]) if scale_factors else 8
latent_pixel_frames = max(1, (int(latent_length) - 1) * max(1, time_scale_factor) + 1)
max_frame = max(0, int(latent_pixel_frames) - 1 - max(0, int(tail_safety_frames)))
images = guide_data.get("images", []) if isinstance(guide_data, dict) else []
insert_frames = guide_data.get("insert_frames", []) if isinstance(guide_data, dict) else []
strengths = guide_data.get("strengths", []) if isinstance(guide_data, dict) else []
labels = guide_data.get("labels", []) if isinstance(guide_data, dict) else []
refs = guide_data.get("reference_indices", []) if isinstance(guide_data, dict) else []
step_sources = guide_data.get("step_transition_sources", []) if isinstance(guide_data, dict) else []
step_targets = guide_data.get("step_transition_targets", []) if isinstance(guide_data, dict) else []
step_protected = guide_data.get("step_transition_protected", []) if isinstance(guide_data, dict) else []
attention_mode = IAMCCS_CineFLFProductorDyno._normalise_attention_mode(guide_attention_mode)
applied = []
skipped = []
attention_entries = 0
attention_skipped = []
for idx, img in enumerate(images):
label = str(labels[idx]) if idx < len(labels) else f"guide_{idx + 1}"
ref = _safe_int(refs[idx], idx + 1) if idx < len(refs) else idx + 1
if not torch.is_tensor(img) or img.ndim != 4 or int(img.shape[0]) <= 0:
skipped.append({"label": label, "reference_index": int(ref), "reason": "empty guide image tensor"})
continue
requested_frame = _safe_int(insert_frames[idx], 0) if idx < len(insert_frames) else 0
frame_idx = requested_frame if requested_frame < 0 else min(max(0, requested_frame), max_frame)
strength = _clamp((_safe_float(strengths[idx], 1.0) if idx < len(strengths) else 1.0) * float(strength_scale), 0.0, 1.0, 1.0)
if strength <= 0:
skipped.append({"label": label, "reference_index": int(ref), "frame": int(frame_idx), "reason": "zero strength"})
continue
try:
image_1, encoded = LTXVAddGuide.encode(vae, latent_width, latent_height, img, scale_factors)
conditioning_frame, latent_idx = LTXVAddGuide.get_latent_index(positive, latent_length, len(image_1), frame_idx, scale_factors)
if latent_idx + encoded.shape[2] > latent_length:
skipped.append({
"label": label,
"reference_index": int(ref),
"frame": int(frame_idx),
"reason": "guide exceeds latent length",
})
continue
positive, negative, latent_image, noise_mask = LTXVAddGuide.append_keyframe(
positive,
negative,
conditioning_frame,
latent_image,
noise_mask,
encoded,
float(strength),
scale_factors,
)
is_step_source = _safe_bool(step_sources[idx], False) if idx < len(step_sources) else False
is_step_target = _safe_bool(step_targets[idx], False) if idx < len(step_targets) else False
if idx > 0 and idx - 1 < len(step_sources):
is_step_target = bool(is_step_target or _safe_bool(step_sources[idx - 1], False))
is_step_protected = _safe_bool(step_protected[idx], False) if idx < len(step_protected) else False
is_step_protected = bool(is_step_protected or is_step_source or is_step_target)
if attention_mode == "off":
did_attention = False
attention_reason = "guide_attention_mode=off"
elif attention_mode == "skip_step_transitions" and is_step_protected:
positive, negative, did_attention, attention_reason = IAMCCS_CineFLFProductorDyno._apply_attention_entry(
positive,
negative,
encoded,
0.0,
)
if did_attention:
attention_reason = "step transition protected; guide_attention_entry kept with zero strength for LTX mask alignment"
else:
positive, negative, did_attention, attention_reason = IAMCCS_CineFLFProductorDyno._apply_attention_entry(
positive,
negative,
encoded,
float(strength),
)
if did_attention:
attention_entries += 1
elif attention_reason:
attention_skipped.append({
"label": label,
"reference_index": int(ref),
"reason": attention_reason,
"step_transition_protected": bool(is_step_protected),
})
except Exception as exc:
skipped.append({"label": label, "reference_index": int(ref), "frame": int(frame_idx), "reason": f"dyno guide apply failed: {exc}"})
continue
applied.append({
"label": label,
"reference_index": int(ref),
"requested_frame": int(requested_frame),
"frame": int(frame_idx),
"strength": float(strength),
"guide_attention_entry": did_attention,
"guide_attention_strength": 0.0 if (attention_mode == "skip_step_transitions" and is_step_protected) else float(strength) if did_attention else None,
"step_transition_protected": bool(is_step_protected),
})
return positive, negative, {"samples": latent_image, "noise_mask": noise_mask}, {
"applied_guides": applied,
"skipped_guides": skipped,
"attention_skipped": attention_skipped,
"latent_pixel_frames": int(latent_pixel_frames),
"guide_attention_available": _append_guide_attention_entry is not None,
"guide_attention_entries": attention_entries > 0,
"guide_attention_entry_count": int(attention_entries),
"guide_attention_mode": attention_mode,
"guide_strength_semantics": "append_keyframe_plus_legacy_guide_attention_entry",
"compatibility_mode": "dyno_motion_conditioning_guide",
}
def execute(
self,
positive,
negative,
vae,
latent,
multi_input,
guide_plan_json,
strength_scale,
tail_safety_frames,
timeline_data="",
duration_seconds=20,
frame_rate=24,
guide_data=None,
guide_attention_mode="skip_step_transitions",
):
attention_mode = self._normalise_attention_mode(guide_attention_mode)
if isinstance(guide_data, dict) and guide_data.get("images"):
positive, negative, current_latent, guide_data_report = self._execute_guide_data(
positive,
negative,
vae,
latent,
guide_data,
float(strength_scale),
int(tail_safety_frames),
attention_mode,
)
report = _json_report({
"node": "IAMCCS_CineFLFProductorDyno",
"mode": "wdc_guide_data_compatible_dyno",
"guide_count": len(guide_data.get("images", [])),
"applied_count": len(guide_data_report.get("applied_guides", [])),
"skipped_count": len(guide_data_report.get("skipped_guides", [])),
**guide_data_report,
"truth": (
"Dyno consumed GUIDE_DATA and appended guide_attention_entries according to guide_attention_mode. "
"Step-transition-protected guides stay on the clean Production guide path by default."
),
})
return positive, negative, current_latent, report
plan = self._parse_plan(guide_plan_json, timeline_data, duration_seconds, frame_rate)
clean_guide_data = self._guide_data_from_plan_and_images(plan, multi_input, float(strength_scale))
positive, negative, current_latent, guide_data_report = self._execute_guide_data(
positive,
negative,
vae,
latent,
clean_guide_data,
1.0,
int(tail_safety_frames),
attention_mode,
)
report = _json_report({
"node": "IAMCCS_CineFLFProductorDyno",
"mode": "explicit_dyno_guide_productor",
"plan_source": plan.get("source", "") if isinstance(plan, dict) else "",
"reference_count": int(multi_input.shape[0]) if torch.is_tensor(multi_input) and multi_input.ndim == 4 else 0,
"guide_count": len(clean_guide_data.get("images", [])),
"applied_count": len(guide_data_report.get("applied_guides", [])),
"skipped_count": len(guide_data_report.get("skipped_guides", [])),
**guide_data_report,
"truth": (
"Dyno uses the same guide plan and image slots as CineFLFProductor, then conditionally appends "
"legacy guide_attention_entries. Default skip_step_transitions protects dolly/bridge pairs."
),
})
return positive, negative, current_latent, report
+2872 -125
View File
File diff suppressed because it is too large Load Diff
+419
View File
@@ -0,0 +1,419 @@
from __future__ import annotations
import base64
import copy
import hashlib
import io
import json
import math
import os
import re
import time
from typing import Any, Dict, Iterable, List, Optional, Tuple
import folder_paths
from PIL import Image, ImageOps
SUPERNODE_LINX_TYPE = "IAMCCS_SUPERNODE_LINX"
_IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tif", ".tiff", ".avif"}
_MOTION_WORDS = {
"sea", "ocean", "water", "wave", "waves", "foam", "surface", "underwater",
"bubble", "bubbles", "particle", "particles", "caustic", "caustics",
"dolly", "push", "push-in", "camera", "glide", "travel", "travels", "move",
"moves", "moving", "motion", "parallax", "drift", "drifts", "roll", "rolls",
"slide", "slides", "flow", "flows",
}
_IDENTITY_WORDS = {
"face", "siren", "sirena", "mermaid", "creature", "character", "eyes", "mouth",
"skin", "portrait", "closeup", "close-up", "macro", "head", "woman",
}
def _clamp(value: Any, lo: float, hi: float, default: float = 0.0) -> float:
try:
number = float(value)
except Exception:
number = float(default)
return max(float(lo), min(float(hi), number))
def _safe_int(value: Any, default: int = 0) -> int:
try:
return int(round(float(value)))
except Exception:
return int(default)
def _safe_float(value: Any, default: float = 0.0) -> float:
try:
return float(value)
except Exception:
return float(default)
def _safe_json_loads(value: Any, fallback: Any) -> Any:
if isinstance(value, (dict, list)):
return copy.deepcopy(value)
try:
return json.loads(str(value or ""))
except Exception:
return copy.deepcopy(fallback)
def _sanitize(value: Any, fallback: str = "cine_resolution_parity") -> str:
clean = re.sub(r'[<>:"/\\|?*\x00-\x1F]+', "_", str(value or fallback).strip())
clean = re.sub(r"\s+", "_", clean).strip("._")
return (clean[:80] or fallback)
def _split_paths(value: Any) -> List[str]:
if isinstance(value, list):
return [str(item).strip() for item in value if str(item or "").strip()]
raw = str(value or "").strip()
if not raw:
return []
parsed = _safe_json_loads(raw, None)
if isinstance(parsed, list):
return [str(item).strip() for item in parsed if str(item or "").strip()]
if "\n" in raw or "\r" in raw:
return [item.strip() for item in raw.splitlines() if item.strip()]
parts = [item.strip() for item in raw.split(",") if item.strip()]
return parts if len(parts) > 1 else [raw]
def _join_paths(paths: Iterable[str]) -> str:
return "\n".join(str(path or "").strip() for path in paths if str(path or "").strip())
def _text_for_segment(seg: Dict[str, Any]) -> str:
parts = [
seg.get("label", ""),
seg.get("prompt", ""),
seg.get("camera", ""),
seg.get("transition", ""),
seg.get("note", ""),
seg.get("relay_prompt", ""),
seg.get("step_transition_prompt", ""),
]
return " ".join(str(part or "").lower() for part in parts)
def _classify_segment(seg: Dict[str, Any]) -> str:
text = _text_for_segment(seg)
identity_hits = sum(1 for word in _IDENTITY_WORDS if word in text)
motion_hits = sum(1 for word in _MOTION_WORDS if word in text)
if identity_hits and identity_hits >= motion_hits:
return "identity"
if motion_hits:
return "motion"
return "neutral"
def _strength_from_segment(seg: Dict[str, Any], default: float = 1.0) -> float:
for key in ("guide_strength", "guideStrength", "strength", "force", "motion_force"):
if key in seg and seg.get(key) is not None:
return _clamp(seg.get(key), 0.0, 1.0, default)
return _clamp(default, 0.0, 1.0, 1.0)
def _set_strength(seg: Dict[str, Any], value: float) -> None:
strength = float(_clamp(value, 0.0, 1.0, 1.0))
for key in ("guideStrength", "guide_strength", "strength", "force", "imageLockStrength", "image_lock_strength"):
if key in seg:
seg[key] = strength
if not any(key in seg for key in ("guideStrength", "guide_strength", "strength", "force")):
seg["guideStrength"] = strength
def _resolve_image_path(path: str) -> Optional[str]:
clean = str(path or "").strip()
if not clean or clean.startswith("data:"):
return None
if os.path.isabs(clean):
return os.path.abspath(os.path.expanduser(clean))
return os.path.abspath(os.path.join(folder_paths.get_input_directory(), clean.replace("/", os.sep)))
def _open_image(path_or_data: str) -> Image.Image:
clean = str(path_or_data or "").strip()
if clean.startswith("data:image"):
_, _, payload = clean.partition(",")
if not payload:
raise ValueError("Invalid image data URL")
return Image.open(io.BytesIO(base64.b64decode(payload)))
source = _resolve_image_path(clean)
if not source or not os.path.isfile(source):
raise FileNotFoundError(f"Reference image not found: {clean}")
return Image.open(source)
def _resize_cover(image: Image.Image, width: int, height: int, resample: int) -> Image.Image:
src_w, src_h = image.size
if src_w <= 0 or src_h <= 0:
return image.resize((width, height), resample)
target_ratio = width / float(height)
src_ratio = src_w / float(src_h)
if src_ratio >= target_ratio:
crop_h = src_h
crop_w = int(round(crop_h * target_ratio))
else:
crop_w = src_w
crop_h = int(round(crop_w / target_ratio))
left = max(0, int(round((src_w - crop_w) / 2.0)))
top = max(0, int(round((src_h - crop_h) / 2.0)))
return image.crop((left, top, min(src_w, left + crop_w), min(src_h, top + crop_h))).resize((width, height), resample)
def _prefilter_image_to_target(
source: str,
target_width: int,
target_height: int,
reference_width: int,
reference_height: int,
method: str,
package_tag: str,
) -> str:
resample = {
"nearest": Image.Resampling.NEAREST,
"bilinear": Image.Resampling.BILINEAR,
"bicubic": Image.Resampling.BICUBIC,
"lanczos": Image.Resampling.LANCZOS,
}.get(str(method or "bicubic").lower(), Image.Resampling.BICUBIC)
with _open_image(source) as raw:
image = ImageOps.exif_transpose(raw).convert("RGB")
semantic = _resize_cover(image, max(64, reference_width), max(64, reference_height), resample)
target = semantic.resize((max(64, target_width), max(64, target_height)), resample)
out_root = os.path.join(folder_paths.get_input_directory(), "IAMCCS_resolution_parity")
os.makedirs(out_root, exist_ok=True)
source_name = os.path.basename(_resolve_image_path(source) or "reference.png")
stem = _sanitize(os.path.splitext(source_name)[0], "ref")[:48]
digest = hashlib.sha1(str(source).encode("utf-8", errors="ignore")).hexdigest()[:10]
filename = f"{package_tag}_{stem}_{digest}_{target_width}x{target_height}.png"
out_path = os.path.join(out_root, filename)
target.save(out_path, "PNG", optimize=True)
return "IAMCCS_resolution_parity/" + filename
def _rewrite_image_refs(obj: Any, path_map: Dict[str, str]) -> None:
if isinstance(obj, dict):
for key, value in list(obj.items()):
if key in {"imageFile", "image_file", "path"} and isinstance(value, str) and value in path_map:
obj[key] = path_map[value]
else:
_rewrite_image_refs(value, path_map)
elif isinstance(obj, list):
for item in obj:
_rewrite_image_refs(item, path_map)
def _apply_strength_translation(segments: List[Dict[str, Any]], motion_multiplier: float, identity_multiplier: float) -> Dict[str, Any]:
counts = {"motion": 0, "identity": 0, "neutral": 0}
values = []
for seg in segments:
if not isinstance(seg, dict):
continue
role = _classify_segment(seg)
counts[role] = counts.get(role, 0) + 1
old = _strength_from_segment(seg, 1.0)
if role == "motion":
new = old * motion_multiplier
elif role == "identity":
new = old * identity_multiplier
else:
new = old
new = _clamp(new, 0.0, 1.0, old)
_set_strength(seg, new)
seg["resolution_parity_role"] = role
seg["resolution_parity_original_strength"] = float(old)
seg["resolution_parity_strength"] = float(new)
values.append({"label": seg.get("label", ""), "role": role, "from": float(old), "to": float(new)})
return {"counts": counts, "values": values}
class IAMCCS_CineResolutionParityTranslator:
"""Translate a proven lower-resolution shotboard into a higher-resolution run."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"cine_linx": (SUPERNODE_LINX_TYPE,),
"enabled": ("BOOLEAN", {"default": True}),
"reference_width": ("INT", {"default": 1024, "min": 64, "max": 8192, "step": 32}),
"reference_height": ("INT", {"default": 576, "min": 64, "max": 8192, "step": 32}),
"guide_prefilter": ("BOOLEAN", {"default": True}),
"prefilter_method": (["bicubic", "lanczos", "bilinear", "nearest"], {"default": "bicubic"}),
"motion_multiplier": ("FLOAT", {
"default": 0.0,
"min": 0.0,
"max": 2.0,
"step": 0.01,
"tooltip": "0 = auto, computed as reference/target linear scale.",
}),
"identity_multiplier": ("FLOAT", {"default": 1.05, "min": 0.0, "max": 2.0, "step": 0.01}),
},
}
RETURN_TYPES = (SUPERNODE_LINX_TYPE, "STRING")
RETURN_NAMES = ("cine_linx", "report")
FUNCTION = "translate"
CATEGORY = "IAMCCS/Cine/02 Single Generation"
@classmethod
def IS_CHANGED(cls, *args, **kwargs):
return float("nan")
def translate(
self,
cine_linx,
enabled=True,
reference_width=1024,
reference_height=576,
guide_prefilter=True,
prefilter_method="bicubic",
motion_multiplier=0.0,
identity_multiplier=1.05,
):
if not isinstance(cine_linx, dict):
return cine_linx, json.dumps({"ok": False, "error": "cine_linx is not a dictionary"}, indent=2)
if not enabled:
return cine_linx, json.dumps({"ok": True, "enabled": False, "note": "Resolution parity translator bypassed."}, indent=2)
out = dict(cine_linx)
original_resources = cine_linx.get("resources") if isinstance(cine_linx.get("resources"), dict) else {}
resources = dict(original_resources)
out["resources"] = resources
outputs = dict(cine_linx.get("outputs") if isinstance(cine_linx.get("outputs"), dict) else {})
out["outputs"] = outputs
payload = resources.get("cine_payload")
if not isinstance(payload, dict):
payload = {}
else:
payload = copy.deepcopy(payload)
resources["cine_payload"] = payload
target_width = _safe_int(resources.get("cine_image_width", outputs.get("width", payload.get("image_width", 0))), 0)
target_height = _safe_int(resources.get("cine_image_height", outputs.get("height", payload.get("image_height", 0))), 0)
if target_width <= 0:
target_width = _safe_int(payload.get("image_width"), 768)
if target_height <= 0:
target_height = _safe_int(payload.get("image_height"), 432)
resources["cine_image_width"] = int(target_width)
resources["cine_image_height"] = int(target_height)
payload["image_width"] = int(target_width)
payload["image_height"] = int(target_height)
outputs["width"] = int(target_width)
outputs["height"] = int(target_height)
ref_w = max(64, _safe_int(reference_width, 1024))
ref_h = max(64, _safe_int(reference_height, 576))
scale_x = target_width / float(ref_w)
scale_y = target_height / float(ref_h)
linear_scale = math.sqrt(max(0.0001, scale_x * scale_y))
auto_motion = 1.0 / linear_scale if linear_scale > 1.0 else 1.0
motion_mul = float(auto_motion if _safe_float(motion_multiplier, 0.0) <= 0 else motion_multiplier)
identity_mul = float(identity_multiplier)
visual_segments = _safe_json_loads(resources.get("cine_visual_segments_json", payload.get("visual_segments", [])), [])
if isinstance(visual_segments, dict):
visual_segments = visual_segments.get("segments", [])
if not isinstance(visual_segments, list):
visual_segments = []
strength_report = _apply_strength_translation(visual_segments, motion_mul, identity_mul)
original_paths = _split_paths(resources.get("cine_image_paths", payload.get("image_paths", "")))
path_map: Dict[str, str] = {}
image_errors = []
package_tag = f"parity_{int(time.time() * 1000)}"
if guide_prefilter and target_width > 0 and target_height > 0:
for path in original_paths:
try:
path_map[path] = _prefilter_image_to_target(
path,
target_width,
target_height,
ref_w,
ref_h,
str(prefilter_method),
package_tag,
)
except Exception as exc:
image_errors.append({"path": path, "error": str(exc)})
for seg in visual_segments:
if not isinstance(seg, dict):
continue
image_file = str(seg.get("imageFile", seg.get("image_file", "")) or "").strip()
if image_file and image_file not in path_map:
try:
path_map[image_file] = _prefilter_image_to_target(
image_file,
target_width,
target_height,
ref_w,
ref_h,
str(prefilter_method),
package_tag,
)
except Exception as exc:
image_errors.append({"path": image_file, "error": str(exc)})
if path_map:
translated_paths = [path_map.get(path, path) for path in original_paths]
resources["cine_image_paths"] = _join_paths(translated_paths)
payload["image_paths"] = resources["cine_image_paths"]
_rewrite_image_refs(visual_segments, path_map)
_rewrite_image_refs(payload.get("rows"), path_map)
_rewrite_image_refs(payload.get("guide_rows"), path_map)
resources["cine_visual_segments_json"] = json.dumps(visual_segments, ensure_ascii=False)
payload["visual_segments"] = visual_segments
payload["resolution_parity"] = {
"enabled": True,
"reference_width": int(ref_w),
"reference_height": int(ref_h),
"target_width": int(target_width),
"target_height": int(target_height),
"linear_scale": float(linear_scale),
"motion_multiplier": float(motion_mul),
"identity_multiplier": float(identity_mul),
"guide_prefilter": bool(guide_prefilter),
"prefilter_method": str(prefilter_method),
"prefiltered_images": len(path_map),
}
resources["cine_resolution_parity"] = payload["resolution_parity"]
stages = []
for stage in out.get("stages", []) if isinstance(out.get("stages"), list) else []:
if isinstance(stage, dict) and isinstance(stage.get("payload"), dict):
stage_copy = dict(stage)
stage_copy["payload"] = payload
stages.append(stage_copy)
else:
stages.append(stage)
if stages:
out["stages"] = stages
resources["cine_payload"] = payload
out["resource_keys"] = sorted(resources.keys())
out["resource_types"] = {key: type(value).__name__ for key, value in resources.items()}
report = {
"ok": True,
"node": "IAMCCS_CineResolutionParityTranslator",
"reference": [int(ref_w), int(ref_h)],
"target": [int(target_width), int(target_height)],
"linear_scale": round(float(linear_scale), 4),
"motion_multiplier": round(float(motion_mul), 4),
"identity_multiplier": round(float(identity_mul), 4),
"guide_prefilter": bool(guide_prefilter),
"prefiltered_images": len(path_map),
"image_errors": image_errors,
"strength_translation": strength_report,
}
return out, json.dumps(report, indent=2, ensure_ascii=False)
+84
View File
@@ -0,0 +1,84 @@
from __future__ import annotations
import json
from typing import Any
class IAMCCS_CineStage2PreviewToggle:
"""Gate KJ/LTX sampling previews on the model branch that feeds Stage 2."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": ("MODEL",),
"stage2_preview_enabled": ("BOOLEAN", {
"default": False,
"tooltip": (
"OFF removes the KJ LTX2 sampling_preview wrapper from this model branch. "
"Use it before Stage 2 to avoid TAELTX preview OOM while keeping the render active."
),
}),
},
}
RETURN_TYPES = ("MODEL", "STRING")
RETURN_NAMES = ("model", "report")
FUNCTION = "toggle"
CATEGORY = "IAMCCS/Cine/02 Single Generation"
@classmethod
def IS_CHANGED(cls, *args, **kwargs):
return float("nan")
@staticmethod
def _count_sampling_preview_wrappers(model: Any) -> int:
try:
import comfy.patcher_extension as patcher_extension
wrappers = model.get_wrappers(patcher_extension.WrappersMP.OUTER_SAMPLE, "sampling_preview")
return len(wrappers or [])
except Exception:
return 0
def toggle(self, model: Any, stage2_preview_enabled: bool = False):
before = self._count_sampling_preview_wrappers(model)
selected_model = model
removed = 0
if not bool(stage2_preview_enabled):
try:
import comfy.patcher_extension as patcher_extension
selected_model = model.clone()
selected_model.remove_wrappers_with_key(
patcher_extension.WrappersMP.OUTER_SAMPLE,
"sampling_preview",
)
after = self._count_sampling_preview_wrappers(selected_model)
removed = max(0, int(before) - int(after))
except Exception as exc:
report = {
"node": "IAMCCS_CineStage2PreviewToggle",
"stage2_preview_enabled": bool(stage2_preview_enabled),
"status": "failed_to_remove_wrapper",
"sampling_preview_wrappers_before": int(before),
"error": str(exc),
}
return model, json.dumps(report, indent=2)
else:
after = before
report = {
"node": "IAMCCS_CineStage2PreviewToggle",
"stage2_preview_enabled": bool(stage2_preview_enabled),
"status": "preview_passthrough" if bool(stage2_preview_enabled) else "stage2_preview_wrapper_removed",
"sampling_preview_wrappers_before": int(before),
"sampling_preview_wrappers_after": int(after),
"sampling_preview_wrappers_removed": int(removed),
"truth": (
"Place this node only on the Stage 2 model branch. OFF disables the KJ/TAELTX sampling "
"preview callback for Stage 2 without disabling PromptRelay, guide data, or Stage 2 sampling."
),
}
return selected_model, json.dumps(report, indent=2)
+83
View File
@@ -0,0 +1,83 @@
from __future__ import annotations
import json
from typing import Any
class IAMCCS_CineStage2BypassSwitch:
"""Lazy switch for choosing Stage 1 or Stage 2 latents before decode."""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"use_second_stage": ("BOOLEAN", {
"default": True,
"tooltip": "True: decode Stage 2 output. False: bypass Stage 2 and decode Stage 1 output.",
}),
},
"optional": {
"stage1_audio_latent": ("LATENT", {"lazy": True}),
"stage1_video_latent": ("LATENT", {"lazy": True}),
"stage2_audio_latent": ("LATENT", {"lazy": True}),
"stage2_video_latent": ("LATENT", {"lazy": True}),
},
}
RETURN_TYPES = ("LATENT", "LATENT", "STRING")
RETURN_NAMES = ("audio_latent", "video_latent", "report")
FUNCTION = "switch"
CATEGORY = "IAMCCS/Cine/02 Single Generation"
@classmethod
def IS_CHANGED(cls, *args, **kwargs):
return float("nan")
def check_lazy_status(
self,
use_second_stage: bool,
stage1_audio_latent: Any = None,
stage1_video_latent: Any = None,
stage2_audio_latent: Any = None,
stage2_video_latent: Any = None,
**kwargs,
):
if bool(use_second_stage):
needed = []
if stage2_audio_latent is None:
needed.append("stage2_audio_latent")
if stage2_video_latent is None:
needed.append("stage2_video_latent")
return needed
needed = []
if stage1_audio_latent is None:
needed.append("stage1_audio_latent")
if stage1_video_latent is None:
needed.append("stage1_video_latent")
return needed
def switch(
self,
use_second_stage: bool,
stage1_audio_latent: Any = None,
stage1_video_latent: Any = None,
stage2_audio_latent: Any = None,
stage2_video_latent: Any = None,
):
if bool(use_second_stage):
if stage2_audio_latent is None or stage2_video_latent is None:
raise ValueError("Stage 2 is enabled, but Stage 2 latents are not connected.")
report = {
"node": "IAMCCS_CineStage2BypassSwitch",
"use_second_stage": True,
"selected": "stage2",
}
return stage2_audio_latent, stage2_video_latent, json.dumps(report, indent=2)
if stage1_audio_latent is None or stage1_video_latent is None:
raise ValueError("Stage 2 bypass is enabled, but Stage 1 latents are not connected.")
report = {
"node": "IAMCCS_CineStage2BypassSwitch",
"use_second_stage": False,
"selected": "stage1_bypass",
}
return stage1_audio_latent, stage1_video_latent, json.dumps(report, indent=2)
+835
View File
@@ -0,0 +1,835 @@
from __future__ import annotations
import json
import math
import importlib.util
import sys
import types
from dataclasses import dataclass
from pathlib import Path
VRAM_PRESETS = ("8GB", "12GB", "16GB", "24GB")
QUALITY_PRESETS = ("preview", "balanced", "quality")
DETAIL_ATELIER_LINX_TYPE = "IAMCCS_DETAIL_ATELIER_LINX"
HARDWARE_MODES = ("auto", "manual")
@dataclass(frozen=True)
class DetailAtelierPreset:
target_long_edge: int
temporal_tile_size: int
temporal_overlap: int
temporal_overlap_cond_strength: float
guiding_strength: float
cond_image_strength: float
horizontal_tiles: int
vertical_tiles: int
spatial_overlap: int
vae_tile_size: int
vae_overlap: int
vae_temporal_size: int
vae_temporal_overlap: int
reserved_vram_gb: float
cleanup_before_decode: bool
run_final_upscale: bool
run_interpolation: bool
final_upscale_mode: str
notes: str
_PRESETS: dict[str, dict[str, DetailAtelierPreset]] = {
"8GB": {
"preview": DetailAtelierPreset(768, 32, 16, 0.45, 0.85, 1.0, 1, 1, 1, 256, 32, 96, 16, 1.5, True, False, False, "off", "Fastest safe pass for 8GB cards."),
"balanced": DetailAtelierPreset(960, 40, 16, 0.50, 0.90, 1.0, 1, 1, 1, 256, 32, 128, 16, 1.5, True, True, False, "rtx_x2_optional", "8GB default: detail modestly, upscale after."),
"quality": DetailAtelierPreset(1280, 40, 16, 0.55, 0.95, 1.0, 1, 1, 1, 256, 32, 128, 16, 2.0, True, True, False, "rtx_x2", "Slow but still conservative for 8GB."),
},
"12GB": {
"preview": DetailAtelierPreset(960, 40, 16, 0.45, 0.90, 1.0, 1, 1, 1, 384, 48, 128, 16, 2.0, True, False, False, "off", "Preview for 12GB cards."),
"balanced": DetailAtelierPreset(1280, 48, 16, 0.50, 1.00, 1.0, 1, 1, 1, 384, 48, 128, 16, 2.0, True, True, False, "rtx_x2", "Recommended 12GB daily preset."),
"quality": DetailAtelierPreset(1536, 56, 24, 0.55, 1.00, 1.0, 1, 1, 1, 384, 48, 128, 16, 2.5, True, True, True, "rtx_x2", "Final-ish 12GB preset; interpolation is optional."),
},
"16GB": {
"preview": DetailAtelierPreset(1280, 48, 16, 0.45, 0.95, 1.0, 1, 1, 1, 384, 48, 128, 16, 2.5, True, False, False, "off", "Quick check for 16GB cards."),
"balanced": DetailAtelierPreset(1536, 56, 24, 0.50, 1.00, 1.0, 1, 1, 1, 512, 64, 192, 24, 3.0, False, True, False, "rtx_x2", "Recommended 16GB default."),
"quality": DetailAtelierPreset(1920, 64, 24, 0.60, 1.00, 1.0, 1, 1, 1, 512, 64, 192, 24, 3.0, False, True, True, "rtx_x2", "High detail before final upscale."),
},
"24GB": {
"preview": DetailAtelierPreset(1536, 56, 24, 0.45, 1.00, 1.0, 1, 1, 1, 512, 64, 192, 24, 3.0, False, False, False, "off", "Fast preview on high VRAM."),
"balanced": DetailAtelierPreset(1920, 64, 24, 0.55, 1.00, 1.0, 1, 1, 1, 512, 64, 256, 32, 3.5, False, True, False, "rtx_x2", "24GB default for clean final passes."),
"quality": DetailAtelierPreset(2160, 80, 32, 0.60, 1.00, 1.0, 2, 1, 2, 512, 64, 256, 32, 4.0, False, True, True, "rtx_x2", "Experimental high-end pass; spatial tiling may be slow."),
},
}
def _round_to_multiple(value: float, multiple: int, mode: str = "nearest") -> int:
value = max(float(multiple), float(value))
if mode == "down":
rounded = math.floor(value / multiple) * multiple
elif mode == "up":
rounded = math.ceil(value / multiple) * multiple
else:
rounded = round(value / multiple) * multiple
return max(multiple, int(rounded))
def _scaled_dims(width: int, height: int, long_edge: int, multiple: int) -> tuple[int, int]:
width = int(width)
height = int(height)
long_edge = int(long_edge)
if width <= 0 or height <= 0:
return (0, 0)
scale = float(long_edge) / float(max(width, height))
out_w = _round_to_multiple(width * scale, multiple)
out_h = _round_to_multiple(height * scale, multiple)
return (out_w, out_h)
def _preset_for(vram_preset: str, quality_mode: str) -> tuple[str, str, DetailAtelierPreset]:
vram = str(vram_preset if vram_preset in _PRESETS else "12GB")
quality = str(quality_mode if quality_mode in _PRESETS[vram] else "balanced")
return vram, quality, _PRESETS[vram][quality]
def _safe_hardware_info() -> dict:
info = {
"cuda_available": False,
"cuda_device_name": None,
"cuda_total_vram_gb": None,
"system_ram_gb": None,
"warnings": [],
}
try:
import torch
if torch.cuda.is_available():
idx = torch.cuda.current_device()
prop = torch.cuda.get_device_properties(idx)
total = getattr(prop, "total_memory", None)
info["cuda_available"] = True
info["cuda_device_name"] = getattr(prop, "name", None)
if total:
info["cuda_total_vram_gb"] = float(total) / (1024.0**3)
except Exception as e:
info["warnings"].append(f"cuda_probe_failed={e!r}")
try:
import psutil # type: ignore
info["system_ram_gb"] = float(psutil.virtual_memory().total) / (1024.0**3)
except Exception as e:
info["warnings"].append(f"ram_probe_failed={e!r}")
return info
def _vram_preset_from_gb(vram_gb: float | None, fallback: str) -> str:
if vram_gb is None:
return fallback if fallback in VRAM_PRESETS else "12GB"
if vram_gb <= 8.5:
return "8GB"
if vram_gb <= 12.5:
return "12GB"
if vram_gb <= 16.5:
return "16GB"
return "24GB"
def _latent_shape_info(latents, vae=None) -> dict:
out = {
"latent_frames": None,
"latent_height": None,
"latent_width": None,
"time_scale_factor": 8,
"estimated_frames": None,
}
try:
samples = latents.get("samples") if isinstance(latents, dict) else None
shape = getattr(samples, "shape", None)
if shape is not None and len(shape) >= 5:
out["latent_frames"] = int(shape[2])
out["latent_height"] = int(shape[3])
out["latent_width"] = int(shape[4])
except Exception:
pass
try:
formula = getattr(vae, "downscale_index_formula", None)
if formula and len(formula) >= 1:
out["time_scale_factor"] = max(1, int(formula[0]))
except Exception:
pass
if out["latent_frames"] is not None:
out["estimated_frames"] = int(out["latent_frames"]) * int(out["time_scale_factor"])
return out
def _estimate_temporal_chunks(latent_frames: int | None, temporal_tile_size: int, temporal_overlap: int, time_scale_factor: int) -> dict:
if not latent_frames or latent_frames <= 0:
return {"chunks": None, "latent_tile": None, "latent_overlap": None, "latent_step": None}
scale = max(1, int(time_scale_factor or 8))
latent_tile = max(1, int(temporal_tile_size) // scale)
latent_overlap = max(0, int(temporal_overlap) // scale)
if latent_overlap >= latent_tile:
latent_overlap = max(0, latent_tile - 1)
latent_step = max(1, latent_tile - latent_overlap)
if int(latent_frames) <= latent_tile:
chunks = 1
else:
chunks = int(math.ceil((int(latent_frames) - latent_tile) / float(latent_step))) + 1
return {
"chunks": int(chunks),
"latent_tile": int(latent_tile),
"latent_overlap": int(latent_overlap),
"latent_step": int(latent_step),
}
def _auto_adjust_loop_values(
*,
selected_vram: str,
quality: str,
temporal_tile_size: int,
temporal_overlap: int,
horizontal_tiles: int,
vertical_tiles: int,
spatial_overlap: int,
prefer_low_ram: bool,
hardware: dict,
latent_info: dict,
) -> tuple[str, dict, list[str]]:
decisions: list[str] = []
detected_vram = _vram_preset_from_gb(hardware.get("cuda_total_vram_gb"), selected_vram)
effective_vram = detected_vram
if detected_vram != selected_vram:
decisions.append(f"vram_preset:auto {selected_vram}->{detected_vram}")
ram_gb = hardware.get("system_ram_gb")
low_ram = bool(prefer_low_ram)
if isinstance(ram_gb, (int, float)) and ram_gb <= 40.0 and effective_vram in {"8GB", "12GB"}:
low_ram = True
decisions.append("low_ram:auto_enabled")
max_tile_by_vram = {
"8GB": 40,
"12GB": 48,
"16GB": 64,
"24GB": 80,
}
min_tile_by_vram = {
"8GB": 32,
"12GB": 40,
"16GB": 48,
"24GB": 56,
}
if low_ram:
max_tile_by_vram["8GB"] = 32
max_tile_by_vram["12GB"] = 48
max_tile = max_tile_by_vram.get(effective_vram, 48)
min_tile = min_tile_by_vram.get(effective_vram, 40)
if int(temporal_tile_size) > max_tile:
decisions.append(f"temporal_tile_size:cap {temporal_tile_size}->{max_tile}")
temporal_tile_size = max_tile
if int(temporal_tile_size) < min_tile and quality != "preview":
decisions.append(f"temporal_tile_size:floor {temporal_tile_size}->{min_tile}")
temporal_tile_size = min_tile
max_overlap = 16 if effective_vram in {"8GB", "12GB"} else 24
if int(temporal_overlap) > max_overlap:
decisions.append(f"temporal_overlap:cap {temporal_overlap}->{max_overlap}")
temporal_overlap = max_overlap
if int(temporal_overlap) >= int(temporal_tile_size):
new_overlap = max(0, int(temporal_tile_size) // 2)
decisions.append(f"temporal_overlap:repair {temporal_overlap}->{new_overlap}")
temporal_overlap = new_overlap
latent_area = None
if latent_info.get("latent_width") and latent_info.get("latent_height"):
latent_area = int(latent_info["latent_width"]) * int(latent_info["latent_height"])
if low_ram or effective_vram in {"8GB", "12GB"}:
if horizontal_tiles != 1 or vertical_tiles != 1 or spatial_overlap != 1:
decisions.append("spatial_tiling:low_ram_force_1x1")
horizontal_tiles = 1
vertical_tiles = 1
spatial_overlap = 1
elif latent_area is not None and latent_area > 4096 and effective_vram == "16GB":
horizontal_tiles = max(1, int(horizontal_tiles))
vertical_tiles = max(1, int(vertical_tiles))
spatial_overlap = max(1, int(spatial_overlap))
chunk_info = _estimate_temporal_chunks(
latent_info.get("latent_frames"),
int(temporal_tile_size),
int(temporal_overlap),
int(latent_info.get("time_scale_factor") or 8),
)
if chunk_info.get("chunks") and chunk_info["chunks"] > 12 and effective_vram in {"8GB", "12GB"}:
decisions.append(f"long_clip:estimated_chunks={chunk_info['chunks']}")
values = {
"temporal_tile_size": int(temporal_tile_size),
"temporal_overlap": int(temporal_overlap),
"horizontal_tiles": int(horizontal_tiles),
"vertical_tiles": int(vertical_tiles),
"spatial_overlap": int(spatial_overlap),
"low_ram_effective": bool(low_ram),
"chunk_info": chunk_info,
}
return effective_vram, values, decisions
def _load_ltxv_looping_sampler_class():
module_name = "_iamccs_ltxvideo_runtime.looping_sampler"
cached = sys.modules.get(module_name)
if cached is not None and hasattr(cached, "LTXVLoopingSampler"):
return cached.LTXVLoopingSampler
custom_nodes_dir = Path(__file__).resolve().parent.parent
ltx_dir = custom_nodes_dir / "ComfyUI-LTXVideo"
looping_path = ltx_dir / "looping_sampler.py"
if not looping_path.exists():
raise ImportError(f"ComfyUI-LTXVideo looping_sampler.py not found at {looping_path}")
package_name = "_iamccs_ltxvideo_runtime"
if package_name not in sys.modules:
package = types.ModuleType(package_name)
package.__path__ = [str(ltx_dir)]
package.__file__ = str(ltx_dir / "__init__.py")
package.__package__ = package_name
sys.modules[package_name] = package
spec = importlib.util.spec_from_file_location(module_name, looping_path)
if spec is None or spec.loader is None:
raise ImportError(f"Unable to load ComfyUI-LTXVideo looping sampler from {looping_path}")
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
return module.LTXVLoopingSampler
class IAMCCS_DetailAtelier:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"vram_preset": (VRAM_PRESETS, {"default": "12GB"}),
"quality_mode": (QUALITY_PRESETS, {"default": "balanced"}),
"source_width": ("INT", {"default": 0, "min": 0, "max": 16384, "step": 8}),
"source_height": ("INT", {"default": 0, "min": 0, "max": 16384, "step": 8}),
"target_override_long_edge": ("INT", {"default": 0, "min": 0, "max": 8192, "step": 8}),
"dimension_multiple": ("INT", {"default": 16, "min": 8, "max": 128, "step": 8}),
"prefer_low_ram": ("BOOLEAN", {"default": True}),
"allow_interpolation": ("BOOLEAN", {"default": True}),
"allow_final_upscale": ("BOOLEAN", {"default": True}),
"pretty_json": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = (
"INT",
"INT",
"INT",
"INT",
"INT",
"FLOAT",
"FLOAT",
"FLOAT",
"INT",
"INT",
"INT",
"INT",
"INT",
"INT",
"INT",
"FLOAT",
"BOOLEAN",
"BOOLEAN",
"BOOLEAN",
"STRING",
"STRING",
"STRING",
)
RETURN_NAMES = (
"target_long_edge",
"target_width",
"target_height",
"temporal_tile_size",
"temporal_overlap",
"temporal_overlap_cond_strength",
"guiding_strength",
"cond_image_strength",
"horizontal_tiles",
"vertical_tiles",
"spatial_overlap",
"vae_tile_size",
"vae_overlap",
"vae_temporal_size",
"vae_temporal_overlap",
"reserved_vram_gb",
"cleanup_before_decode",
"run_final_upscale",
"run_interpolation",
"final_upscale_mode",
"report_json",
"report",
)
FUNCTION = "plan"
CATEGORY = "IAMCCS/LTX-2/Detail Atelier"
def plan(
self,
vram_preset: str,
quality_mode: str,
source_width: int,
source_height: int,
target_override_long_edge: int,
dimension_multiple: int,
prefer_low_ram: bool,
allow_interpolation: bool,
allow_final_upscale: bool,
pretty_json: bool,
):
vram = str(vram_preset if vram_preset in _PRESETS else "12GB")
quality = str(quality_mode if quality_mode in _PRESETS[vram] else "balanced")
preset = _PRESETS[vram][quality]
target_long_edge = int(target_override_long_edge) if int(target_override_long_edge) > 0 else preset.target_long_edge
target_width, target_height = _scaled_dims(
int(source_width),
int(source_height),
target_long_edge,
max(8, int(dimension_multiple)),
)
vae_tile_size = preset.vae_tile_size
vae_overlap = preset.vae_overlap
vae_temporal_size = preset.vae_temporal_size
vae_temporal_overlap = preset.vae_temporal_overlap
cleanup_before_decode = bool(preset.cleanup_before_decode)
if bool(prefer_low_ram):
if vram == "8GB":
vae_tile_size = min(vae_tile_size, 256)
vae_overlap = min(vae_overlap, 32)
cleanup_before_decode = True
elif vram == "12GB":
vae_tile_size = min(vae_tile_size, 384)
vae_overlap = min(vae_overlap, 48)
cleanup_before_decode = True
run_final_upscale = bool(allow_final_upscale and preset.run_final_upscale)
run_interpolation = bool(allow_interpolation and preset.run_interpolation)
final_upscale_mode = preset.final_upscale_mode if run_final_upscale else "off"
data = {
"node": "IAMCCS_DetailAtelier",
"vram_preset": vram,
"quality_mode": quality,
"source": {
"width": int(source_width),
"height": int(source_height),
},
"target": {
"long_edge": int(target_long_edge),
"width": int(target_width),
"height": int(target_height),
"dimension_multiple": int(dimension_multiple),
},
"looper": {
"temporal_tile_size": preset.temporal_tile_size,
"temporal_overlap": preset.temporal_overlap,
"temporal_overlap_cond_strength": preset.temporal_overlap_cond_strength,
"guiding_strength": preset.guiding_strength,
"cond_image_strength": preset.cond_image_strength,
"horizontal_tiles": preset.horizontal_tiles,
"vertical_tiles": preset.vertical_tiles,
"spatial_overlap": preset.spatial_overlap,
},
"decode": {
"vae_tile_size": int(vae_tile_size),
"vae_overlap": int(vae_overlap),
"vae_temporal_size": int(vae_temporal_size),
"vae_temporal_overlap": int(vae_temporal_overlap),
"cleanup_before_decode": bool(cleanup_before_decode),
},
"post": {
"reserved_vram_gb": preset.reserved_vram_gb,
"run_final_upscale": run_final_upscale,
"run_interpolation": run_interpolation,
"final_upscale_mode": final_upscale_mode,
},
"strategy": "detail first, upscale after",
"notes": preset.notes,
}
report_json = json.dumps(data, ensure_ascii=False, indent=2 if bool(pretty_json) else None)
report = (
f"{vram} {quality}: target_edge={target_long_edge}"
f"{f' -> {target_width}x{target_height}' if target_width and target_height else ''}, "
f"looper={preset.temporal_tile_size}/{preset.temporal_overlap}, "
f"decode={vae_tile_size}/{vae_overlap}/{vae_temporal_size}/{vae_temporal_overlap}, "
f"cleanup={cleanup_before_decode}, upscale={final_upscale_mode}, rife={run_interpolation}. "
"Strategy: detail first, upscale after."
)
return (
int(target_long_edge),
int(target_width),
int(target_height),
int(preset.temporal_tile_size),
int(preset.temporal_overlap),
float(preset.temporal_overlap_cond_strength),
float(preset.guiding_strength),
float(preset.cond_image_strength),
int(preset.horizontal_tiles),
int(preset.vertical_tiles),
int(preset.spatial_overlap),
int(vae_tile_size),
int(vae_overlap),
int(vae_temporal_size),
int(vae_temporal_overlap),
float(preset.reserved_vram_gb),
bool(cleanup_before_decode),
bool(run_final_upscale),
bool(run_interpolation),
final_upscale_mode,
report_json,
report,
)
class IAMCCS_DetailAtelierSampler:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"model": ("MODEL", {"tooltip": "Diffusion model used for the detail pass."}),
"vae": ("VAE", {"tooltip": "Video VAE used by LTXVideo."}),
"noise": ("NOISE", {"tooltip": "Noise object; the seed is reused internally."}),
"sampler": ("SAMPLER", {"tooltip": "Sampler selected for the detail pass."}),
"sigmas": ("SIGMAS", {"tooltip": "Sigma schedule for the detail pass."}),
"guider": ("GUIDER", {"tooltip": "CFG/STG guider for the detail pass."}),
"latents": ("LATENT", {"tooltip": "Input video latents to detail."}),
"vram_preset": (VRAM_PRESETS, {"default": "12GB"}),
"quality_mode": (QUALITY_PRESETS, {"default": "balanced"}),
"prefer_low_ram": ("BOOLEAN", {"default": True}),
"allow_spatial_tiling": ("BOOLEAN", {"default": True}),
"adain_factor": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"guiding_start_step": ("INT", {"default": 0, "min": 0, "max": 1000}),
"guiding_end_step": ("INT", {"default": 1000, "min": 0, "max": 1000}),
"optional_cond_image_indices": ("STRING", {"default": "0"}),
"per_tile_seed_offsets": ("STRING", {"default": "0"}),
"pretty_json": ("BOOLEAN", {"default": True}),
"hardware_mode": (HARDWARE_MODES, {"default": "auto"}),
"oom_retry": ("BOOLEAN", {"default": True}),
},
"optional": {
"atelier_advanced": (DETAIL_ATELIER_LINX_TYPE, {}),
"optional_cond_images": ("IMAGE", {}),
"optional_guiding_latents": ("LATENT", {}),
"optional_positive_conditionings": ("CONDITIONING", {}),
"optional_negative_index_latents": ("LATENT", {}),
"optional_normalizing_latents": ("LATENT", {}),
},
}
RETURN_TYPES = ("LATENT", "STRING", "STRING")
RETURN_NAMES = ("denoised_output", "report_json", "report")
FUNCTION = "detail"
CATEGORY = "IAMCCS/LTX-2/Detail Atelier"
def detail(
self,
model,
vae,
noise,
sampler,
sigmas,
guider,
latents,
vram_preset: str,
quality_mode: str,
prefer_low_ram: bool,
allow_spatial_tiling: bool,
adain_factor: float,
guiding_start_step: int,
guiding_end_step: int,
optional_cond_image_indices: str,
per_tile_seed_offsets: str,
pretty_json: bool,
hardware_mode: str = "auto",
oom_retry: bool = True,
atelier_advanced=None,
optional_cond_images=None,
optional_guiding_latents=None,
optional_positive_conditionings=None,
optional_negative_index_latents=None,
optional_normalizing_latents=None,
):
requested_vram, quality, requested_preset = _preset_for(vram_preset, quality_mode)
hardware = _safe_hardware_info()
if str(hardware_mode or "auto") == "auto":
vram = _vram_preset_from_gb(hardware.get("cuda_total_vram_gb"), requested_vram)
_, _, preset = _preset_for(vram, quality)
else:
vram, _, preset = requested_vram, quality, requested_preset
advanced = atelier_advanced if isinstance(atelier_advanced, dict) and atelier_advanced.get("enabled", True) else {}
latent_info = _latent_shape_info(latents, vae)
def adv_int(name: str, default: int) -> int:
try:
value = int(advanced.get(name, 0))
except Exception:
value = 0
return int(value if value > 0 else default)
def adv_float(name: str, default: float) -> float:
try:
value = float(advanced.get(name, -1.0))
except Exception:
value = -1.0
return float(value if value >= 0.0 else default)
temporal_tile_size = adv_int("temporal_tile_size", preset.temporal_tile_size)
temporal_overlap = adv_int("temporal_overlap", preset.temporal_overlap)
guiding_strength = adv_float("guiding_strength", preset.guiding_strength)
temporal_overlap_cond_strength = adv_float(
"temporal_overlap_cond_strength",
preset.temporal_overlap_cond_strength,
)
cond_image_strength = adv_float("cond_image_strength", preset.cond_image_strength)
horizontal_tiles = adv_int("horizontal_tiles", preset.horizontal_tiles)
vertical_tiles = adv_int("vertical_tiles", preset.vertical_tiles)
spatial_overlap = adv_int("spatial_overlap", preset.spatial_overlap)
if not bool(allow_spatial_tiling):
horizontal_tiles = 1
vertical_tiles = 1
spatial_overlap = 1
spatial_mode = str(advanced.get("spatial_tiling_mode", "inherit") or "inherit")
if spatial_mode == "force_off":
horizontal_tiles = 1
vertical_tiles = 1
spatial_overlap = 1
if bool(prefer_low_ram) and vram in {"8GB", "12GB"}:
horizontal_tiles = 1
vertical_tiles = 1
spatial_overlap = min(spatial_overlap, 1)
auto_decisions: list[str] = []
auto_values = {
"chunk_info": _estimate_temporal_chunks(
latent_info.get("latent_frames"),
int(temporal_tile_size),
int(temporal_overlap),
int(latent_info.get("time_scale_factor") or 8),
)
}
if str(hardware_mode or "auto") == "auto":
vram, auto_values, auto_decisions = _auto_adjust_loop_values(
selected_vram=requested_vram,
quality=quality,
temporal_tile_size=int(temporal_tile_size),
temporal_overlap=int(temporal_overlap),
horizontal_tiles=int(horizontal_tiles),
vertical_tiles=int(vertical_tiles),
spatial_overlap=int(spatial_overlap),
prefer_low_ram=bool(prefer_low_ram),
hardware=hardware,
latent_info=latent_info,
)
temporal_tile_size = int(auto_values["temporal_tile_size"])
temporal_overlap = int(auto_values["temporal_overlap"])
horizontal_tiles = int(auto_values["horizontal_tiles"])
vertical_tiles = int(auto_values["vertical_tiles"])
spatial_overlap = int(auto_values["spatial_overlap"])
adain = adv_float("adain_factor", float(adain_factor))
guide_start = adv_int("guiding_start_step", int(guiding_start_step))
guide_end = adv_int("guiding_end_step", int(guiding_end_step))
cond_indices = str(advanced.get("optional_cond_image_indices") or optional_cond_image_indices or "0")
seed_offsets = str(advanced.get("per_tile_seed_offsets") or per_tile_seed_offsets or "0")
looping_sampler_cls = _load_ltxv_looping_sampler_class()
looping_sampler = looping_sampler_cls()
retry_applied = False
def run_loop(ts: int, ov: int, ht: int, vt: int, so: int):
return looping_sampler.sample(
model=model,
vae=vae,
noise=noise,
sampler=sampler,
sigmas=sigmas,
guider=guider,
latents=latents,
guiding_strength=float(guiding_strength),
adain_factor=float(adain),
temporal_tile_size=int(ts),
temporal_overlap=int(ov),
temporal_overlap_cond_strength=float(temporal_overlap_cond_strength),
horizontal_tiles=int(ht),
vertical_tiles=int(vt),
spatial_overlap=int(so),
optional_cond_images=optional_cond_images,
cond_image_strength=float(cond_image_strength),
optional_guiding_latents=optional_guiding_latents,
optional_negative_index_latents=optional_negative_index_latents,
optional_negative_index_strength=1.0,
optional_positive_conditionings=optional_positive_conditionings,
guiding_start_step=int(guide_start),
guiding_end_step=int(guide_end),
optional_cond_image_indices=str(cond_indices),
optional_normalizing_latents=optional_normalizing_latents,
per_tile_seed_offsets=str(seed_offsets),
)
try:
result = run_loop(temporal_tile_size, temporal_overlap, horizontal_tiles, vertical_tiles, spatial_overlap)
except RuntimeError as e:
is_oom = "out of memory" in str(e).lower() or "cuda oom" in str(e).lower()
if not (bool(oom_retry) and str(hardware_mode or "auto") == "auto" and is_oom):
raise
try:
import torch
if torch.cuda.is_available():
torch.cuda.empty_cache()
try:
torch.cuda.ipc_collect()
except Exception:
pass
except Exception:
pass
retry_applied = True
old_ts, old_ov = int(temporal_tile_size), int(temporal_overlap)
temporal_tile_size = max(32, int(temporal_tile_size) - 16)
temporal_overlap = min(16, max(8, int(temporal_tile_size) // 3))
horizontal_tiles = 1
vertical_tiles = 1
spatial_overlap = 1
auto_decisions.append(f"oom_retry:{old_ts}/{old_ov}-> {temporal_tile_size}/{temporal_overlap}, tiles=1x1")
auto_values["chunk_info"] = _estimate_temporal_chunks(
latent_info.get("latent_frames"),
int(temporal_tile_size),
int(temporal_overlap),
int(latent_info.get("time_scale_factor") or 8),
)
result = run_loop(temporal_tile_size, temporal_overlap, horizontal_tiles, vertical_tiles, spatial_overlap)
denoised = result[0] if isinstance(result, tuple) else result
data = {
"node": "IAMCCS_DetailAtelierSampler",
"display": "IAMCCS Detail Atelier",
"hardware_mode": str(hardware_mode or "auto"),
"requested_vram_preset": requested_vram,
"effective_vram_preset": vram,
"quality_mode": quality,
"strategy": "orchestrated internal LTXV loop; detail only, no decode/upscale unification",
"advanced_linx_enabled": bool(advanced),
"hardware": hardware,
"latent": latent_info,
"estimated_chunks": auto_values.get("chunk_info"),
"auto_decisions": auto_decisions,
"oom_retry_applied": bool(retry_applied),
"looper": {
"temporal_tile_size": int(temporal_tile_size),
"temporal_overlap": int(temporal_overlap),
"guiding_strength": float(guiding_strength),
"temporal_overlap_cond_strength": float(temporal_overlap_cond_strength),
"cond_image_strength": float(cond_image_strength),
"horizontal_tiles": int(horizontal_tiles),
"vertical_tiles": int(vertical_tiles),
"spatial_overlap": int(spatial_overlap),
"adain_factor": float(adain),
"guiding_start_step": int(guide_start),
"guiding_end_step": int(guide_end),
"optional_cond_image_indices": str(cond_indices),
"per_tile_seed_offsets": str(seed_offsets),
},
"notes": preset.notes,
}
report_json = json.dumps(data, ensure_ascii=False, indent=2 if bool(pretty_json) else None)
chunks = auto_values.get("chunk_info", {}).get("chunks") if isinstance(auto_values.get("chunk_info"), dict) else None
report = (
f"IAMCCS Detail Atelier {vram}/{quality} ({hardware_mode}): internal looper "
f"{temporal_tile_size}/{temporal_overlap}, "
f"tiles={horizontal_tiles}x{vertical_tiles}, overlap={spatial_overlap}"
f"{f', chunks≈{chunks}' if chunks else ''}."
)
return (denoised, report_json, report)
class IAMCCS_DetailAtelierAdvanced:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"enabled": ("BOOLEAN", {"default": True}),
"temporal_tile_size": ("INT", {"default": 0, "min": 0, "max": 1000, "step": 8}),
"temporal_overlap": ("INT", {"default": 0, "min": 0, "max": 80, "step": 8}),
"guiding_strength": ("FLOAT", {"default": -1.0, "min": -1.0, "max": 1.0, "step": 0.01}),
"temporal_overlap_cond_strength": ("FLOAT", {"default": -1.0, "min": -1.0, "max": 1.0, "step": 0.01}),
"cond_image_strength": ("FLOAT", {"default": -1.0, "min": -1.0, "max": 1.0, "step": 0.01}),
"horizontal_tiles": ("INT", {"default": 0, "min": 0, "max": 6}),
"vertical_tiles": ("INT", {"default": 0, "min": 0, "max": 6}),
"spatial_overlap": ("INT", {"default": 0, "min": 0, "max": 8}),
"spatial_tiling_mode": (["inherit", "force_off"], {"default": "inherit"}),
"adain_factor": ("FLOAT", {"default": -1.0, "min": -1.0, "max": 1.0, "step": 0.01}),
"guiding_start_step": ("INT", {"default": 0, "min": 0, "max": 1000}),
"guiding_end_step": ("INT", {"default": 0, "min": 0, "max": 1000}),
"optional_cond_image_indices": ("STRING", {"default": ""}),
"per_tile_seed_offsets": ("STRING", {"default": ""}),
"pretty_json": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = (DETAIL_ATELIER_LINX_TYPE, "STRING", "STRING")
RETURN_NAMES = ("atelier_advanced", "report_json", "report")
FUNCTION = "build"
CATEGORY = "IAMCCS/LTX-2/Detail Atelier"
def build(
self,
enabled: bool,
temporal_tile_size: int,
temporal_overlap: int,
guiding_strength: float,
temporal_overlap_cond_strength: float,
cond_image_strength: float,
horizontal_tiles: int,
vertical_tiles: int,
spatial_overlap: int,
spatial_tiling_mode: str,
adain_factor: float,
guiding_start_step: int,
guiding_end_step: int,
optional_cond_image_indices: str,
per_tile_seed_offsets: str,
pretty_json: bool,
):
cfg = {
"enabled": bool(enabled),
"temporal_tile_size": int(temporal_tile_size),
"temporal_overlap": int(temporal_overlap),
"guiding_strength": float(guiding_strength),
"temporal_overlap_cond_strength": float(temporal_overlap_cond_strength),
"cond_image_strength": float(cond_image_strength),
"horizontal_tiles": int(horizontal_tiles),
"vertical_tiles": int(vertical_tiles),
"spatial_overlap": int(spatial_overlap),
"spatial_tiling_mode": str(spatial_tiling_mode or "inherit"),
"adain_factor": float(adain_factor),
"guiding_start_step": int(guiding_start_step),
"guiding_end_step": int(guiding_end_step),
"optional_cond_image_indices": str(optional_cond_image_indices or ""),
"per_tile_seed_offsets": str(per_tile_seed_offsets or ""),
}
report_json = json.dumps(cfg, ensure_ascii=False, indent=2 if bool(pretty_json) else None)
active = [k for k, v in cfg.items() if k != "enabled" and v not in (0, -1.0, "", "inherit")]
report = "Detail Atelier Advanced: " + ("disabled" if not enabled else ("overrides " + ", ".join(active) if active else "enabled, no overrides"))
return (cfg, report_json, report)
+8 -44
View File
@@ -9,10 +9,6 @@ import torch
from comfy_api.latest import io as comfy_io
from comfy_extras.nodes_lt import LTXVAddGuide
try:
from comfy_extras.nodes_lt import _append_guide_attention_entry
except Exception:
_append_guide_attention_entry = None
from PIL import Image, ImageOps
@@ -324,16 +320,6 @@ class IAMCCS_WDC_LTXSequencer(LTXVAddGuide):
float(strength),
scale_factors,
)
if _append_guide_attention_entry is not None:
pre_filter_count = t.shape[2] * t.shape[3] * t.shape[4]
guide_latent_shape = list(t.shape[2:])
positive, negative = _append_guide_attention_entry(
positive,
negative,
pre_filter_count,
guide_latent_shape,
strength=float(strength),
)
return comfy_io.NodeOutput(positive, negative, {"samples": latent_image, "noise_mask": noise_mask})
@@ -469,21 +455,19 @@ class IAMCCS_CineLTXSequencerExact(LTXVAddGuide):
float(strength),
scale_factors,
)
if _append_guide_attention_entry is not None:
pre_filter_count = t.shape[2] * t.shape[3] * t.shape[4]
guide_latent_shape = list(t.shape[2:])
positive, negative = _append_guide_attention_entry(
positive,
negative,
pre_filter_count,
guide_latent_shape,
strength=float(strength),
)
return comfy_io.NodeOutput(positive, negative, {"samples": latent_image, "noise_mask": noise_mask})
class IAMCCS_CineFLFEngineSimple(LTXVAddGuide):
@classmethod
def fingerprint_inputs(cls, *args, **kwargs):
return float("nan")
@classmethod
def IS_CHANGED(cls, *args, **kwargs):
return float("nan")
@classmethod
def define_schema(cls):
inputs = [
@@ -735,16 +719,6 @@ class IAMCCS_CineFLFEngineSimple(LTXVAddGuide):
strength,
scale_factors,
)
if _append_guide_attention_entry is not None:
pre_filter_count = t.shape[2] * t.shape[3] * t.shape[4]
guide_latent_shape = list(t.shape[2:])
positive, negative = _append_guide_attention_entry(
positive,
negative,
pre_filter_count,
guide_latent_shape,
strength=strength,
)
return comfy_io.NodeOutput(positive, negative, {"samples": latent_image, "noise_mask": noise_mask})
@@ -846,16 +820,6 @@ class IAMCCS_WDC_LTXSequencerFixed5:
float(strength),
scale_factors,
)
if _append_guide_attention_entry is not None:
pre_filter_count = t.shape[2] * t.shape[3] * t.shape[4]
guide_latent_shape = list(t.shape[2:])
positive, negative = _append_guide_attention_entry(
positive,
negative,
pre_filter_count,
guide_latent_shape,
strength=float(strength),
)
applied += 1
ops.append(f"image_{idx}:f{frame_idx}->t{latent_idx},s={float(strength):.2f}")
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "iamccs-nodes",
"version": "1.4.5",
"version": "1.4.6",
"author": "Carmine Cristallo Scalzi (IAMCCS)",
"description": "IAMCCS nodes for ComfyUI: IAMCCS echosystem for ComfyUI, nodes 4 LoRA, WAN 2.2, WAN 2.1 and LTX-2 pipelines, WANIMAGEMOTION pro added."
}
File diff suppressed because it is too large Load Diff
+308
View File
@@ -0,0 +1,308 @@
// IAMCCS Detail Atelier UI
// Frontend helper for live loop estimates and "Apply Auto" on the Advanced LINX node.
import { app } from "../../scripts/app.js";
import { api } from "../../scripts/api.js";
const ADVANCED_TYPE = "IAMCCS_DetailAtelierAdvanced";
const SAMPLER_TYPE = "IAMCCS_DetailAtelierSampler";
const PREVIEW_WIDGET = "iamccs_detail_atelier_live_preview";
const APPLY_BUTTON = "iamccs_detail_atelier_apply_auto";
const REFRESH_BUTTON = "iamccs_detail_atelier_refresh";
const PRESETS = {
"8GB": {
preview: { temporal_tile_size: 32, temporal_overlap: 16, guiding_strength: 0.85, temporal_overlap_cond_strength: 0.45, cond_image_strength: 1.0 },
balanced: { temporal_tile_size: 40, temporal_overlap: 16, guiding_strength: 0.90, temporal_overlap_cond_strength: 0.50, cond_image_strength: 1.0 },
quality: { temporal_tile_size: 40, temporal_overlap: 16, guiding_strength: 0.95, temporal_overlap_cond_strength: 0.55, cond_image_strength: 1.0 },
},
"12GB": {
preview: { temporal_tile_size: 40, temporal_overlap: 16, guiding_strength: 0.90, temporal_overlap_cond_strength: 0.45, cond_image_strength: 1.0 },
balanced: { temporal_tile_size: 48, temporal_overlap: 16, guiding_strength: 1.00, temporal_overlap_cond_strength: 0.50, cond_image_strength: 1.0 },
quality: { temporal_tile_size: 56, temporal_overlap: 24, guiding_strength: 1.00, temporal_overlap_cond_strength: 0.55, cond_image_strength: 1.0 },
},
"16GB": {
preview: { temporal_tile_size: 48, temporal_overlap: 16, guiding_strength: 0.95, temporal_overlap_cond_strength: 0.45, cond_image_strength: 1.0 },
balanced: { temporal_tile_size: 56, temporal_overlap: 24, guiding_strength: 1.00, temporal_overlap_cond_strength: 0.50, cond_image_strength: 1.0 },
quality: { temporal_tile_size: 64, temporal_overlap: 24, guiding_strength: 1.00, temporal_overlap_cond_strength: 0.60, cond_image_strength: 1.0 },
},
"24GB": {
preview: { temporal_tile_size: 56, temporal_overlap: 24, guiding_strength: 1.00, temporal_overlap_cond_strength: 0.45, cond_image_strength: 1.0 },
balanced: { temporal_tile_size: 64, temporal_overlap: 24, guiding_strength: 1.00, temporal_overlap_cond_strength: 0.55, cond_image_strength: 1.0 },
quality: { temporal_tile_size: 80, temporal_overlap: 32, guiding_strength: 1.00, temporal_overlap_cond_strength: 0.60, cond_image_strength: 1.0 },
},
};
function nodeClass(node) {
return node?.comfyClass || node?.type || "";
}
function widget(node, name) {
return (node?.widgets || []).find((w) => w?.name === name || w?.label === name) || null;
}
function setWidget(node, name, value) {
const w = widget(node, name);
if (!w) return false;
w.value = value;
try {
w.callback?.(value);
} catch {
// ignore
}
return true;
}
function clamp(v, min, max) {
const n = Number(v);
if (!Number.isFinite(n)) return min;
return Math.max(min, Math.min(max, n));
}
function effectiveVramPreset(vramGb, fallback = "12GB") {
const n = Number(vramGb);
if (!Number.isFinite(n)) return fallback;
if (n <= 8.5) return "8GB";
if (n <= 12.5) return "12GB";
if (n <= 16.5) return "16GB";
return "24GB";
}
function getGraphNodeById(id) {
try {
return app.graph?.getNodeById?.(id) || app.graph?._nodes_by_id?.[id] || null;
} catch {
return null;
}
}
function linkedTarget(node) {
try {
const out = node.outputs?.find((o) => o?.type === "IAMCCS_DETAIL_ATELIER_LINX") || node.outputs?.[0];
const linkId = Array.isArray(out?.links) ? out.links[0] : null;
if (linkId == null) return null;
const link = app.graph?.links?.[linkId];
if (!link) return null;
const targetId = link.target_id ?? link[3];
return getGraphNodeById(targetId);
} catch {
return null;
}
}
function linkedAdvancedNode(node) {
try {
const inp = (node.inputs || []).find((i) => i?.type === "IAMCCS_DETAIL_ATELIER_LINX");
const linkId = inp?.link;
if (linkId == null) return null;
const link = app.graph?.links?.[linkId];
if (!link) return null;
const sourceId = link.origin_id ?? link[1];
return getGraphNodeById(sourceId);
} catch {
return null;
}
}
function readLoaderContext() {
const ctx = { frames: null, fps: null, width: null, height: null };
try {
const nodes = app.graph?._nodes || [];
const loader = nodes.find((n) => nodeClass(n) === "VHS_LoadVideoPath");
const info = nodes.find((n) => nodeClass(n) === "VHS_VideoInfoLoaded");
const scaler = nodes.find((n) => nodeClass(n) === "ImageScaleToMaxDimension");
const loaderFrames = widget(loader, "frame_load_cap")?.value ?? loader?.widgets_values?.frame_load_cap;
const loaderFps = widget(loader, "force_rate")?.value ?? loader?.widgets_values?.force_rate;
ctx.frames = Number(loaderFrames) > 0 ? Number(loaderFrames) : null;
ctx.fps = Number(loaderFps) > 0 ? Number(loaderFps) : null;
const infoFps = widget(info, "fps")?.value ?? widget(info, "frame_rate")?.value;
if (!ctx.fps && Number(infoFps) > 0) ctx.fps = Number(infoFps);
const largest = widget(scaler, "largest_size")?.value ?? widget(scaler, "max_dimension")?.value;
if (Number(largest) > 0) {
ctx.width = Number(largest);
ctx.height = Math.round(Number(largest) * 9 / 16);
}
} catch {
// best effort only
}
return ctx;
}
function estimateChunks(frames, temporalTileSize, temporalOverlap, timeScale = 8) {
const f = Number(frames);
if (!Number.isFinite(f) || f <= 0) return null;
const latentFrames = Math.max(1, Math.ceil(f / timeScale));
const latentTile = Math.max(1, Math.floor(Number(temporalTileSize) / timeScale));
let latentOverlap = Math.max(0, Math.floor(Number(temporalOverlap) / timeScale));
if (latentOverlap >= latentTile) latentOverlap = Math.max(0, latentTile - 1);
const step = Math.max(1, latentTile - latentOverlap);
const chunks = latentFrames <= latentTile ? 1 : Math.ceil((latentFrames - latentTile) / step) + 1;
return { chunks, latentFrames, latentTile, latentOverlap, step };
}
async function probeHardware(ctx) {
try {
const params = new URLSearchParams();
if (ctx.width) params.set("width", String(Math.round(ctx.width)));
if (ctx.height) params.set("height", String(Math.round(ctx.height)));
if (ctx.frames) params.set("frames", String(Math.round(ctx.frames)));
if (ctx.fps) params.set("fps", String(ctx.fps));
const res = await api.fetchApi(`/api/iamccs/hw_probe?${params.toString()}`);
if (!res.ok) return null;
return await res.json();
} catch {
return null;
}
}
function chooseAutoValues(hw, ctx, quality = "balanced", fallbackVram = "12GB") {
const vramGb = hw?.hardware?.cuda_total_vram_gb ?? hw?.cuda_total_vram_gb ?? null;
const ramGb = hw?.hardware?.system_ram_gb ?? hw?.system_ram_gb ?? null;
const vram = effectiveVramPreset(vramGb, fallbackVram);
const q = PRESETS[vram]?.[quality] ? quality : "balanced";
const base = { ...(PRESETS[vram]?.[q] || PRESETS["12GB"].balanced) };
const lowRam = Number(ramGb) > 0 && Number(ramGb) <= 40 && (vram === "8GB" || vram === "12GB");
if (lowRam || vram === "8GB" || vram === "12GB") {
base.horizontal_tiles = 1;
base.vertical_tiles = 1;
base.spatial_overlap = 1;
base.spatial_tiling_mode = "force_off";
if (vram === "12GB") {
base.temporal_tile_size = Math.min(base.temporal_tile_size, 48);
base.temporal_overlap = Math.min(base.temporal_overlap, 16);
}
if (vram === "8GB") {
base.temporal_tile_size = Math.min(base.temporal_tile_size, 32);
base.temporal_overlap = Math.min(base.temporal_overlap, 16);
}
} else {
base.horizontal_tiles = 0;
base.vertical_tiles = 0;
base.spatial_overlap = 0;
base.spatial_tiling_mode = "inherit";
}
const chunks = estimateChunks(ctx.frames, base.temporal_tile_size, base.temporal_overlap);
return {
vram,
quality: q,
vramGb,
ramGb,
lowRam,
chunks,
values: base,
};
}
function ensurePreview(node) {
let w = widget(node, PREVIEW_WIDGET);
if (w) return w;
w = node.addWidget("text", "Atelier Live", "", () => {}, { multiline: true });
w.name = PREVIEW_WIDGET;
w.serialize = false;
try {
w.inputEl?.setAttribute?.("readonly", true);
} catch {
// ignore
}
return w;
}
function setPreview(node, text) {
const w = ensurePreview(node);
w.value = text;
node.properties = node.properties || {};
node.properties[PREVIEW_WIDGET] = text;
try {
app.graph.setDirtyCanvas(true, false);
} catch {
// ignore
}
}
async function refreshAdvanced(node, apply = false) {
const ctx = readLoaderContext();
const target = linkedTarget(node);
const fallbackVram = widget(target, "vram_preset")?.value || "12GB";
const quality = widget(target, "quality_mode")?.value || "balanced";
const hw = await probeHardware(ctx);
const plan = chooseAutoValues(hw, ctx, quality, fallbackVram);
if (apply) {
setWidget(node, "enabled", true);
setWidget(node, "temporal_tile_size", plan.values.temporal_tile_size);
setWidget(node, "temporal_overlap", plan.values.temporal_overlap);
setWidget(node, "guiding_strength", plan.values.guiding_strength);
setWidget(node, "temporal_overlap_cond_strength", plan.values.temporal_overlap_cond_strength);
setWidget(node, "cond_image_strength", plan.values.cond_image_strength);
setWidget(node, "horizontal_tiles", plan.values.horizontal_tiles ?? 0);
setWidget(node, "vertical_tiles", plan.values.vertical_tiles ?? 0);
setWidget(node, "spatial_overlap", plan.values.spatial_overlap ?? 0);
setWidget(node, "spatial_tiling_mode", plan.values.spatial_tiling_mode || "inherit");
}
const seconds = ctx.frames && ctx.fps ? (ctx.frames / ctx.fps).toFixed(2) : "?";
const hwLine = plan.vramGb ? `${plan.vram} detected (${Number(plan.vramGb).toFixed(1)}GB VRAM)` : `${plan.vram} fallback`;
const ramLine = plan.ramGb ? `${Number(plan.ramGb).toFixed(1)}GB RAM` : "RAM ?";
const chunkLine = plan.chunks ? `${plan.chunks.chunks} chunks approx, latent tile ${plan.chunks.latentTile}, step ${plan.chunks.step}` : "chunks ?";
setPreview(
node,
[
`Auto: ${hwLine}, ${ramLine}`,
`Video: ${ctx.frames || "?"} frames @ ${ctx.fps || "?"} fps (${seconds}s)`,
`Loop: ${plan.values.temporal_tile_size}/${plan.values.temporal_overlap}, ${chunkLine}`,
`Tiles: ${plan.values.horizontal_tiles ?? 0}x${plan.values.vertical_tiles ?? 0}, overlap ${plan.values.spatial_overlap ?? 0}`,
apply ? "Applied to Advanced overrides." : "Use Apply Auto to write these values.",
].join("\n")
);
}
function installAdvanced(node) {
ensurePreview(node);
if (!widget(node, APPLY_BUTTON)) {
const btn = node.addWidget("button", "Apply Auto", null, () => {
refreshAdvanced(node, true);
});
btn.name = APPLY_BUTTON;
btn.serialize = false;
}
if (!widget(node, REFRESH_BUTTON)) {
const btn = node.addWidget("button", "Refresh Estimate", null, () => {
refreshAdvanced(node, false);
});
btn.name = REFRESH_BUTTON;
btn.serialize = false;
}
setTimeout(() => refreshAdvanced(node, false), 250);
}
function installSampler(node) {
const advanced = linkedAdvancedNode(node);
if (advanced) setTimeout(() => refreshAdvanced(advanced, false), 250);
}
app.registerExtension({
name: "iamccs.detail_atelier.ui",
async beforeRegisterNodeDef(nodeType, nodeData) {
const name = nodeData?.name;
if (name !== ADVANCED_TYPE && name !== SAMPLER_TYPE) return;
const onNodeCreated = nodeType.prototype.onNodeCreated;
nodeType.prototype.onNodeCreated = function () {
const r = onNodeCreated?.apply(this, arguments);
if (name === ADVANCED_TYPE) installAdvanced(this);
if (name === SAMPLER_TYPE) installSampler(this);
return r;
};
},
async nodeCreated(node) {
if (nodeClass(node) === ADVANCED_TYPE) installAdvanced(node);
if (nodeClass(node) === SAMPLER_TYPE) installSampler(node);
},
});
+11 -432
View File
@@ -1,437 +1,16 @@
import { app } from "../../scripts/app.js";
const LEGACY_LOCAL_KEYS = [
"workflow",
"Comfy.OpenWorkflowsPaths",
"Comfy.ActiveWorkflowIndex",
];
const LEGACY_PREFIXES = [
"Comfy.Workflow.Drafts",
"Comfy.Workflow.DraftOrder",
];
const V2_PREFIX = {
draftIndex: "Comfy.Workflow.DraftIndex.v2:",
draftPayload: "Comfy.Workflow.Draft.v2:",
lastActivePath: "Comfy.Workflow.LastActivePath:",
lastOpenPaths: "Comfy.Workflow.LastOpenPaths:",
};
const MAX_DRAFTS_PER_WORKSPACE = 4;
const MAX_DRAFT_CHARS_PER_WORKSPACE = 2_500_000;
const STORAGE_PATCH_FLAG = "__iamccsWorkflowDraftQuotaPatchApplied";
const STORAGE_PROTOTYPE_PATCH_FLAG = "__iamccsWorkflowDraftQuotaPrototypePatchApplied";
function safeParseJson(json) {
if (!json) return null;
try {
return JSON.parse(json);
} catch {
return null;
}
}
function listStorageKeys(storage) {
const keys = [];
try {
for (let index = 0; index < storage.length; index += 1) {
const key = storage.key(index);
if (key) keys.push(key);
}
} catch {
return [];
}
return keys;
}
function removeKey(storage, key) {
try {
storage.removeItem(key);
return true;
} catch {
return false;
}
}
function getPayloadKey(workspaceId, draftKey) {
return `${V2_PREFIX.draftPayload}${workspaceId}:${draftKey}`;
}
function extractWorkspaceIdFromKey(key) {
if (typeof key !== "string") return null;
if (key.startsWith(V2_PREFIX.draftIndex)) {
return key.slice(V2_PREFIX.draftIndex.length);
}
if (key.startsWith(V2_PREFIX.draftPayload)) {
const rest = key.slice(V2_PREFIX.draftPayload.length);
const separatorIndex = rest.indexOf(":");
return separatorIndex === -1 ? null : rest.slice(0, separatorIndex);
}
if (key.startsWith(V2_PREFIX.lastActivePath)) {
return key.slice(V2_PREFIX.lastActivePath.length);
}
if (key.startsWith(V2_PREFIX.lastOpenPaths)) {
return key.slice(V2_PREFIX.lastOpenPaths.length);
}
return null;
}
function isDraftStorageKey(key) {
return typeof key === "string" && (
key.startsWith(V2_PREFIX.draftIndex)
|| key.startsWith(V2_PREFIX.draftPayload)
|| key.startsWith(V2_PREFIX.lastActivePath)
|| key.startsWith(V2_PREFIX.lastOpenPaths)
|| LEGACY_LOCAL_KEYS.includes(key)
|| LEGACY_PREFIXES.some((prefix) => key.startsWith(prefix))
);
}
function isQuotaExceededError(error) {
if (!error) return false;
const name = String(error.name || "");
const message = String(error.message || "").toLowerCase();
return name === "QuotaExceededError"
|| name === "NS_ERROR_DOM_QUOTA_REACHED"
|| message.includes("quota")
|| message.includes("storage is full");
}
function cleanupWorkspaceTrackingStorage(localStorageRef, workspaceId = null) {
let removed = 0;
for (const key of listStorageKeys(localStorageRef)) {
const isTrackingKey = key.startsWith(V2_PREFIX.lastActivePath)
|| key.startsWith(V2_PREFIX.lastOpenPaths);
if (!isTrackingKey) continue;
if (workspaceId !== null && extractWorkspaceIdFromKey(key) !== workspaceId) {
continue;
}
if (removeKey(localStorageRef, key)) removed += 1;
}
return removed;
}
function cleanupLegacyWorkflowStorage(localStorageRef) {
let removed = 0;
for (const key of LEGACY_LOCAL_KEYS) {
if (removeKey(localStorageRef, key)) removed += 1;
}
for (const key of listStorageKeys(localStorageRef)) {
if (LEGACY_PREFIXES.some((prefix) => key.startsWith(prefix))) {
if (removeKey(localStorageRef, key)) removed += 1;
}
}
return removed;
}
function cleanupWorkspaceDrafts(localStorageRef, workspaceId) {
const indexKey = `${V2_PREFIX.draftIndex}${workspaceId}`;
const rawIndex = localStorageRef.getItem(indexKey);
if (!rawIndex) return { removed: 0, rewritten: false };
const index = safeParseJson(rawIndex);
if (!index || typeof index !== "object" || !Array.isArray(index.order) || typeof index.entries !== "object" || index.entries == null) {
let removed = removeKey(localStorageRef, indexKey) ? 1 : 0;
const payloadPrefix = `${V2_PREFIX.draftPayload}${workspaceId}:`;
for (const key of listStorageKeys(localStorageRef)) {
if (key.startsWith(payloadPrefix) && removeKey(localStorageRef, key)) {
removed += 1;
}
}
return { removed, rewritten: false };
}
const payloadPrefix = `${V2_PREFIX.draftPayload}${workspaceId}:`;
const seenPayloads = new Set();
const keptEntries = {};
const keptOrder = [];
let keptChars = 0;
let keptCount = 0;
let removed = 0;
const newestFirst = [...index.order].reverse();
for (const draftKey of newestFirst) {
const entry = index.entries[draftKey];
if (!entry || typeof entry.path !== "string") continue;
const payloadKey = getPayloadKey(workspaceId, draftKey);
const payloadJson = localStorageRef.getItem(payloadKey);
if (!payloadJson) continue;
const payloadChars = payloadJson.length;
const canKeep = keptCount < MAX_DRAFTS_PER_WORKSPACE && (
keptCount === 0 || keptChars + payloadChars <= MAX_DRAFT_CHARS_PER_WORKSPACE
);
if (canKeep) {
keptEntries[draftKey] = entry;
keptOrder.unshift(draftKey);
keptChars += payloadChars;
keptCount += 1;
seenPayloads.add(payloadKey);
} else if (removeKey(localStorageRef, payloadKey)) {
removed += 1;
}
}
for (const key of listStorageKeys(localStorageRef)) {
if (key.startsWith(payloadPrefix) && !seenPayloads.has(key)) {
if (removeKey(localStorageRef, key)) removed += 1;
}
}
const nextIndex = {
...index,
updatedAt: Date.now(),
order: keptOrder,
entries: keptEntries,
};
const nextIndexJson = JSON.stringify(nextIndex);
const rewritten = nextIndexJson !== rawIndex;
if (rewritten) {
try {
localStorageRef.setItem(indexKey, nextIndexJson);
} catch {
// If index rewrite fails, leaving the trimmed payload set is still better
// than keeping localStorage saturated with stale drafts.
}
}
return { removed, rewritten };
}
function cleanupWorkflowPersistenceStorage() {
try {
const localStorageRef = window.localStorage;
if (!localStorageRef) return;
let removed = cleanupLegacyWorkflowStorage(localStorageRef);
removed += cleanupWorkspaceTrackingStorage(localStorageRef);
let rewritten = 0;
const workspaceIds = new Set();
for (const key of listStorageKeys(localStorageRef)) {
if (key.startsWith(V2_PREFIX.draftIndex)) {
workspaceIds.add(key.slice(V2_PREFIX.draftIndex.length));
}
}
for (const workspaceId of workspaceIds) {
const result = cleanupWorkspaceDrafts(localStorageRef, workspaceId);
removed += result.removed;
if (result.rewritten) rewritten += 1;
}
if (removed || rewritten) {
console.info("[IAMCCS] Workflow draft storage cleanup applied", {
removedKeys: removed,
rewrittenIndexes: rewritten,
});
}
} catch (error) {
console.warn("[IAMCCS] Workflow draft storage cleanup skipped", error);
}
}
function recoverDraftStorage(localStorageRef, key) {
let removed = cleanupLegacyWorkflowStorage(localStorageRef);
let rewritten = 0;
const workspaceIds = new Set();
const workspaceId = extractWorkspaceIdFromKey(key);
if (workspaceId) workspaceIds.add(workspaceId);
for (const storageKey of listStorageKeys(localStorageRef)) {
if (storageKey.startsWith(V2_PREFIX.draftIndex)) {
workspaceIds.add(storageKey.slice(V2_PREFIX.draftIndex.length));
}
}
for (const currentWorkspaceId of workspaceIds) {
removed += cleanupWorkspaceTrackingStorage(localStorageRef, currentWorkspaceId);
const result = cleanupWorkspaceDrafts(localStorageRef, currentWorkspaceId);
removed += result.removed;
if (result.rewritten) rewritten += 1;
}
return { removed, rewritten };
}
function aggressiveRecoverDraftStorage(localStorageRef) {
let removed = cleanupLegacyWorkflowStorage(localStorageRef);
removed += cleanupWorkspaceTrackingStorage(localStorageRef);
for (const storageKey of listStorageKeys(localStorageRef)) {
if (
storageKey.startsWith(V2_PREFIX.draftIndex)
|| storageKey.startsWith(V2_PREFIX.draftPayload)
) {
if (removeKey(localStorageRef, storageKey)) removed += 1;
}
}
return { removed, rewritten: 0 };
}
function isStorageLike(storageRef) {
return storageRef != null
&& typeof storageRef.getItem === "function"
&& typeof storageRef.setItem === "function"
&& typeof storageRef.removeItem === "function";
}
function isLocalStorageRef(storageRef) {
try {
return storageRef === window.localStorage;
} catch {
return false;
}
}
function shouldSuppressDraftQuotaFailure(key) {
if (!isDraftStorageKey(key)) return false;
return typeof key === "string" && (
key.startsWith(V2_PREFIX.draftPayload)
|| key.startsWith(V2_PREFIX.draftIndex)
|| key.startsWith(V2_PREFIX.lastActivePath)
|| key.startsWith(V2_PREFIX.lastOpenPaths)
|| LEGACY_LOCAL_KEYS.includes(key)
|| LEGACY_PREFIXES.some((prefix) => key.startsWith(prefix))
);
}
function handleDraftQuotaSetItem(originalSetItem, storageRef, key, value, error) {
if (!isDraftStorageKey(key) || !isQuotaExceededError(error) || !isStorageLike(storageRef)) {
throw error;
}
const canRecover = isLocalStorageRef(storageRef);
const recovery = canRecover ? recoverDraftStorage(storageRef, key) : { removed: 0, rewritten: 0 };
console.warn("[IAMCCS] Workflow draft quota reached, attempting recovery", {
key,
removedKeys: recovery.removed,
rewrittenIndexes: recovery.rewritten,
storage: canRecover ? "localStorage" : "other",
});
try {
const result = Reflect.apply(originalSetItem, storageRef, [key, value]);
console.info("[IAMCCS] Workflow draft save recovered after storage cleanup", {
key,
removedKeys: recovery.removed,
rewrittenIndexes: recovery.rewritten,
storage: canRecover ? "localStorage" : "other",
});
return result;
} catch (retryError) {
if (!isQuotaExceededError(retryError)) {
console.error("[IAMCCS] Workflow draft save failed after cleanup with non-quota error", {
key,
removedKeys: recovery.removed,
rewrittenIndexes: recovery.rewritten,
retryError,
});
throw retryError;
}
const aggressiveRecovery = canRecover
? aggressiveRecoverDraftStorage(storageRef)
: { removed: 0, rewritten: 0 };
if (aggressiveRecovery.removed) {
console.warn("[IAMCCS] Workflow draft quota still exceeded, applying aggressive recovery", {
key,
removedKeys: aggressiveRecovery.removed,
storage: canRecover ? "localStorage" : "other",
});
}
try {
const result = Reflect.apply(originalSetItem, storageRef, [key, value]);
console.info("[IAMCCS] Workflow draft save recovered after aggressive cleanup", {
key,
removedKeys: recovery.removed + aggressiveRecovery.removed,
rewrittenIndexes: recovery.rewritten,
storage: canRecover ? "localStorage" : "other",
});
return result;
} catch (finalRetryError) {
if (!isQuotaExceededError(finalRetryError) || !shouldSuppressDraftQuotaFailure(key)) {
console.error("[IAMCCS] Workflow draft save still exceeds storage quota after aggressive cleanup", {
key,
removedKeys: recovery.removed + aggressiveRecovery.removed,
rewrittenIndexes: recovery.rewritten,
retryError: finalRetryError,
});
throw finalRetryError;
}
console.warn("[IAMCCS] Workflow draft save skipped after quota recovery failed", {
key,
removedKeys: recovery.removed + aggressiveRecovery.removed,
rewrittenIndexes: recovery.rewritten,
storage: canRecover ? "localStorage" : "other",
});
return undefined;
}
}
}
function installDraftStorageQuotaGuard() {
try {
const localStorageRef = window.localStorage;
if (!localStorageRef || localStorageRef[STORAGE_PATCH_FLAG]) return;
const storagePrototype = Object.getPrototypeOf(localStorageRef);
if (!storagePrototype || storagePrototype[STORAGE_PROTOTYPE_PATCH_FLAG]) {
Object.defineProperty(localStorageRef, STORAGE_PATCH_FLAG, {
value: true,
configurable: false,
enumerable: false,
writable: false,
});
return;
}
const originalSetItem = storagePrototype.setItem;
if (typeof originalSetItem !== "function") return;
storagePrototype.setItem = function patchedSetItem(key, value) {
try {
return Reflect.apply(originalSetItem, this, [key, value]);
} catch (error) {
return handleDraftQuotaSetItem(originalSetItem, this, key, value, error);
}
};
Object.defineProperty(storagePrototype, STORAGE_PROTOTYPE_PATCH_FLAG, {
value: true,
configurable: false,
enumerable: false,
writable: false,
});
Object.defineProperty(localStorageRef, STORAGE_PATCH_FLAG, {
value: true,
configurable: false,
enumerable: false,
writable: false,
});
} catch (error) {
console.warn("[IAMCCS] Workflow draft quota guard skipped", error);
}
}
// IAMCCS intentionally does not touch ComfyUI workflow persistence.
//
// ComfyUI frontend v1.43+ restores open workflow tabs through its own browser
// storage keys. Older revisions of this extension tried to clean or recover
// those keys when localStorage quota was exceeded, but that can make ComfyUI
// start with a blank workspace or only one restored workflow.
//
// Keep this extension as a harmless no-op so cached extension lists and older
// installs do not fail to import it, while leaving all workflow/tab persistence
// behavior to ComfyUI itself.
app.registerExtension({
name: "iamccs.workflow_persist_cleanup",
async setup() {
cleanupWorkflowPersistenceStorage();
installDraftStorageQuotaGuard();
},
async setup() {},
});