Skip to content
Draft
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 17 additions & 17 deletions manim/utils/hashing.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ class _Memoizer:
content-equality detection.
"""

_already_processed = set()
_already_processed: set[int] = set()

# Can be changed to whatever string to help debugging the JSon generation.
ALREADY_PROCESSED_PLACEHOLDER = "AP"
Expand Down Expand Up @@ -77,7 +77,7 @@ def layer(func: Callable[[Any], Any]) -> Callable:
if is_method:
return lambda self, obj: cls._handle_already_processed(
obj,
default_function=lambda obj: func(self, obj),
default_function=lambda obj: func(obj),
)
return lambda obj: cls._handle_already_processed(obj, default_function=func)

Expand Down Expand Up @@ -112,7 +112,7 @@ def mark_as_processed(cls: type[_Memoizer], obj: Any) -> None:
The object to mark as processed.
"""
cls._handle_already_processed(obj, lambda x: x)
return cls._return(obj, id, lambda x: x, memoizing=False)
cls._return(obj, id, lambda x: x, memoizing=False)

@classmethod
def _handle_already_processed(
Expand Down Expand Up @@ -235,12 +235,12 @@ def default(self, obj: Any) -> Any:
return str(type(obj))

@overload
def _cleaned_iterable(self, iterable: Sequence[Any]) -> list[Any]: ...
def _cleaned_iterable(self, iterable: dict[Any, Any]) -> dict[Any, Any]: ...

Check notice

Code scanning / CodeQL

Statement has no effect

This statement has no effect.

@overload
def _cleaned_iterable(self, iterable: dict[Any, Any]) -> dict[Any, Any]: ...
def _cleaned_iterable(self, iterable: Sequence[Any]) -> list[Any]: ...

Check notice

Code scanning / CodeQL

Statement has no effect

This statement has no effect.

def _cleaned_iterable(self, iterable):
def _cleaned_iterable(self, iterable: Iterable) -> dict[Any, Any] | list[Any]:
"""Check for circular reference at each iterable that will go through the JSONEncoder, as well as key of the wrong format.

If a key with a bad format is found (i.e not a int, string, or float), it gets replaced by its hash using the same process implemented here.
Expand All @@ -255,17 +255,17 @@ def _cleaned_iterable(self, iterable):
def _key_to_hash(key: Any) -> int:
return zlib.crc32(json.dumps(key, cls=_CustomEncoder).encode())

def _iter_check_list(lst: Sequence[Any]) -> list[Any]:
processed_list = [None] * len(lst)
for i, el in enumerate(lst):
def _iter_check_list(lst: Iterable[Any]) -> list[Any]:
processed_list = []
for el in lst:
el = _Memoizer.check_already_processed(el)
if isinstance(el, (list, tuple)):
new_value = _iter_check_list(el)
elif isinstance(el, dict):
if isinstance(el, dict):
new_value = _iter_check_dict(el)
elif isinstance(el, Iterable) and not isinstance(el, (str, bytes)):
new_value = _iter_check_list(el)
else:
new_value = el
processed_list[i] = new_value
processed_list.append(new_value)
return processed_list

def _iter_check_dict(dct: dict[Any, Any]) -> dict[Any, Any]:
Expand All @@ -281,17 +281,17 @@ def _iter_check_dict(dct: dict[Any, Any]) -> dict[Any, Any]:
k_new = k
if isinstance(v, dict):
new_value = _iter_check_dict(v)
elif isinstance(v, (list, tuple)):
elif isinstance(v, Iterable) and not isinstance(v, (str, bytes)):
new_value = _iter_check_list(v)
else:
new_value = v
processed_dict[k_new] = new_value
return processed_dict

if isinstance(iterable, (list, tuple)):
return _iter_check_list(iterable)
elif isinstance(iterable, dict):
if isinstance(iterable, dict):
return _iter_check_dict(iterable)
elif isinstance(iterable, Iterable):
return _iter_check_list(iterable)
else:
raise TypeError("'iterable' is neither an iterable nor a dictionary.")

Expand Down