import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

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

import transformers
from transformers import (
    CONFIG_MAPPING,
    FEATURE_EXTRACTOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoProcessor,
    AutoTokenizer,
    BertTokenizer,
    ProcessorMixin,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
)
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test
from transformers.tokenization_utils import TOKENIZER_CONFIG_FILE
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_tokenizers_available


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

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402
from test_module.custom_processing import CustomProcessor  # noqa E402
from test_module.custom_tokenization import CustomTokenizer  # noqa E402


_UpperCAmelCase          =						get_tests_dir("""fixtures/dummy_feature_extractor_config.json""")
_UpperCAmelCase          =						get_tests_dir("""fixtures/vocab.json""")
_UpperCAmelCase          =						get_tests_dir("""fixtures""")





class UpperCAmelCase		(   unittest.TestCase    ):


				'''simple docstring'''






				lowerCamelCase_        =   ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						A_						:							List[Any]      = 0
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						A_						:							Optional[int]      = AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'   )
						self.assertIsInstance(lowercase						,      lowercase   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						with tempfile.TemporaryDirectory() as tmpdirname:
								A_						:							str      = WavaVecaConfig()
								A_						:							Tuple      = AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'   )

								# save in new folder
								model_config.save_pretrained(lowercase   )
								processor.save_pretrained(lowercase   )

								A_						:							List[str]      = AutoProcessor.from_pretrained(lowercase   )

						self.assertIsInstance(lowercase						,      lowercase   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						with tempfile.TemporaryDirectory() as tmpdirname:
								# copy relevant files
								copyfile(lowercase						,      os.path.join(lowercase						,      lowercase   )   )
								copyfile(lowercase						,      os.path.join(lowercase						,      'vocab.json'   )   )

								A_						:							Tuple      = AutoProcessor.from_pretrained(lowercase   )

						self.assertIsInstance(lowercase						,      lowercase   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						with tempfile.TemporaryDirectory() as tmpdirname:
								A_						:							Any      = WavaVecaFeatureExtractor()
								A_						:							List[Any]      = AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'   )

								A_						:							Dict      = WavaVecaProcessor(lowercase						,      lowercase   )

								# save in new folder
								processor.save_pretrained(lowercase   )

								# drop `processor_class` in tokenizer
								with open(os.path.join(lowercase						,      lowercase   )						,      'r'   ) as f:
										A_						:							str      = json.load(lowercase   )
										config_dict.pop('processor_class'   )

								with open(os.path.join(lowercase						,      lowercase   )						,      'w'   ) as f:
										f.write(json.dumps(lowercase   )   )

								A_						:							List[Any]      = AutoProcessor.from_pretrained(lowercase   )

						self.assertIsInstance(lowercase						,      lowercase   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						with tempfile.TemporaryDirectory() as tmpdirname:
								A_						:							int      = WavaVecaFeatureExtractor()
								A_						:							List[str]      = AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'   )

								A_						:							Tuple      = WavaVecaProcessor(lowercase						,      lowercase   )

								# save in new folder
								processor.save_pretrained(lowercase   )

								# drop `processor_class` in feature extractor
								with open(os.path.join(lowercase						,      lowercase   )						,      'r'   ) as f:
										A_						:							List[Any]      = json.load(lowercase   )
										config_dict.pop('processor_class'   )

								with open(os.path.join(lowercase						,      lowercase   )						,      'w'   ) as f:
										f.write(json.dumps(lowercase   )   )

								A_						:							Dict      = AutoProcessor.from_pretrained(lowercase   )

						self.assertIsInstance(lowercase						,      lowercase   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						with tempfile.TemporaryDirectory() as tmpdirname:
								A_						:							Any      = WavaVecaConfig(processor_class='Wav2Vec2Processor'   )
								model_config.save_pretrained(lowercase   )
								# copy relevant files
								copyfile(lowercase						,      os.path.join(lowercase						,      'vocab.json'   )   )
								# create emtpy sample processor
								with open(os.path.join(lowercase						,      lowercase   )						,      'w'   ) as f:
										f.write('{}'   )

								A_						:							Union[str, Any]      = AutoProcessor.from_pretrained(lowercase   )

						self.assertIsInstance(lowercase						,      lowercase   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						with self.assertRaises(lowercase   ):
								A_						:							Tuple      = AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'   )
						# If remote code is disabled, we can't load this config.
						with self.assertRaises(lowercase   ):
								A_						:							Any      = AutoProcessor.from_pretrained(
								    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowercase   )

						A_						:							int      = AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowercase   )
						self.assertTrue(processor.special_attribute_present   )
						self.assertEqual(processor.__class__.__name__						,      'NewProcessor'   )

						A_						:							Tuple      = processor.feature_extractor
						self.assertTrue(feature_extractor.special_attribute_present   )
						self.assertEqual(feature_extractor.__class__.__name__						,      'NewFeatureExtractor'   )

						A_						:							Any      = processor.tokenizer
						self.assertTrue(tokenizer.special_attribute_present   )
						if is_tokenizers_available():
								self.assertEqual(tokenizer.__class__.__name__						,      'NewTokenizerFast'   )

								# Test we can also load the slow version
								A_						:							str      = AutoProcessor.from_pretrained(
								    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowercase						,      use_fast=lowercase   )
								A_						:							List[Any]      = new_processor.tokenizer
								self.assertTrue(new_tokenizer.special_attribute_present   )
								self.assertEqual(new_tokenizer.__class__.__name__						,      'NewTokenizer'   )
						else:
								self.assertEqual(tokenizer.__class__.__name__						,      'NewTokenizer'   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						try:
								AutoConfig.register('custom'						,      lowercase   )
								AutoFeatureExtractor.register(lowercase						,      lowercase   )
								AutoTokenizer.register(lowercase						,      slow_tokenizer_class=lowercase   )
								AutoProcessor.register(lowercase						,      lowercase   )
								# Trying to register something existing in the Transformers library will raise an error
								with self.assertRaises(lowercase   ):
										AutoProcessor.register(lowercase						,      lowercase   )

								# Now that the config is registered, it can be used as any other config with the auto-API
								A_						:							str      = CustomFeatureExtractor.from_pretrained(lowercase   )

								with tempfile.TemporaryDirectory() as tmp_dir:
										A_						:							int      = os.path.join(lowercase						,      'vocab.txt'   )
										with open(lowercase						,      'w'						,      encoding='utf-8'   ) as vocab_writer:
												vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]   )   )
										A_						:							List[Any]      = CustomTokenizer(lowercase   )

								A_						:							Optional[int]      = CustomProcessor(lowercase						,      lowercase   )

								with tempfile.TemporaryDirectory() as tmp_dir:
										processor.save_pretrained(lowercase   )
										A_						:							List[Any]      = AutoProcessor.from_pretrained(lowercase   )
										self.assertIsInstance(lowercase						,      lowercase   )

						finally:
								if "custom" in CONFIG_MAPPING._extra_content:
										del CONFIG_MAPPING._extra_content["custom"]
								if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
										del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
								if CustomConfig in TOKENIZER_MAPPING._extra_content:
										del TOKENIZER_MAPPING._extra_content[CustomConfig]
								if CustomConfig in PROCESSOR_MAPPING._extra_content:
										del PROCESSOR_MAPPING._extra_content[CustomConfig]
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						class UpperCAmelCase		(   __A    ):


										'''simple docstring'''






										lowerCamelCase_        =   False





						class UpperCAmelCase		(   __A    ):


										'''simple docstring'''






										lowerCamelCase_        =   False





						class UpperCAmelCase		(   __A    ):


										'''simple docstring'''






										lowerCamelCase_        =   '''AutoFeatureExtractor'''
										lowerCamelCase_        =   '''AutoTokenizer'''
										lowerCamelCase_        =   False

						try:
								AutoConfig.register('custom'						,      lowercase   )
								AutoFeatureExtractor.register(lowercase						,      lowercase   )
								AutoTokenizer.register(lowercase						,      slow_tokenizer_class=lowercase   )
								AutoProcessor.register(lowercase						,      lowercase   )
								# If remote code is not set, the default is to use local classes.
								A_						:							List[str]      = AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'   )
								self.assertEqual(processor.__class__.__name__						,      'NewProcessor'   )
								self.assertFalse(processor.special_attribute_present   )
								self.assertFalse(processor.feature_extractor.special_attribute_present   )
								self.assertFalse(processor.tokenizer.special_attribute_present   )

								# If remote code is disabled, we load the local ones.
								A_						:							Union[str, Any]      = AutoProcessor.from_pretrained(
								    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowercase   )
								self.assertEqual(processor.__class__.__name__						,      'NewProcessor'   )
								self.assertFalse(processor.special_attribute_present   )
								self.assertFalse(processor.feature_extractor.special_attribute_present   )
								self.assertFalse(processor.tokenizer.special_attribute_present   )

								# If remote is enabled, we load from the Hub.
								A_						:							Union[str, Any]      = AutoProcessor.from_pretrained(
								    'hf-internal-testing/test_dynamic_processor'						,      trust_remote_code=lowercase   )
								self.assertEqual(processor.__class__.__name__						,      'NewProcessor'   )
								self.assertTrue(processor.special_attribute_present   )
								self.assertTrue(processor.feature_extractor.special_attribute_present   )
								self.assertTrue(processor.tokenizer.special_attribute_present   )

						finally:
								if "custom" in CONFIG_MAPPING._extra_content:
										del CONFIG_MAPPING._extra_content["custom"]
								if CustomConfig in FEATURE_EXTRACTOR_MAPPING._extra_content:
										del FEATURE_EXTRACTOR_MAPPING._extra_content[CustomConfig]
								if CustomConfig in TOKENIZER_MAPPING._extra_content:
										del TOKENIZER_MAPPING._extra_content[CustomConfig]
								if CustomConfig in PROCESSOR_MAPPING._extra_content:
										del PROCESSOR_MAPPING._extra_content[CustomConfig]
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						A_						:							Union[str, Any]      = AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-bert'   )
						self.assertEqual(processor.__class__.__name__						,      'BertTokenizerFast'   )







				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						A_						:							str      = AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-convnext'   )
						self.assertEqual(processor.__class__.__name__						,      'ConvNextImageProcessor'   )





@is_staging_test
class UpperCAmelCase		(   unittest.TestCase    ):


				'''simple docstring'''






				lowerCamelCase_        =   ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']
				@classmethod
				def     lowerCAmelCase_  (		cls   ):



						"""simple docstring"""





						A_						:							Tuple      = TOKEN
						HfFolder.save_token(lowercase   )
				@classmethod
				def     lowerCAmelCase_  (		cls   ):



						"""simple docstring"""





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

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

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



						"""simple docstring"""





						A_						:							int      = WavaVecaProcessor.from_pretrained(lowercase   )
						with tempfile.TemporaryDirectory() as tmp_dir:
								processor.save_pretrained(
								    os.path.join(lowercase						,      'test-processor'   )						,      push_to_hub=lowercase						,      use_auth_token=self._token   )

								A_						:							List[str]      = WavaVecaProcessor.from_pretrained(F'''{USER}/test-processor'''   )
								for k, v in processor.feature_extractor.__dict__.items():
										self.assertEqual(lowercase						,      getattr(new_processor.feature_extractor						,      lowercase   )   )
								self.assertDictEqual(new_processor.tokenizer.get_vocab()						,      processor.tokenizer.get_vocab()   )
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						A_						:							Optional[Any]      = WavaVecaProcessor.from_pretrained(lowercase   )

						with tempfile.TemporaryDirectory() as tmp_dir:
								processor.save_pretrained(
								    os.path.join(lowercase						,      'test-processor-org'   )						,      push_to_hub=lowercase						,      use_auth_token=self._token						,      organization='valid_org'						,      )

								A_						:							Optional[Any]      = WavaVecaProcessor.from_pretrained('valid_org/test-processor-org'   )
								for k, v in processor.feature_extractor.__dict__.items():
										self.assertEqual(lowercase						,      getattr(new_processor.feature_extractor						,      lowercase   )   )
								self.assertDictEqual(new_processor.tokenizer.get_vocab()						,      processor.tokenizer.get_vocab()   )







				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						CustomFeatureExtractor.register_for_auto_class()
						CustomTokenizer.register_for_auto_class()
						CustomProcessor.register_for_auto_class()

						A_						:							Any      = CustomFeatureExtractor.from_pretrained(lowercase   )

						with tempfile.TemporaryDirectory() as tmp_dir:
								A_						:							str      = os.path.join(lowercase						,      'vocab.txt'   )
								with open(lowercase						,      'w'						,      encoding='utf-8'   ) as vocab_writer:
										vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]   )   )
								A_						:							int      = CustomTokenizer(lowercase   )

						A_						:							Optional[int]      = CustomProcessor(lowercase						,      lowercase   )

						with tempfile.TemporaryDirectory() as tmp_dir:
								create_repo(F'''{USER}/test-dynamic-processor'''						,      token=self._token   )
								A_						:							Union[str, Any]      = Repository(lowercase						,      clone_from=F'''{USER}/test-dynamic-processor'''						,      token=self._token   )
								processor.save_pretrained(lowercase   )

								# This has added the proper auto_map field to the feature extractor config
								self.assertDictEqual(
								    processor.feature_extractor.auto_map						,      {
								        'AutoFeatureExtractor': 'custom_feature_extraction.CustomFeatureExtractor',
								        'AutoProcessor': 'custom_processing.CustomProcessor',
								    }						,      )

								# This has added the proper auto_map field to the tokenizer config
								with open(os.path.join(lowercase						,      'tokenizer_config.json'   )   ) as f:
										A_						:							Tuple      = json.load(lowercase   )
								self.assertDictEqual(
								    tokenizer_config['auto_map']						,      {
								        'AutoTokenizer': ['custom_tokenization.CustomTokenizer', None],
								        'AutoProcessor': 'custom_processing.CustomProcessor',
								    }						,      )

								# The code has been copied from fixtures
								self.assertTrue(os.path.isfile(os.path.join(lowercase						,      'custom_feature_extraction.py'   )   )   )
								self.assertTrue(os.path.isfile(os.path.join(lowercase						,      'custom_tokenization.py'   )   )   )
								self.assertTrue(os.path.isfile(os.path.join(lowercase						,      'custom_processing.py'   )   )   )

								repo.push_to_hub()

						A_						:							List[str]      = AutoProcessor.from_pretrained(F'''{USER}/test-dynamic-processor'''						,      trust_remote_code=lowercase   )
						# Can't make an isinstance check because the new_processor is from the CustomProcessor class of a dynamic module
						self.assertEqual(new_processor.__class__.__name__						,      'CustomProcessor'   )







