1307 lines
43 KiB
Python
1307 lines
43 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import time
|
|
import uuid
|
|
from collections import OrderedDict
|
|
|
|
import numpy as np
|
|
from PIL import Image
|
|
|
|
import folder_paths
|
|
|
|
|
|
_MEDIA_EXT_RE = re.compile(
|
|
r"\.(png|jpe?g|webp|gif|bmp|tiff?|avif|mp4|mov|webm|mkv|m4v)$", re.IGNORECASE
|
|
)
|
|
_DATE_TOKEN_RE = re.compile(r"%date:([^%]+)%")
|
|
_SERVE_TOKENS = OrderedDict()
|
|
_SERVE_CAP = 256
|
|
_H3_PLAN_TYPE = "H3_CHAIN_PLAN"
|
|
_H3_PLAN_IMAGE_BINDINGS_KEY = "_dumas_scene_image_bindings"
|
|
_H3_PLAN_IMAGE_BINDINGS = OrderedDict()
|
|
_H3_PLAN_IMAGE_BINDINGS_CAP = 128
|
|
_H3_PLAN_IMAGE_SLOTS = 9
|
|
_FOLDER_IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".webp", ".bmp", ".gif", ".tiff", ".tif")
|
|
_LOAD_IMAGES_FOLDER_DEFAULT_STATE = {
|
|
"version": 1,
|
|
"folder": "",
|
|
"recursive": False,
|
|
"sort": "name",
|
|
"sort_dir": "asc",
|
|
"selected": [],
|
|
"selection_mode": "selected",
|
|
"first_n": 5,
|
|
}
|
|
|
|
|
|
def _clean_input_token_value(value):
|
|
cleaned = ""
|
|
if value is not None:
|
|
cleaned = value if isinstance(value, str) else str(value)
|
|
cleaned = _MEDIA_EXT_RE.sub("", cleaned.strip())
|
|
cleaned = cleaned.replace("\\", "_").replace("/", "_")
|
|
return cleaned
|
|
|
|
|
|
def _expand_date_tokens(value):
|
|
if not isinstance(value, str) or "%date:" not in value:
|
|
return value
|
|
|
|
now = time.localtime()
|
|
|
|
def pad(number, width):
|
|
return str(number).zfill(width)
|
|
|
|
def repl(match):
|
|
fmt = match.group(1)
|
|
|
|
def swap(token_match):
|
|
token = token_match.group(0)
|
|
if token == "yyyy":
|
|
return pad(now.tm_year, 4)
|
|
if token == "yy":
|
|
return str(now.tm_year)[-2:]
|
|
if token == "MM":
|
|
return pad(now.tm_mon, 2)
|
|
if token == "M":
|
|
return str(now.tm_mon)
|
|
if token == "dd":
|
|
return pad(now.tm_mday, 2)
|
|
if token == "d":
|
|
return str(now.tm_mday)
|
|
if token in ("hh", "HH"):
|
|
return pad(now.tm_hour, 2)
|
|
if token in ("h", "H"):
|
|
return str(now.tm_hour)
|
|
if token == "mm":
|
|
return pad(now.tm_min, 2)
|
|
if token == "m":
|
|
return str(now.tm_min)
|
|
if token == "ss":
|
|
return pad(now.tm_sec, 2)
|
|
if token == "s":
|
|
return str(now.tm_sec)
|
|
return token
|
|
|
|
return re.sub(r"yyyy|yy|MM|M|dd|d|hh|h|HH|H|mm|m|ss|s", swap, fmt)
|
|
|
|
return _DATE_TOKEN_RE.sub(repl, value)
|
|
|
|
|
|
def _expand_native_tokens(value):
|
|
if not isinstance(value, str) or "%" not in value:
|
|
return value
|
|
|
|
now = time.localtime()
|
|
replacements = (
|
|
("%year%", f"{now.tm_year:04}"),
|
|
("%month%", f"{now.tm_mon:02}"),
|
|
("%day%", f"{now.tm_mday:02}"),
|
|
("%hour%", f"{now.tm_hour:02}"),
|
|
("%minute%", f"{now.tm_min:02}"),
|
|
("%second%", f"{now.tm_sec:02}"),
|
|
)
|
|
for token, replacement in replacements:
|
|
value = value.replace(token, replacement)
|
|
return value
|
|
|
|
|
|
def _safe_pattern(value):
|
|
value = str(value or "").replace("\\", "/")
|
|
value = re.sub(r'[<>:"|?*]', "_", value)
|
|
value = re.sub(r"/{2,}", "/", value).strip(" /.") or "image_%counter%"
|
|
return value
|
|
|
|
|
|
def _register_serve_token(path):
|
|
token = uuid.uuid4().hex
|
|
_SERVE_TOKENS[token] = path
|
|
while len(_SERVE_TOKENS) > _SERVE_CAP:
|
|
_SERVE_TOKENS.popitem(last=False)
|
|
return token
|
|
|
|
|
|
def resolve_serve_token(token):
|
|
return _SERVE_TOKENS.get(str(token or ""))
|
|
|
|
|
|
def _touch_plan_image_binding(token):
|
|
token = str(token or "")
|
|
if not token or token not in _H3_PLAN_IMAGE_BINDINGS:
|
|
return
|
|
binding = _H3_PLAN_IMAGE_BINDINGS.pop(token)
|
|
_H3_PLAN_IMAGE_BINDINGS[token] = binding
|
|
|
|
|
|
def _prune_plan_image_bindings():
|
|
while len(_H3_PLAN_IMAGE_BINDINGS) > _H3_PLAN_IMAGE_BINDINGS_CAP:
|
|
_H3_PLAN_IMAGE_BINDINGS.popitem(last=False)
|
|
|
|
|
|
def _clone_h3_plan(plan):
|
|
if not isinstance(plan, dict):
|
|
raise ValueError("Dumas H3 plan helpers require a plan dictionary.")
|
|
shots = plan.get("shots")
|
|
if not isinstance(shots, list):
|
|
raise ValueError("Dumas H3 plan helpers require a plan with a shots list.")
|
|
|
|
cloned = dict(plan)
|
|
cloned["shots"] = [dict(shot) if isinstance(shot, dict) else shot for shot in shots]
|
|
bindings = plan.get(_H3_PLAN_IMAGE_BINDINGS_KEY)
|
|
if isinstance(bindings, dict):
|
|
cloned[_H3_PLAN_IMAGE_BINDINGS_KEY] = {
|
|
"token": str(bindings.get("token") or ""),
|
|
"scene_counts": {
|
|
str(key): int(value)
|
|
for key, value in dict(bindings.get("scene_counts") or {}).items()
|
|
},
|
|
}
|
|
return cloned
|
|
|
|
|
|
def _normalize_h3_scene_index(plan, scene_index):
|
|
shots = plan.get("shots")
|
|
total = len(shots) if isinstance(shots, list) else 0
|
|
index = int(scene_index)
|
|
if index < 1 or index > total:
|
|
raise ValueError(
|
|
f"Dumas H3 scene index {index} is outside the plan's {total} scenes."
|
|
)
|
|
return index
|
|
|
|
|
|
def _h3_plan_binding_entry(plan):
|
|
bindings = plan.get(_H3_PLAN_IMAGE_BINDINGS_KEY)
|
|
if not isinstance(bindings, dict):
|
|
return "", {}
|
|
token = str(bindings.get("token") or "")
|
|
counts = {
|
|
str(key): int(value)
|
|
for key, value in dict(bindings.get("scene_counts") or {}).items()
|
|
}
|
|
return token, counts
|
|
|
|
|
|
def _scene_images_tuple(
|
|
image1=None,
|
|
image2=None,
|
|
image3=None,
|
|
image4=None,
|
|
image5=None,
|
|
image6=None,
|
|
image7=None,
|
|
image8=None,
|
|
image9=None,
|
|
):
|
|
return (image1, image2, image3, image4, image5, image6, image7, image8, image9)
|
|
|
|
|
|
def _connected_image_count(images):
|
|
return sum(1 for image in images if image is not None)
|
|
|
|
|
|
def _is_within_directory(parent_path, child_path):
|
|
try:
|
|
return os.path.commonpath([parent_path, child_path]) == parent_path
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _next_counter(directory, filename_template):
|
|
os.makedirs(directory, exist_ok=True)
|
|
if "%counter%" not in filename_template:
|
|
return 1
|
|
|
|
parts = filename_template.split("%counter%")
|
|
highest = 0
|
|
for entry in os.listdir(directory):
|
|
if not entry.startswith(parts[0]) or not entry.endswith(parts[-1]):
|
|
continue
|
|
middle = entry[len(parts[0]):]
|
|
if parts[-1]:
|
|
middle = middle[: -len(parts[-1])]
|
|
if middle.isdigit():
|
|
highest = max(highest, int(middle))
|
|
return highest + 1
|
|
|
|
|
|
def _next_counter_for_relative_path(base_directory, relative_template):
|
|
os.makedirs(base_directory, exist_ok=True)
|
|
if "%counter%" not in relative_template:
|
|
return 1
|
|
|
|
counter = 1
|
|
while True:
|
|
candidate = relative_template.replace("%counter%", str(counter).zfill(3))
|
|
full_path = os.path.join(base_directory, *[part for part in candidate.split("/") if part])
|
|
if not os.path.exists(full_path):
|
|
return counter
|
|
counter += 1
|
|
|
|
|
|
def _build_pnginfo(prompt=None, extra_pnginfo=None):
|
|
try:
|
|
pnginfo = Image.PngImagePlugin.PngInfo()
|
|
except AttributeError:
|
|
from PIL.PngImagePlugin import PngInfo
|
|
|
|
pnginfo = PngInfo()
|
|
if prompt is not None:
|
|
pnginfo.add_text("prompt", json.dumps(prompt))
|
|
if isinstance(extra_pnginfo, dict):
|
|
for key, value in extra_pnginfo.items():
|
|
pnginfo.add_text(str(key), json.dumps(value))
|
|
return pnginfo
|
|
|
|
|
|
def _tensor_image_to_pil_image(tensor):
|
|
image_tensor = tensor[0]
|
|
if hasattr(image_tensor, "mul") and hasattr(image_tensor, "clamp"):
|
|
image_array = image_tensor.mul(255).clamp(0, 255)
|
|
if hasattr(image_array, "byte"):
|
|
image_array = image_array.byte()
|
|
image_array = image_array.cpu().numpy()
|
|
return Image.fromarray(image_array)
|
|
|
|
image_array = 255.0 * image_tensor.cpu().numpy()
|
|
return Image.fromarray(np.clip(image_array, 0, 255).astype(np.uint8))
|
|
|
|
|
|
def _folder_loader_default_state():
|
|
return dict(_LOAD_IMAGES_FOLDER_DEFAULT_STATE)
|
|
|
|
|
|
def _parse_load_images_folder_state(state_json):
|
|
if not state_json:
|
|
return _folder_loader_default_state()
|
|
try:
|
|
parsed = json.loads(state_json)
|
|
except Exception:
|
|
return _folder_loader_default_state()
|
|
|
|
state = _folder_loader_default_state()
|
|
if isinstance(parsed, dict):
|
|
state.update({key: value for key, value in parsed.items() if key in state})
|
|
return state
|
|
|
|
|
|
def _folder_is_image(name):
|
|
return str(name or "").lower().endswith(_FOLDER_IMAGE_EXTS)
|
|
|
|
|
|
def _list_folder_image_files(real_folder, recursive):
|
|
files = []
|
|
if recursive:
|
|
for root, _dirs, names in os.walk(real_folder):
|
|
for name in names:
|
|
if not _folder_is_image(name):
|
|
continue
|
|
full_path = os.path.join(root, name)
|
|
try:
|
|
stat_result = os.stat(full_path)
|
|
except OSError:
|
|
continue
|
|
rel_path = os.path.relpath(full_path, real_folder).replace("\\", "/")
|
|
files.append(
|
|
{
|
|
"file": rel_path,
|
|
"name": name,
|
|
"size": stat_result.st_size,
|
|
"mtime": stat_result.st_mtime,
|
|
}
|
|
)
|
|
else:
|
|
for name in os.listdir(real_folder):
|
|
full_path = os.path.join(real_folder, name)
|
|
if not os.path.isfile(full_path) or not _folder_is_image(name):
|
|
continue
|
|
try:
|
|
stat_result = os.stat(full_path)
|
|
except OSError:
|
|
continue
|
|
files.append(
|
|
{
|
|
"file": name,
|
|
"name": name,
|
|
"size": stat_result.st_size,
|
|
"mtime": stat_result.st_mtime,
|
|
}
|
|
)
|
|
return files
|
|
|
|
|
|
def _sort_folder_image_files(files, sort_key, sort_dir):
|
|
ordered = list(files or [])
|
|
|
|
def sort_value(entry):
|
|
if sort_key == "date":
|
|
return (float(entry.get("mtime") or 0), str(entry.get("file") or "").lower())
|
|
return str(entry.get("file") or "").lower()
|
|
|
|
ordered.sort(key=sort_value, reverse=str(sort_dir or "").lower() == "desc")
|
|
return ordered
|
|
|
|
|
|
def _resolve_folder_selection(state, files):
|
|
ordered = _sort_folder_image_files(
|
|
files,
|
|
state.get("sort", "name"),
|
|
state.get("sort_dir", "asc"),
|
|
)
|
|
mode = str(state.get("selection_mode") or "selected").lower()
|
|
if mode == "all":
|
|
return [entry["file"] for entry in ordered]
|
|
if mode == "first_n":
|
|
try:
|
|
count = max(0, int(state.get("first_n", 0) or 0))
|
|
except Exception:
|
|
count = 0
|
|
return [entry["file"] for entry in ordered[:count]]
|
|
if mode == "random":
|
|
return [random.choice(ordered)["file"]] if ordered else []
|
|
|
|
present = {entry["file"] for entry in ordered}
|
|
selected = []
|
|
for rel_path in state.get("selected", []) or []:
|
|
if isinstance(rel_path, str) and rel_path in present:
|
|
selected.append(rel_path)
|
|
return selected
|
|
|
|
|
|
def _load_folder_image(path):
|
|
import numpy as np
|
|
|
|
try:
|
|
import torch
|
|
except Exception as exc:
|
|
raise RuntimeError("torch is required to load folder images") from exc
|
|
|
|
from PIL import Image, ImageOps, ImageSequence
|
|
|
|
try:
|
|
import comfy.model_management as comfy_model_management
|
|
|
|
tensor_dtype = comfy_model_management.intermediate_dtype()
|
|
except Exception:
|
|
tensor_dtype = torch.float32
|
|
|
|
try:
|
|
import node_helpers
|
|
|
|
image = node_helpers.pillow(Image.open, path)
|
|
except Exception:
|
|
image = Image.open(path)
|
|
|
|
frame = ImageOps.exif_transpose(next(ImageSequence.Iterator(image)))
|
|
if frame.mode == "I":
|
|
frame = frame.point(lambda px: px * (1 / 255))
|
|
rgb_image = frame.convert("RGB")
|
|
width, height = rgb_image.size
|
|
|
|
if "A" in frame.getbands():
|
|
alpha = np.array(frame.getchannel("A")).astype(np.float32) / 255.0
|
|
mask_image = Image.fromarray(((1.0 - alpha) * 255).astype(np.uint8), mode="L")
|
|
elif frame.mode == "P" and "transparency" in frame.info:
|
|
alpha = np.array(frame.convert("RGBA").getchannel("A")).astype(np.float32) / 255.0
|
|
mask_image = Image.fromarray(((1.0 - alpha) * 255).astype(np.uint8), mode="L")
|
|
else:
|
|
mask_image = Image.new("L", rgb_image.size, 0)
|
|
|
|
image_tensor = torch.from_numpy(np.array(rgb_image).astype(np.float32) / 255.0)[None,].to(
|
|
dtype=tensor_dtype
|
|
)
|
|
mask_tensor = torch.from_numpy(np.array(mask_image).astype(np.float32) / 255.0).unsqueeze(0).to(
|
|
dtype=tensor_dtype
|
|
)
|
|
return image_tensor, mask_tensor, int(width), int(height)
|
|
|
|
|
|
def _normalize_free_text(value):
|
|
return " ".join(str(value or "").split()).strip()
|
|
|
|
|
|
def _label_for_character(name, character_id):
|
|
return _normalize_free_text(name) or _normalize_free_text(character_id) or "the character"
|
|
|
|
|
|
def _format_height_text(feet, inches):
|
|
feet_value = str(feet or "").strip()
|
|
inches_value = str(inches or "").strip()
|
|
if not feet_value and not inches_value:
|
|
return ""
|
|
|
|
parts = []
|
|
if feet_value:
|
|
feet_number = int(feet_value)
|
|
parts.append(f"{feet_number} foot" if feet_number == 1 else f"{feet_number} feet")
|
|
if inches_value:
|
|
inches_number = int(inches_value)
|
|
parts.append(
|
|
f"{inches_number} inch" if inches_number == 1 else f"{inches_number} inches"
|
|
)
|
|
return " ".join(parts)
|
|
|
|
|
|
def _ensure_sentence(value):
|
|
text = _normalize_free_text(value)
|
|
if not text:
|
|
return ""
|
|
if text[-1] not in ".!?":
|
|
text += "."
|
|
return text
|
|
|
|
|
|
def _parse_positive_int(value):
|
|
text = str(value or "").strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
parsed = int(text)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
if parsed <= 0:
|
|
return None
|
|
return parsed
|
|
|
|
|
|
def _indefinite_article(value):
|
|
text = _normalize_free_text(value).lower()
|
|
if not text:
|
|
return "a"
|
|
return "an" if text[0] in "aeiou" else "a"
|
|
|
|
|
|
def _build_character_helper_text(
|
|
primary_picture_id,
|
|
secondary_picture_id,
|
|
character_id,
|
|
name,
|
|
alias,
|
|
gender,
|
|
age,
|
|
nationality,
|
|
occupation,
|
|
height_feet,
|
|
height_inches,
|
|
accent,
|
|
general,
|
|
):
|
|
primary_picture = int(primary_picture_id)
|
|
secondary_picture = int(secondary_picture_id)
|
|
character_name = _normalize_free_text(name)
|
|
character_id = _normalize_free_text(character_id)
|
|
alias = _normalize_free_text(alias)
|
|
gender = _normalize_free_text(gender)
|
|
nationality = _normalize_free_text(nationality)
|
|
occupation = _normalize_free_text(occupation)
|
|
accent = _normalize_free_text(accent)
|
|
general = _ensure_sentence(general)
|
|
age_value = _parse_positive_int(age)
|
|
character_label = _label_for_character(character_name, character_id)
|
|
|
|
if character_name:
|
|
first_line = (
|
|
f"<Picture {primary_picture}> and <Picture {secondary_picture}> reference "
|
|
f"the same character who is called {character_name}."
|
|
)
|
|
elif character_id:
|
|
first_line = (
|
|
f"<Picture {primary_picture}> and <Picture {secondary_picture}> reference "
|
|
f'the same character with ID "{character_id}".'
|
|
)
|
|
else:
|
|
first_line = (
|
|
f"<Picture {primary_picture}> and <Picture {secondary_picture}> reference "
|
|
"the same character."
|
|
)
|
|
|
|
lines = [
|
|
first_line,
|
|
f"<Picture {primary_picture}> is the primary full-body reference for {character_label}.",
|
|
f"<Picture {secondary_picture}> is a frontal facial reference for {character_label}.",
|
|
]
|
|
|
|
fact_fragments = []
|
|
if alias:
|
|
fact_fragments.append(f"is also known as {alias}")
|
|
if gender:
|
|
fact_fragments.append(f"is {gender}")
|
|
if age_value is not None:
|
|
fact_fragments.append(f"is {age_value} years old")
|
|
if nationality:
|
|
fact_fragments.append(f"is {nationality}")
|
|
if occupation:
|
|
fact_fragments.append(f"works as {occupation}")
|
|
height_text = _format_height_text(height_feet, height_inches)
|
|
if height_text:
|
|
fact_fragments.append(f"is {height_text} tall")
|
|
if accent:
|
|
fact_fragments.append(
|
|
f"speaks in {_indefinite_article(accent)} {accent} accent"
|
|
)
|
|
|
|
if fact_fragments:
|
|
lines.append(f"{character_label} {', '.join(fact_fragments)}.")
|
|
|
|
if general:
|
|
lines.append(general)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _build_character_wardrobe_text(wardrobe, character_id, name, alias):
|
|
text = _normalize_free_text(wardrobe)
|
|
if not text:
|
|
return ""
|
|
|
|
if text.lower().startswith("wardrobe:"):
|
|
text = text.split(":", 1)[1].strip()
|
|
if not text:
|
|
return ""
|
|
|
|
# If the user already authored a full H3-style sheet entry, leave it alone.
|
|
if "=" in text or ":" in text:
|
|
return text
|
|
|
|
character_label = (
|
|
_normalize_free_text(name)
|
|
or _normalize_free_text(alias)
|
|
or _normalize_free_text(character_id)
|
|
)
|
|
if character_label:
|
|
return f"{character_label} = {text}"
|
|
return text
|
|
|
|
|
|
class DumasImageCompareNode:
|
|
DESCRIPTION = (
|
|
"Dumas Image Compare shows the difference between two images directly on "
|
|
"the node. Connect one or two IMAGE inputs to compare before/after "
|
|
"results, model variants, or processing stages without breaking a "
|
|
"workflow when one branch is bypassed."
|
|
)
|
|
RETURN_TYPES = ("IMAGE",)
|
|
RETURN_NAMES = ("new image",)
|
|
FUNCTION = "compare_images"
|
|
OUTPUT_NODE = True
|
|
CATEGORY = "Dumas/Image"
|
|
|
|
def __init__(self):
|
|
self.output_dir = folder_paths.get_temp_directory()
|
|
self.type = "temp"
|
|
self.prefix_append = "_dumascmp_" + "".join(
|
|
random.choice("abcdefghijklmnopqrstuvwxyz") for _ in range(5)
|
|
)
|
|
self.compress_level = 4
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"optional": {
|
|
"image1": (
|
|
"IMAGE",
|
|
{
|
|
"tooltip": (
|
|
"First image to compare. Optional so muted or bypassed "
|
|
"branches do not trigger a missing-input error."
|
|
)
|
|
},
|
|
),
|
|
"image2": (
|
|
"IMAGE",
|
|
{
|
|
"tooltip": (
|
|
"Second image to compare. Optional so the node can still "
|
|
"display a single available image."
|
|
)
|
|
},
|
|
),
|
|
}
|
|
}
|
|
|
|
def compare_images(self, image1=None, image2=None):
|
|
present = []
|
|
if image1 is not None:
|
|
present.append((1, image1))
|
|
if image2 is not None:
|
|
present.append((2, image2))
|
|
results = []
|
|
|
|
if present:
|
|
first_tensor = present[0][1]
|
|
prefix = "dumas_compare" + self.prefix_append
|
|
first_image = first_tensor[0]
|
|
full_output_folder, filename, counter, subfolder, _ = folder_paths.get_save_image_path(
|
|
prefix,
|
|
self.output_dir,
|
|
first_image.shape[1],
|
|
first_image.shape[0],
|
|
)
|
|
join_path = os.path.join
|
|
for slot, tensor in present:
|
|
image = _tensor_image_to_pil_image(tensor)
|
|
file_name = f"{filename}_{counter:05}_.png"
|
|
image.save(
|
|
join_path(full_output_folder, file_name),
|
|
compress_level=self.compress_level,
|
|
)
|
|
results.append(
|
|
{
|
|
"filename": file_name,
|
|
"subfolder": subfolder,
|
|
"type": self.type,
|
|
"slot": slot,
|
|
}
|
|
)
|
|
counter += 1
|
|
|
|
new_image = image2 if image2 is not None else image1
|
|
return {"ui": {"images": results}, "result": (new_image,)}
|
|
|
|
|
|
class DumasSaveImageNode:
|
|
DESCRIPTION = (
|
|
"Dumas Save Image writes images to any folder, with filename tokens "
|
|
"such as %input%, %input2%, %date:yyyy-MM-dd%, %counter%, %width%, "
|
|
"%height%, and %batch_num%."
|
|
)
|
|
RETURN_TYPES = ()
|
|
FUNCTION = "save_images"
|
|
OUTPUT_NODE = True
|
|
CATEGORY = "Dumas/Image"
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"images": ("IMAGE", {"tooltip": "Image batch to save."}),
|
|
"folder": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": (
|
|
"Destination folder. Leave empty to save in ComfyUI's "
|
|
"output directory."
|
|
),
|
|
},
|
|
),
|
|
"pattern": (
|
|
"STRING",
|
|
{
|
|
"default": "image_%date:yyyy-MM-dd%_%counter%",
|
|
"multiline": False,
|
|
"tooltip": (
|
|
"Filename pattern with optional subfolders. Tokens: "
|
|
"%input%, %input2%, %date:yyyy-MM-dd%, %counter%, "
|
|
"%width%, %height%, %batch_num%."
|
|
),
|
|
},
|
|
),
|
|
"format": (["png", "jpg"], {"default": "png"}),
|
|
"quality": (
|
|
"INT",
|
|
{"default": 100, "min": 1, "max": 100, "step": 1},
|
|
),
|
|
"embed_workflow": ("BOOLEAN", {"default": True}),
|
|
"save_on_run": ("BOOLEAN", {"default": True}),
|
|
},
|
|
"optional": {
|
|
"name": (
|
|
"STRING",
|
|
{
|
|
"forceInput": True,
|
|
"tooltip": "Optional text inserted by the %input% token.",
|
|
},
|
|
),
|
|
"name_2": (
|
|
"STRING",
|
|
{
|
|
"forceInput": True,
|
|
"tooltip": "Optional text inserted by the %input2% token.",
|
|
},
|
|
),
|
|
},
|
|
"hidden": {
|
|
"prompt": "PROMPT",
|
|
"extra_pnginfo": "EXTRA_PNGINFO",
|
|
},
|
|
}
|
|
|
|
@classmethod
|
|
def IS_CHANGED(cls, **_kwargs):
|
|
return float("nan")
|
|
|
|
def save_images(
|
|
self,
|
|
images,
|
|
folder,
|
|
pattern,
|
|
format,
|
|
quality,
|
|
embed_workflow,
|
|
save_on_run,
|
|
name=None,
|
|
name_2=None,
|
|
prompt=None,
|
|
extra_pnginfo=None,
|
|
):
|
|
if not save_on_run:
|
|
return {"ui": {"images": []}}
|
|
|
|
width = int(images.shape[2])
|
|
height = int(images.shape[1])
|
|
output_dir = folder_paths.get_output_directory()
|
|
target_dir = os.path.abspath(folder.strip()) if str(folder or "").strip() else output_dir
|
|
os.makedirs(target_dir, exist_ok=True)
|
|
|
|
resolved_pattern = str(pattern or "image_%date:yyyy-MM-dd%_%counter%")
|
|
resolved_pattern = resolved_pattern.replace("%input%", _clean_input_token_value(name))
|
|
resolved_pattern = resolved_pattern.replace("%input2%", _clean_input_token_value(name_2))
|
|
resolved_pattern = _expand_date_tokens(resolved_pattern)
|
|
resolved_pattern = _expand_native_tokens(resolved_pattern)
|
|
resolved_pattern = resolved_pattern.replace("%width%", str(width))
|
|
resolved_pattern = resolved_pattern.replace("%height%", str(height))
|
|
resolved_pattern = _safe_pattern(resolved_pattern)
|
|
|
|
extension = ".jpg" if format == "jpg" else ".png"
|
|
quality = max(1, min(100, int(quality)))
|
|
ui_images = []
|
|
|
|
for batch_index in range(images.shape[0]):
|
|
frame_pattern = resolved_pattern.replace("%batch_num%", str(batch_index))
|
|
frame_parts = [part for part in frame_pattern.split("/") if part]
|
|
relative_template = "/".join(frame_parts[:-1] + [((frame_parts[-1] if frame_parts else "image_%counter%") + extension)])
|
|
counter = _next_counter_for_relative_path(target_dir, relative_template)
|
|
resolved_relative = relative_template.replace("%counter%", str(counter).zfill(3))
|
|
resolved_parts = [part for part in resolved_relative.split("/") if part]
|
|
sub_dirs = resolved_parts[:-1]
|
|
filename = resolved_parts[-1] if resolved_parts else f"image_{str(counter).zfill(3)}{extension}"
|
|
frame_dir = os.path.join(target_dir, *sub_dirs)
|
|
os.makedirs(frame_dir, exist_ok=True)
|
|
image = _tensor_image_to_pil_image(images[batch_index : batch_index + 1])
|
|
full_path = os.path.join(frame_dir, filename)
|
|
|
|
if format == "jpg":
|
|
image = image.convert("RGB")
|
|
image.save(full_path, "JPEG", quality=quality)
|
|
else:
|
|
pnginfo = None
|
|
if embed_workflow:
|
|
pnginfo = _build_pnginfo(prompt=prompt, extra_pnginfo=extra_pnginfo)
|
|
image.save(full_path, "PNG", pnginfo=pnginfo)
|
|
|
|
if _is_within_directory(output_dir, full_path):
|
|
subfolder = os.path.relpath(frame_dir, output_dir)
|
|
ui_images.append(
|
|
{
|
|
"filename": filename,
|
|
"subfolder": "" if subfolder == "." else subfolder.replace("\\", "/"),
|
|
"type": "output",
|
|
}
|
|
)
|
|
else:
|
|
ui_images.append(
|
|
{
|
|
"filename": filename,
|
|
"subfolder": frame_dir.replace("\\", "/"),
|
|
"type": "external",
|
|
"token": _register_serve_token(full_path),
|
|
}
|
|
)
|
|
|
|
return {"ui": {"images": ui_images}}
|
|
|
|
|
|
class DumasLoadImagesFolderNode:
|
|
DESCRIPTION = (
|
|
"Load many images from any folder on disk and feed them through your "
|
|
"workflow one at a time. Pick specific images, all images, the first N "
|
|
"images in sort order, or one random image per run."
|
|
)
|
|
RETURN_TYPES = ("IMAGE", "MASK", "INT", "INT", "STRING", "INT", "INT")
|
|
RETURN_NAMES = ("image", "mask", "width", "height", "filename", "index", "total")
|
|
OUTPUT_IS_LIST = (True, True, True, True, True, True, True)
|
|
FUNCTION = "load"
|
|
CATEGORY = "Dumas/Image"
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {},
|
|
"hidden": {
|
|
"LoadImagesFolderState": (
|
|
"STRING",
|
|
{"default": json.dumps(_LOAD_IMAGES_FOLDER_DEFAULT_STATE)},
|
|
),
|
|
},
|
|
}
|
|
|
|
def load(self, LoadImagesFolderState=""):
|
|
state = _parse_load_images_folder_state(LoadImagesFolderState)
|
|
folder = str(state.get("folder") or "").strip()
|
|
recursive = bool(state.get("recursive", False))
|
|
|
|
if not folder or not os.path.isdir(folder):
|
|
raise ValueError(
|
|
"Load Images from Folder: folder not found. Set a folder on the node first."
|
|
)
|
|
|
|
real_folder = os.path.realpath(folder)
|
|
files = _list_folder_image_files(real_folder, recursive)
|
|
selected = _resolve_folder_selection(state, files)
|
|
mode = str(state.get("selection_mode") or "selected").lower()
|
|
if not selected:
|
|
if mode == "random":
|
|
raise ValueError(
|
|
"Load Images from Folder: no images found for random selection."
|
|
)
|
|
raise ValueError(
|
|
"Load Images from Folder: no images selected. Use Pick images on the node."
|
|
)
|
|
|
|
images = []
|
|
masks = []
|
|
widths = []
|
|
heights = []
|
|
names = []
|
|
indices = []
|
|
|
|
count = 0
|
|
for rel_path in selected:
|
|
if not isinstance(rel_path, str) or not rel_path:
|
|
continue
|
|
full_path = os.path.realpath(os.path.join(real_folder, rel_path))
|
|
if not _is_within_directory(real_folder, full_path) or not os.path.isfile(full_path):
|
|
continue
|
|
try:
|
|
image_tensor, mask_tensor, width, height = _load_folder_image(full_path)
|
|
except Exception as exc:
|
|
print(f"[DumasLoadImagesFolder] failed to load {rel_path}: {exc}")
|
|
continue
|
|
|
|
images.append(image_tensor)
|
|
masks.append(mask_tensor)
|
|
widths.append(width)
|
|
heights.append(height)
|
|
if recursive:
|
|
names.append(os.path.splitext(rel_path)[0].replace("/", "_").replace("\\", "_"))
|
|
else:
|
|
names.append(os.path.splitext(os.path.basename(rel_path))[0])
|
|
count += 1
|
|
indices.append(count)
|
|
|
|
if not images:
|
|
raise ValueError(
|
|
"Load Images from Folder: none of the chosen images could be loaded."
|
|
)
|
|
|
|
totals = [count] * len(images)
|
|
return (images, masks, widths, heights, names, indices, totals)
|
|
|
|
@classmethod
|
|
def IS_CHANGED(cls, LoadImagesFolderState=""):
|
|
state = _parse_load_images_folder_state(LoadImagesFolderState)
|
|
folder = str(state.get("folder") or "").strip()
|
|
if not folder or not os.path.isdir(folder):
|
|
return hashlib.sha256((LoadImagesFolderState or "").encode("utf-8")).hexdigest()
|
|
|
|
real_folder = os.path.realpath(folder)
|
|
files = _list_folder_image_files(real_folder, bool(state.get("recursive", False)))
|
|
mode = str(state.get("selection_mode") or "selected").lower()
|
|
parts = [json.dumps({k: v for k, v in state.items() if k != "selected"}, sort_keys=True)]
|
|
|
|
if mode == "random":
|
|
for entry in _sort_folder_image_files(files, state.get("sort", "name"), state.get("sort_dir", "asc")):
|
|
parts.append(f"{entry['file']}:{entry.get('mtime', 0)}")
|
|
parts.append(f"random:{time.time_ns()}")
|
|
else:
|
|
for rel_path in _resolve_folder_selection(state, files):
|
|
full_path = os.path.realpath(os.path.join(real_folder, rel_path))
|
|
if not _is_within_directory(real_folder, full_path):
|
|
parts.append(f"{rel_path}:outside")
|
|
continue
|
|
try:
|
|
parts.append(f"{rel_path}:{os.stat(full_path).st_mtime_ns}")
|
|
except OSError:
|
|
parts.append(f"{rel_path}:missing")
|
|
|
|
return hashlib.sha256("|".join(parts).encode("utf-8")).hexdigest()
|
|
|
|
|
|
class DumasH3PlanAttachSceneImagesNode:
|
|
DESCRIPTION = (
|
|
"Attach up to nine optional IMAGE sockets to one H3 Chain Plan scene "
|
|
"without breaking the upstream plan archive format. Chain multiple "
|
|
"copies of this node to bind different scene indexes."
|
|
)
|
|
RETURN_TYPES = (_H3_PLAN_TYPE, "INT")
|
|
RETURN_NAMES = ("plan", "connected_images")
|
|
FUNCTION = "attach"
|
|
CATEGORY = "Dumas/MiniMax"
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
optional = {}
|
|
for slot in range(1, _H3_PLAN_IMAGE_SLOTS + 1):
|
|
optional[f"image{slot}"] = (
|
|
"IMAGE",
|
|
{
|
|
"tooltip": (
|
|
f"Optional image for slot {slot} on the selected H3 plan scene."
|
|
)
|
|
},
|
|
)
|
|
return {
|
|
"required": {
|
|
"plan": (
|
|
_H3_PLAN_TYPE,
|
|
{
|
|
"tooltip": (
|
|
"Validated MiniMax H3 chain plan to enrich with scene-level "
|
|
"image bindings."
|
|
)
|
|
},
|
|
),
|
|
"scene_index": (
|
|
"INT",
|
|
{
|
|
"default": 1,
|
|
"min": 1,
|
|
"max": 9999,
|
|
"step": 1,
|
|
"tooltip": (
|
|
"1-based scene index inside the H3 plan. Use one node per "
|
|
"scene that needs up to nine image sockets."
|
|
),
|
|
},
|
|
),
|
|
},
|
|
"optional": optional,
|
|
}
|
|
|
|
def attach(
|
|
self,
|
|
plan,
|
|
scene_index,
|
|
image1=None,
|
|
image2=None,
|
|
image3=None,
|
|
image4=None,
|
|
image5=None,
|
|
image6=None,
|
|
image7=None,
|
|
image8=None,
|
|
image9=None,
|
|
):
|
|
updated_plan = _clone_h3_plan(plan)
|
|
scene_index = _normalize_h3_scene_index(updated_plan, scene_index)
|
|
images = _scene_images_tuple(
|
|
image1,
|
|
image2,
|
|
image3,
|
|
image4,
|
|
image5,
|
|
image6,
|
|
image7,
|
|
image8,
|
|
image9,
|
|
)
|
|
connected_count = _connected_image_count(images)
|
|
|
|
token, scene_counts = _h3_plan_binding_entry(updated_plan)
|
|
if not token:
|
|
token = uuid.uuid4().hex
|
|
registry = _H3_PLAN_IMAGE_BINDINGS.setdefault(token, {})
|
|
_touch_plan_image_binding(token)
|
|
|
|
if connected_count:
|
|
registry[int(scene_index)] = images
|
|
scene_counts[str(scene_index)] = connected_count
|
|
else:
|
|
registry.pop(int(scene_index), None)
|
|
scene_counts.pop(str(scene_index), None)
|
|
|
|
if registry:
|
|
updated_plan[_H3_PLAN_IMAGE_BINDINGS_KEY] = {
|
|
"token": token,
|
|
"scene_counts": scene_counts,
|
|
}
|
|
else:
|
|
_H3_PLAN_IMAGE_BINDINGS.pop(token, None)
|
|
updated_plan.pop(_H3_PLAN_IMAGE_BINDINGS_KEY, None)
|
|
|
|
_prune_plan_image_bindings()
|
|
return (updated_plan, connected_count)
|
|
|
|
|
|
class DumasH3PlanExtractSceneImagesNode:
|
|
DESCRIPTION = (
|
|
"Read back the nine optional image bindings for one H3 Chain Plan scene. "
|
|
"Connect clip_index or another scene selector to recover the matching "
|
|
"scene images downstream."
|
|
)
|
|
RETURN_TYPES = (
|
|
_H3_PLAN_TYPE,
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"IMAGE",
|
|
"INT",
|
|
)
|
|
RETURN_NAMES = (
|
|
"plan",
|
|
"image1",
|
|
"image2",
|
|
"image3",
|
|
"image4",
|
|
"image5",
|
|
"image6",
|
|
"image7",
|
|
"image8",
|
|
"image9",
|
|
"connected_images",
|
|
)
|
|
FUNCTION = "extract"
|
|
CATEGORY = "Dumas/MiniMax"
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"plan": (
|
|
_H3_PLAN_TYPE,
|
|
{
|
|
"tooltip": (
|
|
"H3 plan previously enriched by Dumas H3 Plan Attach Scene Images."
|
|
)
|
|
},
|
|
),
|
|
"scene_index": (
|
|
"INT",
|
|
{
|
|
"default": 1,
|
|
"min": 1,
|
|
"max": 9999,
|
|
"step": 1,
|
|
"tooltip": (
|
|
"1-based scene index to retrieve. Connect Current Shot "
|
|
"clip_index to get the active scene's images."
|
|
),
|
|
},
|
|
),
|
|
}
|
|
}
|
|
|
|
def extract(self, plan, scene_index):
|
|
passthrough_plan = _clone_h3_plan(plan)
|
|
scene_index = _normalize_h3_scene_index(passthrough_plan, scene_index)
|
|
token, _scene_counts = _h3_plan_binding_entry(passthrough_plan)
|
|
if not token:
|
|
return (passthrough_plan, *(None for _ in range(_H3_PLAN_IMAGE_SLOTS)), 0)
|
|
|
|
registry = _H3_PLAN_IMAGE_BINDINGS.get(token) or {}
|
|
_touch_plan_image_binding(token)
|
|
images = registry.get(int(scene_index)) or (None,) * _H3_PLAN_IMAGE_SLOTS
|
|
return (passthrough_plan, *images, _connected_image_count(images))
|
|
|
|
|
|
class DumasCharacterHelperNode:
|
|
DESCRIPTION = (
|
|
"Build a MiniMax H3-ready character reference prompt and wardrobe sheet "
|
|
"from two IMAGE sockets plus simple identity fields, while passing both "
|
|
"images through unchanged."
|
|
)
|
|
RETURN_TYPES = ("IMAGE", "IMAGE", "STRING", "STRING")
|
|
RETURN_NAMES = ("image1", "image2", "reference_prompt", "wardrobe")
|
|
FUNCTION = "build_character_text"
|
|
CATEGORY = "Dumas/MiniMax"
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"image1": ("IMAGE", {"tooltip": "Primary image to pass through and describe."}),
|
|
"image2": ("IMAGE", {"tooltip": "Secondary image to pass through and describe."}),
|
|
"image1_picture_id": (
|
|
["1", "2", "3", "4", "5", "6", "7", "8", "9"],
|
|
{
|
|
"default": "1",
|
|
"tooltip": "Picture number to mention for image1 in the H3 reference prompt.",
|
|
},
|
|
),
|
|
"image2_picture_id": (
|
|
["1", "2", "3", "4", "5", "6", "7", "8", "9"],
|
|
{
|
|
"default": "2",
|
|
"tooltip": "Picture number to mention for image2 in the H3 reference prompt.",
|
|
},
|
|
),
|
|
"character_id": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Optional character ID string to include in the output text.",
|
|
},
|
|
),
|
|
"name": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Character name used in the main reference sentences.",
|
|
},
|
|
),
|
|
"alias": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Optional alternate name, codename, or nickname.",
|
|
},
|
|
),
|
|
"gender": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Optional gender field for non-visual character facts.",
|
|
},
|
|
),
|
|
"age": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Optional numeric age. Invalid values are omitted.",
|
|
},
|
|
),
|
|
"nationality": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Optional nationality, origin, or cultural background.",
|
|
},
|
|
),
|
|
"occupation": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Optional job, role, or function that is not visually obvious.",
|
|
},
|
|
),
|
|
"height_feet": (
|
|
["", "3", "4", "5", "6", "7", "8"],
|
|
{
|
|
"default": "",
|
|
"tooltip": "Optional feet component for the character's height.",
|
|
},
|
|
),
|
|
"height_inches": (
|
|
["", "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11"],
|
|
{
|
|
"default": "",
|
|
"tooltip": "Optional inches component for the character's height.",
|
|
},
|
|
),
|
|
"accent": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": False,
|
|
"tooltip": "Optional short accent description.",
|
|
},
|
|
),
|
|
"general": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": True,
|
|
"tooltip": "Optional non-clothing details appended as the last sentence of the H3 reference prompt.",
|
|
},
|
|
),
|
|
"wardrobe": (
|
|
"STRING",
|
|
{
|
|
"default": "",
|
|
"multiline": True,
|
|
"tooltip": "Optional H3 wardrobe/channel text. Plain clothing lists are auto-wrapped as 'Name = ...' when a name, alias, or character ID is present.",
|
|
},
|
|
),
|
|
}
|
|
}
|
|
|
|
def build_character_text(
|
|
self,
|
|
image1,
|
|
image2,
|
|
image1_picture_id,
|
|
image2_picture_id,
|
|
character_id,
|
|
name,
|
|
alias,
|
|
gender,
|
|
age,
|
|
nationality,
|
|
occupation,
|
|
height_feet,
|
|
height_inches,
|
|
accent,
|
|
general,
|
|
wardrobe,
|
|
):
|
|
text = _build_character_helper_text(
|
|
image1_picture_id,
|
|
image2_picture_id,
|
|
character_id,
|
|
name,
|
|
alias,
|
|
gender,
|
|
age,
|
|
nationality,
|
|
occupation,
|
|
height_feet,
|
|
height_inches,
|
|
accent,
|
|
general,
|
|
)
|
|
wardrobe_text = _build_character_wardrobe_text(
|
|
wardrobe,
|
|
character_id,
|
|
name,
|
|
alias,
|
|
)
|
|
return (image1, image2, text, wardrobe_text)
|
|
|
|
|
|
NODE_CLASS_MAPPINGS = {
|
|
"DumasImageCompare": DumasImageCompareNode,
|
|
"DumasSaveImage": DumasSaveImageNode,
|
|
"DumasLoadImagesFolder": DumasLoadImagesFolderNode,
|
|
"DumasH3PlanAttachSceneImages": DumasH3PlanAttachSceneImagesNode,
|
|
"DumasH3PlanExtractSceneImages": DumasH3PlanExtractSceneImagesNode,
|
|
"DumasCharacterHelper": DumasCharacterHelperNode,
|
|
"DumasH3CharacterHelper": DumasCharacterHelperNode,
|
|
}
|
|
|
|
NODE_DISPLAY_NAME_MAPPINGS = {
|
|
"DumasImageCompare": "Dumas Image Compare",
|
|
"DumasSaveImage": "Save Image Dumas",
|
|
"DumasLoadImagesFolder": "Load Images from Folder Dumas",
|
|
"DumasH3PlanAttachSceneImages": "Dumas H3 Plan Attach Scene Images",
|
|
"DumasH3PlanExtractSceneImages": "Dumas H3 Plan Extract Scene Images",
|
|
"DumasCharacterHelper": "Dumas H3 Character Helper",
|
|
"DumasH3CharacterHelper": "Dumas H3 Character Helper",
|
|
}
|