del nodes moved to Dict Tools
This commit is contained in:
@@ -5,19 +5,14 @@
|
||||
from comfy_api.latest import ComfyExtension as _ComfyExtension, io as _io
|
||||
|
||||
from ._formatter_class import Formatter
|
||||
from .node_dict_from_text import *
|
||||
from .node_formatter import *
|
||||
from .node_validate import *
|
||||
from .nodes_dict import *
|
||||
|
||||
class StringConstructorExtension(_ComfyExtension):
|
||||
async def get_node_list(self) -> list[type[_io.ComfyNode]]:
|
||||
return [
|
||||
DictFromText, DictFromTextOld1,
|
||||
StringFormatter, StringFormatterOld1,
|
||||
ValidateKeys, ValidateKeysOld1,
|
||||
DictAddAny, DictAddString, DictExtractString,
|
||||
DictAddAnyOld1, DictAddStringOld1, DictExtractStringOld1
|
||||
]
|
||||
|
||||
async def comfy_entrypoint() -> _ComfyExtension:
|
||||
|
||||
@@ -4,6 +4,5 @@ Metadata-related module.
|
||||
"""
|
||||
|
||||
category: str = "String Constructor"
|
||||
category_dict: str = f"{category}/Format-Dict"
|
||||
category_old: str = f"{category}/_old"
|
||||
pack_id_suffix: str = " [String Constructor]"
|
||||
|
||||
@@ -7,22 +7,9 @@ import typing as _t
|
||||
|
||||
import re as _re
|
||||
|
||||
from frozendict import deepfreeze as _deepfreeze, frozendict as _frozendict
|
||||
from frozendict import frozendict as _frozendict
|
||||
|
||||
from server import PromptServer as _PromptServer
|
||||
|
||||
from .__typing import T as _T, T2 as _T2
|
||||
|
||||
|
||||
def _show_text_on_node(text: str = None, unique_id: str = None):
|
||||
if not text:
|
||||
# TODO: Planned for the future - currently, there's no point removing the text since it's box is shown anyway
|
||||
# An odd workaround since `send_progress_text()` doesn't want to update text when '' passed
|
||||
text = '<span></span>'
|
||||
# print(f"{unique_id} text: {text!r}")
|
||||
|
||||
# Snatched from: https://github.com/comfyanonymous/ComfyUI/blob/27870ec3c30e56be9707d89a120eb7f0e2836be1/comfy_extras/nodes_images.py#L581-L582
|
||||
_PromptServer.instance.send_progress_text(text, unique_id)
|
||||
from .__typing import T as _T
|
||||
|
||||
|
||||
_re_valid_key_match = _re.compile("[a-zA-Z_][a-zA-Z_0-9]*$").match
|
||||
@@ -84,49 +71,3 @@ def _verify_input_dict(input_dict: _t.Dict[str, _T] = None, error_if_none=False)
|
||||
errors_dict,
|
||||
"Invalid input-dict: {}", "Invalid keys (string names) in input-dict: {}"
|
||||
)
|
||||
|
||||
|
||||
def _verify_input_dict_into_new(input_dict: _t.Dict[str, _T] = None) -> _t.Dict[str, _T]:
|
||||
"""
|
||||
Verify input dict to have only valid keys. Raises errors if invalid ones found.
|
||||
Always returns a new dict instance (shallow copy if non-empty dict passed).
|
||||
"""
|
||||
if input_dict is None:
|
||||
return dict()
|
||||
_verify_input_dict(input_dict)
|
||||
return dict(input_dict)
|
||||
|
||||
|
||||
def _new_updated_dict(
|
||||
input_dict: _t.Union[_t.Dict[str, _T], None], *updating_dicts: _t.Dict[str, _T2],
|
||||
sort=True, frozen=True, validate_new_keys=True
|
||||
) -> _t.Dict[str, _t.Union[_T, _T2]]:
|
||||
"""Return a new dict being a union of the input one and the new (updating) ones."""
|
||||
out_dict: _t.Dict[str, _t.Union[_T, _T2]] = _verify_input_dict_into_new(input_dict)
|
||||
|
||||
if validate_new_keys:
|
||||
errors_dict: _t.Dict[_t.Any, str] = dict()
|
||||
for upd_d in updating_dicts:
|
||||
upd_d = dict(upd_d)
|
||||
for key in list(upd_d.keys()):
|
||||
val = upd_d.pop(key)
|
||||
key = _validate_key(key, errors_dict)
|
||||
upd_d[key] = val
|
||||
out_dict.update(upd_d)
|
||||
_raise_from_errors_dict(errors_dict)
|
||||
else:
|
||||
for upd_d in updating_dicts:
|
||||
out_dict.update(upd_d)
|
||||
|
||||
if sort:
|
||||
out_dict = {k: v for k, v in sorted(out_dict.items())} # rely on implied ordered dicts in newer py3 versions
|
||||
return _deepfreeze(out_dict) if frozen else out_dict
|
||||
|
||||
|
||||
def _new_dict_with_updated_key(
|
||||
input_dict: _t.Union[_t.Dict[str, _T], None], key: str, value: _T2, sort=True, frozen=True, validate_key=True
|
||||
) -> _t.Dict[str, _t.Union[_T, _T2]]:
|
||||
"""Return a new dict being a union of the input one and the new (updating) key/value."""
|
||||
return _new_updated_dict(
|
||||
input_dict, {key: value}, sort=sort, frozen=frozen, validate_new_keys=validate_key
|
||||
)
|
||||
|
||||
@@ -1,159 +0,0 @@
|
||||
# encoding: utf-8
|
||||
"""Code for the ``Dict from Text`` node."""
|
||||
|
||||
from inspect import cleandoc as _cleandoc
|
||||
|
||||
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, FormatDict as _FormatDict
|
||||
from ._dict 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
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
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:
|
||||
return line_stripped
|
||||
|
||||
|
||||
def _parsed_kv_pairs_gen(multiline_str: str, strip_lines=True):
|
||||
"""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).
|
||||
"""
|
||||
if not multiline_str:
|
||||
return
|
||||
multiline_str = str(multiline_str)
|
||||
if not multiline_str.strip():
|
||||
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)
|
||||
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)
|
||||
|
||||
for line in multiline_str.splitlines():
|
||||
line_stripped: str = line.strip()
|
||||
if line_stripped:
|
||||
cur_chunk.append(appended_line_f(line, line_stripped))
|
||||
continue
|
||||
|
||||
assert not line_stripped
|
||||
if cur_chunk:
|
||||
yield dump_chunk()
|
||||
cur_chunk = list()
|
||||
|
||||
# We simply skip all the empty lines entirely
|
||||
|
||||
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).
|
||||
"""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictFromText{_pack_id}',
|
||||
display_name='Dict from Text',
|
||||
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.",
|
||||
default=True,
|
||||
label_on='leading/trailing 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"
|
||||
"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"
|
||||
),
|
||||
multiline=True,
|
||||
),
|
||||
_io.Boolean.Input(
|
||||
'show_status',
|
||||
tooltip="Show detected string names on the node itself?",
|
||||
default=False,
|
||||
label_on='detected keys',
|
||||
label_off='no',
|
||||
),
|
||||
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
# hidden=[_io.Hidden.unique_id],
|
||||
is_output_node=True, # Just to show the status message even if not connected to anything
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
cleanup: bool, dict_text: str, show_status: bool = False,
|
||||
dict: _O[_FormatDict] = None
|
||||
) -> _io.NodeOutput:
|
||||
parsed_items = _parsed_kv_pairs_gen(dict_text, strip_lines=cleanup)
|
||||
new_dict = {k: v for k, v in parsed_items}
|
||||
|
||||
if not new_dict:
|
||||
# No need to create another dict instance if we add nothing:
|
||||
out_dict = _dict() if dict is None else dict
|
||||
out_ui = '' if show_status else None
|
||||
return _io.NodeOutput(out_dict, ui=out_ui)
|
||||
|
||||
out_dict = _new_updated_dict(dict, new_dict)
|
||||
# These two ↑↓ must be in this specific order:
|
||||
# `_new_updated_dict()` also checks both dicts and might throw an error
|
||||
out_ui = None
|
||||
if show_status:
|
||||
status_text = ','.join(new_dict.keys()) if new_dict else ''
|
||||
out_ui = _ui.PreviewText(status_text)
|
||||
|
||||
return _io.NodeOutput(out_dict, ui=out_ui)
|
||||
|
||||
# ==========================================================
|
||||
# Deprecated node with old ID (for backwards compatibility)
|
||||
|
||||
class DictFromTextOld1(_BaseNode):
|
||||
# noinspection PyProtectedMember
|
||||
_schema = _schema_old_name(
|
||||
DictFromText._schema,
|
||||
'StringConstructorDictFromText',
|
||||
inputs=[
|
||||
_input_override(x, id='strings') if x.id == 'dict_text' else x
|
||||
for x in DictFromText._schema.inputs
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
cleanup: bool, strings: str, show_status: bool = False,
|
||||
dict: _O[_FormatDict] = None
|
||||
) -> _io.NodeOutput:
|
||||
return DictFromText.execute(
|
||||
cleanup=cleanup, dict_text=strings, show_status=show_status,
|
||||
dict=dict
|
||||
)
|
||||
+2
-2
@@ -6,7 +6,7 @@ from inspect import cleandoc as _cleandoc
|
||||
from comfy_api.latest import io as _io
|
||||
|
||||
from .__meta import (
|
||||
category_dict as _category_dict,
|
||||
category as _category,
|
||||
pack_id_suffix as _pack_id
|
||||
)
|
||||
from .__typing import _A, _U, _O, _t, T as _T, FormatDict as _FormatDict
|
||||
@@ -28,7 +28,7 @@ class ValidateKeys(_BaseNode):
|
||||
_schema = _io.Schema(
|
||||
node_id=f'ValidateKeys{_pack_id}',
|
||||
display_name='Validate Dict',
|
||||
category=_category_dict,
|
||||
category=_category,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[_DICT_INPUT_OPTIONAL],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
|
||||
-231
@@ -1,231 +0,0 @@
|
||||
# encoding: utf-8
|
||||
"""Service nodes for dict preparation."""
|
||||
|
||||
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_dict as _category_dict,
|
||||
pack_id_suffix as _pack_id
|
||||
)
|
||||
from .__typing import _A, _U, _O, _t, T as _T, FormatDict as _FormatDict
|
||||
from ._dict import _new_dict_with_updated_key
|
||||
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_dict,
|
||||
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: str, dict: _O[_FormatDict] = None, value: _A = None
|
||||
) -> _io.NodeOutput:
|
||||
"""Update/append an item of any type to the dict."""
|
||||
if value is None:
|
||||
if dict is None:
|
||||
dict = _frozendict()
|
||||
# No need to create another dict instance if we add nothing:
|
||||
result = dict
|
||||
else:
|
||||
result = _new_dict_with_updated_key(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_dict,
|
||||
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[_FormatDict] = 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_dict_with_updated_key(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_dict,
|
||||
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[_FormatDict] = 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[_FormatDict] = None
|
||||
) -> _io.NodeOutput:
|
||||
return DictExtractString.execute(
|
||||
cleanup_key=False, key=name, show_status=show_status, dict=dict
|
||||
)
|
||||
Reference in New Issue
Block a user