diff --git a/CHANGELOG.md b/CHANGELOG.md index 47de817..6e769f3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,11 @@ The pack has already started to sip into dictionary-editing domain. Only the val - ~~Multi-item nodes.~~ - ~~Bulk `Search & Replace` within **keys** in dictionary.~~ +# v1.0.3 + +- [fix] Nodes made compatible with built-in preview nodes. +- Some internal refactoring/cleanup. + # v1.0.2 - New node: `Validate Format-Dict` diff --git a/enums.py b/enums.py index 14ede54..a04c4ab 100644 --- a/enums.py +++ b/enums.py @@ -12,6 +12,7 @@ except ImportError: T = _t.TypeVar('T') +T2 = _t.TypeVar('T2') class __BaseEnum(_StrEnum): diff --git a/funcs_common.py b/funcs_common.py index 0ffd27b..cca9b4f 100644 --- a/funcs_common.py +++ b/funcs_common.py @@ -7,9 +7,11 @@ import typing as _t import re as _re +from frozendict import deepfreeze as _deepfreeze + from server import PromptServer as _PromptServer -from .enums import T as _T +from .enums import T as _T, T2 as _T2 def _show_text_on_node(text: str = None, unique_id: str = None): @@ -48,7 +50,7 @@ def _validate_key(key: str, errors_dict: _t.Dict[_t.Any, str]) -> _t.Union[str, def _raise_from_errors_dict( errors_dict: _t.Dict[_t.Any, str], single_error_format: str = '{}', - multi_errors_format: str = "Invalid keys (string names):\n{}", + multi_errors_format: str = "Invalid keys (string names): {}", ): if not errors_dict: return @@ -56,7 +58,7 @@ def _raise_from_errors_dict( if len(errors_dict) == 1: msg = single_error_format.format(next(iter(errors_dict.values()))) else: - msg = multi_errors_format.format('\n'.join(repr(x) for x in errors_dict.keys())) + msg = multi_errors_format.format(', '.join(repr(x) for x in errors_dict.keys())) raise KeyError(msg) @@ -78,7 +80,7 @@ def _verify_input_dict(input_dict: _t.Dict[str, _T] = None, error_if_none=False) _raise_from_errors_dict( errors_dict, - "Invalid input-dict:\n{}", "Invalid keys (string names) in input-dict:\n{}" + "Invalid input-dict: {}", "Invalid keys (string names) in input-dict: {}" ) @@ -91,3 +93,38 @@ def _verify_input_dict_into_new(input_dict: _t.Dict[str, _T] = None) -> _t.Dict[ 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/funcs_dict_add_any.py b/funcs_dict_add_any.py deleted file mode 100644 index aa00f3d..0000000 --- a/funcs_dict_add_any.py +++ /dev/null @@ -1,27 +0,0 @@ -# encoding: utf-8 -""" -The actual code (MVC pattern) for `StringConstructorDictAddAny` node. -""" - -import typing as _t - -from frozendict import deepfreeze as _deepfreeze - -from .funcs_common import _raise_from_errors_dict, _validate_key, _verify_input_dict_into_new - - -def set_dict_any_type_item( - name: str, value: _t.Any = None, in_dict: _t.Dict[str, _t.Any] = None, - # unique_id: str = None, -) -> _t.Tuple[_t.Dict[str, _t.Any]]: - """Update/append an item of any type to the dict.""" - out_dict: _t.Dict[str, _t.Any] = _verify_input_dict_into_new(in_dict) - if value is None: - return (_deepfreeze(in_dict), ) # No need to create another dict instance if it's already frozen - - errors_dict: _t.Dict[_t.Any, str] = dict() - name = _validate_key(name, errors_dict) - _raise_from_errors_dict(errors_dict) - new_dict: _t.Dict[str, _t.Any] = {name: value} - out_dict.update(new_dict) - return (_deepfreeze(out_dict), ) diff --git a/funcs_dict_add_string.py b/funcs_dict_add_string.py deleted file mode 100644 index a396d32..0000000 --- a/funcs_dict_add_string.py +++ /dev/null @@ -1,30 +0,0 @@ -# encoding: utf-8 -""" -The actual code (MVC pattern) for `StringConstructorDictString` node. -""" - -import typing as _t - -from frozendict import deepfreeze as _deepfreeze - -from .enums import T as _T -from .funcs_common import _raise_from_errors_dict, _validate_key, _verify_input_dict_into_new - - -def set_dict_string( - name: str, value: str, strip_lines=True, in_dict: _t.Dict[str, _T] = None, - # unique_id: str = None, -) -> _t.Tuple[_t.Dict[str, _t.Union[_T, str]]]: - """Update/append a string to the dict.""" - out_dict: _t.Dict[str, _t.Union[_T, str]] = _verify_input_dict_into_new(in_dict) - errors_dict: _t.Dict[_t.Any, str] = dict() - name = _validate_key(name, errors_dict) - _raise_from_errors_dict(errors_dict) - value = str(value) - if strip_lines: - value = '\n'.join( - x.strip() for x in value.splitlines() - ) - new_dict: _t.Dict[str, str] = {name: value} - out_dict.update(new_dict) - return (_deepfreeze(out_dict), ) diff --git a/funcs_dict_from_text.py b/funcs_dict_from_text.py deleted file mode 100644 index 5ed9bac..0000000 --- a/funcs_dict_from_text.py +++ /dev/null @@ -1,81 +0,0 @@ -# encoding: utf-8 -""" -The actual code (MVC pattern) for `StringConstructorDictFromText` node. -""" - -import typing as _t - -from frozendict import frozendict as _frozendict, deepfreeze as _deepfreeze - -from .enums import T as _T -from .funcs_common import _show_text_on_node, _raise_from_errors_dict, _validate_key, _verify_input_dict_into_new - - -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): - """ - 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() - - -def parse_dict_from_text( - multiline_str: str, strip_lines=True, - show: bool = True, in_dict: _t.Dict[str, _T] = None, unique_id: str = None -) -> _t.Tuple[_t.Dict[str, _T]]: - out_dict: _t.Dict[str, _T] = _verify_input_dict_into_new(in_dict) - - errors_dict: _t.Dict[_t.Any, str] = dict() - new_dict: _t.Dict[str, str] = { - _k: _v for _k, _v in ( - (_validate_key(k, errors_dict), v) - for k, v in _parsed_kv_pairs_gen(multiline_str, strip_lines=strip_lines) - ) if _k is not None - } - _raise_from_errors_dict(errors_dict) - - out_dict.update(new_dict) - - if show and unique_id: - status_text = ','.join(new_dict.keys()) if new_dict else '' - _show_text_on_node(status_text, unique_id) - - return (_deepfreeze(out_dict), ) diff --git a/funcs_dict_preview.py b/funcs_dict_preview.py deleted file mode 100644 index de1ab0e..0000000 --- a/funcs_dict_preview.py +++ /dev/null @@ -1,47 +0,0 @@ -# encoding: utf-8 -""" -The actual code (MVC pattern) for `StringConstructorDictPreview` node. -""" - -import typing as _t - -from pprint import pformat as _pformat - -from .funcs_common import _show_text_on_node, _verify_input_dict_into_new - - -def _to_regular_dict_recursive_copy(input_dict: dict = None) -> dict: - """ - To simplify preview, turns frozen dicts back to regular ones. - Copies the dict in process and does so recursively, but not truly deep: if value is another dict, - it will be processed, too. But dicts within other objects (like lists) won't be found. - """ - if not input_dict: - return dict() - return { - k: (_to_regular_dict_recursive_copy(v) if isinstance(v, dict) else v) - for k, v in input_dict.items() - } - - -def _preview_single_entry(key: str, value) -> str: - if isinstance(value, str): - return f"{key}:\n{value}" - value_repr = _pformat(value) - return f"{key}:\n{value_repr}" - - -def _preview_message(input_dict: dict = None): - input_dict = _to_regular_dict_recursive_copy(_verify_input_dict_into_new(input_dict)) - parts: _t.List[str] = [ - _preview_single_entry(k, v) for k, v in input_dict.items() - ] - return '\n\n'.join(parts) - - -def preview_str_dict(input_dict: dict, unique_id: str = None): - if not unique_id: - return (input_dict, ) - status_text = _preview_message(input_dict) - _show_text_on_node(status_text, unique_id) - return (input_dict, ) diff --git a/funcs_formatter.py b/funcs_formatter.py deleted file mode 100644 index 9b36a37..0000000 --- a/funcs_formatter.py +++ /dev/null @@ -1,59 +0,0 @@ -# encoding: utf-8 -""" -The actual code (MVC pattern) for `StringConstructorFormatter` node. -""" - -import typing as _t - -import sys as _sys - -from .funcs_common import _show_text_on_node, _verify_input_dict_into_new - - -_RECURSION_LIMIT = max(int(_sys.getrecursionlimit()), 10) # You can externally monkey-patch it... but if it blows up, your fault 🤷🏻‍♂️ - - -def _recursive_format(template: str, format_dict: _t.Dict[str, _t.Any], show: bool = True, unique_id: str = None) -> str: - """ - It's not actually recursive - because, you know, any recursion could be turned into iteration, - and good boys do that. 😊 - """ - assert isinstance(_RECURSION_LIMIT, int) and _RECURSION_LIMIT > 0 - - prev: str = '' - new: str = template - for i in range(_RECURSION_LIMIT): - if prev == new: - break - prev = new - new = new.format_map(format_dict) - if prev == new: - return new - - msg = ( - f"Recursion limit ({_RECURSION_LIMIT}) reached on attempt to format a string: {template!r}\n" - f"Last two formatting attempts:\n{prev!r}\n{new!r}" - ) - if show and unique_id: - _show_text_on_node(msg, unique_id) - raise RecursionError(msg) - # noinspection PyUnreachableCode - return '' # just to be safe - - -def formatter( - template: str, recursive: bool = False, format_dict: _t.Dict[str, _t.Any] = None, - show: bool = True, unique_id: str = None -) -> _t.Tuple[str]: - format_dict = _verify_input_dict_into_new(format_dict) - if not isinstance(template, str): - raise TypeError(f"Not a string: {template!r}") - - out_text = ( - _recursive_format(template, format_dict, show, unique_id) - if recursive - else template.format_map(format_dict) - ) - if show and unique_id: - _show_text_on_node(out_text, unique_id) - return (out_text, ) diff --git a/node_dict_add_any.py b/node_dict_add_any.py index 6416969..b4bd4b6 100644 --- a/node_dict_add_any.py +++ b/node_dict_add_any.py @@ -13,10 +13,12 @@ from comfy.comfy_types.node_typing import IO as _IO from . import _meta from .docstring_formatter import format_docstring as _format_docstring from .enums import DataTypes as _DataTypes -from .funcs_dict_add_any import set_dict_any_type_item as _set_dict_any_type_item +from .funcs_common import _new_dict_with_updated_key from .node_dict_add_string import _input_types as _input_types_str +_dict = dict + # A tiny optimization by reusing the same immutable dict: _input_types = _deepfreeze({ 'required': { @@ -50,9 +52,14 @@ class StringConstructorDictAddAny: def INPUT_TYPES(cls): return _input_types + @staticmethod def main( - self, name: str, value: _t.Any = None, - dict: _t.Dict[str, _t.Any] = None, + name: str, dict: _t.Dict[str, _t.Any] = None, value: _t.Any = None, # unique_id: str = None, - ): - return _set_dict_any_type_item(name, value, in_dict=dict) + ) -> _t.Tuple[_t.Dict[str, _t.Any]]: + """Update/append an item of any type to the dict.""" + if value is None: + if dict is None: + dict = _dict() + return (dict, ) # No need to create another dict instance if we add nothing + return (_new_dict_with_updated_key(dict, name, value), ) diff --git a/node_dict_add_string.py b/node_dict_add_string.py index 265d979..1245797 100644 --- a/node_dict_add_string.py +++ b/node_dict_add_string.py @@ -13,7 +13,7 @@ from comfy.comfy_types.node_typing import IO as _IO from . import _meta from .docstring_formatter import format_docstring as _format_docstring from .enums import DataTypes as _DataTypes -from .funcs_dict_add_string import set_dict_string as _set_dict_string +from .funcs_common import _new_dict_with_updated_key, _T # A tiny optimization by reusing the same immutable dict: @@ -52,9 +52,16 @@ class StringConstructorDictAddString: def INPUT_TYPES(cls): return _input_types + @staticmethod def main( - self, name: str, cleanup: bool, string: str, - dict: _t.Dict[str, _t.Any] = None, + name: str, cleanup: bool, string: str, + dict: _t.Dict[str, _T] = None, # unique_id: str = None, - ): - return _set_dict_string(name, string, strip_lines=cleanup, in_dict=dict) + ) -> _t.Tuple[_t.Dict[str, _t.Union[_T, str]]]: + """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() + ) + return (_new_dict_with_updated_key(dict, name, string), ) diff --git a/node_dict_from_text.py b/node_dict_from_text.py index e92fba0..1d7681c 100644 --- a/node_dict_from_text.py +++ b/node_dict_from_text.py @@ -13,11 +13,59 @@ from comfy.comfy_types.node_typing import IO as _IO from . import _meta from .docstring_formatter import format_docstring as _format_docstring from .enums import DataTypes as _DataTypes -from .funcs_dict_from_text import parse_dict_from_text as _parse_dict_from_text +from .funcs_common import _show_text_on_node, _new_updated_dict, _T from .node_dict_add_string import _input_types as _input_types_str -# A tiny optimization by reusing the same immutable 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): + """ + 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() + +# -------------------------------------- + +_dict = dict + _input_types = _deepfreeze({ 'required': { 'cleanup': (_IO.BOOLEAN, {'default': True, 'label_on': 'leading/trailing spaces', 'label_off': 'no', 'tooltip': ( @@ -60,9 +108,22 @@ class StringConstructorDictFromText: def INPUT_TYPES(cls): return _input_types + @staticmethod def main( - self, cleanup: bool, strings: str, show_status: bool, - dict: _t.Dict[str, _t.Any] = None, - unique_id: str = None, - ): - return _parse_dict_from_text(strings, strip_lines=cleanup, show=show_status, in_dict=dict, unique_id=unique_id) + cleanup: bool, strings: str, show_status: bool = True, + dict: _t.Dict[str, _T] = None, unique_id: str = None, + ) -> _t.Tuple[_t.Dict[str, _t.Union[_T, str]]]: + new_dict = {k: v for k, v in _parsed_kv_pairs_gen(strings, strip_lines=cleanup)} + + if not new_dict: + if dict is None: + dict = _dict() + return (dict, ) # No need to create another dict instance if we add nothing + + out_dict = _new_updated_dict(dict, new_dict) + # These two ↑↓ must be in this specific order: `_new_updated_dict()` also checks both dicts + if show_status and unique_id: + status_text = ','.join(new_dict.keys()) if new_dict else '' + _show_text_on_node(status_text, unique_id) + + return (out_dict, ) diff --git a/node_dict_preview.py b/node_dict_preview.py index c59068b..16c23a4 100644 --- a/node_dict_preview.py +++ b/node_dict_preview.py @@ -5,21 +5,53 @@ import typing as _t from inspect import cleandoc as _cleandoc +from pprint import pformat as _pformat from frozendict import deepfreeze as _deepfreeze from . import _meta from .docstring_formatter import format_docstring as _format_docstring from .enums import DataTypes as _DataTypes -from .funcs_dict_preview import preview_str_dict as _preview_str_dict +from .funcs_common import _show_text_on_node, _verify_input_dict_into_new -# A tiny optimization by reusing the same immutable dict: +def _to_regular_dict_recursive_copy(input_dict: dict = None) -> dict: + """ + To simplify preview, turns frozen dicts back to regular ones. + Copies the dict in process and does so recursively, but not truly deep: if value is another dict, + it will be processed, too. But dicts within other objects (like lists) won't be found. + """ + if not input_dict: + return dict() + return { + k: (_to_regular_dict_recursive_copy(v) if isinstance(v, dict) else v) + for k, v in input_dict.items() + } + + +def _preview_single_entry(key: str, value) -> str: + if isinstance(value, str): + return f"{key}:\n{value}" + value_repr = _pformat(value) + return f"{key}:\n{value_repr}" + + +def _preview_message(input_dict: dict = None): + input_dict = _to_regular_dict_recursive_copy(_verify_input_dict_into_new(input_dict)) + parts: _t.List[str] = [ + _preview_single_entry(k, v) for k, v in input_dict.items() + ] + return '\n\n'.join(parts) + +# -------------------------------------- + +_dict = dict + _input_types = _deepfreeze({ - 'required': { + 'required': {}, + 'optional': { 'dict': _DataTypes.input_dict(tooltip="A Format-Dictionary."), }, - # 'optional': {}, 'hidden': { 'unique_id': 'UNIQUE_ID', # used for text display at the bottom of the node }, @@ -45,8 +77,14 @@ class StringConstructorDictPreview: def INPUT_TYPES(cls): return _input_types + @staticmethod def main( - self, dict: _t.Dict[str, _t.Any], + dict: _t.Dict[str, _t.Any] = None, unique_id: str = None, ): - return _preview_str_dict(dict, unique_id=unique_id) + if dict is None: + dict = _dict() + if unique_id: + status_text = _preview_message(dict) + _show_text_on_node(status_text, unique_id) + return (dict, ) diff --git a/node_formatter.py b/node_formatter.py index aec9c2e..ce2210d 100644 --- a/node_formatter.py +++ b/node_formatter.py @@ -5,6 +5,7 @@ import typing as _t from inspect import cleandoc as _cleandoc +import sys as _sys from frozendict import deepfreeze as _deepfreeze @@ -13,10 +14,44 @@ from comfy.comfy_types.node_typing import IO as _IO from . import _meta from .docstring_formatter import format_docstring as _format_docstring from .enums import DataTypes as _DataTypes -from .funcs_formatter import formatter as _formatter +from .funcs_common import _show_text_on_node, _verify_input_dict +# from .funcs_formatter import formatter as _formatter -# A tiny optimization by reusing the same immutable dict: +_RECURSION_LIMIT = max(int(_sys.getrecursionlimit()), 1) # You can externally monkey-patch it... but if it blows up, your fault 🤷🏻‍♂️ + + +def _recursive_format(template: str, format_dict: _t.Dict[str, _t.Any], show: bool = True, unique_id: str = None) -> str: + """ + It's not actually recursive - because, you know, any recursion could be turned into iteration, + and good boys do that. 😊 + """ + assert isinstance(_RECURSION_LIMIT, int) and _RECURSION_LIMIT > 0 + + prev: str = '' + new: str = template + for i in range(_RECURSION_LIMIT): + if prev == new: + break + prev = new + new = new.format_map(format_dict) + if prev == new: + return new + + msg = ( + f"Recursion limit ({_RECURSION_LIMIT}) reached on attempt to format a string: {template!r}\n" + f"Last two formatting attempts:\n{prev!r}\n{new!r}" + ) + if show and unique_id: + _show_text_on_node(msg, unique_id) + raise RecursionError(msg) + # noinspection PyUnreachableCode + return '' # just to be safe + +# -------------------------------------- + +_dict = dict + _input_types = _deepfreeze({ 'required': { 'template': (_IO.STRING, {'multiline': True, 'tooltip': ( @@ -69,9 +104,22 @@ class StringConstructorFormatter: def INPUT_TYPES(cls): return _input_types + @staticmethod def main( - self, template: str, recursive_format: bool, show_status: bool, - dict: _t.Dict[str, _t.Any] = None, - unique_id: str = None, - ): - return _formatter(template, recursive_format, dict, show=show_status, unique_id=unique_id) + template: str, recursive_format: bool = False, show_status: bool = False, dict: _t.Dict[str, _t.Any] = None, + unique_id: str = None + ) -> _t.Tuple[str]: + if dict is None: + dict = _dict() + _verify_input_dict(dict) + if not isinstance(template, str): + raise TypeError(f"Not a string: {template!r}") + + out_text = ( + _recursive_format(template, dict, show_status, unique_id) + if recursive_format + else template.format_map(dict) + ) + if show_status and unique_id: + _show_text_on_node(out_text, unique_id) + return (out_text, )