def __call__(self, x: np.ndarray, y: np.ndarray | None = None) -> tuple[np.ndarray, np.ndarray | None]:
        """
        Apply video compression to sample `x`.

        :param x: Sample to compress of shape NCFHW or NFHWC. `x` values are expected to be either in range [0, 1] or
                  [0, 255].
        :param y: Labels of the sample `x`. This function does not affect them in any way.
        :return: Compressed sample.
        """

        def compress_video(x: np.ndarray, video_format: str, constant_rate_factor: int, dir_: str = ""):
            """
            Apply video compression to video input of shape (frames, height, width, channel).
            """
            import ffmpeg

            video_path = os.path.join(dir_, f"tmp_video.{video_format}")
            _, height, width, _ = x.shape

            if (height % 2) != 0 or (width % 2) != 0:
                warnings.warn("Codec might require even number of pixels in height and width.")

            # numpy to local video file
            process = (
                ffmpeg.input("pipe:", format="rawvideo", pix_fmt="rgb24", s=f"{width}x{height}")
                .output(video_path, pix_fmt="yuv420p", vcodec="libx264", crf=constant_rate_factor)
                .overwrite_output()
                .run_async(pipe_stdin=True, quiet=True)
            )
            process.stdin.write(x.flatten().astype(np.uint8).tobytes())
            process.stdin.close()
            process.wait()

            # local video file to numpy
            stdout, _ = (
                ffmpeg.input(video_path)
                .output("pipe:", format="rawvideo", pix_fmt="rgb24")
                .run(capture_stdout=True, quiet=True)
            )
            return np.frombuffer(stdout, np.uint8).reshape(x.shape)

        if x.ndim != 5:
            raise ValueError("Video compression can only be applied to spatio-temporal data.")

        if self.channels_first:
            x = np.transpose(x, (0, 2, 3, 4, 1))

        # apply video compression per video item
        scale = 1
        if x.min() >= 0 and x.max() <= 1.0:
            scale = 255

        x_compressed = x.copy()
        with TemporaryDirectory(dir=config.ART_DATA_PATH) as tmp_dir:
            for i, x_i in enumerate(tqdm(x, desc="Video compression", disable=not self.verbose)):
                x_i *= scale
                x_compressed[i] = compress_video(x_i, self.video_format, self.constant_rate_factor, dir_=tmp_dir)

        x_compressed = x_compressed / scale
        x_compressed = x_compressed.astype(x.dtype)

        if self.channels_first:
            x_compressed = np.transpose(x_compressed, (0, 4, 1, 2, 3))

        return x_compressed, y

        Apply video compression to sample `x`.

        :param x: Sample to compress of shape NCFHW or NFHWC. `x` values are expected to be either in range [0, 1] or
                  [0, 255].
        :param y: Labels of the sample `x`. This function does not affect them in any way.
        :return: Compressed sample.
        
art/defences/preprocessor/video_compression.py