Repository: ManimCommunity/manim
File: manim/scene/scene_file_writer.py (Python)

Code before:

    def end_animation(self, allow_write: bool = False):
        """
        Internally used by Manim to stop streaming to
        FFMPEG gracefully.

        Parameters
        ----------
        allow_write
            Whether or not to write to a video file.
        """
        if write_to_movie() and allow_write:
            self.close_partial_movie_stream()

    def listen_and_write(self):
        """
        For internal use only: blocks until new frame is available on the queue.
        """
        while True:
            num_frames, frame_data = self.queue.get()
            if frame_data is None:
                break

            self.encode_and_write_frame(frame_data, num_frames)

    def encode_and_write_frame(self, frame: np.ndarray, num_frames: int) -> None:
        """
        For internal use only: takes a given frame in `np.ndarray` format and
        write it to the stream
        """
        for _ in range(num_frames):
            # Notes: precomputing reusing packets does not work!
            # I.e., you cannot do `packets = encode(...)`
            # and reuse it, as it seems that `mux(...)`
            # consumes the packet.
            # The same issue applies for `av_frame`,
            # reusing it renders weird-looking frames.
            av_frame = av.VideoFrame.from_ndarray(frame, format="rgba")
            for packet in self.video_stream.encode(av_frame):
                self.video_container.mux(packet)

    def write_frame(
        self, frame_or_renderer: np.ndarray | OpenGLRenderer, num_frames: int = 1
    ):
        """
        Used internally by Manim to write a frame to
        the FFMPEG input buffer.

        Parameters
        ----------
        frame_or_renderer

Review comment (suggestion):
Perfect time to fix some typehints!
```suggestion
    def encode_and_write_frame(self, frame: PixelArray, num_frames: int) -> None:
```
(`PixelArray` is from `manim.typing` - [docs](https://docs.manim.community/en/latest/reference/manim.typing.html#manim.typing.PixelArray))

Code after:
            self.open_partial_movie_stream(file_path=file_path)

    def end_animation(self, allow_write: bool = False):
        """
        Internally used by Manim to stop streaming to
        FFMPEG gracefully.

        Parameters
        ----------
        allow_write
            Whether or not to write to a video file.
        """
        if write_to_movie() and allow_write:
            self.close_partial_movie_stream()

    def listen_and_write(self):
        """
        For internal use only: blocks until new frame is available on the queue.
        """
        while True:
            num_frames, frame_data = self.queue.get()
            if frame_data is None:
                break

            self.encode_and_write_frame(frame_data, num_frames)

    def encode_and_write_frame(self, frame: PixelArray, num_frames: int) -> None:
        """
        For internal use only: takes a given frame in ``np.ndarray`` format and
        write it to the stream
        """
        for _ in range(num_frames):
            # Notes: precomputing reusing packets does not work!
            # I.e., you cannot do `packets = encode(...)`
            # and reuse it, as it seems that `mux(...)`
            # consumes the packet.
            # The same issue applies for `av_frame`,
            # reusing it renders weird-looking frames.
            av_frame = av.VideoFrame.from_ndarray(frame, format="rgba")
            for packet in self.video_stream.encode(av_frame):
                self.video_container.mux(packet)

    def write_frame(
        self, frame_or_renderer: np.ndarray | OpenGLRenderer, num_frames: int = 1
    ):
        """
        Used internally by Manim to write a frame to
        the FFMPEG input buffer.

        Parameters
        ----------