diff --git a/__init__.py b/__init__.py index 9242c64..be47061 100644 --- a/__init__.py +++ b/__init__.py @@ -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: diff --git a/__meta.py b/__meta.py index eac65e6..a290eaf 100644 --- a/__meta.py +++ b/__meta.py @@ -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]" diff --git a/_dict.py b/_dict.py index 8b70e61..3b8bec9 100644 --- a/_dict.py +++ b/_dict.py @@ -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 = '' - # 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 - ) diff --git a/node_dict_from_text.py b/node_dict_from_text.py deleted file mode 100644 index 1212fb4..0000000 --- a/node_dict_from_text.py +++ /dev/null @@ -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 - ) diff --git a/node_validate.py b/node_validate.py index ce3a333..cba9a7f 100644 --- a/node_validate.py +++ b/node_validate.py @@ -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], diff --git a/nodes_dict.py b/nodes_dict.py deleted file mode 100644 index 77b9e64..0000000 --- a/nodes_dict.py +++ /dev/null @@ -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 - )