refactor: func modules -> static methods right within nodes
This commit is contained in:
@@ -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`
|
||||
|
||||
@@ -12,6 +12,7 @@ except ImportError:
|
||||
|
||||
|
||||
T = _t.TypeVar('T')
|
||||
T2 = _t.TypeVar('T2')
|
||||
|
||||
|
||||
class __BaseEnum(_StrEnum):
|
||||
|
||||
+41
-4
@@ -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
|
||||
)
|
||||
|
||||
@@ -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), )
|
||||
@@ -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), )
|
||||
@@ -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), )
|
||||
@@ -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, )
|
||||
@@ -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, )
|
||||
+12
-5
@@ -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), )
|
||||
|
||||
+12
-5
@@ -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), )
|
||||
|
||||
+68
-7
@@ -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, )
|
||||
|
||||
+44
-6
@@ -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, )
|
||||
|
||||
+55
-7
@@ -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, )
|
||||
|
||||
Reference in New Issue
Block a user