diff --git a/manim/renderer/cairo_renderer.py b/manim/renderer/cairo_renderer.py index 93be76803f..bcbe3a4fc7 100644 --- a/manim/renderer/cairo_renderer.py +++ b/manim/renderer/cairo_renderer.py @@ -41,7 +41,7 @@ def __init__( camera_class: type[Camera] | None = None, skip_animations: bool = False, **kwargs: Any, - ) -> None: + ): # All of the following are set to EITHER the value passed via kwargs, # OR the value stored in the global config dict at the time of # _instance construction_. @@ -50,10 +50,10 @@ def __init__( self.camera = camera_cls() self._original_skipping_status = skip_animations self.skip_animations = skip_animations - self.animations_hashes = [] + self.animations_hashes: list[str | None] = [] self.num_plays = 0 - self.time = 0 - self.static_image = None + self.time = 0.0 + self.static_image: PixelArray | None = None def init_scene(self, scene: Scene) -> None: self.file_writer: Any = self._file_writer_class( @@ -65,8 +65,8 @@ def play( self, scene: Scene, *args: Animation | Mobject | _AnimationBuilder, - **kwargs, - ): + **kwargs: Any, + ) -> None: # Reset skip_animations to the original state. # Needed when rendering only some animations, and skipping others. self.skip_animations = self._original_skipping_status @@ -83,6 +83,7 @@ def play( logger.info("Caching disabled.") hash_current_animation = f"uncached_{self.num_plays:05}" else: + assert scene.animations is not None hash_current_animation = get_hash_from_play_call( scene, self.camera, @@ -159,7 +160,12 @@ def update_frame( # TODO Description in Docstring kwargs["include_submobjects"] = include_submobjects self.camera.capture_mobjects(mobjects, **kwargs) - def render(self, scene, time, moving_mobjects): + def render( + self, + scene: Scene, + time: float, + moving_mobjects: Iterable[Mobject] | None = None, + ) -> None: self.update_frame(scene, moving_mobjects) self.add_frame(self.get_frame()) @@ -168,13 +174,13 @@ def get_frame(self) -> PixelArray: Returns ------- - np.array + PixelArray NumPy array of pixel values of each pixel in screen. The shape of the array is height x width x 3. """ return np.array(self.camera.pixel_array) - def add_frame(self, frame: np.ndarray, num_frames: int = 1): + def add_frame(self, frame: PixelArray, num_frames: int = 1) -> None: """Adds a frame to the video_file_stream Parameters @@ -190,7 +196,7 @@ def add_frame(self, frame: np.ndarray, num_frames: int = 1): self.time += num_frames * dt self.file_writer.write_frame(frame, num_frames=num_frames) - def freeze_current_frame(self, duration: float): + def freeze_current_frame(self, duration: float) -> None: """Adds a static frame to the movie for a given duration. The static frame is the current frame. Parameters @@ -204,16 +210,18 @@ def freeze_current_frame(self, duration: float): num_frames=int(duration / dt), ) - def show_frame(self): - """Opens the current frame in the Default Image Viewer of your system.""" - self.update_frame(ignore_skipping=True) + def show_frame(self, scene: Scene) -> None: + """Opens the current frame in the Default Image Viewer + of your system. + """ + self.update_frame(scene, ignore_skipping=True) self.camera.get_image().show() def save_static_frame_data( self, scene: Scene, static_mobjects: Iterable[Mobject], - ) -> Iterable[Mobject] | None: + ) -> PixelArray | None: """Compute and save the static frame, that will be reused at each frame to avoid unnecessarily computing static mobjects. @@ -226,8 +234,8 @@ def save_static_frame_data( Returns ------- - typing.Iterable[Mobject] - The static image computed. + PixelArray | None + The static image computed. The return value is None if there are no static mobjects in the scene. """ self.static_image = None if not static_mobjects: @@ -236,9 +244,8 @@ def save_static_frame_data( self.static_image = self.get_frame() return self.static_image - def update_skipping_status(self): - """ - This method is used internally to check if the current + def update_skipping_status(self) -> None: + """This method is used internally to check if the current animation needs to be skipped or not. It also checks if the number of animations that were played correspond to the number of animations that need to be played, and diff --git a/manim/scene/scene.py b/manim/scene/scene.py index 9ea3e31749..73faf6003a 100644 --- a/manim/scene/scene.py +++ b/manim/scene/scene.py @@ -186,7 +186,7 @@ def __init__( self.moving_mobjects: list[Mobject] = [] self.static_mobjects: list[Mobject] = [] self.time_progression: tqdm[float] | None = None - self.duration: float | None = None + self.duration: float = 0.0 self.last_t = 0.0 self.queue: Queue[SceneInteractAction] = Queue() self.skip_animation_preview = False diff --git a/manim/utils/hashing.py b/manim/utils/hashing.py index be680aef61..d9824ef83f 100644 --- a/manim/utils/hashing.py +++ b/manim/utils/hashing.py @@ -6,10 +6,10 @@ import inspect import json import zlib -from collections.abc import Callable, Hashable, Iterable +from collections.abc import Callable, Hashable, Iterable, Sequence from time import perf_counter from types import FunctionType, MappingProxyType, MethodType, ModuleType -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, overload import numpy as np @@ -54,14 +54,16 @@ class _Memoizer: THRESHOLD_WARNING = 170_000 @classmethod - def reset_already_processed(cls): + def reset_already_processed(cls: type[_Memoizer]) -> None: cls._already_processed.clear() @classmethod - def check_already_processed_decorator(cls: _Memoizer, is_method: bool = False): + def check_already_processed_decorator( + cls: type[_Memoizer], is_method: bool = False + ) -> Callable: """Decorator to handle the arguments that goes through the decorated function. - Returns _ALREADY_PROCESSED_PLACEHOLDER if the obj has been processed, or lets - the decorated function call go ahead. + Returns the value of ALREADY_PROCESSED_PLACEHOLDER if the obj has been processed, + or lets the decorated function call go ahead. Parameters ---------- @@ -69,7 +71,7 @@ def check_already_processed_decorator(cls: _Memoizer, is_method: bool = False): Whether the function passed is a method, by default False. """ - def layer(func): + def layer(func: Callable[[Any], Any]) -> Callable: # NOTE : There is probably a better way to separate both case when func is # a method or a function. if is_method: @@ -82,9 +84,9 @@ def layer(func): return layer @classmethod - def check_already_processed(cls, obj: Any) -> Any: + def check_already_processed(cls: type[_Memoizer], obj: Any) -> Any: """Checks if obj has been already processed. Returns itself if it has not been, - or the value of _ALREADY_PROCESSED_PLACEHOLDER if it has. + or the value of ALREADY_PROCESSED_PLACEHOLDER if it has. Marks the object as processed in the second case. Parameters @@ -101,7 +103,7 @@ def check_already_processed(cls, obj: Any) -> Any: return cls._handle_already_processed(obj, lambda x: x) @classmethod - def mark_as_processed(cls, obj: Any) -> None: + def mark_as_processed(cls: type[_Memoizer], obj: Any) -> None: """Marks an object as processed. Parameters @@ -114,10 +116,10 @@ def mark_as_processed(cls, obj: Any) -> None: @classmethod def _handle_already_processed( - cls, - obj, + cls: type[_Memoizer], + obj: Any, default_function: Callable[[Any], Any], - ): + ) -> str | Any: if isinstance( obj, ( @@ -142,11 +144,11 @@ def _handle_already_processed( @classmethod def _return( - cls, + cls: type[_Memoizer], obj: Any, obj_to_membership_sign: Callable[[Any], int], - default_func, - memoizing=True, + default_func: Callable[[Any], Any], + memoizing: bool = True, ) -> str | Any: obj_membership_sign = obj_to_membership_sign(obj) if obj_membership_sign in cls._already_processed: @@ -172,9 +174,8 @@ def _return( class _CustomEncoder(json.JSONEncoder): - def default(self, obj: Any): - """ - This method is used to serialize objects to JSON format. + def default(self, obj: Any) -> Any: + """This method is used to serialize objects to JSON format. If obj is a function, then it will return a dict with two keys : 'code', for the code source, and 'nonlocals' for all nonlocalsvalues. (including nonlocals @@ -219,7 +220,7 @@ def default(self, obj: Any): if obj.size > 1000: obj = np.resize(obj, (100, 100)) return f"TRUNCATED ARRAY: {repr(obj)}" - # We return the repr and not a list to avoid the JsonEncoder to iterate over it. + # We return the repr and not a list to avoid the JSONEncoder to iterate over it. return repr(obj) elif hasattr(obj, "__dict__"): temp = obj.__dict__ @@ -233,11 +234,17 @@ def default(self, obj: Any): # Serialize it with only the type of the object. You can change this to whatever string when debugging the serialization process. return str(type(obj)) - def _cleaned_iterable(self, iterable: Iterable[Any]): + @overload + def _cleaned_iterable(self, iterable: Sequence[Any]) -> list[Any]: ... + + @overload + def _cleaned_iterable(self, iterable: dict[Any, Any]) -> dict[Any, Any]: ... + + def _cleaned_iterable(self, iterable): """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 byt its hash using the same process implemented here. - If a circular reference is found within the iterable, it will be replaced by the string "already processed". + 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. + If a circular reference is found within the iterable, it will be replaced by the value of ALREADY_PROCESSED_PLACEHOLDER. Parameters ---------- @@ -245,10 +252,10 @@ def _cleaned_iterable(self, iterable: Iterable[Any]): The iterable to check. """ - def _key_to_hash(key): + def _key_to_hash(key: Any) -> int: return zlib.crc32(json.dumps(key, cls=_CustomEncoder).encode()) - def _iter_check_list(lst): + def _iter_check_list(lst: Sequence[Any]) -> list[Any]: processed_list = [None] * len(lst) for i, el in enumerate(lst): el = _Memoizer.check_already_processed(el) @@ -261,13 +268,13 @@ def _iter_check_list(lst): processed_list[i] = new_value return processed_list - def _iter_check_dict(dct): + def _iter_check_dict(dct: dict[Any, Any]) -> dict[Any, Any]: processed_dict = {} for k, v in dct.items(): v = _Memoizer.check_already_processed(v) if k in KEYS_TO_FILTER_OUT: continue - # We check if the k is of the right format (supporter by Json) + # We check if the k is of the right format (supported by JSON) if not isinstance(k, (str, int, float, bool)) and k is not None: k_new = _key_to_hash(k) else: @@ -285,8 +292,10 @@ def _iter_check_dict(dct): return _iter_check_list(iterable) elif isinstance(iterable, dict): return _iter_check_dict(iterable) + else: + raise TypeError("'iterable' is neither an iterable nor a dictionary.") - def encode(self, obj: Any): + def encode(self, obj: Any) -> str: """Overriding of :meth:`JSONEncoder.encode`, to make our own process. Parameters @@ -305,7 +314,7 @@ def encode(self, obj: Any): return super().encode(obj) -def get_json(obj: dict): +def get_json(obj: Any) -> str: """Recursively serialize `object` to JSON using the :class:`CustomEncoder` class. Parameters diff --git a/mypy.ini b/mypy.ini index 27017a7afa..0d93a46c37 100644 --- a/mypy.ini +++ b/mypy.ini @@ -126,9 +126,6 @@ ignore_errors = True [mypy-manim.mobject.vector_field] ignore_errors = True -[mypy-manim.renderer.cairo_renderer] -ignore_errors = True - [mypy-manim.renderer.opengl_renderer] ignore_errors = True