diff --git a/__init__.py b/__init__.py index d04d282..ed2ba9b 100644 --- a/__init__.py +++ b/__init__.py @@ -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]]: diff --git a/__meta.py b/__meta.py index 5b8b6c2..0944892 100644 --- a/__meta.py +++ b/__meta.py @@ -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]" diff --git a/_io_custom.py b/_io_custom.py index 9dd9acd..777c184 100644 --- a/_io_custom.py +++ b/_io_custom.py @@ -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) diff --git a/node_dict_from_text.py b/node_dict_from_text.py index ef7c0f3..d683bf7 100644 --- a/node_dict_from_text.py +++ b/node_dict_from_text.py @@ -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 ) diff --git a/nodes_add.py b/nodes_add.py new file mode 100644 index 0000000..43719a4 --- /dev/null +++ b/nodes_add.py @@ -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 + ) diff --git a/nodes_dict.py b/nodes_dict.py deleted file mode 100644 index 769ab63..0000000 --- a/nodes_dict.py +++ /dev/null @@ -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 - ) diff --git a/nodes_extract.py b/nodes_extract.py new file mode 100644 index 0000000..fff93ed --- /dev/null +++ b/nodes_extract.py @@ -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 + )