from .integrations import (
    is_optuna_available,
    is_ray_available,
    is_sigopt_available,
    is_wandb_available,
    run_hp_search_optuna,
    run_hp_search_ray,
    run_hp_search_sigopt,
    run_hp_search_wandb,
)
from .trainer_utils import (
    HPSearchBackend,
    default_hp_space_optuna,
    default_hp_space_ray,
    default_hp_space_sigopt,
    default_hp_space_wandb,
)
from .utils import logging


_UpperCAmelCase          =						logging.get_logger(__name__)





class UpperCAmelCase		:


				'''simple docstring'''






				lowerCamelCase_        =   42
				lowerCamelCase_        =   None
				@staticmethod
				def     lowerCAmelCase_  (		):



						"""simple docstring"""





						raise NotImplementedError
				def     lowerCAmelCase_  (		self						,      lowercase						,      lowercase						,      lowercase						,      **lowercase   ):



						"""simple docstring"""





						raise NotImplementedError
				def     lowerCAmelCase_  (		self						,      lowercase   ):



						"""simple docstring"""





						raise NotImplementedError
				def     lowerCAmelCase_  (		self   ):



						"""simple docstring"""





						if not self.is_available():
								raise RuntimeError(
								    F'''You picked the {self.name} backend, but it is not installed. Run {self.pip_install()}.'''   )







				@classmethod
				def     lowerCAmelCase_  (		cls   ):



						"""simple docstring"""





						return F'''`pip install {cls.pip_package or cls.name}`'''





