






"""simple docstring"""



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 AutoImageProcessor, ViTImageProcessor
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_image_processing import CustomImageProcessor  # noqa E402


__A		:       Union[str, Any]  				=		get_tests_dir('''fixtures''')





class 					_UpperCAmelCase       ( unittest.TestCase							):




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

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




					def   A  (				self     :						Optional[Any]   )  ->  int:
							# This test is for deprecated behavior and can be removed in v5
							lowercase_ : Union[str, Any]							       =   ViTImageProcessor.from_pretrained(
							    '''https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json'''   )






					def   A  (				self     :						str   )  ->  List[str]:
							with self.assertRaises(A   ):
									# config is in subfolder, the following should not work without specifying the subfolder
									lowercase_ : str							       =   AutoImageProcessor.from_pretrained('''hf-internal-testing/stable-diffusion-all-variants'''   )

							lowercase_ : Optional[Any]							       =   AutoImageProcessor.from_pretrained(
							    '''hf-internal-testing/stable-diffusion-all-variants'''    ,						subfolder='''feature_extractor'''   )

							self.assertIsNotNone(A   )





@is_staging_test
class 					_UpperCAmelCase       ( unittest.TestCase							):




					@classmethod
					def   A  (				cls     :						Union[str, Any]   )  ->  int:
							lowercase_ : Dict							       =   TOKEN
							HfFolder.save_token(A   )




					@classmethod
					def   A  (				cls     :						Any   )  ->  int:
							try:
									delete_repo(token=cls._token    ,						repo_id='''test-image-processor'''   )
							except HTTPError:
									pass

							try:
									delete_repo(token=cls._token    ,						repo_id='''valid_org/test-image-processor-org'''   )
							except HTTPError:
									pass

							try:
									delete_repo(token=cls._token    ,						repo_id='''test-dynamic-image-processor'''   )
							except HTTPError:
									pass




					def   A  (				self     :						List[str]   )  ->  Dict:
							lowercase_ : List[Any]							       =   ViTImageProcessor.from_pretrained(A   )
							image_processor.push_to_hub('''test-image-processor'''    ,						use_auth_token=self._token   )

							lowercase_ : Union[str, Any]							       =   ViTImageProcessor.from_pretrained(F'''{USER}/test-image-processor'''   )
							for k, v in image_processor.__dict__.items():
									self.assertEqual(A    ,						getattr(A    ,						A   )   )

							# Reset repo
							delete_repo(token=self._token    ,						repo_id='''test-image-processor'''   )

							# Push to hub via save_pretrained
							with tempfile.TemporaryDirectory() as tmp_dir:
									image_processor.save_pretrained(
									    A    ,						repo_id='''test-image-processor'''    ,						push_to_hub=A    ,						use_auth_token=self._token   )

							lowercase_ : str							       =   ViTImageProcessor.from_pretrained(F'''{USER}/test-image-processor'''   )
							for k, v in image_processor.__dict__.items():
									self.assertEqual(A    ,						getattr(A    ,						A   )   )




					def   A  (				self     :						str   )  ->  List[Any]:
							lowercase_ : List[Any]							       =   ViTImageProcessor.from_pretrained(A   )
							image_processor.push_to_hub('''valid_org/test-image-processor'''    ,						use_auth_token=self._token   )

							lowercase_ : Union[str, Any]							       =   ViTImageProcessor.from_pretrained('''valid_org/test-image-processor'''   )
							for k, v in image_processor.__dict__.items():
									self.assertEqual(A    ,						getattr(A    ,						A   )   )

							# Reset repo
							delete_repo(token=self._token    ,						repo_id='''valid_org/test-image-processor'''   )

							# Push to hub via save_pretrained
							with tempfile.TemporaryDirectory() as tmp_dir:
									image_processor.save_pretrained(
									    A    ,						repo_id='''valid_org/test-image-processor-org'''    ,						push_to_hub=A    ,						use_auth_token=self._token   )

							lowercase_ : str							       =   ViTImageProcessor.from_pretrained('''valid_org/test-image-processor-org'''   )
							for k, v in image_processor.__dict__.items():
									self.assertEqual(A    ,						getattr(A    ,						A   )   )






					def   A  (				self     :						Union[str, Any]   )  ->  Tuple:
							CustomImageProcessor.register_for_auto_class()
							lowercase_ : List[str]							       =   CustomImageProcessor.from_pretrained(A   )

							image_processor.push_to_hub('''test-dynamic-image-processor'''    ,						use_auth_token=self._token   )

							# This has added the proper auto_map field to the config
							self.assertDictEqual(
							    image_processor.auto_map    ,						{'''AutoImageProcessor''': '''custom_image_processing.CustomImageProcessor'''}    ,						)

							lowercase_ : Dict							       =   AutoImageProcessor.from_pretrained(
							    F'''{USER}/test-dynamic-image-processor'''    ,						trust_remote_code=A   )
							# Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module
							self.assertEqual(new_image_processor.__class__.__name__    ,						'''CustomImageProcessor'''   )











from abc import ABC, abstractmethod
from typing import Optional, Union

from .. import Dataset, DatasetDict, Features, IterableDataset, IterableDatasetDict, NamedSplit
from ..utils.typing import NestedDataStructureLike, PathLike
class     a    ( lowercase__   ):



      """simple docstring"""


      def __init__(      self					:			Any		,      __lowercase					:			Optional[NestedDataStructureLike[PathLike]] = None		,      __lowercase					:			Optional[NamedSplit] = None		,      __lowercase					:			Optional[Features] = None		,      __lowercase					:			str = None		,      __lowercase					:			bool = False		,      __lowercase					:			bool = False		,      __lowercase					:			Optional[int] = None		,      **__lowercase					:			List[str]		,      )							->      Tuple:
        __UpperCAmelCase							:	Any           =  path_or_paths
        __UpperCAmelCase							:	Dict           =  split if split or isinstance(__lowercase		,      __lowercase					) else """train"""
        __UpperCAmelCase							:	Optional[int]           =  features
        __UpperCAmelCase							:	str           =  cache_dir
        __UpperCAmelCase							:	str           =  keep_in_memory
        __UpperCAmelCase							:	Optional[int]           =  streaming
        __UpperCAmelCase							:	Dict           =  num_proc
        __UpperCAmelCase							:	Tuple           =  kwargs




      @abstractmethod
      def 	UpperCAmelCase      (      self					:			List[Any]					)							->      Union[Dataset, DatasetDict, IterableDataset, IterableDatasetDict]:
        pass


class     a    ( lowercase__   ):



      """simple docstring"""


      def __init__(      self					:			List[str]		,      __lowercase					:			Optional[Features] = None		,      __lowercase					:			str = None		,      __lowercase					:			bool = False		,      __lowercase					:			bool = False		,      __lowercase					:			Optional[int] = None		,      **__lowercase					:			Optional[Any]		,      )							->      Optional[int]:
        __UpperCAmelCase							:	Optional[Any]           =  features
        __UpperCAmelCase							:	str           =  cache_dir
        __UpperCAmelCase							:	Optional[int]           =  keep_in_memory
        __UpperCAmelCase							:	Dict           =  streaming
        __UpperCAmelCase							:	Optional[Any]           =  num_proc
        __UpperCAmelCase							:	Union[str, Any]           =  kwargs




      @abstractmethod
      def 	UpperCAmelCase      (      self					:			List[str]					)							->      Union[Dataset, IterableDataset]:
        pass


