class_docstr: Implement FFmpeg wrapper for video compression defence based on H.264/MPEG-4 AVC.
  
  Video compression uses H.264 video encoding. The video quality is controlled with the constant rate factor
  parameter. More information on the constant rate factor: https://trac.ffmpeg.org/wiki/Encode/H.264.
class_docstr_tok: ["Implement","FFmpeg","wrapper","for","video","compression","defence","based","on","H.264\/MPEG-4","AVC",".","Video","compression","uses","H.264","video","encoding",".","The","video","quality","is","controlled","with","the","constant","rate","factor","parameter",".","More","information","on","the","constant","rate","factor",":","https",":","\/\/trac.ffmpeg.org\/wiki\/Encode\/H.264","."]
docstr: 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.
docstr_tok: ["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","."]
code: def __call__(
      self, x: np.ndarray, y: Optional[np.ndarray] = None
  ) -> Tuple[np.ndarray, Optional[np.ndarray]]:
      """
      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
  
file_path: adversarial-robustness-toolbox/art/defences/preprocessor/video_compression.py
imports: from __future__ import absolute_import, division, print_function, unicode_literals
  import logging
  import os
  from tempfile import TemporaryDirectory
  from typing import Optional, Tuple
  import warnings
  import numpy
  from tqdm.auto import tqdm
  from art import config
  from art.defences.preprocessor.preprocessor import Preprocessor
label_desc: Use image node_id 2 for calling the VideoCompression obj's underlying member method code with example usage: obj.__call__(x, y) and returns: x_compressed,  y, np