class UpperCAmelCase		(   __A    ):


				'''simple docstring'''






				lowerCamelCase_        =   '''optuna'''
				@staticmethod
				def     lowerCAmelCase_  (		):



						"""simple docstring"""





						return is_optuna_available()
				def     lowerCAmelCase_  (		self						,      lowercase						,      lowercase						,      lowercase						,      **lowercase   ):



						"""simple docstring"""





						return run_hp_search_optuna(lowercase						,      lowercase						,      lowercase						,      **lowercase   )







				def     lowerCAmelCase_  (		self						,      lowercase   ):



						"""simple docstring"""





						return default_hp_space_optuna(lowercase   )





class UpperCAmelCase		(   __A    ):


				'''simple docstring'''






				lowerCamelCase_        =   '''ray'''
				lowerCamelCase_        =   '''\'ray[tune]\''''
				@staticmethod
				def     lowerCAmelCase_  (		):



						"""simple docstring"""





						return is_ray_available()
				def     lowerCAmelCase_  (		self						,      lowercase						,      lowercase						,      lowercase						,      **lowercase   ):



						"""simple docstring"""





						return run_hp_search_ray(lowercase						,      lowercase						,      lowercase						,      **lowercase   )







				def     lowerCAmelCase_  (		self						,      lowercase   ):



						"""simple docstring"""





						return default_hp_space_ray(lowercase   )





