



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


A__								=						get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
A__								=						get_tests_dir('''fixtures/vocab.json''')
A__								=						get_tests_dir('''fixtures''')


class       a							(			unittest.TestCase				):
		__lowerCAmelCase	:    Tuple											=					["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]



		def 	__lowerCamelCase			(     self     :Any  ):
						snake_case__	:      Any     		=	0



		def 	__lowerCamelCase			(     self     :Optional[int]  ):
						snake_case__	:      Optional[int]     		=	AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''  )
						self.assertIsInstance(__lowercase   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :List[Any]  ):
						with tempfile.TemporaryDirectory() as tmpdirname:
										snake_case__	:      Union[str, Any]     		=	WavaVecaConfig()
										snake_case__	:      Tuple     		=	AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''  )

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

										snake_case__	:      List[Any]     		=	AutoProcessor.from_pretrained(__lowercase  )

						self.assertIsInstance(__lowercase   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :Optional[Any]  ):
						with tempfile.TemporaryDirectory() as tmpdirname:
										# copy relevant files
										copyfile(__lowercase   ,os.path.join(__lowercase   ,__lowercase  )  )
										copyfile(__lowercase   ,os.path.join(__lowercase   ,'''vocab.json'''  )  )

										snake_case__	:      Optional[int]     		=	AutoProcessor.from_pretrained(__lowercase  )

						self.assertIsInstance(__lowercase   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :str  ):
						with tempfile.TemporaryDirectory() as tmpdirname:
										snake_case__	:      str     		=	WavaVecaFeatureExtractor()
										snake_case__	:      Tuple     		=	AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''  )

										snake_case__	:      Tuple     		=	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:
														snake_case__	:      Tuple     		=	json.load(__lowercase  )
														config_dict.pop('''processor_class'''  )

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

										snake_case__	:      List[Any]     		=	AutoProcessor.from_pretrained(__lowercase  )

						self.assertIsInstance(__lowercase   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :Optional[int]  ):
						with tempfile.TemporaryDirectory() as tmpdirname:
										snake_case__	:      str     		=	WavaVecaFeatureExtractor()
										snake_case__	:      Union[str, Any]     		=	AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''  )

										snake_case__	:      Union[str, Any]     		=	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:
														snake_case__	:      Any     		=	json.load(__lowercase  )
														config_dict.pop('''processor_class'''  )

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

										snake_case__	:      Optional[Any]     		=	AutoProcessor.from_pretrained(__lowercase  )

						self.assertIsInstance(__lowercase   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :Any  ):
						with tempfile.TemporaryDirectory() as tmpdirname:
										snake_case__	:      List[str]     		=	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('''{}'''  )

										snake_case__	:      int     		=	AutoProcessor.from_pretrained(__lowercase  )

						self.assertIsInstance(__lowercase   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :int  ):
						# If remote code is not set, we will time out when asking whether to load the model.
						with self.assertRaises(__lowercase  ):
										snake_case__	:      List[Any]     		=	AutoProcessor.from_pretrained('''hf-internal-testing/test_dynamic_processor'''  )
						# If remote code is disabled, we can't load this config.
						with self.assertRaises(__lowercase  ):
										snake_case__	:      Optional[Any]     		=	AutoProcessor.from_pretrained(
										    '''hf-internal-testing/test_dynamic_processor'''   ,trust_remote_code=__lowercase  )

						snake_case__	:      str     		=	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'''  )

						snake_case__	:      Optional[int]     		=	processor.feature_extractor
						self.assertTrue(feature_extractor.special_attribute_present  )
						self.assertEqual(feature_extractor.__class__.__name__   ,'''NewFeatureExtractor'''  )

						snake_case__	:      Dict     		=	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
										snake_case__	:      Optional[int]     		=	AutoProcessor.from_pretrained(
										    '''hf-internal-testing/test_dynamic_processor'''   ,trust_remote_code=__lowercase   ,use_fast=__lowercase  )
										snake_case__	:      str     		=	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     :int  ):
						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
										snake_case__	:      List[Any]     		=	CustomFeatureExtractor.from_pretrained(__lowercase  )

										with tempfile.TemporaryDirectory() as tmp_dir:
														snake_case__	:      Union[str, Any]     		=	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]  )  )
														snake_case__	:      List[str]     		=	CustomTokenizer(__lowercase  )

										snake_case__	:      Tuple     		=	CustomProcessor(__lowercase   ,__lowercase  )

										with tempfile.TemporaryDirectory() as tmp_dir:
														processor.save_pretrained(__lowercase  )
														snake_case__	:      Dict     		=	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     :Optional[int]  ):


						class       a							(			__lowerCamelCase				):
								__lowerCAmelCase	:    Union[str, Any]											=					False


						class       a							(			__lowerCamelCase				):
								__lowerCAmelCase	:    Optional[int]											=					False


						class       a							(			__lowerCamelCase				):
								__lowerCAmelCase	:    str											=					"""AutoFeatureExtractor"""
								__lowerCAmelCase	:    str											=					"""AutoTokenizer"""
								__lowerCAmelCase	:    Optional[Any]											=					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.
										snake_case__	:      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.
										snake_case__	:      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.
										snake_case__	:      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     :Any  ):
						snake_case__	:      List[str]     		=	AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-bert'''  )
						self.assertEqual(processor.__class__.__name__   ,'''BertTokenizerFast'''  )






		def 	__lowerCamelCase			(     self     :Tuple  ):
						snake_case__	:      Optional[Any]     		=	AutoProcessor.from_pretrained('''hf-internal-testing/tiny-random-convnext'''  )
						self.assertEqual(processor.__class__.__name__   ,'''ConvNextImageProcessor'''  )


