

from typing import Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature
from ...image_transforms import get_image_size, pad, rescale, to_channel_dimension_format
from ...image_utils import ChannelDimension, ImageInput, make_list_of_images, to_numpy_array, valid_images
from ...utils import TensorType, logging


lowerCAmelCase__       = logging.get_logger(__name__)






class 						__magic_name__    (       _snake_case    ):
 UpperCAmelCase							    =						["""pixel_values"""]


 def __init__(						self		: Optional[Any]			,						lowerCAmelCase__		: bool = True			,						lowerCAmelCase__		: Union[int, float] = 1 / 2_5_5			,						lowerCAmelCase__		: bool = True			,						lowerCAmelCase__		: int = 8			,						**lowerCAmelCase__		: Optional[Any]			,						)    ->      None:
   super().__init__(**lowerCAmelCase__    )

   UpperCAmelCase					      =  do_rescale
   UpperCAmelCase					      =  rescale_factor
   UpperCAmelCase					      =  do_pad
   UpperCAmelCase					      =  pad_size


 def _UpperCamelCase	(						self		: Dict			,						lowerCAmelCase__		: np.ndarray			,						lowerCAmelCase__		: float			,						lowerCAmelCase__		: Optional[Union[str, ChannelDimension]] = None			,						**lowerCAmelCase__		: int    )    ->      np.ndarray:
   return rescale(lowerCAmelCase__			,						scale=lowerCAmelCase__			,						data_format=lowerCAmelCase__			,						**lowerCAmelCase__    )


 def _UpperCamelCase	(						self		: List[str]			,						lowerCAmelCase__		: np.ndarray			,						lowerCAmelCase__		: int			,						lowerCAmelCase__		: Optional[Union[str, ChannelDimension]] = None    )    ->      Dict:
   UpperCAmelCase      ,	UpperCAmelCase					      =  get_image_size(lowerCAmelCase__    )
   UpperCAmelCase					      =  (old_height // size + 1) * size - old_height
   UpperCAmelCase					      =  (old_width // size + 1) * size - old_width

   return pad(lowerCAmelCase__			,						((0, pad_height), (0, pad_width))			,						mode="symmetric"			,						data_format=lowerCAmelCase__    )


 def _UpperCamelCase	(						self		: List[str]			,						lowerCAmelCase__		: ImageInput			,						lowerCAmelCase__		: Optional[bool] = None			,						lowerCAmelCase__		: Optional[float] = None			,						lowerCAmelCase__		: Optional[bool] = None			,						lowerCAmelCase__		: Optional[int] = None			,						lowerCAmelCase__		: Optional[Union[str, TensorType]] = None			,						lowerCAmelCase__		: Union[str, ChannelDimension] = ChannelDimension.FIRST			,						**lowerCAmelCase__		: Union[str, Any]			,						)    ->      str:
   UpperCAmelCase					      =  do_rescale if do_rescale is not None else self.do_rescale
   UpperCAmelCase					      =  rescale_factor if rescale_factor is not None else self.rescale_factor
   UpperCAmelCase					      =  do_pad if do_pad is not None else self.do_pad
   UpperCAmelCase					      =  pad_size if pad_size is not None else self.pad_size

   UpperCAmelCase					      =  make_list_of_images(lowerCAmelCase__    )

   if not valid_images(lowerCAmelCase__    ):
     raise ValueError(
         "Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, "
         "torch.Tensor, tf.Tensor or jax.ndarray."    )

   if do_rescale and rescale_factor is None:
     raise ValueError("Rescale factor must be specified if do_rescale is True."    )

   # All transformations expect numpy arrays.
   UpperCAmelCase					      =  [to_numpy_array(lowerCAmelCase__    ) for image in images]

   if do_rescale:
     UpperCAmelCase					      =  [self.rescale(image=lowerCAmelCase__			,						scale=lowerCAmelCase__    ) for image in images]

   if do_pad:
     UpperCAmelCase					      =  [self.pad(lowerCAmelCase__			,						size=lowerCAmelCase__    ) for image in images]

   UpperCAmelCase					      =  [to_channel_dimension_format(lowerCAmelCase__			,						lowerCAmelCase__    ) for image in images]

   UpperCAmelCase					      =  {"pixel_values": images}
   return BatchFeature(data=lowerCAmelCase__			,						tensor_type=lowerCAmelCase__    )






import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowerCAmelCase__       = get_tests_dir("fixtures")






class 						__magic_name__    (       unittest.TestCase    ):


 def _UpperCamelCase	(						self		: Tuple    )    ->      Union[str, Any]:
   # A mock response for an HTTP head request to emulate server down
   UpperCAmelCase					      =  mock.Mock()
   UpperCAmelCase					      =  5_0_0
   UpperCAmelCase					      =  {}
   UpperCAmelCase					      =  HTTPError
   UpperCAmelCase					      =  {}

   # Download this model to make sure it's in the cache.
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"    )
   # Under the mock environment we get a 500 error when trying to reach the model.
   with mock.patch("requests.Session.request"			,						return_value=lowerCAmelCase__    ) as mock_head:
     UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"    )
     # This check we did call the fake head request
     mock_head.assert_called()


 def _UpperCamelCase	(						self		: List[Any]    )    ->      Dict:
   # This test is for deprecated behavior and can be removed in v5
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(
       "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"    )






@is_staging_test
class 						__magic_name__    (       unittest.TestCase    ):


 @classmethod
 def _UpperCamelCase	(						cls		: List[str]    )    ->      List[Any]:
   UpperCAmelCase					      =  TOKEN
   HfFolder.save_token(lowerCAmelCase__    )


 @classmethod
 def _UpperCamelCase	(						cls		: Optional[int]    )    ->      Union[str, Any]:
   try:
     delete_repo(token=cls._token			,						repo_id="test-feature-extractor"    )
   except HTTPError:
     pass

   try:
     delete_repo(token=cls._token			,						repo_id="valid_org/test-feature-extractor-org"    )
   except HTTPError:
     pass

   try:
     delete_repo(token=cls._token			,						repo_id="test-dynamic-feature-extractor"    )
   except HTTPError:
     pass


 def _UpperCamelCase	(						self		: Any    )    ->      Any:
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__    )
   feature_extractor.push_to_hub("test-feature-extractor"			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )

   # Reset repo
   delete_repo(token=self._token			,						repo_id="test-feature-extractor"    )

   # Push to hub via save_pretrained
   with tempfile.TemporaryDirectory() as tmp_dir:
     feature_extractor.save_pretrained(
         lowerCAmelCase__			,						repo_id="test-feature-extractor"			,						push_to_hub=lowerCAmelCase__			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )


 def _UpperCamelCase	(						self		: List[Any]    )    ->      Tuple:
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__    )
   feature_extractor.push_to_hub("valid_org/test-feature-extractor"			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )

   # Reset repo
   delete_repo(token=self._token			,						repo_id="valid_org/test-feature-extractor"    )

   # Push to hub via save_pretrained
   with tempfile.TemporaryDirectory() as tmp_dir:
     feature_extractor.save_pretrained(
         lowerCAmelCase__			,						repo_id="valid_org/test-feature-extractor-org"			,						push_to_hub=lowerCAmelCase__			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )


 def _UpperCamelCase	(						self		: Dict    )    ->      List[str]:
   CustomFeatureExtractor.register_for_auto_class()
   UpperCAmelCase					      =  CustomFeatureExtractor.from_pretrained(lowerCAmelCase__    )

   feature_extractor.push_to_hub("test-dynamic-feature-extractor"			,						use_auth_token=self._token    )

   # This has added the proper auto_map field to the config
   self.assertDictEqual(
       feature_extractor.auto_map			,						{"AutoFeatureExtractor": "custom_feature_extraction.CustomFeatureExtractor"}			,						)

   UpperCAmelCase					      =  AutoFeatureExtractor.from_pretrained(
       f"{USER}/test-dynamic-feature-extractor"			,						trust_remote_code=lowerCAmelCase__    )
   # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
   self.assertEqual(new_feature_extractor.__class__.__name__			,						"CustomFeatureExtractor"    )



