Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dec733bd0a | ||
|
|
8fb304f86e | ||
|
|
63f38879a6 | ||
|
|
513267c7b7 | ||
|
|
cc13def5ee | ||
|
|
2d625b4ea1 | ||
|
|
d0029bd3e0 | ||
|
|
7b81f8f63b |
+20
-4
@@ -5,14 +5,30 @@
|
||||
from comfy_api.latest import ComfyExtension as _ComfyExtension, io as _io
|
||||
|
||||
from .node_dict_from_text import *
|
||||
from .nodes_dict import *
|
||||
from .nodes_add import *
|
||||
from .nodes_extract import *
|
||||
|
||||
class DictToolsExtension(_ComfyExtension):
|
||||
async def get_node_list(self) -> list[type[_io.ComfyNode]]:
|
||||
return [
|
||||
DictFromText, DictFromTextOld1,
|
||||
DictAddAny, DictAddString, DictExtractString,
|
||||
DictAddAnyOld1, DictAddStringOld1, DictExtractStringOld1
|
||||
# Add:
|
||||
DictAddAny, DictAddAnyKey,
|
||||
DictAddBool, DictAddFloat, DictAddInt, DictAddString,
|
||||
DictAddCond, DictAddImage, DictAddLatent, DictAddMask,
|
||||
DictAddGuider, DictAddNoise, DictAddSampler, DictAddSigmas,
|
||||
DictAddClip, DictAddClipVision, DictAddControlNet, DictAddGligen, DictAddLora, DictAddModel, DictAddUpscale, DictAddVae,
|
||||
DictAddAnyOld1, DictAddStringOld1,
|
||||
|
||||
# Extract:
|
||||
DictExtractAny, DictExtractAnyKey,
|
||||
DictExtractBool, DictExtractFloat, DictExtractInt, DictExtractString,
|
||||
DictExtractCond, DictExtractImage, DictExtractLatent, DictExtractMask,
|
||||
DictExtractGuider, DictExtractNoise, DictExtractSampler, DictExtractSigmas,
|
||||
DictExtractClip, DictExtractClipVision, DictExtractControlNet, DictExtractGligen, DictExtractLora, DictExtractModel, DictExtractUpscale, DictExtractVae,
|
||||
DictExtractStringOld1,
|
||||
|
||||
# Root category:
|
||||
TextToDict, TextToDictOld1, TextToDictOld2,
|
||||
]
|
||||
|
||||
async def comfy_entrypoint() -> _ComfyExtension:
|
||||
|
||||
@@ -4,5 +4,14 @@ Metadata-related module.
|
||||
"""
|
||||
|
||||
category: str = "🗂️ Dict Tools"
|
||||
category_add: str = f"{category}/Add"
|
||||
category_add_adv: str = f"{category_add}/Advanced"
|
||||
category_add_mdl: str = f"{category_add}/Models"
|
||||
|
||||
category_extract: str = f"{category}/Extract"
|
||||
category_extract_adv: str = f"{category_extract}/Advanced"
|
||||
category_extract_mdl: str = f"{category_extract}/Models"
|
||||
|
||||
category_old: str = f"{category}/_old"
|
||||
|
||||
pack_id_suffix: str = " [Dict Tools]"
|
||||
|
||||
+90
-1
@@ -3,8 +3,97 @@
|
||||
"""
|
||||
|
||||
import typing as _t
|
||||
from typing import Any as _A, Optional as _O, Union as _U, TypeVar as _TypeVar
|
||||
from typing import (
|
||||
Any as _A, Callable as _C, Optional as _O, Union as _U,
|
||||
TypeVar as _TypeVar
|
||||
)
|
||||
|
||||
from sys import float_info as _float_info
|
||||
|
||||
from frozendict import frozendict as _frozendict
|
||||
|
||||
from comfy_api.latest import io as _io
|
||||
# Not used in this module per se, but used for input-type checking:
|
||||
from comfy.clip_vision import ClipVisionModel as _ClipVisionModel
|
||||
from comfy.controlnet import ControlNet as _ControlNet
|
||||
from comfy.model_patcher import ModelPatcher as _ModelPatcher
|
||||
from comfy.samplers import CFGGuider as _CFGGuider, Sampler as _Sampler
|
||||
from comfy.sd import CLIP as _CLIP, VAE as _VAE
|
||||
from spandrel import ImageModelDescriptor as _ImageModelDescriptor
|
||||
|
||||
T = _TypeVar('T')
|
||||
T2 = _TypeVar('T2')
|
||||
DictMap = _TypeVar('DictMap', bound=_U[_t.Dict, _t.Mapping])
|
||||
T_ComfyNode = _TypeVar('T_ComfyNode', bound=_io.ComfyNode)
|
||||
|
||||
INT_MAX: int = 9223372036854775807
|
||||
FLOAT_MAX: float = _float_info.max
|
||||
|
||||
|
||||
@_t.runtime_checkable
|
||||
class NoiseProtocol(_t.Protocol):
|
||||
"""The expected type of ``Noise`` input."""
|
||||
# Seed should be only an int - but just to be safe:
|
||||
seed: _U[int, float, str]
|
||||
|
||||
def generate_noise(self, *args, **kwargs) -> _A: pass
|
||||
|
||||
|
||||
# noinspection PyShadowingBuiltins
|
||||
def _validate_type_factory(
|
||||
type: _t.Type[T], type_name: str = None, what: str = 'Value', a='a'
|
||||
) -> _C[[_A], T]:
|
||||
"""Build a function to type-check an input against a specific type."""
|
||||
if not what:
|
||||
what = 'Value'
|
||||
what = str(what)
|
||||
what = what[0].upper() + what[1:] # Make the first letter uppercase
|
||||
|
||||
if not a:
|
||||
a = 'a'
|
||||
a = str(a)
|
||||
|
||||
if not type_name:
|
||||
type_name = type.__name__
|
||||
type_name = str(type_name)
|
||||
|
||||
error_template = f"{what} isn't {a} <{type_name}> instance: {{}}"
|
||||
|
||||
def validate_type(value: _A) -> T:
|
||||
"""Type-check an input."""
|
||||
if not isinstance(value, type):
|
||||
raise TypeError(error_template.format(repr(value)))
|
||||
return value
|
||||
|
||||
return validate_type
|
||||
|
||||
|
||||
def _validate_required_dict(dict: _A) -> DictMap:
|
||||
if dict is None:
|
||||
raise ValueError("No Dict provided.")
|
||||
if not isinstance(dict, (_t.Dict, _frozendict, _t.Mapping)):
|
||||
raise TypeError(f"Not a Dict: {dict!r}")
|
||||
if not dict:
|
||||
raise ValueError("Dict is empty.")
|
||||
return dict
|
||||
|
||||
|
||||
def _validate_str_key(key: _A, strip: bool = False) -> str:
|
||||
if key is None:
|
||||
raise ValueError("No key provided.")
|
||||
key = str(key)
|
||||
|
||||
if not strip:
|
||||
return key
|
||||
|
||||
key_raw = key
|
||||
key = ''
|
||||
for line in key_raw.split():
|
||||
line = line.strip() # just in case \r is left
|
||||
if line:
|
||||
key = line
|
||||
break
|
||||
return key
|
||||
|
||||
|
||||
_validate_str_value = _validate_type_factory(str, type_name='string')
|
||||
|
||||
+119
-3
@@ -2,12 +2,15 @@
|
||||
"""
|
||||
"""
|
||||
|
||||
from dataclasses import fields as __dataclass_fields
|
||||
from dataclasses import dataclass as _dataclass, fields as __dataclass_fields
|
||||
|
||||
from comfy_api.latest import io as _io
|
||||
|
||||
from .__meta import category_old as _category_old
|
||||
from .__typing import _t, T as _T
|
||||
from .__typing import (
|
||||
_t, _A, _C, _O, _U, _TypeVar,
|
||||
T as _T, T_ComfyNode as _T_ComfyNode
|
||||
)
|
||||
|
||||
|
||||
_DICT = _io.Custom("DICT")
|
||||
@@ -16,14 +19,78 @@ _DICT_INPUT_OPTIONAL = _DICT.Input(
|
||||
optional=True,
|
||||
tooltip="An (optional) Dictionary to work with."
|
||||
)
|
||||
_DICT_INPUT_REQUIRED = _DICT.Input(
|
||||
'dict',
|
||||
tooltip="A Dictionary to work with."
|
||||
)
|
||||
_DICT_OUTPUT = _DICT.Output(
|
||||
'DICT',
|
||||
display_name='dict'
|
||||
)
|
||||
_KEY_CLEANUP_INPUT = _io.Boolean.Input(
|
||||
'cleanup_key', display_name='clean key',
|
||||
tooltip=(
|
||||
"Automatically remove leading/trailing spaces and extra newlines from the key."
|
||||
),
|
||||
default=True,
|
||||
label_on='strip spaces',
|
||||
label_off='no',
|
||||
)
|
||||
_KEY_INPUT_ADD = _io.String.Input(
|
||||
'key',
|
||||
tooltip="Key (name) of the item inserted into the dict.",
|
||||
)
|
||||
_KEY_INPUT_EXTRACT = _io.String.Input(
|
||||
'key',
|
||||
tooltip="Key (name) of the item extracted from the dict.",
|
||||
)
|
||||
|
||||
_T_Input = _t.TypeVar('T_Input', bound=_io.Input)
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
|
||||
@_dataclass
|
||||
class _InputsConverter:
|
||||
"""
|
||||
Config-like class used to rebuild schema's inputs for old version of a node.
|
||||
|
||||
- Only the inputs with IDs in ``preserved`` set are kept (or all, if empty).
|
||||
- Then, they're optionally renamed according to ``renames`` dict.
|
||||
"""
|
||||
preserved: _O[_t.Sequence[str]] = None
|
||||
renames: _O[_t.Dict[str, str]] = None
|
||||
|
||||
def convert(self, inputs: _t.Sequence[_T_Input]) -> _t.List[_T_Input]:
|
||||
preserved = self.preserved
|
||||
renames = self.renames
|
||||
if not(preserved or renames):
|
||||
return list(inputs)
|
||||
|
||||
def _out_item_with_renames(inp: _T_Input) -> _T_Input:
|
||||
_id = inp.id
|
||||
if _id in renames:
|
||||
return _input_override(inp, id=renames[_id])
|
||||
return inp
|
||||
|
||||
def _out_item_as_is(inp: _T_Input) -> _T_Input:
|
||||
return inp
|
||||
|
||||
out_item = _out_item_with_renames if renames else _out_item_as_is
|
||||
|
||||
if not preserved:
|
||||
# We keep all the inputs in the original order:
|
||||
return [out_item(x) for x in inputs]
|
||||
|
||||
# Instead, we filter and possibly reorder them:
|
||||
orig_inputs_dict = {x.id: x for x in inputs}
|
||||
return [out_item(orig_inputs_dict[_id]) for _id in preserved]
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
|
||||
def __dataclass_fields_dict(obj) -> _t.Dict[str, _t.Any]:
|
||||
"""Return all the field values from a dataclass - as dict. No deep-copy."""
|
||||
field_names = (field.name for field in __dataclass_fields(obj))
|
||||
@@ -56,8 +123,17 @@ def _input_override(input_obj: _T_Input, **overrides) -> _T_Input:
|
||||
return input_type(new_id, **input_args)
|
||||
|
||||
|
||||
def _schema_old_name(schema: _io.Schema, node_id: str, **overrides):
|
||||
def _schema_old_node(
|
||||
schema: _io.Schema, node_id: str,
|
||||
inputs_converter: _InputsConverter = None,
|
||||
**overrides
|
||||
):
|
||||
"""Build a deprecated schema for old node-ID."""
|
||||
if inputs_converter is not None:
|
||||
inputs: _t.List[_T_Input] = overrides.get('inputs', schema.inputs)
|
||||
inputs = inputs_converter.convert(inputs)
|
||||
overrides['inputs'] = inputs
|
||||
|
||||
schema_dict = __dataclass_fields_dict(schema)
|
||||
if 'node_id' in schema_dict:
|
||||
del schema_dict['node_id']
|
||||
@@ -83,3 +159,43 @@ class _BaseNode(_io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls) -> _io.Schema:
|
||||
return cls._schema
|
||||
|
||||
|
||||
_T_BaseNode = _TypeVar('T_BaseNode', bound=_BaseNode)
|
||||
|
||||
|
||||
def _attach_execute_method(cls: _t.Type[_T_ComfyNode], execute: _C, docstring: str = None) -> _t.Type[_T_ComfyNode]:
|
||||
"""Attach a function as ``execute`` method override."""
|
||||
if not docstring:
|
||||
try:
|
||||
docstring = execute.__doc__
|
||||
except AttributeError:
|
||||
docstring = None
|
||||
if not docstring:
|
||||
try:
|
||||
docstring = cls.execute.__doc__
|
||||
except AttributeError:
|
||||
docstring = None
|
||||
if docstring:
|
||||
execute.__doc__ = docstring
|
||||
|
||||
# Mimic the func's metadata to make it look as if it was
|
||||
# actually defined as an in-class method:
|
||||
execute.__module__ = cls.__module__
|
||||
execute.__name__ = 'execute'
|
||||
execute.__qualname__ = f"{cls.__qualname__}.execute"
|
||||
|
||||
# Make it a class method (has to be done after this ^)
|
||||
# and attach to the class:
|
||||
cls.execute = classmethod(execute)
|
||||
|
||||
# Since any decorator works after ABCMeta has done its job,
|
||||
# and `execute` might be previously missing (kept as abstract method),
|
||||
# we need to manually exclude it:
|
||||
if getattr(cls, "__abstractmethods__", None):
|
||||
cls.__abstractmethods__ = frozenset(
|
||||
name for name in cls.__abstractmethods__
|
||||
if name != 'execute'
|
||||
)
|
||||
|
||||
return cls
|
||||
|
||||
+63
-32
@@ -16,21 +16,23 @@ from ._dict_funcs import _new_updated_dict
|
||||
from ._io_custom import (
|
||||
_BaseNode,
|
||||
_DICT_INPUT_OPTIONAL, _DICT_OUTPUT,
|
||||
_input_override, _schema_old_name
|
||||
_InputsConverter, _schema_old_node
|
||||
)
|
||||
from .docstring_formatter import format_docstring as _format_docstring
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
def _return_line_raw(line_raw: str, line_stripped:str) -> str:
|
||||
def __return_line_raw(line_raw: str, line_stripped:str) -> str:
|
||||
return line_raw
|
||||
|
||||
|
||||
def _return_line_stripped(line_raw: str, line_stripped:str) -> str:
|
||||
def __return_line_stripped(line_raw: str, line_stripped:str) -> str:
|
||||
return line_stripped
|
||||
|
||||
|
||||
def _parsed_kv_pairs_gen(multiline_str: str, strip_lines=True):
|
||||
def _parsed_kv_pairs_gen(
|
||||
multiline_str: _O[str], pre_strip_lines=True
|
||||
) -> _t.Generator[_t.Tuple[str, str], _A, None]:
|
||||
"""Parse the "dict text" into sequence of key-value pairs:
|
||||
Given a multiline string, extract keywords (first non-empty line in each chunk)
|
||||
and their substrings (all the following non-empty lines).
|
||||
@@ -38,23 +40,34 @@ def _parsed_kv_pairs_gen(multiline_str: str, strip_lines=True):
|
||||
if not multiline_str:
|
||||
return
|
||||
multiline_str = str(multiline_str)
|
||||
if not multiline_str.strip():
|
||||
|
||||
choose_appended_line = __return_line_raw
|
||||
if pre_strip_lines:
|
||||
multiline_str = multiline_str.strip()
|
||||
choose_appended_line = __return_line_stripped
|
||||
|
||||
if not multiline_str:
|
||||
return
|
||||
|
||||
appended_line_f = _return_line_stripped if strip_lines else _return_line_raw
|
||||
cur_chunk: _t.List[str] = list()
|
||||
|
||||
def dump_chunk():
|
||||
cur_chunk_iter = iter(cur_chunk) # It's more backwards-compatible than `a, *b = x`
|
||||
chunk_key = next(cur_chunk_iter)
|
||||
chunk_key = next(cur_chunk_iter) # Might have just whitespaces when no strip
|
||||
cur_chunk_lines = list(cur_chunk_iter)
|
||||
# If there are no actual lines, join() would return an empty string:
|
||||
return chunk_key.strip(), '\n'.join(cur_chunk_lines)
|
||||
return chunk_key, '\n'.join(cur_chunk_lines)
|
||||
|
||||
for line in multiline_str.splitlines():
|
||||
line_stripped: str = line.strip()
|
||||
if line_stripped:
|
||||
cur_chunk.append(appended_line_f(line, line_stripped))
|
||||
line = choose_appended_line(line, line_stripped)
|
||||
if (
|
||||
line_stripped
|
||||
# Even if we don't strip,
|
||||
# whitespace-only lines are included only for keys:
|
||||
or (line and not cur_chunk)
|
||||
):
|
||||
cur_chunk.append(line)
|
||||
continue
|
||||
|
||||
assert not line_stripped
|
||||
@@ -62,34 +75,41 @@ def _parsed_kv_pairs_gen(multiline_str: str, strip_lines=True):
|
||||
yield dump_chunk()
|
||||
cur_chunk = list()
|
||||
|
||||
# We simply skip all the empty lines entirely
|
||||
# We simply skip all the empty lines between blocks
|
||||
|
||||
if cur_chunk:
|
||||
yield dump_chunk()
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
class DictFromText(_BaseNode):
|
||||
"""
|
||||
Build a dict of named sub-strings to be used later in string formatting (text construction).
|
||||
"""
|
||||
class TextToDict(_BaseNode):
|
||||
"""Parse raw text into key-value pairs."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictFromText{_pack_id}',
|
||||
display_name='Dict from Text',
|
||||
node_id=f'TextToDict{_pack_id}',
|
||||
display_name='📃Text → Dict🗂️',
|
||||
category=_category,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_io.Boolean.Input(
|
||||
'cleanup',
|
||||
tooltip="When enabled, each line in each sub-string is stripped from any spaces at its start and end.",
|
||||
'pre_cleanup', display_name='cleanup',
|
||||
tooltip=(
|
||||
"When enabled, each line is individually pre-stripped "
|
||||
"of any leading/trailing spaces before parsing the text."
|
||||
),
|
||||
default=True,
|
||||
label_on='leading/trailing spaces',
|
||||
label_on='strip spaces',
|
||||
label_off='no',
|
||||
),
|
||||
_io.String.Input(
|
||||
'dict_text', display_name='dict-items text',
|
||||
tooltip=(
|
||||
"Sub-string names followed by their text. Different sub-string chunks are separated by empty lines. Example:\n\n"
|
||||
"A wall of key-value pairs:\n"
|
||||
"• keys (item names) on their own line,\n"
|
||||
"• followed by their text.\n\n"
|
||||
"Different items are separated by empty lines."
|
||||
),
|
||||
placeholder=(
|
||||
"Example:\n\n"
|
||||
"char1_short\n1boy, blond, short hair\n\n"
|
||||
"char1_long\n1boy, smiling, blue eyes, blond, short hair,\nwearing a leather jacket, sitting on a bike"
|
||||
),
|
||||
@@ -112,10 +132,12 @@ class DictFromText(_BaseNode):
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
cleanup: bool, dict_text: str, show_status: bool = False,
|
||||
pre_cleanup: bool, dict_text: _O[str],
|
||||
show_status: bool = False,
|
||||
dict: _O[_DictMap] = None
|
||||
) -> _io.NodeOutput:
|
||||
parsed_items = _parsed_kv_pairs_gen(dict_text, strip_lines=cleanup)
|
||||
"""Parse raw text into key-value pairs."""
|
||||
parsed_items = _parsed_kv_pairs_gen(dict_text, pre_strip_lines=pre_cleanup)
|
||||
new_dict = {k: v for k, v in parsed_items}
|
||||
|
||||
if not new_dict:
|
||||
@@ -137,15 +159,18 @@ class DictFromText(_BaseNode):
|
||||
# ==========================================================
|
||||
# Deprecated node with old ID (for backwards compatibility)
|
||||
|
||||
class DictFromTextOld1(_BaseNode):
|
||||
class TextToDictOld1(_BaseNode):
|
||||
# noinspection PyProtectedMember
|
||||
_schema = _schema_old_name(
|
||||
DictFromText._schema,
|
||||
_schema = _schema_old_node(
|
||||
TextToDict._schema,
|
||||
'StringConstructorDictFromText',
|
||||
inputs=[
|
||||
_input_override(x, id='strings') if x.id == 'dict_text' else x
|
||||
for x in DictFromText._schema.inputs
|
||||
],
|
||||
inputs_converter=_InputsConverter(
|
||||
preserved=('pre_cleanup', 'dict_text', 'show_status', 'dict'),
|
||||
renames={
|
||||
'pre_cleanup': 'cleanup',
|
||||
'dict_text': 'strings',
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
@@ -153,7 +178,13 @@ class DictFromTextOld1(_BaseNode):
|
||||
cleanup: bool, strings: str, show_status: bool = False,
|
||||
dict: _O[_DictMap] = None
|
||||
) -> _io.NodeOutput:
|
||||
return DictFromText.execute(
|
||||
cleanup=cleanup, dict_text=strings, show_status=show_status,
|
||||
return TextToDict.execute(
|
||||
pre_cleanup=cleanup, dict_text=strings, show_status=show_status,
|
||||
dict=dict
|
||||
)
|
||||
|
||||
class TextToDictOld2(TextToDict):
|
||||
_schema = _schema_old_node(
|
||||
TextToDict._schema,
|
||||
f'DictFromText{_pack_id}',
|
||||
)
|
||||
|
||||
+446
@@ -0,0 +1,446 @@
|
||||
# encoding: utf-8
|
||||
"""Nodes to add/update individual items to a dict (``Add`` category)."""
|
||||
|
||||
from inspect import cleandoc as _cleandoc
|
||||
|
||||
from frozendict import frozendict as _frozendict
|
||||
|
||||
from comfy_api.latest import io as _io, ui as _ui
|
||||
|
||||
from .__meta import (
|
||||
category_add as _category_add,
|
||||
category_add_adv as _category_add_adv,
|
||||
category_add_mdl as _category_add_mdl,
|
||||
pack_id_suffix as _pack_id
|
||||
)
|
||||
from .__typing import (
|
||||
_t, _A, _O, _U, _validate_type_factory, _validate_str_key,
|
||||
T as _T, INT_MAX as _INT_MAX, FLOAT_MAX as _FLOAT_MAX,
|
||||
DictMap as _DictMap, NoiseProtocol as _NoiseProtocol,
|
||||
_ClipVisionModel, _ControlNet, _ModelPatcher, _CFGGuider, _Sampler,
|
||||
_CLIP, _VAE, _ImageModelDescriptor
|
||||
)
|
||||
from ._dict_funcs import _new_updated_dict
|
||||
from ._io_custom import (
|
||||
_BaseNode, _T_BaseNode,
|
||||
_DICT_INPUT_OPTIONAL, _DICT_OUTPUT, _KEY_CLEANUP_INPUT, _KEY_INPUT_ADD,
|
||||
_InputsConverter, _schema_old_node, _attach_execute_method
|
||||
)
|
||||
from .docstring_formatter import format_docstring as _format_docstring
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
def _adding_node(
|
||||
type_name: str,
|
||||
value_type: _t.Type[_T],
|
||||
value_input_cls: _t.Type[_io.Input],
|
||||
prefix: str = '',
|
||||
a='a', # a/an <type> for docstring
|
||||
convert_type: bool = False,
|
||||
category=_category_add,
|
||||
**input_kwargs
|
||||
):
|
||||
"""Get class decorator building a ``DictAdd*`` node for a specific type."""
|
||||
type_name = str(type_name)
|
||||
assert issubclass(value_input_cls, _io.Input)
|
||||
if not prefix:
|
||||
prefix = ''
|
||||
if not a:
|
||||
a = 'a'
|
||||
|
||||
def class_decorator(cls: _t.Type[_T_BaseNode]):
|
||||
docstring = f"Add/update {a} {type_name} item to a Dict."
|
||||
if not cls.__doc__:
|
||||
cls.__doc__ = docstring
|
||||
|
||||
class_name = str(cls.__name__).split('.')[-1]
|
||||
tooltip = f"The actual {type_name}-type item to add into the Dict."
|
||||
cls._schema = _io.Schema(
|
||||
node_id=f'{class_name}{_pack_id}',
|
||||
display_name=f'{prefix}{type_name} → Dict🗂️',
|
||||
category=category,
|
||||
description=_format_docstring(_cleandoc(cls.__doc__)),
|
||||
inputs=[
|
||||
_KEY_INPUT_ADD,
|
||||
_KEY_CLEANUP_INPUT,
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
value_input_cls('value', tooltip=tooltip, **input_kwargs),
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
_type_convert = value_type if convert_type else _validate_type_factory(type=value_type, type_name=type_name, a=a)
|
||||
|
||||
def execute(
|
||||
cls: _t.Type[_T_BaseNode],
|
||||
key: str, value: _O[_T] = None, cleanup_key: bool = True,
|
||||
dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
if value is None and not convert_type:
|
||||
return _io.NodeOutput(
|
||||
_frozendict() if dict is None else dict
|
||||
)
|
||||
value = _type_convert(value)
|
||||
key = _validate_str_key(key, strip=cleanup_key)
|
||||
result = _new_updated_dict(dict, {key: value})
|
||||
return _io.NodeOutput(result)
|
||||
|
||||
cls = _attach_execute_method(cls, execute, docstring)
|
||||
|
||||
return cls
|
||||
|
||||
return class_decorator
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
# Basic types
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'bool', _io.Boolean.Type, _io.Boolean.Input, convert_type=True,
|
||||
default=False
|
||||
)
|
||||
class DictAddBool(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'float', _io.Float.Type, _io.Float.Input, convert_type=True,
|
||||
default=0.0, min=-_FLOAT_MAX, max=_FLOAT_MAX, step=0.01,
|
||||
)
|
||||
class DictAddFloat(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'int', _io.Int.Type, _io.Int.Input, convert_type=True, a='an',
|
||||
default=0, min=-_INT_MAX, max=_INT_MAX, step=1,
|
||||
control_after_generate=False,
|
||||
)
|
||||
class DictAddInt(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Cond', list, _io.Conditioning.Input, prefix='🟠',
|
||||
optional=True,
|
||||
)
|
||||
class DictAddCond(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Image', _io.Image.Type, _io.Image.Input, a='an', prefix='🔵',
|
||||
optional=True,
|
||||
)
|
||||
class DictAddImage(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Latent', _U[_t.Dict, _t.Mapping], _io.Latent.Input, prefix='🟣',
|
||||
optional=True,
|
||||
)
|
||||
class DictAddLatent(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Mask', _io.Mask.Type, _io.Mask.Input, prefix='🟢',
|
||||
optional=True,
|
||||
)
|
||||
class DictAddMask(_BaseNode): pass
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
# Advanced types
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Guider', _CFGGuider, _io.Guider.Input,
|
||||
category=_category_add_adv,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddGuider(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Noise', _NoiseProtocol, _io.Noise.Input,
|
||||
category=_category_add_adv,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddNoise(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Sampler', _Sampler, _io.Sampler.Input,
|
||||
category=_category_add_adv,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddSampler(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Sigmas', _io.Sigmas.Type, _io.Sigmas.Input,
|
||||
category=_category_add_adv,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddSigmas(_BaseNode): pass
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
# Model types
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'CLIP', _CLIP, _io.Clip.Input, prefix='🟡',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddClip(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'CLIP-vision', _ClipVisionModel, _io.ClipVision.Input, prefix='🔵',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddClipVision(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'ControlNet', _ControlNet, _io.ControlNet.Input, prefix='🟢',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddControlNet(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'GLIGEN', _ModelPatcher, _io.Gligen.Input, prefix='🟤',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddGligen(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'LoRA', _io.LoraModel.Type, _io.LoraModel.Input, prefix='⚪',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddLora(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Model', _ModelPatcher, _io.Model.Input, prefix='🟣',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddModel(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'Upscale', _ImageModelDescriptor, _io.UpscaleModel.Input, a='an', prefix='🟢',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddUpscale(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_adding_node(
|
||||
'VAE', _VAE, _io.Vae.Input, prefix='🔴',
|
||||
category=_category_add_mdl,
|
||||
optional=True,
|
||||
)
|
||||
class DictAddVae(_BaseNode): pass
|
||||
|
||||
|
||||
# ==========================================================
|
||||
# Nodes with custom implementation
|
||||
|
||||
|
||||
class DictAddAny(_BaseNode):
|
||||
"""Add/update any-type item to a Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictAddAny{_pack_id}',
|
||||
display_name='◯ANY → Dict🗂️',
|
||||
category=_category_add,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_KEY_INPUT_ADD,
|
||||
_KEY_CLEANUP_INPUT,
|
||||
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
_io.AnyType.Input(
|
||||
'value',
|
||||
optional=True,
|
||||
tooltip="The actual any-type item to add into the Dict."
|
||||
),
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
# noinspection PyShadowingBuiltins
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
key: str, value: _A = None, cleanup_key: bool = True,
|
||||
dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
"""Add/update any-type item to a Dict."""
|
||||
key = _validate_str_key(key, strip=cleanup_key)
|
||||
# No value-type check
|
||||
result = _new_updated_dict(dict, {key: value})
|
||||
return _io.NodeOutput(result)
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
|
||||
class DictAddAnyKey(_BaseNode):
|
||||
"""Add/update any-type item with any-type key to a Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictAddAnyKey{_pack_id}',
|
||||
display_name='◯ANY-Key◯ → Dict🗂️',
|
||||
category=_category_add,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
_io.AnyType.Input(
|
||||
'key',
|
||||
optional=True,
|
||||
tooltip="Key (of ANY type) for the item inserted into the dict.",
|
||||
),
|
||||
_io.AnyType.Input(
|
||||
'value',
|
||||
optional=True,
|
||||
tooltip="The actual any-type item to add into the Dict."
|
||||
),
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
# noinspection PyShadowingBuiltins
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
key: _A = None, value: _A = None,
|
||||
dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
"""Add/update any-type item with any-type key to a Dict."""
|
||||
# No type check: neither for the key, nor for value
|
||||
result = _new_updated_dict(dict, {key: value})
|
||||
return _io.NodeOutput(result)
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
|
||||
class DictAddString(_BaseNode):
|
||||
"""Add/update a string item to a Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictAddString{_pack_id}',
|
||||
display_name='string → Dict🗂️',
|
||||
category=_category_add,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_KEY_INPUT_ADD,
|
||||
_KEY_CLEANUP_INPUT,
|
||||
_io.String.Input(
|
||||
'value',
|
||||
tooltip="The actual string to add into the Dict.",
|
||||
multiline=True,
|
||||
),
|
||||
_io.Boolean.Input(
|
||||
'cleanup_value', display_name='clean val',
|
||||
tooltip=(
|
||||
"When enabled, each line in the value-string is "
|
||||
"individually stripped of any leading/trailing spaces."
|
||||
),
|
||||
default=True,
|
||||
label_on='strip spaces',
|
||||
label_off='no',
|
||||
),
|
||||
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
key: str, value: str,
|
||||
cleanup_key: bool = True, cleanup_value: bool = True,
|
||||
dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
"""Add/update a string item to a Dict."""
|
||||
key = _validate_str_key(key, strip=cleanup_key)
|
||||
|
||||
value = '' if value is None else str(value)
|
||||
if cleanup_value:
|
||||
value = '\n'.join(
|
||||
x.strip() for x in value.splitlines()
|
||||
)
|
||||
result= _new_updated_dict(dict, {key: value})
|
||||
return _io.NodeOutput(result)
|
||||
|
||||
|
||||
# ==========================================================
|
||||
# Deprecated nodes with old IDs (for backwards compatibility)
|
||||
|
||||
|
||||
class DictAddAnyOld1(_BaseNode):
|
||||
# noinspection PyProtectedMember
|
||||
_schema = _schema_old_node(
|
||||
DictAddAny._schema,
|
||||
'StringConstructorDictAddAny',
|
||||
inputs_converter=_InputsConverter(
|
||||
preserved=('key', 'dict', 'value'),
|
||||
renames={'key': 'name'},
|
||||
),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
name: _O[str], dict: _O[_DictMap] = None, value: _A = None
|
||||
) -> _io.NodeOutput:
|
||||
return DictAddAny.execute(key=name, dict=dict, value=value)
|
||||
|
||||
|
||||
class DictAddStringOld1(_BaseNode):
|
||||
# noinspection PyProtectedMember
|
||||
_schema = _schema_old_node(
|
||||
DictAddString._schema,
|
||||
'StringConstructorDictAddString',
|
||||
inputs_converter=_InputsConverter(
|
||||
preserved=('key', 'cleanup_value', 'value', 'dict'),
|
||||
renames={
|
||||
'key': 'name',
|
||||
'cleanup_value': 'cleanup',
|
||||
'value': 'string',
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
name: str, cleanup: bool, string: str,
|
||||
dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
return DictAddString.execute(
|
||||
key=name, value=string,
|
||||
cleanup_key=False, cleanup_value=cleanup,
|
||||
dict=dict
|
||||
)
|
||||
-231
@@ -1,231 +0,0 @@
|
||||
# encoding: utf-8
|
||||
"""The main module for simple dict nodes."""
|
||||
|
||||
from inspect import cleandoc as _cleandoc
|
||||
|
||||
from frozendict import frozendict as _frozendict
|
||||
|
||||
from comfy_api.latest import io as _io, ui as _ui
|
||||
|
||||
from .__meta import (
|
||||
category as _category,
|
||||
pack_id_suffix as _pack_id
|
||||
)
|
||||
from .__typing import _A, _U, _O, _t, T as _T, DictMap as _DictMap
|
||||
from ._dict_funcs import _new_updated_dict
|
||||
from ._io_custom import (
|
||||
_BaseNode,
|
||||
_DICT_INPUT_OPTIONAL, _DICT_OUTPUT,
|
||||
_input_override, _schema_old_name
|
||||
)
|
||||
from .docstring_formatter import format_docstring as _format_docstring
|
||||
|
||||
_dict = dict
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
class DictAddAny(_BaseNode):
|
||||
"""Add/update any-type item to a Format-Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictAddAny{_pack_id}',
|
||||
display_name='Add ANY to Dict',
|
||||
category=_category,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_io.String.Input(
|
||||
'name',
|
||||
tooltip=(
|
||||
"Name (key) of the item inserted into the dict. "
|
||||
"It must comprise only of latin letters, digits and underscores + it can't start with a digit."
|
||||
),
|
||||
),
|
||||
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
_io.AnyType.Input(
|
||||
'value',
|
||||
optional=True,
|
||||
tooltip="The actual any-type item to add into the dict."
|
||||
),
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
name: _O[str], dict: _O[_DictMap] = None, value: _A = None
|
||||
) -> _io.NodeOutput:
|
||||
"""Update/append an item of any type to the dict."""
|
||||
if name is None:
|
||||
if dict is None:
|
||||
dict = _frozendict()
|
||||
# No need to create another dict instance if we add nothing:
|
||||
result = dict
|
||||
else:
|
||||
result = _new_updated_dict(dict, {name: value})
|
||||
return _io.NodeOutput(result)
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
class DictAddString(_BaseNode):
|
||||
"""Add/update a string to a Format-Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictAddString{_pack_id}',
|
||||
display_name='Add String to Dict',
|
||||
category=_category,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_io.String.Input(
|
||||
'name',
|
||||
tooltip=(
|
||||
"Name (key) of the string inserted into the dict. "
|
||||
"It must comprise only of latin letters, digits and underscores + it can't start with a digit."
|
||||
),
|
||||
),
|
||||
_io.Boolean.Input(
|
||||
'cleanup',
|
||||
tooltip=(
|
||||
"When enabled, each line in the sub-string is stripped "
|
||||
"from any spaces at its start and end."
|
||||
),
|
||||
default=True,
|
||||
label_on='leading/trailing spaces',
|
||||
label_off='no',
|
||||
),
|
||||
_io.String.Input(
|
||||
'string',
|
||||
tooltip="The actual string to add into the dict.",
|
||||
multiline=True,
|
||||
),
|
||||
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
name: str, cleanup: bool, string: str,
|
||||
dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
"""Update/append a string to the dict."""
|
||||
string = '' if string is None else str(string)
|
||||
if cleanup:
|
||||
string = '\n'.join(
|
||||
x.strip() for x in string.splitlines()
|
||||
)
|
||||
result= _new_updated_dict(dict, {name: string})
|
||||
return _io.NodeOutput(result)
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
class DictExtractString(_BaseNode):
|
||||
"""Extract a single string from a Format-Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictExtractString{_pack_id}',
|
||||
display_name='Extract String from Dict',
|
||||
category=_category,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_io.Boolean.Input(
|
||||
'cleanup_key', display_name='cleanup',
|
||||
tooltip=(
|
||||
"Automatically remove leading/trailing spaces and newlines from the key."
|
||||
),
|
||||
default=True,
|
||||
label_on='leading/trailing spaces',
|
||||
label_off='no',
|
||||
),
|
||||
_io.String.Input(
|
||||
'key',
|
||||
tooltip=(
|
||||
"Key (unique name) of a string to extract from dictionary.\n"
|
||||
"If the element with such key isn't a string, it will be turned to one.\n"
|
||||
"If no such key exists in the dict, an empty string returned."
|
||||
),
|
||||
),
|
||||
_io.Boolean.Input(
|
||||
'show_status',
|
||||
tooltip="Show the extracted string on the node itself?",
|
||||
default=False,
|
||||
label_on='value',
|
||||
label_off='no',
|
||||
),
|
||||
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
],
|
||||
outputs=[_io.String.Output('string')],
|
||||
# hidden=[_io.Hidden.unique_id],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
cleanup_key: bool, key: str, show_status: bool = False,
|
||||
dict: _O[_DictMap] = None
|
||||
) -> _io.NodeOutput:
|
||||
"""Extract a single string from the Format-Dict."""
|
||||
if cleanup_key:
|
||||
for line in key.split():
|
||||
line = line.strip() # just in case \r is left
|
||||
if line:
|
||||
key = line
|
||||
break
|
||||
|
||||
# The passed dict might not actually be a dict but ANY mapping,
|
||||
# which doesn't have a `get()` method.
|
||||
# So direct key access is better:
|
||||
string: str = ''
|
||||
# noinspection PyBroadException
|
||||
try:
|
||||
string = dict[key]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if string is None:
|
||||
string = ''
|
||||
elif not isinstance(string, str):
|
||||
string = repr(string)
|
||||
|
||||
return _io.NodeOutput(
|
||||
string,
|
||||
ui=_ui.PreviewText(string) if show_status else None
|
||||
)
|
||||
|
||||
|
||||
# ==========================================================
|
||||
# Deprecated nodes with old IDs (for backwards compatibility)
|
||||
|
||||
|
||||
class DictAddAnyOld1(DictAddAny):
|
||||
_schema = _schema_old_name(
|
||||
DictAddAny._schema,
|
||||
'StringConstructorDictAddAny'
|
||||
)
|
||||
|
||||
|
||||
class DictAddStringOld1(DictAddString):
|
||||
_schema = _schema_old_name(
|
||||
DictAddString._schema,
|
||||
'StringConstructorDictAddString'
|
||||
)
|
||||
|
||||
|
||||
class DictExtractStringOld1(_BaseNode):
|
||||
# noinspection PyProtectedMember
|
||||
_schema = _schema_old_name(
|
||||
DictExtractString._schema,
|
||||
'StringConstructorDictExtractString',
|
||||
inputs=[
|
||||
_input_override(x, id='name') if x.id == 'key' else x
|
||||
for x in DictExtractString._schema.inputs
|
||||
if x.id != 'cleanup_key'
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
name: str, show_status: bool = False,
|
||||
dict: _O[_DictMap] = None
|
||||
) -> _io.NodeOutput:
|
||||
return DictExtractString.execute(
|
||||
cleanup_key=False, key=name, show_status=show_status, dict=dict
|
||||
)
|
||||
@@ -0,0 +1,415 @@
|
||||
# encoding: utf-8
|
||||
"""Nodes to extract individual items from a dict (``Extract`` category)."""
|
||||
|
||||
from inspect import cleandoc as _cleandoc
|
||||
|
||||
from comfy_api.latest import io as _io, ui as _ui
|
||||
|
||||
from .__meta import (
|
||||
category_extract as _category_extract,
|
||||
category_extract_adv as _category_extract_adv,
|
||||
category_extract_mdl as _category_extract_mdl,
|
||||
pack_id_suffix as _pack_id
|
||||
)
|
||||
from .__typing import (
|
||||
_t, _A, _O, _U,
|
||||
_validate_type_factory, _validate_required_dict, _validate_str_key,
|
||||
T as _T, DictMap as _DictMap, NoiseProtocol as _NoiseProtocol,
|
||||
_ClipVisionModel, _ControlNet, _ModelPatcher, _CFGGuider, _Sampler,
|
||||
_CLIP, _VAE, _ImageModelDescriptor
|
||||
)
|
||||
from ._io_custom import (
|
||||
_BaseNode, _T_BaseNode,
|
||||
_DICT_INPUT_REQUIRED, _KEY_CLEANUP_INPUT, _KEY_INPUT_EXTRACT,
|
||||
_InputsConverter, _schema_old_node, _attach_execute_method
|
||||
)
|
||||
from .docstring_formatter import format_docstring as _format_docstring
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
# noinspection PyShadowingBuiltins
|
||||
def _get_key(dict: _DictMap, key: _A) -> _A:
|
||||
try:
|
||||
return dict[key]
|
||||
except KeyError:
|
||||
raise KeyError(f"No such key in Dict: {key!r}")
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
def _extracting_node(
|
||||
type_name: str,
|
||||
value_type: _t.Type[_T],
|
||||
value_output_cls: _t.Type[_io.Output],
|
||||
suffix: str = '',
|
||||
a='a', # a/an <type> for docstring
|
||||
convert_type: bool = False,
|
||||
category=_category_extract,
|
||||
**output_kwargs
|
||||
):
|
||||
"""Get class decorator building a ``DictExtract*`` node for a specific type."""
|
||||
type_name = str(type_name)
|
||||
assert issubclass(value_output_cls, _io.Output)
|
||||
if not suffix:
|
||||
suffix = ''
|
||||
if not a:
|
||||
a = 'a'
|
||||
|
||||
def class_decorator(cls: _t.Type[_T_BaseNode]):
|
||||
docstring = f"Extract {a} {type_name} item from a Dict."
|
||||
if not cls.__doc__:
|
||||
cls.__doc__ = docstring
|
||||
|
||||
class_name = str(cls.__name__).split('.')[-1]
|
||||
tooltip = f"The actual {type_name}-type item extracted from the Dict."
|
||||
cls._schema = _io.Schema(
|
||||
node_id=f'{class_name}{_pack_id}',
|
||||
display_name=f'🗂️Dict → {type_name}{suffix}',
|
||||
category=category,
|
||||
description=_format_docstring(_cleandoc(cls.__doc__)),
|
||||
inputs=[
|
||||
_DICT_INPUT_REQUIRED,
|
||||
_KEY_INPUT_EXTRACT,
|
||||
_KEY_CLEANUP_INPUT,
|
||||
],
|
||||
outputs=[
|
||||
value_output_cls('value', tooltip=tooltip, **output_kwargs),
|
||||
],
|
||||
)
|
||||
|
||||
_type_convert = value_type if convert_type else _validate_type_factory(type=value_type, type_name=type_name, a=a)
|
||||
|
||||
# noinspection PyShadowingBuiltins
|
||||
def execute(
|
||||
cls: _t.Type[_T_BaseNode],
|
||||
dict: _DictMap, key: str, cleanup_key: bool = True,
|
||||
) -> _io.NodeOutput:
|
||||
dict = _validate_required_dict(dict)
|
||||
key = _validate_str_key(key, strip=cleanup_key)
|
||||
value = _get_key(dict, key)
|
||||
value = _type_convert(value)
|
||||
return _io.NodeOutput(value)
|
||||
|
||||
cls = _attach_execute_method(cls, execute, docstring)
|
||||
|
||||
return cls
|
||||
|
||||
return class_decorator
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
# Basic types
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'bool', _io.Boolean.Type, _io.Boolean.Output, convert_type=True,
|
||||
)
|
||||
class DictExtractBool(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'float', _io.Float.Type, _io.Float.Output, convert_type=True,
|
||||
)
|
||||
class DictExtractFloat(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'int', _io.Int.Type, _io.Int.Output, a='an',convert_type=True,
|
||||
)
|
||||
class DictExtractInt(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Cond', list, _io.Conditioning.Output, suffix='🟠',
|
||||
)
|
||||
class DictExtractCond(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Image', _io.Image.Type, _io.Image.Output, a='an', suffix='🔵',
|
||||
)
|
||||
class DictExtractImage(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Latent', _U[_t.Dict, _t.Mapping], _io.Latent.Output, suffix='🟣',
|
||||
)
|
||||
class DictExtractLatent(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Mask', _io.Mask.Type, _io.Mask.Output, suffix='🟢',
|
||||
)
|
||||
class DictExtractMask(_BaseNode): pass
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
# Advanced types
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Guider', _CFGGuider, _io.Guider.Output,
|
||||
category=_category_extract_adv,
|
||||
)
|
||||
class DictExtractGuider(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Noise', _NoiseProtocol, _io.Noise.Output,
|
||||
category=_category_extract_adv,
|
||||
)
|
||||
class DictExtractNoise(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Sampler', _Sampler, _io.Sampler.Output,
|
||||
category=_category_extract_adv,
|
||||
)
|
||||
class DictExtractSampler(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Sigmas', _io.Sigmas.Type, _io.Sigmas.Output,
|
||||
category=_category_extract_adv,
|
||||
)
|
||||
class DictExtractSigmas(_BaseNode): pass
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
# Model types
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'CLIP', _CLIP, _io.Clip.Output, suffix='🟡',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractClip(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'CLIP-vision', _ClipVisionModel, _io.ClipVision.Output, suffix='🔵',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractClipVision(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'ControlNet', _ControlNet, _io.ControlNet.Output, suffix='🟢',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractControlNet(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'GLIGEN', _ModelPatcher, _io.Gligen.Output, suffix='🟤',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractGligen(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'LoRA', _io.LoraModel.Type, _io.LoraModel.Output, suffix='⚪',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractLora(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Model', _ModelPatcher, _io.Model.Output, suffix='🟣',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractModel(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'Upscale', _ImageModelDescriptor, _io.UpscaleModel.Output, a='an', suffix='🟢',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractUpscale(_BaseNode): pass
|
||||
|
||||
|
||||
# noinspection PyAbstractClass
|
||||
@_extracting_node(
|
||||
'VAE', _VAE, _io.Vae.Output, suffix='🔴',
|
||||
category=_category_extract_mdl,
|
||||
)
|
||||
class DictExtractVae(_BaseNode): pass
|
||||
|
||||
|
||||
# ==========================================================
|
||||
# Nodes with custom implementation
|
||||
|
||||
|
||||
class DictExtractAny(_BaseNode):
|
||||
"""Extract any-type item from a Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictExtractAny{_pack_id}',
|
||||
display_name='🗂️Dict → ANY◯',
|
||||
category=_category_extract,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_DICT_INPUT_REQUIRED,
|
||||
_KEY_INPUT_EXTRACT,
|
||||
_KEY_CLEANUP_INPUT,
|
||||
],
|
||||
outputs=[
|
||||
_io.AnyType.Output(
|
||||
'value',
|
||||
tooltip="The actual any-type item extracted from the Dict."
|
||||
),
|
||||
],
|
||||
# hidden=[_io.Hidden.unique_id],
|
||||
)
|
||||
|
||||
# noinspection PyShadowingBuiltins
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
dict: _DictMap, key: str, cleanup_key: bool = True,
|
||||
) -> _io.NodeOutput:
|
||||
"""Extract any-type item from a Dict."""
|
||||
dict = _validate_required_dict(dict)
|
||||
key = _validate_str_key(key, strip=cleanup_key)
|
||||
value = _get_key(dict, key)
|
||||
# No output-type check
|
||||
return _io.NodeOutput(value)
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
|
||||
class DictExtractAnyKey(_BaseNode):
|
||||
"""Extract any-type item with any-type key from a Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictExtractAnyKey{_pack_id}',
|
||||
display_name='🗂️Dict → ◯ANY-Key◯',
|
||||
category=_category_extract,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_DICT_INPUT_REQUIRED,
|
||||
_io.AnyType.Input(
|
||||
'key',
|
||||
optional=True,
|
||||
tooltip="Key (of ANY type) for the item extracted from the dict.",
|
||||
),
|
||||
],
|
||||
outputs=[
|
||||
_io.AnyType.Output(
|
||||
'value',
|
||||
tooltip="The actual any-type item extracted from the Dict."
|
||||
),
|
||||
],
|
||||
# hidden=[_io.Hidden.unique_id],
|
||||
)
|
||||
|
||||
# noinspection PyShadowingBuiltins
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
dict: _DictMap, key: _A = None,
|
||||
) -> _io.NodeOutput:
|
||||
"""Extract any-type item with any-type key from a Dict."""
|
||||
dict = _validate_required_dict(dict)
|
||||
# No key-check
|
||||
value = _get_key(dict, key)
|
||||
# No output-type check
|
||||
return _io.NodeOutput(value)
|
||||
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
|
||||
class DictExtractString(_BaseNode):
|
||||
"""Extract a string item from a Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictExtractString{_pack_id}',
|
||||
display_name='🗂️Dict → string',
|
||||
category=_category_extract,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_DICT_INPUT_REQUIRED,
|
||||
_io.String.Input(
|
||||
'key',
|
||||
tooltip=(
|
||||
"Key (unique name) of a string to extract from dictionary.\n"
|
||||
"If the element with such key isn't a string, it will be turned to one.\n"
|
||||
"If no such key exists in the dict, an empty string returned."
|
||||
),
|
||||
),
|
||||
_KEY_CLEANUP_INPUT,
|
||||
_io.Boolean.Input(
|
||||
'show_status',
|
||||
tooltip="Show the extracted string on the node itself?",
|
||||
default=False,
|
||||
label_on='value',
|
||||
label_off='no',
|
||||
),
|
||||
],
|
||||
outputs=[
|
||||
_io.String.Output(
|
||||
'value',
|
||||
tooltip="The actual string item extracted from the Dict."
|
||||
)
|
||||
],
|
||||
# hidden=[_io.Hidden.unique_id],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
dict: _DictMap, key: str,
|
||||
cleanup_key: bool = True, show_status: bool = False,
|
||||
) -> _io.NodeOutput:
|
||||
"""Extract a string item from a Dict."""
|
||||
dict = _validate_required_dict(dict)
|
||||
key = _validate_str_key(key, strip=cleanup_key)
|
||||
value = _get_key(dict, key)
|
||||
|
||||
if value is None:
|
||||
value = ''
|
||||
elif not isinstance(value, str):
|
||||
value = repr(value)
|
||||
|
||||
return _io.NodeOutput(
|
||||
value,
|
||||
ui=_ui.PreviewText(value) if show_status else None
|
||||
)
|
||||
|
||||
|
||||
# ==========================================================
|
||||
# Deprecated nodes with old IDs (for backwards compatibility)
|
||||
|
||||
|
||||
class DictExtractStringOld1(_BaseNode):
|
||||
# noinspection PyProtectedMember
|
||||
_schema = _schema_old_node(
|
||||
DictExtractString._schema,
|
||||
'StringConstructorDictExtractString',
|
||||
inputs_converter=_InputsConverter(
|
||||
preserved=('key', 'show_status', 'dict'),
|
||||
renames={'key': 'name'},
|
||||
),
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
name: str, show_status: bool = False,
|
||||
dict: _O[_DictMap] = None
|
||||
) -> _io.NodeOutput:
|
||||
return DictExtractString.execute(
|
||||
dict=dict, key=name, cleanup_key=False, show_status=show_status
|
||||
)
|
||||
+2
-2
@@ -1,13 +1,13 @@
|
||||
[project]
|
||||
name = "dict_tools"
|
||||
version = "3.1.1"
|
||||
version = "3.2.1"
|
||||
description = "Essential nodes to use dictionaries in ComfyUI: for smart prompt-formatting, general organization (passing a single connection instead of spaghetti), or anything else."
|
||||
license = {file = "LICENSE.md"}
|
||||
readme = "README.md"
|
||||
authors = [
|
||||
{name = "Lex Darlog"}
|
||||
]
|
||||
requires-python = ">=3.7"
|
||||
requires-python = ">=3.8" # Protocols are used
|
||||
dependencies = ["frozendict"]
|
||||
classifiers = [
|
||||
# https://docs.comfy.org/registry/specifications#classifiers-recommended
|
||||
|
||||
Reference in New Issue
Block a user