@is_staging_test
class       a							(			unittest.TestCase				):
		__lowerCAmelCase	:    Any											=					["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]



		@classmethod
		def 	__lowerCamelCase			(     cls     :List[str]  ):
						snake_case__	:      str     		=	TOKEN
						HfFolder.save_token(__lowercase  )



		@classmethod
		def 	__lowerCamelCase			(     cls     :List[str]  ):
						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     :Optional[Any]  ):
						snake_case__	:      Dict     		=	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  )

										snake_case__	:      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     :List[Any]  ):
						snake_case__	:      Optional[int]     		=	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'''   ,)

										snake_case__	:      List[str]     		=	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     :Optional[Any]  ):
						CustomFeatureExtractor.register_for_auto_class()
						CustomTokenizer.register_for_auto_class()
						CustomProcessor.register_for_auto_class()

						snake_case__	:      int     		=	CustomFeatureExtractor.from_pretrained(__lowercase  )

						with tempfile.TemporaryDirectory() as tmp_dir:
										snake_case__	:      Dict     		=	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]  )  )
										snake_case__	:      Any     		=	CustomTokenizer(__lowercase  )

						snake_case__	:      int     		=	CustomProcessor(__lowercase   ,__lowercase  )

						with tempfile.TemporaryDirectory() as tmp_dir:
										create_repo(F"""{USER}/test-dynamic-processor"""   ,token=self._token  )
										snake_case__	:      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:
														snake_case__	:      Optional[Any]     		=	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()

						snake_case__	:      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 importlib import import_module

from .logging import get_logger


A__								=						get_logger(__name__)


class       a							:



		def __init__(     self     :Optional[int]   ,__lowercase     :List[str]   ,__lowercase     :Any=None  ):
						snake_case__	:      List[str]     		=	attrs or []
						if module is not None:
										for key in module.__dict__:
														if key in attrs or not key.startswith('''__'''  ):
																		setattr(self   ,__lowercase   ,getattr(__lowercase   ,__lowercase  )  )
						snake_case__	:      Optional[Any]     		=	module._original_module if isinstance(__lowercase   ,_PatchedModuleObj  ) else module


class       a							:
		__lowerCAmelCase	:    Any											=					[]



		def __init__(     self     :List[str]   ,__lowercase     :Optional[Any]   ,__lowercase     :str   ,__lowercase     :Dict   ,__lowercase     :Any=None  ):
						snake_case__	:      Dict     		=	obj
						snake_case__	:      Dict     		=	target
						snake_case__	:      List[str]     		=	new
						snake_case__	:      int     		=	target.split('''.'''  )[0]
						snake_case__	:      List[str]     		=	{}
						snake_case__	:      Any     		=	attrs or []



		def __enter__(     self     :Tuple  ):
						*snake_case__		, snake_case__	:      str     		=	self.target.split('''.'''  )

						# Patch modules:
						# it's used to patch attributes of submodules like "os.path.join";
						# in this case we need to patch "os" and "os.path"

						for i in range(len(__lowercase  )  ):
										try:
														snake_case__	:      Any     		=	import_module('''.'''.join(submodules[: i + 1]  )  )
										except ModuleNotFoundError:
														continue
										# We iterate over all the globals in self.obj in case we find "os" or "os.path"
										for attr in self.obj.__dir__():
														snake_case__	:      Optional[int]     		=	getattr(self.obj   ,__lowercase  )
														# We don't check for the name of the global, but rather if its value *is* "os" or "os.path".
														# This allows to patch renamed modules like "from os import path as ospath".
														if obj_attr is submodule or (
														    (isinstance(__lowercase   ,_PatchedModuleObj  ) and obj_attr._original_module is submodule)
														):
																		snake_case__	:      List[Any]     		=	obj_attr
																		# patch at top level
																		setattr(self.obj   ,__lowercase   ,_PatchedModuleObj(__lowercase   ,attrs=self.attrs  )  )
																		snake_case__	:      List[Any]     		=	getattr(self.obj   ,__lowercase  )
																		# construct lower levels patches
																		for key in submodules[i + 1 :]:
																						setattr(__lowercase   ,__lowercase   ,_PatchedModuleObj(getattr(__lowercase   ,__lowercase   ,__lowercase  )   ,attrs=self.attrs  )  )
																						snake_case__	:      List[Any]     		=	getattr(__lowercase   ,__lowercase  )
																		# finally set the target attribute
																		setattr(__lowercase   ,__lowercase   ,self.new  )

        # Patch attribute itself:
        # it's used for builtins like "open",
        # and also to patch "os.path.join" we may also need to patch "join"
        # itself if it was imported as "from os.path import join".

						if submodules:  # if it's an attribute of a submodule like "os.path.join"
										try:
														snake_case__	:      int     		=	getattr(import_module('''.'''.join(__lowercase  )  )   ,__lowercase  )
										except (AttributeError, ModuleNotFoundError):
														return
										# We iterate over all the globals in self.obj in case we find "os.path.join"
										for attr in self.obj.__dir__():
														# We don't check for the name of the global, but rather if its value *is* "os.path.join".
														# This allows to patch renamed attributes like "from os.path import join as pjoin".
														if getattr(self.obj   ,__lowercase  ) is attr_value:
																		snake_case__	:      str     		=	getattr(self.obj   ,__lowercase  )
																		setattr(self.obj   ,__lowercase   ,self.new  )
						elif target_attr in globals()["__builtins__"]:  # if it'a s builtin like "open"
										snake_case__	:      str     		=	globals()['''__builtins__'''][target_attr]
										setattr(self.obj   ,__lowercase   ,self.new  )
						else:
										raise RuntimeError(F"""Tried to patch attribute {target_attr} instead of a submodule."""  )



		def __exit__(     self     :Tuple   ,*__lowercase     :Optional[int]  ):
						for attr in list(self.original  ):
										setattr(self.obj   ,__lowercase   ,self.original.pop(__lowercase  )  )



		def 	__lowerCamelCase			(     self     :Tuple  ):
						self.__enter__()
						self._active_patches.append(self  )






		def 	__lowerCamelCase			(     self     :Dict  ):
						try:
										self._active_patches.remove(self  )
						except ValueError:
										# If the patch hasn't been started this will fail
										return None

						return self.__exit__()

