837 lines
28 KiB
Python
837 lines
28 KiB
Python
import base64
|
||
import io
|
||
import json
|
||
import math
|
||
import os
|
||
import pprint
|
||
import random
|
||
import re
|
||
import time
|
||
import ast
|
||
import operator as op
|
||
import torch
|
||
import logging
|
||
from comfy_execution.graph import ExecutionBlocker
|
||
|
||
_CATEGORY = "KYNode/Utils"
|
||
|
||
|
||
def is_integer(n):
|
||
if n % 1 == 0:
|
||
return True
|
||
else:
|
||
return False
|
||
|
||
|
||
class AnyType(str):
|
||
def __ne__(self, __value: object) -> bool:
|
||
return False
|
||
|
||
|
||
any_typ = AnyType("*")
|
||
|
||
def safe_get_bbox(lst, index, default=None):
|
||
if 0 <= index < len(lst):
|
||
return [[lst[index]]]
|
||
return default
|
||
|
||
def is_deep_empty(obj):
|
||
if obj is None:
|
||
return True
|
||
if isinstance(obj, str):
|
||
return obj.strip() == ''
|
||
if isinstance(obj, torch.Tensor) and torch.all(obj == 0): # 判断是否为空 image 或者 mask
|
||
print('判断是否为空 image 或者 mask sinstance(obj, torch.Tensor) and torch.all(obj == 0)')
|
||
return True
|
||
if isinstance(obj, (list, tuple, set)):
|
||
if len(obj) == 0:
|
||
return True
|
||
return all(is_deep_empty(item) for item in obj)
|
||
if isinstance(obj, dict):
|
||
return len(obj) == 0 or all(is_deep_empty(v) for v in obj.values())
|
||
if hasattr(obj, 'numel'): # PyTorch Tensor
|
||
return obj.numel() == 0
|
||
if hasattr(obj, 'size'): # NumPy Array
|
||
return obj.size == 0
|
||
if isinstance(obj, (int, float, complex)):
|
||
return obj == 0
|
||
return False # 非空值（数字、非空字符串、对象等）
|
||
|
||
class KY_JoinToString:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"optional": {
|
||
"prefix": ("STRING", {"default": ""}),
|
||
"any1": (any_typ,),
|
||
"any1_any2_gap": ("STRING", {"default": ""}),
|
||
"any2": (any_typ,),
|
||
"any2_any3_gap": ("STRING", {"default": ""}),
|
||
"any3": (any_typ,),
|
||
"suffix": ("STRING", {"default": ""}),
|
||
}
|
||
}
|
||
|
||
RETURN_TYPES = (
|
||
"STRING",
|
||
"STRING",
|
||
"STRING",
|
||
)
|
||
RETURN_NAMES = ("prefix+all-text+suffix", "prefix+any1+gap+suffix", "prefix+any2+gap+suffix")
|
||
FUNCTION = "stringifyAny"
|
||
CATEGORY = _CATEGORY
|
||
DESCRIPTION = """
|
||
Converts any type to a string.
|
||
"""
|
||
|
||
def stringifyAny(
|
||
self,
|
||
any1="",
|
||
any2="",
|
||
any3="",
|
||
prefix="",
|
||
any1_any2_gap="",
|
||
any2_any3_gap="",
|
||
suffix="",
|
||
):
|
||
stringified = ""
|
||
|
||
if isinstance(any1, (int, float, bool, str)):
|
||
stringified = str(any1)
|
||
elif isinstance(any1, list):
|
||
stringified = ", ".join(str(item) for item in any1)
|
||
any1text = stringified
|
||
any2text = ""
|
||
if any1_any2_gap: # Check if first_gap is not empty
|
||
stringified = stringified + any1_any2_gap # Add the first_gap
|
||
|
||
if isinstance(any2, (int, float, bool, str)):
|
||
stringified = stringified + str(any2)
|
||
any2text = str(any2)
|
||
elif isinstance(any2, list):
|
||
stringified = stringified + ", ".join(str(item) for item in any2)
|
||
any2text = ", ".join(str(item) for item in any2)
|
||
|
||
if any2_any3_gap: # Check if second_gap is not empty
|
||
stringified = stringified + any2_any3_gap # Add the second_gap
|
||
|
||
if isinstance(any3, (int, float, bool, str)):
|
||
stringified = stringified + str(any3)
|
||
elif isinstance(any3, list):
|
||
stringified = stringified + ", ".join(str(item) for item in any3)
|
||
|
||
if prefix: # Check if prefix is not empty
|
||
stringified = prefix + stringified # Add the prefix
|
||
if suffix: # Check if suffix is not empty
|
||
stringified = stringified + suffix # Add the suffix
|
||
return (
|
||
stringified,
|
||
prefix + any1text + any1_any2_gap + suffix,
|
||
prefix + any2text + any2_any3_gap + suffix,
|
||
)
|
||
|
||
|
||
class KY_RegexReplace:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"input_string": (
|
||
"STRING",
|
||
{
|
||
"default": "",
|
||
"multiline": True,
|
||
},
|
||
),
|
||
"regex1": ("STRING", {"default": ""}),
|
||
"replace1": ("STRING", {"default": ""}),
|
||
},
|
||
"optional": {
|
||
"any1": (any_typ,),
|
||
"any2": (any_typ,),
|
||
"regex2": ("STRING", {"default": ""}),
|
||
"replace2": ("STRING", {"default": ""}),
|
||
},
|
||
}
|
||
|
||
RETURN_TYPES = (
|
||
"STRING",
|
||
"STRING",
|
||
)
|
||
RETURN_NAMES = (
|
||
"regex1-result",
|
||
"final-result",
|
||
)
|
||
FUNCTION = "replace"
|
||
CATEGORY = _CATEGORY
|
||
DESCRIPTION = """
|
||
Converts any type to a string.
|
||
"""
|
||
|
||
def replace(
|
||
self,
|
||
any1="",
|
||
any2="",
|
||
regex1="",
|
||
replace1="",
|
||
input_string="",
|
||
regex2="",
|
||
replace2="",
|
||
):
|
||
stringified = ""
|
||
if isinstance(any1, (int, float, bool, str)):
|
||
stringified = str(any1)
|
||
elif isinstance(any1, list):
|
||
stringified = ", ".join(str(item) for item in any1)
|
||
|
||
if isinstance(any2, (int, float, bool, str)):
|
||
stringified = stringified + str(any2)
|
||
elif isinstance(any2, list):
|
||
stringified = stringified + ", ".join(str(item) for item in any2)
|
||
|
||
if input_string: # Check if prefix is not empty
|
||
stringified = stringified + input_string # Add the prefix
|
||
|
||
result1 = re.sub(regex1, replace1, stringified)
|
||
result2 = ""
|
||
if regex2:
|
||
result2 = re.sub(regex2, replace2, result1)
|
||
return (
|
||
result1,
|
||
result2,
|
||
)
|
||
|
||
|
||
class KY_RegexExtractor:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"regex": ("STRING", {"default": ""}),
|
||
"group_number": ("INT", {"default": 0, "min": 0, "max": 100}),
|
||
},
|
||
"optional": {
|
||
"any1": (any_typ,),
|
||
"any2": (any_typ,),
|
||
"input_string": (
|
||
"STRING",
|
||
{
|
||
"default": "",
|
||
"multiline": True,
|
||
},
|
||
),
|
||
},
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING", "STRING", "STRING")
|
||
RETURN_NAMES = (
|
||
"group_N",
|
||
"group_1",
|
||
"group_2",
|
||
)
|
||
FUNCTION = "execute"
|
||
CATEGORY = _CATEGORY
|
||
DESCRIPTION = "使用正则表达式从输入字符串中提取文本"
|
||
|
||
def execute(self, input_string="", regex="", group_number=1, any1="", any2=""):
|
||
stringified = ""
|
||
if isinstance(any1, (int, float, bool, str)):
|
||
stringified = str(any1)
|
||
elif isinstance(any1, list):
|
||
stringified = ", ".join(str(item) for item in any1)
|
||
|
||
if isinstance(any2, (int, float, bool, str)):
|
||
stringified = stringified + str(any2)
|
||
elif isinstance(any2, list):
|
||
stringified = stringified + ", ".join(str(item) for item in any2)
|
||
stringified = stringified + input_string
|
||
try:
|
||
match = re.search(regex, stringified, re.MULTILINE | re.DOTALL)
|
||
if match:
|
||
groups = match.groups()
|
||
count = len(groups)
|
||
if count >= 2:
|
||
return (match.group(group_number), match.group(1), match.group(2))
|
||
elif count >= 1:
|
||
return (
|
||
match.group(group_number),
|
||
match.group(1),
|
||
"",
|
||
)
|
||
elif 0 <= group_number <= len(groups):
|
||
return (match.group(group_number), "", "")
|
||
else:
|
||
return ("", "", "")
|
||
else:
|
||
return ("", "", "")
|
||
except re.error:
|
||
return ("无效的正则表达式", "", "")
|
||
|
||
|
||
operators = {
|
||
ast.Add: op.add,
|
||
ast.Sub: op.sub,
|
||
ast.Mult: op.mul,
|
||
ast.Div: op.truediv,
|
||
ast.FloorDiv: op.floordiv,
|
||
ast.Pow: op.pow,
|
||
ast.BitXor: op.xor,
|
||
ast.USub: op.neg,
|
||
ast.Mod: op.mod,
|
||
ast.BitAnd: op.and_,
|
||
ast.BitOr: op.or_,
|
||
ast.Invert: op.invert,
|
||
ast.And: lambda a, b: 1 if a and b else 0,
|
||
ast.Or: lambda a, b: 1 if a or b else 0,
|
||
ast.Not: lambda a: 0 if a else 1,
|
||
ast.RShift: op.rshift,
|
||
ast.LShift: op.lshift,
|
||
}
|
||
|
||
# TODO: restructure args to provide more info, generate hint based on args to save duplication
|
||
functions = {
|
||
"round": {
|
||
"args": (1, 2),
|
||
"call": lambda a, b=None: round(a, b),
|
||
"hint": "number, dp? = 0",
|
||
},
|
||
"ceil": {"args": (1, 1), "call": lambda a: math.ceil(a), "hint": "number"},
|
||
"floor": {"args": (1, 1), "call": lambda a: math.floor(a), "hint": "number"},
|
||
"min": {"args": (2, None), "call": lambda *args: min(*args), "hint": "...numbers"},
|
||
"max": {"args": (2, None), "call": lambda *args: max(*args), "hint": "...numbers"},
|
||
"randomint": {
|
||
"args": (2, 2),
|
||
"call": lambda a, b: random.randint(a, b),
|
||
"hint": "min, max",
|
||
},
|
||
"randomchoice": {
|
||
"args": (2, None),
|
||
"call": lambda *args: random.choice(args),
|
||
"hint": "...numbers",
|
||
},
|
||
"sqrt": {"args": (1, 1), "call": lambda a: math.sqrt(a), "hint": "number"},
|
||
"int": {"args": (1, 1), "call": lambda a=None: int(a), "hint": "number"},
|
||
"iif": {
|
||
"args": (3, 3),
|
||
"call": lambda a, b, c=None: b if a else c,
|
||
"hint": "value, truepart, falsepart",
|
||
},
|
||
}
|
||
|
||
|
||
class KY_MathExpression:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"d": (
|
||
"FLOAT",
|
||
{
|
||
"default": 0,
|
||
# "step": 0.5,
|
||
},
|
||
),
|
||
"expression": (
|
||
"STRING",
|
||
{
|
||
"multiline": False,
|
||
"dynamicPrompts": False,
|
||
},
|
||
),
|
||
},
|
||
"optional": {
|
||
"a": (any_typ,),
|
||
"b": (any_typ,),
|
||
"c": (any_typ,),
|
||
},
|
||
"hidden": {"extra_pnginfo": "EXTRA_PNGINFO", "prompt": "PROMPT"},
|
||
}
|
||
|
||
RETURN_TYPES = ("INT", "FLOAT", "STRING")
|
||
RETURN_NAMES = ("int", "float", "text")
|
||
FUNCTION = "evaluate"
|
||
CATEGORY = _CATEGORY
|
||
OUTPUT_NODE = True
|
||
DESCRIPTION = """
|
||
KY_MathExpression accept any value can be convert into numbers:
|
||
- round(n), ceil(n), floor(n), int(n), sqrt(n)
|
||
- min(n1, n2, n3...),
|
||
- max(n1, n2, n3...)
|
||
- randomint(min_int_number, max_int_number):
|
||
- randomchoice(n1, n2, n4, n4...): random select arguments
|
||
- iif(bool-eval, is_true_val, is_false_val): iif(1>2, c, d) = return d
|
||
|
||
"""
|
||
|
||
@classmethod
|
||
def IS_CHANGED(s, expression, **kwargs):
|
||
if "random" in expression:
|
||
return float("nan")
|
||
return expression
|
||
|
||
def get_widget_value(self, extra_pnginfo, prompt, node_name, widget_name):
|
||
workflow = (
|
||
extra_pnginfo["workflow"] if "workflow" in extra_pnginfo else {"nodes": []}
|
||
)
|
||
node_id = None
|
||
for node in workflow["nodes"]:
|
||
name = node["type"]
|
||
if "properties" in node:
|
||
if "Node name for S&R" in node["properties"]:
|
||
name = node["properties"]["Node name for S&R"]
|
||
if name == node_name:
|
||
node_id = node["id"]
|
||
break
|
||
if "title" in node:
|
||
name = node["title"]
|
||
if name == node_name:
|
||
node_id = node["id"]
|
||
break
|
||
if node_id is not None:
|
||
values = prompt[str(node_id)]
|
||
if "inputs" in values:
|
||
if widget_name in values["inputs"]:
|
||
value = values["inputs"][widget_name]
|
||
if isinstance(value, list):
|
||
raise ValueError(
|
||
"Converted widgets are not supported via named reference, use the inputs instead."
|
||
)
|
||
return value
|
||
raise NameError(f"Widget not found: {node_name}.{widget_name}")
|
||
raise NameError(f"Node not found: {node_name}.{widget_name}")
|
||
|
||
def get_size(self, target, property):
|
||
if isinstance(target, dict) and "samples" in target:
|
||
# Latent
|
||
if property == "width":
|
||
return target["samples"].shape[3] * 8
|
||
return target["samples"].shape[2] * 8
|
||
else:
|
||
# Image
|
||
if property == "width":
|
||
return target.shape[2]
|
||
return target.shape[1]
|
||
|
||
def evaluate(
|
||
self, expression, prompt, extra_pnginfo={}, a=None, b=None, c=None, d=0
|
||
):
|
||
expression = expression.replace("\n", " ").replace("\r", "")
|
||
node = ast.parse(expression, mode="eval").body
|
||
|
||
lookup = {"a": a, "b": b, "c": c, "d": d}
|
||
|
||
def eval_op(node, l, r):
|
||
l = eval_expr(l)
|
||
r = eval_expr(r)
|
||
l = l if isinstance(l, int) else float(l)
|
||
r = r if isinstance(r, int) else float(r)
|
||
return operators[type(node.op)](l, r)
|
||
|
||
def eval_expr(node):
|
||
if isinstance(node, ast.Constant) or isinstance(node, ast.Num):
|
||
return node.n
|
||
elif isinstance(node, ast.BinOp):
|
||
return eval_op(node, node.left, node.right)
|
||
elif isinstance(node, ast.BoolOp):
|
||
return eval_op(node, node.values[0], node.values[1])
|
||
elif isinstance(node, ast.UnaryOp):
|
||
return operators[type(node.op)](eval_expr(node.operand))
|
||
elif isinstance(node, ast.Attribute):
|
||
if node.value.id in lookup:
|
||
if node.attr == "width" or node.attr == "height":
|
||
return self.get_size(lookup[node.value.id], node.attr)
|
||
|
||
return self.get_widget_value(
|
||
extra_pnginfo, prompt, node.value.id, node.attr
|
||
)
|
||
elif isinstance(node, ast.Name):
|
||
if node.id in lookup:
|
||
val = lookup[node.id]
|
||
if isinstance(val, (int, float, complex)):
|
||
return val
|
||
else:
|
||
raise TypeError(
|
||
f"Compex types (LATENT/IMAGE) need to reference their width/height, e.g. {node.id}.width"
|
||
)
|
||
raise NameError(f"Name not found: {node.id}")
|
||
elif isinstance(node, ast.Call):
|
||
if node.func.id in functions:
|
||
fn = functions[node.func.id]
|
||
l = len(node.args)
|
||
if l < fn["args"][0] or (
|
||
fn["args"][1] is not None and l > fn["args"][1]
|
||
):
|
||
if fn["args"][1] is None:
|
||
toErr = " or more"
|
||
else:
|
||
toErr = f" to {fn['args'][1]}"
|
||
raise SyntaxError(
|
||
f"Invalid function call: {node.func.id} requires {fn['args'][0]}{toErr} arguments"
|
||
)
|
||
args = []
|
||
for arg in node.args:
|
||
args.append(eval_expr(arg))
|
||
return fn["call"](*args)
|
||
raise NameError(f"Invalid function call: {node.func.id}")
|
||
elif isinstance(node, ast.Compare):
|
||
l = eval_expr(node.left)
|
||
r = eval_expr(node.comparators[0])
|
||
if isinstance(node.ops[0], ast.Eq):
|
||
return 1 if l == r else 0
|
||
if isinstance(node.ops[0], ast.NotEq):
|
||
return 1 if l != r else 0
|
||
if isinstance(node.ops[0], ast.Gt):
|
||
return 1 if l > r else 0
|
||
if isinstance(node.ops[0], ast.GtE):
|
||
return 1 if l >= r else 0
|
||
if isinstance(node.ops[0], ast.Lt):
|
||
return 1 if l < r else 0
|
||
if isinstance(node.ops[0], ast.LtE):
|
||
return 1 if l <= r else 0
|
||
raise NotImplementedError(
|
||
"Operator " + node.ops[0].__class__.__name__ + " not supported."
|
||
)
|
||
else:
|
||
raise TypeError(node)
|
||
|
||
r = eval_expr(node)
|
||
if is_integer(r):
|
||
s = str(int(r))
|
||
else:
|
||
s = str(r)
|
||
|
||
return {
|
||
"ui": {"value": [r]},
|
||
"result": (
|
||
int(r),
|
||
float(r),
|
||
s,
|
||
),
|
||
}
|
||
|
||
class KY_AnyByIndex:
|
||
@classmethod
|
||
def INPUT_TYPES(self):
|
||
return {
|
||
"required": {
|
||
"ANY": (any_typ, {"forceInput": True}),
|
||
"index": ("INT", {"forceInput": False, "default": 0}),
|
||
}
|
||
}
|
||
|
||
TITLE = "Get item from list"
|
||
RETURN_TYPES = (any_typ, )
|
||
INPUT_IS_LIST = True
|
||
FUNCTION = "run"
|
||
CATEGORY = _CATEGORY
|
||
DESCRIPTION = """
|
||
Get item from any list or batch by index
|
||
- For single item (int/float/bool/str): return itself
|
||
- For list (str/int/float/bool): return item by index
|
||
- For image batch: return single image (keep batch dimension)
|
||
- For image list: return single image (keep batch dimension)
|
||
"""
|
||
|
||
def run(self, ANY, index):
|
||
idx = index[0]
|
||
|
||
# 处理输入列表
|
||
if isinstance(ANY, list):
|
||
# 如果是空列表，直接返回
|
||
if not ANY:
|
||
return (None,)
|
||
|
||
# 获取第一个元素（因为 INPUT_IS_LIST = True，ANY 总是列表）
|
||
first_item = ANY[0]
|
||
|
||
# 如果第一个元素是列表（说明输入是列表的列表）
|
||
if isinstance(first_item, list):
|
||
if idx >= len(first_item):
|
||
raise ValueError(f"Index {idx} out of range for list size {len(first_item)}")
|
||
return (first_item[idx],)
|
||
if isinstance(first_item, tuple):
|
||
return (first_item[idx],)
|
||
|
||
# 如果第一个元素是 tensor
|
||
if isinstance(first_item, torch.Tensor):
|
||
if len(first_item.shape) == 4: # 图像列表
|
||
# 检查是否为图像列表的列表
|
||
if isinstance(ANY, list) and len(ANY) > 1 and all(isinstance(x, torch.Tensor) and len(x.shape) == 4 for x in ANY):
|
||
if idx >= len(ANY):
|
||
raise ValueError(f"Index {idx} out of range for image list size {len(ANY)}")
|
||
return (ANY[idx],)
|
||
# 单个图像的情况
|
||
return (first_item[idx:idx+1],)
|
||
|
||
# 如果是基本类型列表
|
||
if isinstance(first_item, (int, float, bool, str)):
|
||
if idx >= len(ANY):
|
||
raise ValueError(f"Index {idx} out of range for list size {len(ANY)}")
|
||
return (ANY[idx],)
|
||
|
||
return (first_item,)
|
||
|
||
# 处理基本类型
|
||
if isinstance(ANY, (int, float, bool, str)):
|
||
return (ANY,)
|
||
|
||
# 其他情况直接返回
|
||
return (ANY,)
|
||
|
||
class KY_AnyToList:
|
||
@classmethod
|
||
def INPUT_TYPES(self):
|
||
return {
|
||
"required": {
|
||
"any1": (any_typ, {"forceInput": True}),
|
||
},
|
||
"optional": {
|
||
"any2": (any_typ,),
|
||
"any3": (any_typ,),
|
||
}
|
||
}
|
||
|
||
TITLE = "Anything to List"
|
||
RETURN_TYPES = (any_typ, "INT")
|
||
RETURN_NAMES = ("merged_list", "total_items")
|
||
OUTPUT_IS_LIST = (True, False)
|
||
INPUT_IS_LIST = True
|
||
FUNCTION = "run"
|
||
CATEGORY = _CATEGORY
|
||
DESCRIPTION = """
|
||
Merge up to 3 inputs into a single list:
|
||
- For basic types (int/float/bool/str): add as single item
|
||
- For lists: extend the result list
|
||
- For image batch: convert to list of single images
|
||
- For tensor batch: convert to list of single tensors
|
||
Returns:
|
||
- merged_list: Combined list of all inputs
|
||
- total_items: Total number of items in the merged list
|
||
"""
|
||
|
||
def process_input(self, input_value):
|
||
if input_value is None:
|
||
return []
|
||
|
||
# 因为INPUT_IS_LIST=True，input_value总是列表
|
||
result = []
|
||
for item in input_value:
|
||
result.extend(self.process_single_input(item))
|
||
return result
|
||
|
||
def process_single_input(self, value):
|
||
# 如果是列表，直接返回
|
||
if isinstance(value, list):
|
||
return value
|
||
|
||
# 如果是元组，转换为列表
|
||
if isinstance(value, tuple):
|
||
return list(value)
|
||
|
||
# 处理图像或张量批次
|
||
if isinstance(value, torch.Tensor):
|
||
if len(value.shape) == 4: # 图像批次
|
||
return [value[i:i+1] for i in range(value.shape[0])]
|
||
elif len(value.shape) > 1: # 其他张量批次
|
||
return [value[i] for i in range(value.shape[0])]
|
||
else: # 单个张量
|
||
return [value]
|
||
|
||
# 处理基本类型
|
||
if isinstance(value, (int, float, bool, str)):
|
||
return [value]
|
||
|
||
# 其他类型作为单个元素处理
|
||
return [value]
|
||
|
||
def run(self, any1, any2=None, any3=None):
|
||
result = []
|
||
|
||
# 处理所有输入
|
||
result.extend(self.process_input(any1))
|
||
result.extend(self.process_input(any2))
|
||
result.extend(self.process_input(any3))
|
||
|
||
|
||
# 因为INPUT_IS_LIST=True，我们需要返回单个元素的列表
|
||
return ([result], len(result))
|
||
|
||
class KY_isNone_blocker:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"any": (any_typ,)
|
||
},
|
||
"optional": {
|
||
"reverse": ("BOOLEAN", {"default": False}),
|
||
"block_execution": ("BOOLEAN", {"default": False}),
|
||
}
|
||
}
|
||
RETURN_TYPES = (any_typ, "BOOLEAN")
|
||
RETURN_NAMES = ("out", "is_empty",)
|
||
FUNCTION = "execute"
|
||
CATEGORY = _CATEGORY
|
||
|
||
def execute(self, any, reverse, block_execution):
|
||
is_empty = is_deep_empty(any)
|
||
if block_execution and is_empty and not reverse:
|
||
return (ExecutionBlocker(None), is_empty)
|
||
if reverse:
|
||
return (any, not is_empty,)
|
||
else:
|
||
return (any, is_empty,)
|
||
|
||
class KY_First_NOT_EMPTY:
|
||
@classmethod
|
||
def INPUT_TYPES(s):
|
||
return {
|
||
"required": {
|
||
"Fallback": (any_typ,)
|
||
},
|
||
"optional": {
|
||
"A": (any_typ,),
|
||
"B": (any_typ,),
|
||
"C": (any_typ,),
|
||
"D": (any_typ,),
|
||
"force_fallback": ("BOOLEAN", {"default": True}),
|
||
"block_if_all_empty": ("BOOLEAN", {"default": True}),
|
||
}
|
||
}
|
||
RETURN_TYPES = (any_typ, "BOOLEAN")
|
||
RETURN_NAMES = ("out", "is_blocked",)
|
||
FUNCTION = "execute"
|
||
CATEGORY = _CATEGORY
|
||
DESCRIPTION = """
|
||
Return first not NONE value/Image/Mask :
|
||
- Empty list, Dictionary, Turple
|
||
- Empty String
|
||
- 0 float int complex
|
||
- empty tensor
|
||
"""
|
||
|
||
def execute(self, A = None, B = None, C = None, D = None, Fallback = None, force_fallback = False, block_if_all_empty = False):
|
||
if force_fallback:
|
||
if is_deep_empty(Fallback) and block_if_all_empty:
|
||
return (ExecutionBlocker(None), True)
|
||
return (Fallback, False)
|
||
val = None
|
||
if not is_deep_empty(A):
|
||
val = A
|
||
elif not is_deep_empty(B):
|
||
val = B
|
||
elif not is_deep_empty(C):
|
||
val = C
|
||
elif not is_deep_empty(D):
|
||
val = D
|
||
else:
|
||
val = Fallback
|
||
bol = is_deep_empty(val)
|
||
|
||
# return (kwargs['in'] if kwargs['continue'] else ExecutionBlocker(None),)
|
||
if block_if_all_empty and bol:
|
||
return (ExecutionBlocker(None), True)
|
||
else:
|
||
return (val, False,)
|
||
|
||
class KY_MergeToJSON:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"json_input_1": ("STRING", {"multiline": True}),
|
||
"json_input_2": ("STRING", {"multiline": True}),
|
||
"merge_strategy": (["override", "preserve", "concat"], {"default": "override"}),
|
||
},
|
||
}
|
||
|
||
RETURN_TYPES = ("STRING",)
|
||
RETURN_NAMES = ("merged_json",)
|
||
FUNCTION = "merge_json"
|
||
CATEGORY = _CATEGORY
|
||
|
||
def merge_json(self, json_input_1: str, json_input_2: str, merge_strategy: str) -> tuple[str]:
|
||
try:
|
||
data1 = json.loads(json_input_1)
|
||
data2 = json.loads(json_input_2)
|
||
|
||
if isinstance(data1, list) and isinstance(data2, list):
|
||
result = data1 + data2
|
||
elif isinstance(data1, dict) and isinstance(data2, dict):
|
||
result = self._merge_dicts(data1, data2, merge_strategy)
|
||
else:
|
||
raise ValueError("Both inputs must be of the same type (either objects or arrays)")
|
||
|
||
return (json.dumps(result, indent=2),)
|
||
except json.JSONDecodeError:
|
||
raise ValueError("Invalid JSON input")
|
||
|
||
def _merge_dicts(self, dict1: dict, dict2: dict, strategy: str) -> dict:
|
||
result = dict1.copy()
|
||
|
||
for key, value in dict2.items():
|
||
if key not in result:
|
||
result[key] = value
|
||
else:
|
||
if strategy == "override":
|
||
result[key] = value
|
||
elif strategy == "preserve":
|
||
continue
|
||
elif isinstance(result[key], dict) and isinstance(value, dict):
|
||
result[key] = self._merge_dicts(result[key], value, strategy)
|
||
elif isinstance(result[key], list) and isinstance(value, list):
|
||
result[key] = result[key] + value
|
||
|
||
return result
|
||
|
||
class KY_DummyOut:
|
||
@classmethod
|
||
def INPUT_TYPES(cls):
|
||
return {
|
||
"required": {
|
||
"any1": (any_typ, ),
|
||
"any2": (any_typ, ),
|
||
"any3": (any_typ, ),
|
||
"any4": (any_typ, ),
|
||
"any5": (any_typ, ),
|
||
},
|
||
}
|
||
|
||
RETURN_TYPES = (any_typ,any_typ,any_typ,any_typ,any_typ,)
|
||
RETURN_NAMES = ("any1", "any2", "any3", "any4", "any5",)
|
||
FUNCTION = "dummy"
|
||
CATEGORY = "KYNode/misc"
|
||
OUTPUT_NODE = True
|
||
DESCRIPTION = """
|
||
Does nothing, used to trigger generic workflow output.
|
||
A way to get previews in the UI without saving anything to disk.
|
||
"""
|
||
|
||
def dummy(self, any1, any2, any3, any4, any5):
|
||
return (any1, any2, any3, any4, any5,)
|
||
NODE_CLASS_MAPPINGS = {
|
||
"KY_JoinToString": KY_JoinToString,
|
||
"KY_RegexReplace": KY_RegexReplace,
|
||
"KY_RegexExtractor": KY_RegexExtractor,
|
||
"KY_MathExpression": KY_MathExpression,
|
||
"KY_AnyByIndex": KY_AnyByIndex,
|
||
"KY_AnyToList": KY_AnyToList,
|
||
"KY_isNone": KY_isNone_blocker,
|
||
"KY_MergeToJSON": KY_MergeToJSON,
|
||
"KY_First_NOT_EMPTY": KY_First_NOT_EMPTY,
|
||
"KY_DummyOut": KY_DummyOut,
|
||
}
|
||
|
||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||
"KY_JoinToString": "Join any into string",
|
||
"KY_RegexReplace": "Replace text by regex",
|
||
"KY_RegexExtractor": "Extract text by regex",
|
||
"KY_MathExpression": "Math expression eval",
|
||
"KY_AnyByIndex": "Anything Get By Index",
|
||
"KY_AnyToList": "Anything To List",
|
||
"KY_isNone": "Block if None or empty ",
|
||
"KY_MergeToJSON": "Merge Json",
|
||
"KY_First_NOT_EMPTY": "First NOT NONE",
|
||
"KY_DummyOut": "Dummy Out to trigger workflow",
|
||
}
|