code: import os
  import warnings
  from pathlib import Path
  from typing import Optional, Union
  
  from .dynamic_module_utils import custom_object_save
  from .tokenization_utils_base import PreTrainedTokenizerBase
  from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging
  
  
  UpperCAmelCase					=				logging.get_logger(__name__)
  
  # Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
  UpperCAmelCase					=				direct_transformers_import(Path(__file__).parent)
  
  
  UpperCAmelCase					=				{
      '''AutoTokenizer''': '''PreTrainedTokenizerBase''',
      '''AutoFeatureExtractor''': '''FeatureExtractionMixin''',
      '''AutoImageProcessor''': '''ImageProcessingMixin''',
  }
  
  
  
  
  class    __lowerCAmelCase  (       a_						):
  
  
  
   '''simple docstring'''
   _lowerCamelCase						:	Tuple				    = ['feature_extractor', 'tokenizer']
   # Names need to be attr_class for attr in attributes
   _lowerCamelCase						:	List[str]				    = None
   _lowerCamelCase						:	Union[str, Any]				    = None
   _lowerCamelCase						:	Optional[Any]				    = None
  
   def __init__(			self       :		Dict       ,			*lowerCamelCase       :		Union[str, Any]       ,			**lowerCamelCase       :		Union[str, Any]  ) ->							List[Any]:
          # Sanitize args and kwargs
          for key in kwargs:
                 if key not in self.attributes:
                        raise TypeError(f"Unexpected keyword argument {key}."  )
          for arg, attribute_name in zip(lowerCamelCase       ,			self.attributes  ):
                 if attribute_name in kwargs:
                        raise TypeError(f"Got multiple values for argument {attribute_name}."  )
                 else:
                        _UpperCamelCase     :       int  						=  arg
  
          if len(lowerCamelCase  ) != len(self.attributes  ):
                 raise ValueError(
                     f"This processor requires {len(self.attributes  )} arguments: {', '.join(self.attributes  )}. Got "
                     f"{len(lowerCamelCase  )} arguments instead."  )
  
          # Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
          for attribute_name, arg in kwargs.items():
                 _UpperCamelCase     :       List[Any]  						=  getattr(self       ,			f"{attribute_name}_class"  )
                 # Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
                 _UpperCamelCase     :       Optional[Any]  						=  AUTO_TO_BASE_CLASS_MAPPING.get(lowerCamelCase       ,			lowerCamelCase  )
                 if isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                        _UpperCamelCase     :       str  						=  tuple(getattr(lowerCamelCase       ,			lowerCamelCase  ) for n in class_name if n is not None  )
                 else:
                        _UpperCamelCase     :       Optional[Any]  						=  getattr(lowerCamelCase       ,			lowerCamelCase  )
  
                 if not isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                        raise ValueError(
                            f"Received a {type(lowerCamelCase  ).__name__} for argument {attribute_name}, but a {class_name} was expected."  )
  
                 setattr(self       ,			lowerCamelCase       ,			lowerCamelCase  )
  
   def __repr__(			self       :		Union[str, Any]  ) ->							str:
          _UpperCamelCase     :       int  						=  [f"- {name}: {repr(getattr(self       ,			lowerCamelCase  )  )}" for name in self.attributes]
          _UpperCamelCase     :       List[Any]  						=  '\n'.join(lowerCamelCase  )
          return f"{self.__class__.__name__}:\n{attributes_repr}"
  
   def 					__snake_case					(			self       :		str       ,			lowerCamelCase       :		Dict       ,			lowerCamelCase       :		bool = False       ,			**lowerCamelCase       :		str  ) ->							Union[str, Any]:
          os.makedirs(lowerCamelCase       ,			exist_ok=lowerCamelCase  )
  
          if push_to_hub:
                 _UpperCamelCase     :       int  						=  kwargs.pop('commit_message'       ,			lowerCamelCase  )
                 _UpperCamelCase     :       Any  						=  kwargs.pop('repo_id'       ,			save_directory.split(os.path.sep  )[-1]  )
                 _UpperCamelCase     :       List[Any]  						=  self._create_repo(lowerCamelCase       ,			**lowerCamelCase  )
                 _UpperCamelCase     :       List[Any]  						=  self._get_files_timestamps(lowerCamelCase  )
          # If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
          # loaded from the Hub.
          if self._auto_class is not None:
                 _UpperCamelCase     :       Dict  						=  [getattr(self       ,			lowerCamelCase  ) for attribute_name in self.attributes]
                 _UpperCamelCase     :       Any  						=  [(a.init_kwargs if isinstance(lowerCamelCase       ,			lowerCamelCase  ) else a) for a in attrs]
                 custom_object_save(self       ,			lowerCamelCase       ,			config=lowerCamelCase  )
  
          for attribute_name in self.attributes:
                 _UpperCamelCase     :       int  						=  getattr(self       ,			lowerCamelCase  )
                 # Include the processor class in the attribute config so this processor can then be reloaded with the
                 # `AutoProcessor` API.
                 if hasattr(lowerCamelCase       ,			'_set_processor_class'  ):
                        attribute._set_processor_class(self.__class__.__name__  )
                 attribute.save_pretrained(lowerCamelCase  )
  
          if self._auto_class is not None:
                 # We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
                 for attribute_name in self.attributes:
                        _UpperCamelCase     :       List[Any]  						=  getattr(self       ,			lowerCamelCase  )
                        if isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                               del attribute.init_kwargs["auto_map"]
  
          if push_to_hub:
                 self._upload_modified_files(
                     lowerCamelCase       ,			lowerCamelCase       ,			lowerCamelCase       ,			commit_message=lowerCamelCase       ,			token=kwargs.get('use_auth_token'  )       ,			)
  
   @classmethod
   def 					__snake_case					(			cls       :		Union[str, Any]       ,			lowerCamelCase       :		Union[str, os.PathLike]       ,			lowerCamelCase       :		Optional[Union[str, os.PathLike]] = None       ,			lowerCamelCase       :		bool = False       ,			lowerCamelCase       :		bool = False       ,			lowerCamelCase       :		Optional[Union[str, bool]] = None       ,			lowerCamelCase       :		str = "main"       ,			**lowerCamelCase       :		List[Any]       ,			) ->							Any:
          _UpperCamelCase     :       Dict  						=  cache_dir
          _UpperCamelCase     :       int  						=  force_download
          _UpperCamelCase     :       str  						=  local_files_only
          _UpperCamelCase     :       Optional[int]  						=  revision
  
          _UpperCamelCase     :       Any  						=  kwargs.pop('use_auth_token'       ,			lowerCamelCase  )
          if use_auth_token is not None:
                 warnings.warn(
                     'The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers.'       ,			lowerCamelCase  )
                 if token is not None:
                        raise ValueError(
                            '`token` and `use_auth_token` are both specified. Please set only the argument `token`.'  )
                 _UpperCamelCase     :       Optional[Any]  						=  use_auth_token
  
          if token is not None:
                 # change to `token` in a follow-up PR
                 _UpperCamelCase     :       Any  						=  token
  
          _UpperCamelCase     :       Any  						=  cls._get_arguments_from_pretrained(lowerCamelCase       ,			**lowerCamelCase  )
          return cls(*lowerCamelCase  )
  
   @classmethod
   def 					__snake_case					(			cls       :		Optional[Any]       ,			lowerCamelCase       :		Union[str, Any]="AutoProcessor"  ) ->							List[str]:
          if not isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                 _UpperCamelCase     :       Optional[Any]  						=  auto_class.__name__
  
          import transformers.models.auto as auto_module
  
          if not hasattr(lowerCamelCase       ,			lowerCamelCase  ):
                 raise ValueError(f"{auto_class} is not a valid auto class."  )
  
          _UpperCamelCase     :       Optional[int]  						=  auto_class
  
   @classmethod
   def 					__snake_case					(			cls       :		Dict       ,			lowerCamelCase       :		List[str]       ,			**lowerCamelCase       :		List[Any]  ) ->							Dict:
          _UpperCamelCase     :       List[Any]  						=  []
          for attribute_name in cls.attributes:
                 _UpperCamelCase     :       Any  						=  getattr(cls       ,			f"{attribute_name}_class"  )
                 if isinstance(lowerCamelCase       ,			lowerCamelCase  ):
                        _UpperCamelCase     :       Optional[int]  						=  tuple(getattr(lowerCamelCase       ,			lowerCamelCase  ) if n is not None else None for n in class_name  )
                        _UpperCamelCase     :       Any  						=  kwargs.get('use_fast'       ,			lowerCamelCase  )
                        if use_fast and classes[1] is not None:
                               _UpperCamelCase     :       List[str]  						=  classes[1]
                        else:
                               _UpperCamelCase     :       Tuple  						=  classes[0]
                 else:
                        _UpperCamelCase     :       str  						=  getattr(lowerCamelCase       ,			lowerCamelCase  )
  
                 args.append(attribute_class.from_pretrained(lowerCamelCase       ,			**lowerCamelCase  )  )
          return args
  
   @property
   def 					__snake_case					(			self       :		List[Any]  ) ->							int:
          _UpperCamelCase     :       List[Any]  						=  getattr(self       ,			self.attributes[0]  )
          return getattr(lowerCamelCase       ,			'model_input_names'       ,			lowerCamelCase  )
  
  
  
  
  
  
  
  UpperCAmelCase					=				copy_func(ProcessorMixin.push_to_hub)
  if ProcessorMixin.push_to_hub.__doc__ is not None:
   UpperCAmelCase					=				ProcessorMixin.push_to_hub.__doc__.format(
       object='''processor''', object_class='''AutoProcessor''', object_files='''processor files'''
   )
  
  
  
  
