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 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, ): return (image1, image2, image3, image4, image5, image6, image7) 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 _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, pronouns, 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) pronouns = _normalize_free_text(pronouns) 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" and reference " f"the same character who is called {character_name}." ) elif character_id: first_line = ( f" and reference " f'the same character with ID "{character_id}".' ) else: first_line = ( f" and reference " "the same character." ) lines = [ first_line, f" is the primary full-body reference for {character_label}.", f" is a frontal facial reference for {character_label}.", ] if character_id: lines.append(f'The character ID string is "{character_id}".') if alias: lines.append(f"{character_label} is also known as {alias}.") if pronouns: lines.append(f"{character_label} uses {pronouns} pronouns.") if age_value is not None: lines.append(f"{character_label} is {age_value} years old.") if nationality: lines.append(f"{character_label} is {nationality}.") if occupation: lines.append(f"{character_label} works as {occupation}.") height_text = _format_height_text(height_feet, height_inches) if height_text: lines.append(f"{character_label} is {height_text} tall.") if accent: lines.append( f"{character_label} speaks in {_indefinite_article(accent)} {accent} accent." ) if general: lines.append(general) return "\n".join(lines) 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 DumasH3PlanAttachSceneImagesNode: DESCRIPTION = ( "Attach up to seven 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, 8): 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 seven image sockets." ), }, ), }, "optional": optional, } def attach( self, plan, scene_index, image1=None, image2=None, image3=None, image4=None, image5=None, image6=None, image7=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) 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 seven 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", "INT", ) RETURN_NAMES = ( "plan", "image1", "image2", "image3", "image4", "image5", "image6", "image7", "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, None, None, None, None, None, None, 0) registry = _H3_PLAN_IMAGE_BINDINGS.get(token) or {} _touch_plan_image_binding(token) images = registry.get(int(scene_index)) or (None, None, None, None, None, None, None) return (passthrough_plan, *images, _connected_image_count(images)) class DumasCharacterHelperNode: DESCRIPTION = ( "Build a reusable character reference string from two IMAGE sockets and " "simple identity fields, while passing both images through unchanged." ) RETURN_TYPES = ("IMAGE", "IMAGE", "STRING") RETURN_NAMES = ("image1", "image2", "character_text") FUNCTION = "build_character_text" CATEGORY = "Dumas/String" @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": ( ["2", "3", "4", "5", "6", "7"], { "default": "2", "tooltip": "Picture number to mention for image1 in the output string.", }, ), "image2_picture_id": ( ["2", "3", "4", "5", "6", "7"], { "default": "3", "tooltip": "Picture number to mention for image2 in the output string.", }, ), "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.", }, ), "pronouns": ( "STRING", { "default": "", "multiline": False, "tooltip": "Optional pronouns such as he/him or she/her.", }, ), "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 freeform details appended as the last sentence.", }, ), } } def build_character_text( self, image1, image2, image1_picture_id, image2_picture_id, character_id, name, alias, pronouns, age, nationality, occupation, height_feet, height_inches, accent, general, ): text = _build_character_helper_text( image1_picture_id, image2_picture_id, character_id, name, alias, pronouns, age, nationality, occupation, height_feet, height_inches, accent, general, ) return (image1, image2, text) NODE_CLASS_MAPPINGS = { "DumasImageCompare": DumasImageCompareNode, "DumasSaveImage": DumasSaveImageNode, "DumasH3PlanAttachSceneImages": DumasH3PlanAttachSceneImagesNode, "DumasH3PlanExtractSceneImages": DumasH3PlanExtractSceneImagesNode, "DumasCharacterHelper": DumasCharacterHelperNode, } NODE_DISPLAY_NAME_MAPPINGS = { "DumasImageCompare": "Dumas Image Compare", "DumasSaveImage": "Save Image Dumas", "DumasH3PlanAttachSceneImages": "Dumas H3 Plan Attach Scene Images", "DumasH3PlanExtractSceneImages": "Dumas H3 Plan Extract Scene Images", "DumasCharacterHelper": "Dumas Character Helper", }