Merge pull request #2 from Lex-DRL/dev
refactor: Change node signatures + internal tools for better backwards compatibility
This commit is contained in:
Lex Darlog
+2
-1
@@ -5,7 +5,8 @@
|
||||
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]]:
|
||||
|
||||
@@ -4,5 +4,9 @@ Metadata-related module.
|
||||
"""
|
||||
|
||||
category: str = "🗂️ Dict Tools"
|
||||
category_add: str = f"{category}/Add"
|
||||
category_extract: str = f"{category}/Extract"
|
||||
|
||||
category_old: str = f"{category}/_old"
|
||||
|
||||
pack_id_suffix: str = " [Dict Tools]"
|
||||
|
||||
+62
-3
@@ -2,12 +2,12 @@
|
||||
"""
|
||||
"""
|
||||
|
||||
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, T as _T, _A, _O, _U, _TypeVar
|
||||
|
||||
|
||||
_DICT = _io.Custom("DICT")
|
||||
@@ -20,10 +20,57 @@ _DICT_OUTPUT = _DICT.Output(
|
||||
'DICT',
|
||||
display_name='dict'
|
||||
)
|
||||
_KEY_INPUT_ADD = _io.String.Input(
|
||||
'key',
|
||||
tooltip="Key (name) of the item inserted into 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 +103,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 +139,6 @@ class _BaseNode(_io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls) -> _io.Schema:
|
||||
return cls._schema
|
||||
|
||||
|
||||
_T_BaseNode = _TypeVar('T_BaseNode', bound=_BaseNode)
|
||||
|
||||
+38
-21
@@ -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,7 +75,7 @@ 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()
|
||||
@@ -80,8 +93,8 @@ class DictFromText(_BaseNode):
|
||||
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 stripped from any leading/trailing spaces before parsing the text.",
|
||||
default=True,
|
||||
label_on='leading/trailing spaces',
|
||||
label_off='no',
|
||||
@@ -112,10 +125,11 @@ class DictFromText(_BaseNode):
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
cleanup: bool, dict_text: str, show_status: bool = False,
|
||||
pre_cleanup: bool,
|
||||
dict_text: str = None, show_status: bool = False,
|
||||
dict: _O[_DictMap] = None
|
||||
) -> _io.NodeOutput:
|
||||
parsed_items = _parsed_kv_pairs_gen(dict_text, strip_lines=cleanup)
|
||||
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:
|
||||
@@ -139,13 +153,16 @@ class DictFromText(_BaseNode):
|
||||
|
||||
class DictFromTextOld1(_BaseNode):
|
||||
# noinspection PyProtectedMember
|
||||
_schema = _schema_old_name(
|
||||
_schema = _schema_old_node(
|
||||
DictFromText._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
|
||||
@@ -154,6 +171,6 @@ class DictFromTextOld1(_BaseNode):
|
||||
dict: _O[_DictMap] = None
|
||||
) -> _io.NodeOutput:
|
||||
return DictFromText.execute(
|
||||
cleanup=cleanup, dict_text=strings, show_status=show_status,
|
||||
pre_cleanup=cleanup, dict_text=strings, show_status=show_status,
|
||||
dict=dict
|
||||
)
|
||||
|
||||
+150
@@ -0,0 +1,150 @@
|
||||
# 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,
|
||||
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, _KEY_INPUT_ADD,
|
||||
_InputsConverter, _schema_old_node
|
||||
)
|
||||
from .docstring_formatter import format_docstring as _format_docstring
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
class DictAddAny(_BaseNode):
|
||||
"""Add/update any-type item to a Format-Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictAddAny{_pack_id}',
|
||||
display_name='ANY to Dict',
|
||||
category=_category_add,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_KEY_INPUT_ADD,
|
||||
|
||||
_io.AnyType.Input(
|
||||
'value',
|
||||
optional=True,
|
||||
tooltip="The actual any-type item to add into the dict."
|
||||
),
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
key: _O[str], value: _A = None, dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
"""Update/append an item of any type to the dict."""
|
||||
if key 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, {key: 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='STRING to Dict',
|
||||
category=_category_add,
|
||||
description=_format_docstring(_cleandoc(__doc__)),
|
||||
inputs=[
|
||||
_KEY_INPUT_ADD,
|
||||
_io.Boolean.Input(
|
||||
'cleanup_value',
|
||||
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(
|
||||
'value',
|
||||
tooltip="The actual string to add into the dict.",
|
||||
multiline=True,
|
||||
),
|
||||
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
],
|
||||
outputs=[_DICT_OUTPUT],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls,
|
||||
key: str, cleanup_value: bool, value: str = None,
|
||||
dict: _O[_DictMap] = None,
|
||||
) -> _io.NodeOutput:
|
||||
"""Update/append a string to the dict."""
|
||||
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, cleanup_value=cleanup, value=string, 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,117 @@
|
||||
# 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,
|
||||
pack_id_suffix as _pack_id
|
||||
)
|
||||
from .__typing import _A, _U, _O, _t, T as _T, DictMap as _DictMap
|
||||
from ._io_custom import (
|
||||
_BaseNode,
|
||||
_DICT_INPUT_OPTIONAL,
|
||||
_InputsConverter, _schema_old_node
|
||||
)
|
||||
from .docstring_formatter import format_docstring as _format_docstring
|
||||
|
||||
# ----------------------------------------------------------
|
||||
|
||||
class DictExtractString(_BaseNode):
|
||||
"""Extract a single string from a Format-Dict."""
|
||||
_schema = _io.Schema(
|
||||
node_id=f'DictExtractString{_pack_id}',
|
||||
display_name='STRING from Dict',
|
||||
category=_category_extract,
|
||||
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 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(
|
||||
cleanup_key=False, key=name, show_status=show_status, dict=dict
|
||||
)
|
||||
Reference in New Issue
Block a user