style_context: import unittest
  
  import numpy as np
  
  from transformers.testing_utils import require_torch, require_vision
  from transformers.utils import is_torch_available, is_vision_available
  
  from ...test_image_processing_common import ImageProcessingSavingTestMixin
  
  
  if is_torch_available():
   import torch
  
  if is_vision_available():
   from PIL import Image
  
   from transformers import TvltImageProcessor
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	Tuple		,						lowercase_:	Optional[int]=10		,						lowercase_:	Optional[Any]=10		,						lowercase_:	Optional[Any]=False		,						lowercase_:	Any=False				):
         _UpperCamelCase     :       List[Any]  						=  []
         for i in range(image_processor_tester.num_frames				):
                video.append(np.random.randint(255		,						size=(image_processor_tester.num_channels, width, height)		,						dtype=np.uinta				)				)
  
         if not numpify and not torchify:
                # PIL expects the channel dimension as last dimension
                _UpperCamelCase     :       Tuple  						=  [Image.fromarray(np.moveaxis(lowercase_		,						0		,						-1				)				) for frame in video]
  
         if torchify:
                _UpperCamelCase     :       Union[str, Any]  						=  [torch.from_numpy(lowercase_				) for frame in video]
  
         return video
  
  
  def SCREAMING_SNAKE_CASE				(     lowercase_:	List[str]		,						lowercase_:	Optional[Any]=False		,						lowercase_:	Tuple=False		,						lowercase_:	Union[str, Any]=False				):
         assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"
  
         _UpperCamelCase     :       Any  						=  []
         for i in range(image_processor_tester.batch_size				):
                if equal_resolution:
                       _UpperCamelCase     :       str  						=  image_processor_tester.max_resolution
                else:
                       _UpperCamelCase						,      _UpperCamelCase     :       int  						=  np.random.choice(
                           np.arange(image_processor_tester.min_resolution		,						image_processor_tester.max_resolution				)		,						2				)
                       _UpperCamelCase     :       Optional[Any]  						=  prepare_video(
                           image_processor_tester=lowercase_		,						width=lowercase_		,						height=lowercase_		,						numpify=lowercase_		,						torchify=lowercase_		,						)
                video_inputs.append(lowercase_				)
  
         return video_inputs
  
  
  
  
  class    __lowerCAmelCase  (       unittest.TestCase						):
  
  
  
   '''simple docstring'''
  
   def __init__(			self       :		Optional[Any]       ,			lowerCamelCase       :		Union[str, Any]       ,			lowerCamelCase       :		Optional[int]=7       ,			lowerCamelCase       :		List[Any]=3       ,			lowerCamelCase       :		str=4       ,			lowerCamelCase       :		Union[str, Any]=1_8       ,			lowerCamelCase       :		str=3_0       ,			lowerCamelCase       :		List[str]=4_0_0       ,			lowerCamelCase       :		Tuple=True       ,			lowerCamelCase       :		Union[str, Any]=None       ,			lowerCamelCase       :		List[str]=True       ,			lowerCamelCase       :		Optional[Any]=[0.5, 0.5, 0.5]       ,			lowerCamelCase       :		Any=[0.5, 0.5, 0.5]       ,			lowerCamelCase       :		str=True       ,			lowerCamelCase       :		Optional[Any]=None       ,			) ->							Union[str, Any]:
          _UpperCamelCase     :       Optional[int]  						=  size if size is not None else {'shortest_edge': 1_8}
          _UpperCamelCase     :       Dict  						=  crop_size if crop_size is not None else {'height': 1_8, 'width': 1_8}
  
          _UpperCamelCase     :       Any  						=  parent
          _UpperCamelCase     :       Optional[Any]  						=  batch_size
          _UpperCamelCase     :       List[Any]  						=  num_channels
          _UpperCamelCase     :       List[Any]  						=  num_frames
          _UpperCamelCase     :       Union[str, Any]  						=  image_size
          _UpperCamelCase     :       Dict  						=  min_resolution
          _UpperCamelCase     :       int  						=  max_resolution
          _UpperCamelCase     :       Union[str, Any]  						=  do_resize
          _UpperCamelCase     :       str  						=  size
          _UpperCamelCase     :       Dict  						=  do_normalize
          _UpperCamelCase     :       Union[str, Any]  						=  image_mean
          _UpperCamelCase     :       Optional[int]  						=  image_std
          _UpperCamelCase     :       Tuple  						=  do_center_crop
          _UpperCamelCase     :       Any  						=  crop_size
  
   def 					__snake_case					(			self       :		Tuple  ) ->							Optional[Any]:
          return {
              "image_mean": self.image_mean,
              "image_std": self.image_std,
              "do_normalize": self.do_normalize,
              "do_resize": self.do_resize,
              "size": self.size,
              "do_center_crop": self.do_center_crop,
              "crop_size": self.crop_size,
          }
  
  
  
  
  
  
  
  @require_torch
  @require_vision
  class    __lowerCAmelCase  (       a_,	unittest.TestCase						):
  
  
  
   '''simple docstring'''
   _lowerCamelCase						:	Any				    = TvltImageProcessor if is_vision_available() else None
  
   def 					__snake_case					(			self       :		int  ) ->							Optional[int]:
          _UpperCamelCase     :       Optional[Any]  						=  TvltImageProcessorTester(self  )
  
   @property
   def 					__snake_case					(			self       :		Optional[Any]  ) ->							Any:
          return self.image_processor_tester.prepare_image_processor_dict()
  
   def 					__snake_case					(			self       :		Union[str, Any]  ) ->							str:
          _UpperCamelCase     :       Tuple  						=  self.image_processing_class(**self.image_processor_dict  )
          self.assertTrue(hasattr(lowerCamelCase       ,			'image_mean'  )  )
          self.assertTrue(hasattr(lowerCamelCase       ,			'image_std'  )  )
          self.assertTrue(hasattr(lowerCamelCase       ,			'do_normalize'  )  )
          self.assertTrue(hasattr(lowerCamelCase       ,			'do_resize'  )  )
          self.assertTrue(hasattr(lowerCamelCase       ,			'do_center_crop'  )  )
          self.assertTrue(hasattr(lowerCamelCase       ,			'size'  )  )
  
   def 					__snake_case					(			self       :		Dict  ) ->							Optional[int]:
          # Initialize image_processor
          _UpperCamelCase     :       str  						=  self.image_processing_class(**self.image_processor_dict  )
          # create random PIL videos
          _UpperCamelCase     :       int  						=  prepare_video_inputs(self.image_processor_tester       ,			equal_resolution=lowerCamelCase  )
          for video in video_inputs:
                 self.assertIsInstance(lowerCamelCase       ,			lowerCamelCase  )
                 self.assertIsInstance(video[0]       ,			Image.Image  )
  
          # Test not batched input
          _UpperCamelCase     :       Dict  						=  image_processor(video_inputs[0]       ,			return_tensors='pt'  ).pixel_values
          self.assertEqual(
              encoded_videos.shape       ,			(
                  1,
                  self.image_processor_tester.num_frames,
                  self.image_processor_tester.num_channels,
                  self.image_processor_tester.crop_size['height'],
                  self.image_processor_tester.crop_size['width'],
              )       ,			)
  
          # Test batched
          _UpperCamelCase     :       Any  						=  image_processor(lowerCamelCase       ,			return_tensors='pt'  ).pixel_values
          self.assertEqual(
              encoded_videos.shape       ,			(
                  self.image_processor_tester.batch_size,
                  self.image_processor_tester.num_frames,
                  self.image_processor_tester.num_channels,
                  self.image_processor_tester.crop_size['height'],
                  self.image_processor_tester.crop_size['width'],
              )       ,			)
  
   def 					__snake_case					(			self       :		Union[str, Any]  ) ->							List[str]:
          # Initialize image_processor
          _UpperCamelCase     :       Optional[int]  						=  self.image_processing_class(**self.image_processor_dict  )
          # create random numpy tensors
          _UpperCamelCase     :       int  						=  prepare_video_inputs(self.image_processor_tester       ,			equal_resolution=lowerCamelCase       ,			numpify=lowerCamelCase  )
          for video in video_inputs:
                 self.assertIsInstance(lowerCamelCase       ,			lowerCamelCase  )
                 self.assertIsInstance(video[0]       ,			np.ndarray  )
  
          # Test not batched input
          _UpperCamelCase     :       Optional[int]  						=  image_processor(video_inputs[0]       ,			return_tensors='pt'  ).pixel_values
          self.assertEqual(
              encoded_videos.shape       ,			(
                  1,
                  self.image_processor_tester.num_frames,
                  self.image_processor_tester.num_channels,
                  self.image_processor_tester.crop_size['height'],
                  self.image_processor_tester.crop_size['width'],
              )       ,			)
  
          # Test batched
          _UpperCamelCase     :       int  						=  image_processor(lowerCamelCase       ,			return_tensors='pt'  ).pixel_values
          self.assertEqual(
              encoded_videos.shape       ,			(
                  self.image_processor_tester.batch_size,
                  self.image_processor_tester.num_frames,
                  self.image_processor_tester.num_channels,
                  self.image_processor_tester.crop_size['height'],
                  self.image_processor_tester.crop_size['width'],
              )       ,			)
  
   def 					__snake_case					(			self       :		Optional[Any]  ) ->							Any:
          # Initialize image_processor
          _UpperCamelCase     :       int  						=  self.image_processing_class(**self.image_processor_dict  )
          # create random PyTorch tensors
          _UpperCamelCase     :       Optional[Any]  						=  prepare_video_inputs(self.image_processor_tester       ,			equal_resolution=lowerCamelCase       ,			torchify=lowerCamelCase  )
          for video in video_inputs:
                 self.assertIsInstance(lowerCamelCase       ,			lowerCamelCase  )
                 self.assertIsInstance(video[0]       ,			torch.Tensor  )
  
          # Test not batched input
          _UpperCamelCase     :       Dict  						=  image_processor(video_inputs[0]       ,			return_tensors='pt'  ).pixel_values
          self.assertEqual(
              encoded_videos.shape       ,			(
                  1,
                  self.image_processor_tester.num_frames,
                  self.image_processor_tester.num_channels,
                  self.image_processor_tester.crop_size['height'],
                  self.image_processor_tester.crop_size['width'],
              )       ,			)
  
          # Test batched
          _UpperCamelCase     :       Optional[int]  						=  image_processor(lowerCamelCase       ,			return_tensors='pt'  ).pixel_values
          self.assertEqual(
              encoded_videos.shape       ,			(
                  self.image_processor_tester.batch_size,
                  self.image_processor_tester.num_frames,
                  self.image_processor_tester.num_channels,
                  self.image_processor_tester.crop_size['height'],
                  self.image_processor_tester.crop_size['width'],
              )       ,			)
  
  
  
  