Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 copilot-swe-agent[bot]andali1234
|
3b1cc65fd5 | ||
|
copilot-swe-agent[bot]andali1234
|
c761d43b7e | ||
|
Alistair Buxton
|
1abe898567 |
+3
-12
@@ -1,16 +1,7 @@
|
||||
NODE_CLASS_MAPPINGS = {}
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {}
|
||||
from .registry import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS, set_pack_options
|
||||
|
||||
set_pack_options('jobiter', 'Job Iterator')
|
||||
|
||||
def register_node(c):
|
||||
assert not isinstance(c.RETURN_TYPES, str), "Error: string found instead of tuple."
|
||||
assert not isinstance(c.RETURN_NAMES, str), "Error: string found instead of tuple."
|
||||
NODE_CLASS_MAPPINGS[c.__name__] = c
|
||||
NODE_DISPLAY_NAME_MAPPINGS[c.__name__] = c.__name__
|
||||
return c
|
||||
|
||||
|
||||
from . import sequence, paths, job, image, debug
|
||||
|
||||
from . import sequence, job, debug
|
||||
|
||||
|
||||
|
||||
@@ -3,31 +3,12 @@ import os
|
||||
import textwrap
|
||||
import code
|
||||
|
||||
from . import register_node
|
||||
|
||||
|
||||
@register_node
|
||||
class Stringify:
|
||||
"""Convert any input to str/repr."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"x": ("*", ),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING", "STRING")
|
||||
RETURN_NAMES = ("str", "repr")
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/debug"
|
||||
|
||||
def go(self, x):
|
||||
return (str(x), repr(x))
|
||||
from .registry import register_node
|
||||
from .types import Any, Variadic
|
||||
|
||||
|
||||
class RestoreStdStreams(object):
|
||||
# ComfyUI-Manager patches sys.stdout and sys.stder
|
||||
# ComfyUI-Manager patches sys.stdout and sys.stderr
|
||||
# which breaks GNU Readline support and makes the
|
||||
# REPL annoying to use. This context manager temporarily
|
||||
# puts back the originals.
|
||||
@@ -58,39 +39,26 @@ class Quitter:
|
||||
print(MESSAGE)
|
||||
|
||||
|
||||
@register_node
|
||||
class Interact:
|
||||
@register_node(category='debug', output=True, display_name='Interact')
|
||||
def Interact(inputs: Variadic(Any)) -> ():
|
||||
"""Opens an interactive REPL whenever the node is evaluated."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {},
|
||||
"optional": {x: "*" for x in ('a', 'b', 'c', 'd')},
|
||||
}
|
||||
if sys.__stdout__.isatty():
|
||||
with RestoreStdStreams():
|
||||
code.interact(
|
||||
banner=textwrap.dedent(f"""
|
||||
Interactive debugging started.
|
||||
inputs = {inputs!r}
|
||||
{MESSAGE}
|
||||
"""),
|
||||
exitmsg="Resuming workflow...",
|
||||
local={
|
||||
'inputs': inputs,
|
||||
'quit': Quitter(),
|
||||
'exit': Quitter(),
|
||||
}
|
||||
)
|
||||
else:
|
||||
# Don't block the server if there is no tty.
|
||||
print("Skipping interactive prompt because there is no tty.")
|
||||
return ()
|
||||
|
||||
RETURN_TYPES = ()
|
||||
RETURN_NAMES = ()
|
||||
FUNCTION = "interact"
|
||||
CATEGORY = "ali1234/debug"
|
||||
OUTPUT_NODE = True
|
||||
|
||||
def interact(self, **kwargs):
|
||||
if sys.__stdout__.isatty():
|
||||
with RestoreStdStreams():
|
||||
code.interact(
|
||||
banner=textwrap.dedent(f"""
|
||||
Interactive debugging started.
|
||||
Try `print(a)`.
|
||||
{MESSAGE}
|
||||
"""),
|
||||
exitmsg="Resuming workflow...",
|
||||
local={
|
||||
**kwargs,
|
||||
'quit': Quitter(),
|
||||
'exit': Quitter()
|
||||
}
|
||||
)
|
||||
else:
|
||||
# Don't block the server if there is no tty.
|
||||
print("Skipping interactive prompt because there is no tty.")
|
||||
return ()
|
||||
|
||||
@@ -1,180 +1,83 @@
|
||||
import collections
|
||||
import itertools
|
||||
import math
|
||||
|
||||
from execution import PromptExecutor
|
||||
from .registry import register_node
|
||||
from .types import *
|
||||
|
||||
from . import register_node
|
||||
class NamedSeq:
|
||||
def nth(self, n):
|
||||
raise NotImplementedError
|
||||
|
||||
def __getitem__(self, n):
|
||||
return dict(self.nth(n))
|
||||
|
||||
class SingleSeq(NamedSeq):
|
||||
def __init__(self, name, seq):
|
||||
self._name = name
|
||||
self._seq = seq
|
||||
|
||||
def nth(self, n):
|
||||
n = n % len(self)
|
||||
yield (self._name, self._seq[n])
|
||||
|
||||
def __len__(self):
|
||||
return len(self._seq)
|
||||
|
||||
|
||||
@register_node
|
||||
class MakeJob:
|
||||
"""Turns a sequence into a job with one attribute."""
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"sequence": ("SEQUENCE", ),
|
||||
"name": ("STRING", {"default": ''}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("JOB", "INT")
|
||||
RETURN_NAMES = ("job", "count")
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/job"
|
||||
|
||||
def merge_dicts(self, *dicts):
|
||||
#return collections.ChainMap(*reversed(dicts))
|
||||
return dict(itertools.chain.from_iterable(d.items() for d in dicts))
|
||||
|
||||
def go(self, sequence, name):
|
||||
result = [{name: value} for value in sequence]
|
||||
return (result, len(result))
|
||||
class CombinedSeq(NamedSeq):
|
||||
def __init__(self, *args):
|
||||
self._seqs = args
|
||||
|
||||
|
||||
@register_node
|
||||
class CombineJobs(MakeJob):
|
||||
"""Combines multiple jobs."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"a": ("JOB", ),
|
||||
"method": (("zip", "product"), {"default": "zip"}),
|
||||
},
|
||||
"optional": {
|
||||
x: ("JOB", ) for x in ('b', 'c', 'd', 'e')
|
||||
}
|
||||
}
|
||||
class ProductSeq(CombinedSeq):
|
||||
def nth(self, n):
|
||||
n = n % len(self)
|
||||
indices = []
|
||||
for s in reversed(self._seqs):
|
||||
indices.append(n % len(s))
|
||||
n //= len(s)
|
||||
|
||||
def go(self, method, **kwargs):
|
||||
method = {'product': itertools.product, 'zip': zip}[method]
|
||||
result = [self.merge_dicts(*steps) for steps in method(*kwargs.values())]
|
||||
return (result, len(result))
|
||||
for i, s in zip(reversed(indices), self._seqs):
|
||||
yield from s.nth(i)
|
||||
|
||||
def __len__(self):
|
||||
return math.prod(len(s) for s in self._seqs)
|
||||
|
||||
|
||||
@register_node
|
||||
class EnumerateJob(MakeJob):
|
||||
"""Combines multiple jobs."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"job": ("JOB", ),
|
||||
"name": ("STRING", {"default": ''}),
|
||||
},
|
||||
}
|
||||
class ZipSeq(CombinedSeq):
|
||||
def nth(self, n):
|
||||
n = n % len(self)
|
||||
for s in self._seqs:
|
||||
yield from s.nth(n)
|
||||
|
||||
def go(self, job, name):
|
||||
result = [self.merge_dicts(step, {name: n}) for n, step in enumerate(job)]
|
||||
return (result, len(result))
|
||||
def __len__(self):
|
||||
return min(len(s) for s in self._seqs)
|
||||
|
||||
|
||||
@register_node
|
||||
class GetJobStep:
|
||||
"""Gets the job step by number."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"job": ("JOB", ),
|
||||
"step": ("INT", {"default": 0}),
|
||||
"wrap": (("repeat", "clamp"), {"default": "repeat"}),
|
||||
},
|
||||
}
|
||||
RETURN_TYPES = ("ATTRIBUTES", )
|
||||
RETURN_NAMES = ("attributes", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/job"
|
||||
@register_node(display_name="Make Job")
|
||||
def MakeJob(sequence: Sequence(), name: String() = "") -> (Job(), ):
|
||||
return SingleSeq(name, sequence),
|
||||
|
||||
def go(self, job, step, wrap):
|
||||
if wrap == 'repeat':
|
||||
while step < 0:
|
||||
step += len(job)
|
||||
step = step % len(job)
|
||||
elif wrap == 'clamp':
|
||||
step = max(min(step, len(job)), 0)
|
||||
return (job[step], )
|
||||
combine_modes = {
|
||||
'zip': ZipSeq,
|
||||
'product': ProductSeq,
|
||||
}
|
||||
|
||||
@register_node(display_name="Combine Jobs")
|
||||
def CombineJobs(a: Job(), b: Job(), method: Combo(choices=combine_modes) = 'product') -> (Job(), ):
|
||||
return method(a, b),
|
||||
|
||||
|
||||
@register_node
|
||||
class FormatAttributes:
|
||||
"""Applies attributes to a format string."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"attributes": ("ATTRIBUTES",),
|
||||
"format": ("STRING", {'default': '', 'multiline': True, "dynamicPrompts": False})
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("STRING", )
|
||||
RETURN_NAMES = ("string", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/job"
|
||||
|
||||
def go(self, attributes, format):
|
||||
return (format.format(**attributes), )
|
||||
@register_node(display_name="Enumerate Job")
|
||||
def EnumerateJob(job: Job(), name: String() = "") -> (Job(), ):
|
||||
return ZipSeq(SingleSeq(name, range(len(job))), job),
|
||||
|
||||
|
||||
class AnyType(str):
|
||||
def __ne__(self, __value: object) -> bool:
|
||||
return False
|
||||
@register_node(display_name="Get Job Step")
|
||||
def GetJobStep(job: Job(), n: Int()=0) -> (JobStep(), ):
|
||||
return (job[n], )
|
||||
|
||||
|
||||
@register_node
|
||||
class GetAttribute:
|
||||
"""Gets a named attribute from a step."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"attributes": ("ATTRIBUTES", ),
|
||||
"name": ("STRING", {"default": ''}),
|
||||
},
|
||||
}
|
||||
@register_node(display_name="Get Step Attribute")
|
||||
def GetStepAttribute(step: JobStep(), name: String() = "") -> (Any, ):
|
||||
return (step[name], )
|
||||
|
||||
RETURN_TYPES = (AnyType("*"), )
|
||||
RETURN_NAMES = ("value", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/job"
|
||||
|
||||
def go(self, attributes, name):
|
||||
return (attributes[name], )
|
||||
|
||||
|
||||
# Dynamically register typed attribute getters to avoid wildcard bug
|
||||
# https://github.com/comfyanonymous/ComfyUI/pull/770
|
||||
for t in ('INT', 'FLOAT', 'STRING'):
|
||||
register_node(type(
|
||||
'GetAttribute'+t.title(),
|
||||
(GetAttribute, ),
|
||||
{
|
||||
'RETURN_TYPES': (t, ),
|
||||
}
|
||||
))
|
||||
|
||||
|
||||
|
||||
|
||||
@register_node
|
||||
class JobToList:
|
||||
"""Converts a job into a list."""
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"job": ("JOB",),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("ATTRIBUTES",)
|
||||
RETURN_NAMES = ("attributes",)
|
||||
OUTPUT_IS_LIST = (True,)
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/job"
|
||||
|
||||
def go(self, job):
|
||||
return (job,)
|
||||
|
||||
+127
@@ -0,0 +1,127 @@
|
||||
import inspect
|
||||
from functools import wraps
|
||||
|
||||
from .types import ComfyWidgetType, Combo, Variadic
|
||||
|
||||
NODE_CLASS_MAPPINGS = {}
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {}
|
||||
|
||||
PACK_BASE_CATEGORY = None
|
||||
PACK_UID = None
|
||||
|
||||
|
||||
class NodeTemplate:
|
||||
_INPUT_TYPES = None
|
||||
FUNCTION = "exec"
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return cls._INPUT_TYPES
|
||||
|
||||
|
||||
def set_pack_options(uid: str, category: str = None):
|
||||
global PACK_BASE_CATEGORY, PACK_UID
|
||||
PACK_BASE_CATEGORY = category
|
||||
PACK_UID = uid
|
||||
|
||||
|
||||
def get_nodes():
|
||||
return {k: v.exec.__doc__ for k, v in NODE_CLASS_MAPPINGS.items()}
|
||||
|
||||
|
||||
def register_node(category=None, version=0, display_name=None, output=False):
|
||||
def decorator(f):
|
||||
node_attrs = {}
|
||||
node_attrs['OUTPUT_NODE'] = output
|
||||
node_attrs['_INPUT_TYPES'] = {'required': {}, 'optional': {}}
|
||||
|
||||
sig = inspect.signature(f)
|
||||
|
||||
node_attrs['RETURN_TYPES'] = tuple(x.type if isinstance(x, ComfyWidgetType) else x for x in sig.return_annotation)
|
||||
|
||||
for k, v in sig.parameters.items():
|
||||
t = v.annotation
|
||||
opts = {}
|
||||
req = 'required'
|
||||
|
||||
if isinstance(t, Variadic):
|
||||
# Variadic parameter: emit numbered optional input sockets.
|
||||
for i in range(t.max):
|
||||
node_attrs['_INPUT_TYPES']['optional'][f'{k}_{i}'] = (t.type, t.opts())
|
||||
continue
|
||||
|
||||
if isinstance(t, ComfyWidgetType):
|
||||
opts = t.opts()
|
||||
t = t.type
|
||||
if v.default is inspect._empty:
|
||||
opts['forceInput'] = True
|
||||
else:
|
||||
opts['default'] = v.default
|
||||
else:
|
||||
if v.default is inspect._empty:
|
||||
req = 'optional'
|
||||
else:
|
||||
opts['default'] = v.default
|
||||
|
||||
node_attrs['_INPUT_TYPES'][req][k] = (t, opts)
|
||||
|
||||
cat_list = []
|
||||
if PACK_BASE_CATEGORY is not None:
|
||||
cat_list.append(PACK_BASE_CATEGORY)
|
||||
if category is not None:
|
||||
cat_list.append(category)
|
||||
if cat_list:
|
||||
node_attrs['CATEGORY'] = '/'.join(cat_list)
|
||||
else:
|
||||
print(f"WARNING: No category specified for {f.__name__} and no base category. It won't be shown in menus.")
|
||||
|
||||
@wraps(f)
|
||||
def exec(**kwargs):
|
||||
call_kwargs = {}
|
||||
for param_name, param in sig.parameters.items():
|
||||
if isinstance(param.annotation, Variadic):
|
||||
# Collect all {param_name}_N kwargs, sorted by index.
|
||||
prefix = param_name + '_'
|
||||
numbered = {}
|
||||
for k in kwargs:
|
||||
if k.startswith(prefix) and k[len(prefix):].isdigit():
|
||||
numbered[int(k[len(prefix):])] = kwargs[k]
|
||||
values = [numbered[i] for i in sorted(numbered)]
|
||||
if isinstance(param.annotation.inner, ComfyWidgetType):
|
||||
values = [param.annotation.inner[v] for v in values]
|
||||
call_kwargs[param_name] = values
|
||||
else:
|
||||
if param_name in kwargs:
|
||||
v = kwargs[param_name]
|
||||
if isinstance(param.annotation, ComfyWidgetType):
|
||||
# Look up Combo value from mapping
|
||||
v = param.annotation[v]
|
||||
call_kwargs[param_name] = v
|
||||
return f(**call_kwargs)
|
||||
|
||||
node_attrs['exec'] = staticmethod(exec)
|
||||
|
||||
if PACK_UID is None:
|
||||
raise Exception("PACK_UID is not set. Call set_pack_options in __init__.py to set it.")
|
||||
|
||||
unique_name = f'{PACK_UID}_{version}_{f.__name__}'
|
||||
node_class = type(unique_name, (NodeTemplate,), node_attrs)
|
||||
NODE_CLASS_MAPPINGS[unique_name] = node_class
|
||||
if display_name is not None:
|
||||
NODE_DISPLAY_NAME_MAPPINGS[unique_name] = display_name
|
||||
else:
|
||||
NODE_DISPLAY_NAME_MAPPINGS[unique_name] = f.__name__
|
||||
return f
|
||||
return decorator
|
||||
|
||||
|
||||
def scrape_module(m, sig_len):
|
||||
for name in dir(m):
|
||||
v = getattr(m, name)
|
||||
if callable(v):
|
||||
try:
|
||||
sig = inspect.signature(v)
|
||||
if len(sig.parameters) == sig_len:
|
||||
yield v
|
||||
except ValueError:
|
||||
pass
|
||||
+37
-136
@@ -1,159 +1,60 @@
|
||||
import ast
|
||||
import functools
|
||||
import itertools
|
||||
|
||||
from . import register_node
|
||||
|
||||
from .registry import register_node
|
||||
from .types import *
|
||||
|
||||
# Sequence creation.
|
||||
|
||||
|
||||
@register_node
|
||||
class Range:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"start": ("INT", {"default": 0, "min": -9999999, "max": 9999999, "step": 1}),
|
||||
"stop": ("INT", {"default": 10, "min": -9999999, "max": 9999999, "step": 1}),
|
||||
"step": ("INT", {"default": 1, "min": -9999999, "max": 9999999, "step": 1}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("SEQUENCE", )
|
||||
RETURN_NAMES = ("sequence", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/sequence"
|
||||
|
||||
def go(self, start, stop, step):
|
||||
@register_node()
|
||||
def Range(start: Int() = 0, stop: Int() = 10, step: Int() = 1) -> (Sequence(), ):
|
||||
return (range(start, stop, step), )
|
||||
|
||||
|
||||
@register_node
|
||||
class Literal:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"literal": ("STRING", {'default': '', 'multiline': True})
|
||||
},
|
||||
}
|
||||
@register_node()
|
||||
def Literal(literal: String(multiline = True) = "") -> (Any, ):
|
||||
return (ast.literal_eval(literal), )
|
||||
|
||||
RETURN_TYPES = ("SEQUENCE", )
|
||||
RETURN_NAMES = ("sequence", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/sequence"
|
||||
|
||||
def go(self, literal):
|
||||
return (ast.literal_eval(literal), )
|
||||
|
||||
|
||||
# Processing of existing sequences.
|
||||
|
||||
@register_node()
|
||||
def Combinatorics(sequence: Sequence(), min: Int() = 3, max: Int() = 4, replacement: Bool() = False, permutations: Bool() = False) -> (Sequence(), ):
|
||||
"""Sequence combinatorics."""
|
||||
comb = itertools.combinations_with_replacement if replacement else itertools.combinations
|
||||
combs = itertools.chain(*(comb(sequence, n) for n in range(min, max+1)))
|
||||
if permutations:
|
||||
combs = itertools.chain(*(itertools.permutations(x) for x in combs))
|
||||
return (list(combs), )
|
||||
|
||||
@register_node
|
||||
class Reorder:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"sequence": ("SEQUENCE", ),
|
||||
"method": (("reverse", "sort", "reverse sort"), {"default": "sort"}),
|
||||
},
|
||||
}
|
||||
format_modes = {
|
||||
'mapping': lambda x, y: x.format(**y),
|
||||
'iterable': lambda x, y: x.format(*y),
|
||||
'single': lambda x, y: x.format(y)
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("SEQUENCE", )
|
||||
RETURN_NAMES = ("sequence", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/sequence"
|
||||
|
||||
def go(self, sequence, method):
|
||||
f = {"reverse": reversed, "sort": sorted, "reverse_sort": lambda x: sorted(x, reverse=True)}[method]
|
||||
return (f(sequence), )
|
||||
@register_node()
|
||||
def Format(vars: Any, string: String(multiline=True) = "", mode: Combo(choices=format_modes) = 'mapping') -> (String(), ):
|
||||
"""Performs string replacement using the standard Python format() method."""
|
||||
return (mode(string, vars), )
|
||||
|
||||
|
||||
@register_node
|
||||
class Combinations:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"sequence": ("SEQUENCE", ),
|
||||
"count": ("INT", {"default": 0, "min": 0, "max": 9999999, "step": 1}),
|
||||
"replacement": ("BOOLEAN", {"default": False, "label_on": "Yes", "label_off": "No"}),
|
||||
},
|
||||
}
|
||||
@register_node()
|
||||
def Join(sequence: Sequence(), sep: String() = ", ") -> (String(), ):
|
||||
"""Joins an interable of strings into a single string."""
|
||||
return (sep.join(str(x) for x in sequence), )
|
||||
|
||||
RETURN_TYPES = ("SEQUENCE", )
|
||||
RETURN_NAMES = ("sequence", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/sequence"
|
||||
@register_node(display_name="Mapped Join")
|
||||
def MappedJoin(sequence: Sequence(), sep: String() = ", ") -> (Sequence(), ):
|
||||
"""Joins a sequence of iterables of strings into a sequence of single strings."""
|
||||
t = (sep.join(str(x) for x in s) for s in sequence)
|
||||
return (list(t), )
|
||||
|
||||
def go(self, sequence, count, replacement):
|
||||
if replacement:
|
||||
return (itertools.combinations_with_replacement(sequence, count), )
|
||||
else:
|
||||
return (itertools.combinations(sequence, count), )
|
||||
@register_node(display_name="Make Sequence")
|
||||
def MakeSequence(inputs: Variadic(Any)) -> (Sequence(), ):
|
||||
"""Collects variadic inputs into a single Sequence."""
|
||||
return (inputs, )
|
||||
|
||||
|
||||
@register_node
|
||||
class Permutations:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"sequence": ("SEQUENCE", ),
|
||||
"count": ("INT", {"default": 0, "min": 0, "max": 9999999, "step": 1}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("SEQUENCE", )
|
||||
RETURN_NAMES = ("sequence", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/sequence"
|
||||
|
||||
def go(self, sequence, count):
|
||||
return ([x for x in itertools.permutations(sequence, count)], )
|
||||
|
||||
|
||||
@register_node
|
||||
class Slice:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"sequence": ("SEQUENCE",),
|
||||
"start": ("INT", {"default": 0, "min": -9999999, "max": 9999999, "step": 1}),
|
||||
"stop": ("INT", {"default": 9999999, "min": -9999999, "max": 9999999, "step": 1}),
|
||||
"step": ("INT", {"default": 1, "min": -9999999, "max": 9999999, "step": 1}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("SEQUENCE", )
|
||||
RETURN_NAMES = ("sequence", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/sequence"
|
||||
|
||||
def go(self, sequence, start, stop, step):
|
||||
return (sequence[start:stop:step], )
|
||||
|
||||
|
||||
|
||||
@register_node
|
||||
class Join:
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"sequence": ("SEQUENCE", ),
|
||||
"join_str": ("STRING", {'default': ', '}),
|
||||
},
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("SEQUENCE", )
|
||||
RETURN_NAMES = ("sequence", )
|
||||
FUNCTION = "go"
|
||||
CATEGORY = "ali1234/sequence"
|
||||
|
||||
def go(self, sequence, join_str):
|
||||
return ([join_str.join(x) for x in sequence], )
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
import typing
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class ComfyWidgetType(BaseModel):
|
||||
"""Base type for ComfyUI types that have options controlling how they are displayed."""
|
||||
TYPE: typing.ClassVar
|
||||
|
||||
def opts(self):
|
||||
return self.model_dump(mode='python', exclude_none=True)
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
return self.TYPE
|
||||
|
||||
def __getitem__(self, item):
|
||||
return item
|
||||
|
||||
|
||||
class Int(ComfyWidgetType):
|
||||
TYPE = 'INT'
|
||||
min: int = None
|
||||
max: int = None
|
||||
step: int = None
|
||||
display: typing.Literal["number", "slider"] = None
|
||||
|
||||
|
||||
class Float(ComfyWidgetType):
|
||||
TYPE = 'FLOAT'
|
||||
min: float = None
|
||||
max: float = None
|
||||
step: float = None
|
||||
display: typing.Literal["number", "slider"] = None
|
||||
|
||||
|
||||
class String(ComfyWidgetType):
|
||||
TYPE = 'STRING'
|
||||
multiline: bool = None
|
||||
|
||||
|
||||
class Bool(ComfyWidgetType):
|
||||
TYPE = 'BOOLEAN'
|
||||
label_on: str = None
|
||||
label_off: str = None
|
||||
|
||||
|
||||
class Color(ComfyWidgetType):
|
||||
"""Widget only available if you have MTB node pack"""
|
||||
TYPE = 'COLOR'
|
||||
|
||||
|
||||
class Combo(ComfyWidgetType):
|
||||
TYPE = 'COMBO'
|
||||
choices: typing.Mapping[str, typing.Any]
|
||||
|
||||
def opts(self):
|
||||
return self.model_dump(mode='python', exclude_none=True, exclude={'choices'})
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
return list(self.choices.keys())
|
||||
|
||||
def __getitem__(self, item):
|
||||
return self.choices[item]
|
||||
|
||||
# custom ones
|
||||
|
||||
class Sequence(ComfyWidgetType):
|
||||
TYPE = 'SEQUENCE'
|
||||
|
||||
class Job(ComfyWidgetType):
|
||||
TYPE = 'JOB'
|
||||
|
||||
class JobStep(ComfyWidgetType):
|
||||
TYPE = 'JOBSTEP'
|
||||
|
||||
# Workaround ComfyUI #257
|
||||
Any = type('AnyType', (str, ), {'__ne__': lambda self, value: False})("*")
|
||||
|
||||
|
||||
class Variadic:
|
||||
"""Annotation wrapper for variadic (multi-socket) inputs in the node decorator.
|
||||
|
||||
Use as a parameter annotation to declare that a node accepts up to *max*
|
||||
numbered optional input sockets all sharing the same type::
|
||||
|
||||
@register_node()
|
||||
def MyNode(items: Variadic(Int())) -> (Sequence(),):
|
||||
return (list(items),)
|
||||
|
||||
The decorated function receives *items* as a plain Python list containing
|
||||
the values of whichever numbered sockets are connected.
|
||||
"""
|
||||
|
||||
def __init__(self, inner, max=8):
|
||||
self.inner = inner
|
||||
self.max = max
|
||||
|
||||
@property
|
||||
def type(self):
|
||||
if isinstance(self.inner, ComfyWidgetType):
|
||||
return self.inner.type
|
||||
return self.inner
|
||||
|
||||
def opts(self):
|
||||
if isinstance(self.inner, ComfyWidgetType):
|
||||
return self.inner.opts()
|
||||
return {}
|
||||
|
||||
|
||||
__all__ = ['Int', 'Float', 'String', 'Bool', 'Color', 'Combo', 'Any', 'Sequence', 'Job', 'JobStep', 'Variadic']
|
||||
Reference in New Issue
Block a user