class UpperCAmelCase		(   __A    ):


				'''simple docstring'''






				lowerCamelCase_        =   '''sigopt'''
				@staticmethod
				def     lowerCAmelCase_  (		):



						"""simple docstring"""





						return is_sigopt_available()
				def     lowerCAmelCase_  (		self						,      lowercase						,      lowercase						,      lowercase						,      **lowercase   ):



						"""simple docstring"""





						return run_hp_search_sigopt(lowercase						,      lowercase						,      lowercase						,      **lowercase   )







				def     lowerCAmelCase_  (		self						,      lowercase   ):



						"""simple docstring"""





						return default_hp_space_sigopt(lowercase   )





class UpperCAmelCase		(   __A    ):


				'''simple docstring'''






				lowerCamelCase_        =   '''wandb'''
				@staticmethod
				def     lowerCAmelCase_  (		):



						"""simple docstring"""





						return is_wandb_available()
				def     lowerCAmelCase_  (		self						,      lowercase						,      lowercase						,      lowercase						,      **lowercase   ):



						"""simple docstring"""





						return run_hp_search_wandb(lowercase						,      lowercase						,      lowercase						,      **lowercase   )







				def     lowerCAmelCase_  (		self						,      lowercase   ):



						"""simple docstring"""





						return default_hp_space_wandb(lowercase   )

_UpperCAmelCase          =						{
    HPSearchBackend(backend.name): backend for backend in [OptunaBackend, RayTuneBackend, SigOptBackend, WandbBackend]
}




def 							UpperCamelCase		(			):
		'''simple docstring'''




		A_						:							List[str]      = [backend for backend in ALL_HYPERPARAMETER_SEARCH_BACKENDS.values() if backend.is_available()]
		if len(__lowercase      ) > 0:
				A_						:							List[str]      = available_backends[0].name
				if len(__lowercase      ) > 1:
						logger.info(
						    f'''{len(__lowercase      )} hyperparameter search backends available. Using {name} as the default.'''      )
				return name
		raise RuntimeError(
		    'No hyperparameter search backend available.\n'
		    + '\n'.join(
		        f''' - To install {backend.name} run {backend.pip_install()}'''
		        for backend in ALL_HYPERPARAMETER_SEARCH_BACKENDS.values()      )      )






