

'''simple docstring'''
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


_snake_case     						=						logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
_snake_case     						=						direct_transformers_import(Path(__file__).parent)


_snake_case     						=						{
    'AutoTokenizer': 'PreTrainedTokenizerBase',
    'AutoFeatureExtractor': 'FeatureExtractionMixin',
    'AutoImageProcessor': 'ImageProcessingMixin',
}


class      a__						(							lowerCamelCase_						):
   _SCREAMING_SNAKE_CASE					:  Optional[Any]			 =							['feature_extractor', 'tokenizer']
   # Names need to be attr_class for attr in attributes
   _SCREAMING_SNAKE_CASE					:  Union[str, Any]			 =							None
   _SCREAMING_SNAKE_CASE					:  Any			 =							None
   _SCREAMING_SNAKE_CASE					:  str			 =							None


   def __init__(				self    ,       *_UpperCamelCase    ,       **_UpperCamelCase						):





      """simple docstring"""





      for key in kwargs:
         if key not in self.attributes:
            raise TypeError(f'''Unexpected keyword argument {key}.'''						)
      for arg, attribute_name in zip(_UpperCamelCase    ,       self.attributes						):
         if attribute_name in kwargs:
            raise TypeError(f'''Got multiple values for argument {attribute_name}.'''						)
         else:
            _lowercase    :   List[Any]      =     arg

      if len(_UpperCamelCase						) != len(self.attributes						):
         raise ValueError(
             f'''This processor requires {len(self.attributes						)} arguments: {', '.join(self.attributes						)}. Got '''
             f'''{len(_UpperCamelCase						)} 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():
         _lowercase    :   str      =     getattr(self    ,       f'''{attribute_name}_class'''						)
         # Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
         _lowercase    :   List[Any]      =     AUTO_TO_BASE_CLASS_MAPPING.get(_UpperCamelCase    ,       _UpperCamelCase						)
         if isinstance(_UpperCamelCase    ,       _UpperCamelCase						):
            _lowercase    :   int      =     tuple(getattr(_UpperCamelCase    ,       _UpperCamelCase						) for n in class_name if n is not None						)
         else:
            _lowercase    :   List[Any]      =     getattr(_UpperCamelCase    ,       _UpperCamelCase						)

         if not isinstance(_UpperCamelCase    ,       _UpperCamelCase						):
            raise ValueError(
                f'''Received a {type(_UpperCamelCase						).__name__} for argument {attribute_name}, but a {class_name} was expected.'''						)

         setattr(self    ,       _UpperCamelCase    ,       _UpperCamelCase						)


   def __repr__(				self						):





      """simple docstring"""





      _lowercase    :   List[Any]      =     [f'''- {name}: {repr(getattr(self    ,       _UpperCamelCase						)						)}''' for name in self.attributes]
      _lowercase    :   Optional[int]      =     "\n".join(_UpperCamelCase						)
      return f'''{self.__class__.__name__}:\n{attributes_repr}'''


   def 		_lowerCamelCase     (				self    ,       _UpperCamelCase    ,       _UpperCamelCase = False    ,       **_UpperCamelCase						):





      """simple docstring"""





      os.makedirs(_UpperCamelCase    ,       exist_ok=_UpperCamelCase						)

      if push_to_hub:
         _lowercase    :   Dict      =     kwargs.pop("commit_message"    ,       _UpperCamelCase						)
         _lowercase    :   Union[str, Any]      =     kwargs.pop("repo_id"    ,       save_directory.split(os.path.sep						)[-1]						)
         _lowercase    :   List[Any]      =     self._create_repo(_UpperCamelCase    ,       **_UpperCamelCase						)
         _lowercase    :   List[Any]      =     self._get_files_timestamps(_UpperCamelCase						)
      # 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:
         _lowercase    :   Optional[int]      =     [getattr(self    ,       _UpperCamelCase						) for attribute_name in self.attributes]
         _lowercase    :   Optional[int]      =     [(a.init_kwargs if isinstance(_UpperCamelCase    ,       _UpperCamelCase						) else a) for a in attrs]
         custom_object_save(self    ,       _UpperCamelCase    ,       config=_UpperCamelCase						)

      for attribute_name in self.attributes:
         _lowercase    :   List[str]      =     getattr(self    ,       _UpperCamelCase						)
         # Include the processor class in the attribute config so this processor can then be reloaded with the
         # `AutoProcessor` API.
         if hasattr(_UpperCamelCase    ,       "_set_processor_class"						):
            attribute._set_processor_class(self.__class__.__name__						)
         attribute.save_pretrained(_UpperCamelCase						)

      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:
            _lowercase    :   Any      =     getattr(self    ,       _UpperCamelCase						)
            if isinstance(_UpperCamelCase    ,       _UpperCamelCase						):
               del attribute.init_kwargs["auto_map"]

      if push_to_hub:
         self._upload_modified_files(
             _UpperCamelCase    ,       _UpperCamelCase    ,       _UpperCamelCase    ,       commit_message=_UpperCamelCase    ,       token=kwargs.get("use_auth_token"						)    ,       )


   @classmethod
   def 		_lowerCamelCase     (				cls    ,       _UpperCamelCase    ,       _UpperCamelCase = None    ,       _UpperCamelCase = False    ,       _UpperCamelCase = False    ,       _UpperCamelCase = None    ,       _UpperCamelCase = "main"    ,       **_UpperCamelCase    ,       ):





      """simple docstring"""





      _lowercase    :   Optional[Any]      =     cache_dir
      _lowercase    :   Any      =     force_download
      _lowercase    :   Tuple      =     local_files_only
      _lowercase    :   List[Any]      =     revision

      _lowercase    :   List[str]      =     kwargs.pop("use_auth_token"    ,       _UpperCamelCase						)
      if use_auth_token is not None:
         warnings.warn(
             "The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers."    ,       _UpperCamelCase						)
         if token is not None:
            raise ValueError(
                "`token` and `use_auth_token` are both specified. Please set only the argument `token`."						)
         _lowercase    :   Tuple      =     use_auth_token

      if token is not None:
         # change to `token` in a follow-up PR
         _lowercase    :   List[Any]      =     token

      _lowercase    :   Tuple      =     cls._get_arguments_from_pretrained(_UpperCamelCase    ,       **_UpperCamelCase						)
      return cls(*_UpperCamelCase						)


   @classmethod
   def 		_lowerCamelCase     (				cls    ,       _UpperCamelCase="AutoProcessor"						):





      """simple docstring"""





      if not isinstance(_UpperCamelCase    ,       _UpperCamelCase						):
         _lowercase    :   Union[str, Any]      =     auto_class.__name__

      import transformers.models.auto as auto_module

      if not hasattr(_UpperCamelCase    ,       _UpperCamelCase						):
         raise ValueError(f'''{auto_class} is not a valid auto class.'''						)

      _lowercase    :   Any      =     auto_class


   @classmethod
   def 		_lowerCamelCase     (				cls    ,       _UpperCamelCase    ,       **_UpperCamelCase						):





      """simple docstring"""





      _lowercase    :   Dict      =     []
      for attribute_name in cls.attributes:
         _lowercase    :   str      =     getattr(cls    ,       f'''{attribute_name}_class'''						)
         if isinstance(_UpperCamelCase    ,       _UpperCamelCase						):
            _lowercase    :   Optional[int]      =     tuple(getattr(_UpperCamelCase    ,       _UpperCamelCase						) if n is not None else None for n in class_name						)
            _lowercase    :   Dict      =     kwargs.get("use_fast"    ,       _UpperCamelCase						)
            if use_fast and classes[1] is not None:
               _lowercase    :   int      =     classes[1]
            else:
               _lowercase    :   int      =     classes[0]
         else:
            _lowercase    :   Optional[int]      =     getattr(_UpperCamelCase    ,       _UpperCamelCase						)

         args.append(attribute_class.from_pretrained(_UpperCamelCase    ,       **_UpperCamelCase						)						)
      return args






   @property
   def 		_lowerCamelCase     (				self						):





      """simple docstring"""





      _lowercase    :   List[Any]      =     getattr(self    ,       self.attributes[0]						)
      return getattr(_UpperCamelCase    ,       "model_input_names"    ,       _UpperCamelCase						)




_snake_case     						=						copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
    _snake_case     						=						ProcessorMixin.push_to_hub.__doc__.format(
        object='processor', object_class='AutoProcessor', object_files='processor files'
    )









'''simple docstring'''
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       _A	(			snake_case	,							snake_case=10	,							snake_case=10	,							snake_case=False	,							snake_case=False  )    ->			List[Any]:
   _lowercase    :   Dict      =     []
   for i in range(image_processor_tester.num_frames  ):
      video.append(np.random.randint(2_55	,							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
      _lowercase    :   Union[str, Any]      =     [Image.fromarray(np.moveaxis(snake_case	,							0	,							-1  )  ) for frame in video]

   if torchify:
      _lowercase    :   Any      =     [torch.from_numpy(snake_case  ) for frame in video]

   return video







def       _A	(			snake_case	,							snake_case=False	,							snake_case=False	,							snake_case=False  )    ->			Tuple:
   assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

   _lowercase    :   Optional[Any]      =     []
   for i in range(image_processor_tester.batch_size  ):
      if equal_resolution:
         _lowercase    :   List[Any]      =     image_processor_tester.max_resolution
      else:
         _lowercase  ,   _lowercase    :   Dict      =     np.random.choice(
             np.arange(image_processor_tester.min_resolution	,							image_processor_tester.max_resolution  )	,							2  )
         _lowercase    :   List[Any]      =     prepare_video(
             image_processor_tester=snake_case	,							width=snake_case	,							height=snake_case	,							numpify=snake_case	,							torchify=snake_case	,							)
      video_inputs.append(snake_case  )

   return video_inputs


class      a__						(							unittest.TestCase						):


   def __init__(				self    ,       _UpperCamelCase    ,       _UpperCamelCase=7    ,       _UpperCamelCase=3    ,       _UpperCamelCase=4    ,       _UpperCamelCase=18    ,       _UpperCamelCase=30    ,       _UpperCamelCase=400    ,       _UpperCamelCase=True    ,       _UpperCamelCase=None    ,       _UpperCamelCase=True    ,       _UpperCamelCase=[0.5, 0.5, 0.5]    ,       _UpperCamelCase=[0.5, 0.5, 0.5]    ,       _UpperCamelCase=True    ,       _UpperCamelCase=None    ,       ):





      """simple docstring"""





      _lowercase    :   List[str]      =     size if size is not None else {"shortest_edge": 18}
      _lowercase    :   Any      =     crop_size if crop_size is not None else {"height": 18, "width": 18}

      _lowercase    :   Dict      =     parent
      _lowercase    :   Optional[Any]      =     batch_size
      _lowercase    :   List[Any]      =     num_channels
      _lowercase    :   Optional[Any]      =     num_frames
      _lowercase    :   int      =     image_size
      _lowercase    :   Optional[int]      =     min_resolution
      _lowercase    :   Optional[int]      =     max_resolution
      _lowercase    :   Optional[Any]      =     do_resize
      _lowercase    :   Dict      =     size
      _lowercase    :   Any      =     do_normalize
      _lowercase    :   Tuple      =     image_mean
      _lowercase    :   Union[str, Any]      =     image_std
      _lowercase    :   Tuple      =     do_center_crop
      _lowercase    :   str      =     crop_size






   def 		_lowerCamelCase     (				self						):





      """simple docstring"""





      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      a__						(							lowerCamelCase_						,   unittest.TestCase						):
   _SCREAMING_SNAKE_CASE					:  List[str]			 =							TvltImageProcessor if is_vision_available() else None


   def 		_lowerCamelCase     (				self						):





      """simple docstring"""





      _lowercase    :   Optional[Any]      =     TvltImageProcessorTester(self						)


   @property
   def 		_lowerCamelCase     (				self						):





      """simple docstring"""





      return self.image_processor_tester.prepare_image_processor_dict()


   def 		_lowerCamelCase     (				self						):





      """simple docstring"""





      _lowercase    :   int      =     self.image_processing_class(**self.image_processor_dict						)
      self.assertTrue(hasattr(_UpperCamelCase    ,       "image_mean"						)						)
      self.assertTrue(hasattr(_UpperCamelCase    ,       "image_std"						)						)
      self.assertTrue(hasattr(_UpperCamelCase    ,       "do_normalize"						)						)
      self.assertTrue(hasattr(_UpperCamelCase    ,       "do_resize"						)						)
      self.assertTrue(hasattr(_UpperCamelCase    ,       "do_center_crop"						)						)
      self.assertTrue(hasattr(_UpperCamelCase    ,       "size"						)						)


   def 		_lowerCamelCase     (				self						):





      """simple docstring"""





      _lowercase    :   Any      =     self.image_processing_class(**self.image_processor_dict						)
      # create random PIL videos
      _lowercase    :   Any      =     prepare_video_inputs(self.image_processor_tester    ,       equal_resolution=_UpperCamelCase						)
      for video in video_inputs:
         self.assertIsInstance(_UpperCamelCase    ,       _UpperCamelCase						)
         self.assertIsInstance(video[0]    ,       Image.Image						)

      # Test not batched input
      _lowercase    :   Any      =     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
      _lowercase    :   Union[str, Any]      =     image_processor(_UpperCamelCase    ,       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 		_lowerCamelCase     (				self						):





      """simple docstring"""





      _lowercase    :   Optional[Any]      =     self.image_processing_class(**self.image_processor_dict						)
      # create random numpy tensors
      _lowercase    :   List[str]      =     prepare_video_inputs(self.image_processor_tester    ,       equal_resolution=_UpperCamelCase    ,       numpify=_UpperCamelCase						)
      for video in video_inputs:
         self.assertIsInstance(_UpperCamelCase    ,       _UpperCamelCase						)
         self.assertIsInstance(video[0]    ,       np.ndarray						)

      # Test not batched input
      _lowercase    :   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
      _lowercase    :   List[Any]      =     image_processor(_UpperCamelCase    ,       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 		_lowerCamelCase     (				self						):





      """simple docstring"""





      _lowercase    :   Dict      =     self.image_processing_class(**self.image_processor_dict						)
      # create random PyTorch tensors
      _lowercase    :   Optional[int]      =     prepare_video_inputs(self.image_processor_tester    ,       equal_resolution=_UpperCamelCase    ,       torchify=_UpperCamelCase						)
      for video in video_inputs:
         self.assertIsInstance(_UpperCamelCase    ,       _UpperCamelCase						)
         self.assertIsInstance(video[0]    ,       torch.Tensor						)

      # Test not batched input
      _lowercase    :   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
      _lowercase    :   List[Any]      =     image_processor(_UpperCamelCase    ,       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"],
          )    ,       )






