Files
2026-10-20 12:38:56 +03:00

595 lines
23 KiB
Python

import os, re, io
import json
from os.path import realpath, join, dirname, isabs, splitext, basename
from datetime import datetime
from typing import Any, Tuple
import folder_paths
import torch
from .image_nodes import (LoadImageFromPathEnhanced, ImageComposer, ImageSharpener,
ResolutionScale, ImageFromString)
from .minimax import (H3MotionContextClipStitcher, H3ClipRefiner,
H3ContextLatentConverter,
H3MotionContextClipPurge, H3AVLatentFromVideo)
MANIFEST = {"name": "noEmbryo Nodes",
"version": (1, 9, 6),
"author": "noEmbryo",
"project": "https://github.com/noembryo/ComfyUI-noEmbryo",
"description": "Nodes for ComfyUI",
"license": "MIT",
}
__author__ = "noEmbryo"
__version__ = "1.9.6"
LISTS_PATH = join(dirname(realpath(__file__)), "TermLists")
class JsonPromptLoader:
data = {}
data_labels = ["None"]
json_path = ""
def __init__(self):
super(JsonPromptLoader, self).__init__()
self.name = type(self).__name__
@classmethod
def load_data(cls, json_path):
cls.json_path = ""
if not splitext(json_path)[1].lower() == ".json":
return
if json_path:
try:
with io.open(json_path, mode="r", encoding="utf-8") as f:
cls.data.clear()
cls.data["None"] = ""
cls.data.update(json.load(f))
cls.data_labels[:] = list(cls.data.keys())
cls.json_path = json_path
except (FileNotFoundError, json.JSONDecodeError):
cls.data.clear()
cls.data.update({})
cls.data_labels[:] = ["None"]
if os.stat(json_path).st_size == 0: # empty json files
cls.json_path = json_path
else: # no path given
cls.data.clear()
cls.data.update({})
cls.data_labels[:] = ["None"]
@classmethod
def INPUT_TYPES(cls):
return {"required": {"json_path": ("STRING", {"default": "",
"tooltip": "Path to a JSON file with "
"`item`:`prompt` pairs"}),
"selected_item": (cls.data_labels, cls.data), # Options will be updated by JS
"variable": ("STRING", {"default": "{subject}",
"tooltip": "If this variable exists in the selected item's prompt,\n"
"it will be replaced with the custom_prompt text"}),
"custom_prompt": ("STRING", {"multiline": True, "default": "",
"tooltip": "Text to replace the variable in the selected prompt.\n"
"You can also use it to save a new item or update an existing one.\n"
"To do that you should use the following format:\n"
"item=... ...\n"
"value=.... .... ...\n"
"To delete an existing item, use an empty value:\n"
"item=... ...\n"
"value="})},
}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("Prompt",)
FUNCTION = "run"
CATEGORY = "noEmbryo/Prompt"
DESCRIPTION = ("A node that can load a `.json` file with `item:prompt` pairs and outputs "
"the selected item's prompt, while combining it with a custom prompt.\n"
"It can load `.json` files from any directory, not just the node's directory.")
def run(self, json_path, selected_item, variable, custom_prompt):
self.load_data(json_path)
if custom_prompt:
message = self.edit_data(custom_prompt)
if message: # if the custom_prompt was saved successfully
return (message,)
if selected_item in self.data and selected_item != "None":
prompt = self.data[selected_item]
# if variable and "{" + variable + "}" in prompt:
if variable and variable in prompt:
prompt = prompt.replace(variable, custom_prompt)
# prompt = prompt.replace("{" + variable + "}", custom_prompt)
elif custom_prompt:
prompt += " " + custom_prompt
else:
prompt = custom_prompt
return (prompt,)
def edit_data(self, text):
""" Parses the json values from the custom_prompt and changes the json file
:type text: str
:param text: The custom_prompt text
"""
lines = text.splitlines()
if len(lines) >= 2:
if all((lines[0].startswith("item="), lines[1].startswith("value="))):
if not self.json_path:
return False
item = lines[0][5:]
lines_txt = "\n".join(lines[1:])
value = lines_txt[6:]
filename = basename(self.json_path)
if item == "None": # cannot change None
msg = f'{filename}: The item "{item}" cannot be changed!'
return msg
if not value: # delete item
if item in self.data:
del self.data[item]
msg = f'{filename}: The item "{item}" was deleted!'
self.save_json_file()
else:
msg = f'{filename}: The item "{item}" does not exist!'
else: # save/update item
if item in self.data:
msg = f'{filename}: The item "{item}" was updated!'
else:
msg = f'{filename}: The item "{item}" was added!'
self.data[item] = value
self.save_json_file()
return msg
return False
def save_json_file(self):
with io.open(self.json_path, mode="w", encoding="utf-8") as f:
data = self.data.copy()
if "None" in data:
del data["None"]
# noinspection PyTypeChecker
json.dump(data, f, ensure_ascii=False, indent=4)
class PromptTermList:
idx = 0
data = {"None": ""}
data_labels = []
has_error = False
input_error = ("Trying to store invalid input!\nUse the format:\n"
"label=... ...\nvalue=.... .... ...")
def __init__(self):
super(PromptTermList, self).__init__()
self.name = type(self).__name__
@classmethod
def load_data_from_json(cls, json_file_path):
""" Loads a json file from a path
:type json_file_path: str
:param json_file_path: The path to the json file
"""
try:
with io.open(json_file_path, mode="r", encoding="utf-8") as f:
cls.data = json.load(f)
cls.data_labels = list(cls.data.items())
except FileNotFoundError:
pass
@classmethod
def INPUT_TYPES(cls):
list_path = join(LISTS_PATH, f"TermList{cls.idx}.json")
cls.load_data_from_json(list_path)
term_list = [i[0] for i in cls.data_labels]
# noinspection SqlNoDataSourceInspection,SqlResolve
return {"required": {"terms": (term_list,{"tooltip": "Choose a term from the "
"TermList with the "
"corresponding number"}), },
"optional": {"text": ("STRING", {"forceInput": True,
"tooltip": "Input text to store in the "
"TermList\nUse the format:\n"
"label=... ...\n"
"value=.... .... ..."}),
# The round value representing the precision to round to,
# will be set to the step value by default.
# Can be set to False to disable rounding.
"strength": ("FLOAT", {"default": 1.0,
"min": 0.05,
"max": 2.0,
"step": 0.05,
"round": 0.01,
"display": "number",
"tooltip": "Controls how much the "
"image is allowed to change.\n"
"0.0 = almost no change\n"
"1.0 = maximum creativity"}),
"store_input": ("BOOLEAN",
{"default": False,
"tooltip": "Store the input text in the "
"TermList\nUse the format:\n"
"label=... ...\nvalue=.... .... ..."})},
}
def save_data_from_input(self, text):
""" Extracts the json values from the input text and stores them in the json file
:type text: str
:param text: The text input
"""
lines = text.splitlines()
if not len(lines) > 1:
self.has_error = True
print(f"{self.name}:", self.input_error)
return
if not all((lines[0].startswith("label="), lines[1].startswith("value="))):
self.has_error = True
print(f"{self.name}:", self.input_error)
return
label = lines[0][6:]
lines_txt = "\n".join(lines[1:])
value = lines_txt[6:]
if label == "None":
print(f'{self.name}: The label "{label}" cannot be changed!')
return
if not value:
if label in self.data:
del self.data[label]
print(f'{self.name}: The label "{label}" was deleted!')
else:
print(f'{self.name}: The label "{label}" does not exist!')
return
else:
if label in self.data:
print(f'{self.name}: The label "{label}" is updated!')
else:
print(f'{self.name}: The label "{label}" is saved!')
self.data[label] = value
with io.open(join(LISTS_PATH, "TermList{}.json".format(self.idx)), mode="w",
encoding="utf-8") as f:
# noinspection PyTypeChecker
json.dump(self.data, f, ensure_ascii=False, indent=4)
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("Term",)
# OUTPUT_NODE = True
CATEGORY = "noEmbryo/Prompt/Term Nodes"
FUNCTION = "run"
def run(self, terms, strength, store_input, text=None):
selected = terms[:len(terms)]
text_out = ""
for i in self.data_labels:
if i[0] == selected:
text_out = f"{i[1]} "
break
if selected != "None" and strength != 1.0:
text_out = f"({text_out}:{strength})"
if text:
if store_input:
self.save_data_from_input(text)
if not self.has_error:
text_out = ""
else:
self.has_error = False
text_out = self.input_error
else:
if text_out:
text_out = f"{text_out}, {text}"
else:
text_out = text
return (text_out, )
class PromptTermList1(PromptTermList):
idx = 1
class PromptTermList2(PromptTermList):
idx = 2
class PromptTermList3(PromptTermList):
idx = 3
class PromptTermList4(PromptTermList):
idx = 4
class PromptTermList5(PromptTermList):
idx = 5
class PromptTermList6(PromptTermList):
idx = 6
class RegExTextChopper:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(cls):
return {"required": {"text": ("STRING", {"forceInput": True,
"tooltip": "The text that we'll parse"}),
"regex": ("STRING", {"tooltip": "The RegEx pattern"})},
"optional": {},
}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("Part 1", "Part 2", "Part 3", "Part 4", "All parts")
FUNCTION = "run"
CATEGORY = "noEmbryo/Utils"
@staticmethod
def is_valid_regex(regex_from_user: str) -> bool:
try:
re.compile(re.escape(regex_from_user))
is_valid = True
except re.error:
is_valid = False
return is_valid
def run(self, text, regex):
if self.is_valid_regex(regex):
obj = re.compile(regex, re.MULTILINE)
result = obj.findall(text)
try:
text1 = result[0]
except IndexError:
text1 = ""
try:
text2 = result[1]
except IndexError:
text2 = ""
try:
text3 = result[2]
except IndexError:
text3 = ""
try:
text4 = result[3]
except IndexError:
text4 = ""
text_all = "\n\n".join(result)
else:
text1 = text2 = text3 = text4 = ""
text_all = text
return text1, text2, text3, text4, text_all
class AutoSaveWorkflow:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"save_directory": ("STRING", {
"default": "saved_workflows",
"tooltip": "Relative to ComfyUI output directory or absolute path"
}),
"filename": ("STRING", {
"default": "workflow_{timestamp}",
"tooltip": "Filename (include {timestamp} for unique timestamps)"
}),
"trigger": ("BOOLEAN", {
"default": True,
"label_on": "Enabled",
"label_off": "Disabled",
"tooltip": "Save the workflow if Enabled"
}),
},
"hidden": {
"prompt": "PROMPT",
"extra_pnginfo": "EXTRA_PNGINFO",
"trigger": "BOOLEAN", # Hidden trigger input
},
}
RETURN_TYPES = ("STRING", "BOOLEAN")
RETURN_NAMES = ("status", "✳️trigger")
OUTPUT_TOOLTIPS = ("Get a status report text",
"Dammy output, to trigger execution if nothing is connected")
FUNCTION = "execute"
CATEGORY = "noEmbryo/Utils"
OUTPUT_NODE = True
# noinspection PyUnusedLocal
@staticmethod
def execute(trigger, save_directory, filename, prompt=None, extra_pnginfo=None):
status = "Trigger disabled - workflow not saved"
if trigger:
try:
workflow_data = extra_pnginfo.get("workflow", {}) if extra_pnginfo else {}
# Process save directory
if isabs(save_directory):
output_dir = save_directory
else:
output_dir = join(folder_paths.get_output_directory(), save_directory)
os.makedirs(output_dir, exist_ok=True)
# Process filename with timestamp
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
processed_filename = filename.replace("{timestamp}", timestamp)
# Ensure .json extension
if not processed_filename.lower().endswith('.json'):
processed_filename += '.json'
save_path = join(output_dir, processed_filename)
# Save workflow to JSON
with open(save_path, "w", encoding="utf-8") as f:
# noinspection PyTypeChecker
json.dump(workflow_data, f, indent=4)
status = f"Workflow saved to: {save_path}"
except Exception as e:
status = f"Error saving workflow: {str(e)}"
return (status,)
class AnyType(str):
"""A special class that is always equal in not equal comparisons."""
def __ne__(self, __value: object) -> bool:
return False
any_type = AnyType("*")
# copied from https://github.com/Suzie1/ComfyUI_Comfyroll_CustomNodes/wiki/Conversion-Nodes#cr-string-to-combo
class StringToCombo:
""" Convert comma-separated string to combo selection (first item)
Useful for converting the output of a `Custom Combo` (string) to actual combo value
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"string": ("STRING", {
"multiline": False,
"default": "",
"tooltip": "Comma-separated string to convert to combo"
}),
},
}
RETURN_TYPES = (any_type,)
RETURN_NAMES = ("any",)
FUNCTION = "convert"
CATEGORY = "noEmbryo/Utils"
DESCRIPTION = ("Convert comma-separated string to combo selection (first item).\n"
"Useful for converting the output of a `Custom Combo` (string) "
"to actual combo value.")
@staticmethod
def convert(string: str) -> Tuple[Any]:
""" Convert string to combo value (first item in comma-separated list)
"""
text_list = tuple()
if string != "":
values = string.split(',')
text_list = values[0].strip() # Take first item and strip whitespace
return (text_list,)
class StringToInt:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"string": ("STRING", {"default": "", "forceInput": True,
"tooltip": "whitespace will be "
"stripped before parsing"}),
"default": ("INT", {"default": 0,
"tooltip": "used if the string can't be parsed as an "
"integer"}), }}
RETURN_TYPES = ("INT",)
FUNCTION = "func"
CATEGORY = "noEmbryo/Utils"
DESCRIPTION = "Convert a string to an int."
@staticmethod
def func(string: str, default: int):
# noinspection PyBroadException
try:
return (int(string.strip()),)
except Exception:
return (default,)
class EmbellishmentNode:
""" Embellishment Node - Combines a Divider and a Spacer functionality.
When height_px is 0, acts as visual divider line.
When height_px > 0, acts as vertical spacing.
Inputs:
- color: STRING widget, divider color in hex (e.g. "#ffffff")
- height_px: INT widget, height in pixels (0 = Divider, >0 = Spacer)
- decorator: UI_SUB_DIVIDER (the single connection, exposed to subgraph outer node)
"""
@classmethod
def INPUT_TYPES(cls):
return {
"required": {}, "optional": {"decorator": ("UI_SUB_DIVIDER", {
"tooltip": "Single connection — acts as divider when height_px=0, "
"as spacer when height_px>0"}), }}
RETURN_TYPES = ()
FUNCTION = "out"
CATEGORY = "noEmbryo/Utils"
OUTPUT_NODE = True
DESCRIPTION = ("Embellishment Node - Combines a Divider and a Spacer functionality.\n"
" When height_px is 0, acts as Divider.\n"
" When height_px > 0, acts as Spacer.")
# noinspection PyMethodMayBeStatic,PyUnusedLocal
def out(self, decorator=None):
return ()
NODE_CLASS_MAPPINGS = {
f"JsonPromptLoader -{__author__}": JsonPromptLoader,
f"Resolution Scale -{__author__}": ResolutionScale,
f"Regex Text Chopper -{__author__}": RegExTextChopper,
f"StringToCombo -{__author__}": StringToCombo,
f"StringToInt -{__author__}": StringToInt,
f"Embellishment -{__author__}": EmbellishmentNode,
f"Auto Save Workflow -{__author__}": AutoSaveWorkflow,
f"Load Image (from path) -{__author__}": LoadImageFromPathEnhanced,
f"Image Composer -{__author__}": ImageComposer,
f"Image Sharpener -{__author__}": ImageSharpener,
f"Image From String -{__author__}": ImageFromString,
f"H3MotionContextClipStitcher -{__author__}": H3MotionContextClipStitcher,
f"H3ClipRefiner -{__author__}": H3ClipRefiner,
f"H3MotionContextClipPurge -{__author__}": H3MotionContextClipPurge,
f"H3ContextLatentConverter -{__author__}": H3ContextLatentConverter,
f"H3AVLatentFromVideo -{__author__}": H3AVLatentFromVideo,
"PromptTermList1": PromptTermList1,
"PromptTermList2": PromptTermList2,
"PromptTermList3": PromptTermList3,
"PromptTermList4": PromptTermList4,
"PromptTermList5": PromptTermList5,
"PromptTermList6": PromptTermList6,
}
NODE_DISPLAY_NAME_MAPPINGS = {
f"JsonPromptLoader -{__author__}": f"Json Prompt Loader /{__author__}",
f"Resolution Scale -{__author__}": f"Resolution Scale /{__author__}",
f"Regex Text Chopper -{__author__}": f"Regex Text Chopper /{__author__}",
f"StringToCombo -{__author__}": f"String to Combo /{__author__}",
f"StringToInt -{__author__}": f"String to Int /{__author__}",
f"Embellishment -{__author__}": f"Embellishment /{__author__}",
f"Auto Save Workflow -{__author__}": f"Auto Save Workflow /{__author__}",
f"Load Image (from path) -{__author__}": f"Load Image (from path) /{__author__}",
f"Image Composer -{__author__}": f"Image Composer /{__author__}",
f"Image Sharpener -{__author__}": f"Image Sharpener /{__author__}",
f"Image From String -{__author__}": f"Image From String /{__author__}",
f"H3MotionContextClipStitcher -{__author__}": f"H3 Motion Context Clip Stitcher /{__author__}",
f"H3ClipRefiner -{__author__}": f"H3 Clip Refiner /{__author__}",
f"H3MotionContextClipPurge -{__author__}": f"H3 Motion Context Clip Purge /{__author__}",
f"H3ContextLatentConverter -{__author__}": f"H3 Context Latent Converter /{__author__}",
f"H3AVLatentFromVideo -{__author__}": f"H3 AV Latent from Video /{__author__}",
"PromptTermList1": f"PromptTermList 1 /{__author__}",
"PromptTermList2": f"PromptTermList 2 /{__author__}",
"PromptTermList3": f"PromptTermList 3 /{__author__}",
"PromptTermList4": f"PromptTermList 4 /{__author__}",
"PromptTermList5": f"PromptTermList 5 /{__author__}",
"PromptTermList6": f"PromptTermList 6 /{__author__}",
}
# # 2check COMMENT OUT BEFORE UPLOADING
# from .xtra import NODE_CLASS_MAPPINGS as xtra_NODE_CLASS_MAPPINGS
# from .xtra import NODE_DISPLAY_NAME_MAPPINGS as xtra_NODE_DISPLAY_NAME_MAPPINGS
# NODE_CLASS_MAPPINGS.update(xtra_NODE_CLASS_MAPPINGS)
# NODE_DISPLAY_NAME_MAPPINGS.update(xtra_NODE_DISPLAY_NAME_MAPPINGS)