






"""simple docstring"""






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


SCREAMING_SNAKE_CASE						:						Optional[Any]					   =       get_tests_dir('''fixtures/dummy_feature_extractor_config.json''')
SCREAMING_SNAKE_CASE						:						List[Any]					   =       get_tests_dir('''fixtures/vocab.json''')
SCREAMING_SNAKE_CASE						:						Optional[Any]					   =       get_tests_dir('''fixtures''')





class       snake_case_       (    unittest.TestCase       ):



	"""simple docstring"""

	SCREAMING_SNAKE_CASE_:							Any   			=					["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       0


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       AutoProcessor.from_pretrained('facebook/wav2vec2-base-960h'						)
					self.assertIsInstance(__a  ,					__a						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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

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

									A__							 =       AutoProcessor.from_pretrained(__a						)

					self.assertIsInstance(__a  ,					__a						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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

									A__							 =       AutoProcessor.from_pretrained(__a						)

					self.assertIsInstance(__a  ,					__a						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					with tempfile.TemporaryDirectory() as tmpdirname:
									A__							 =       WavaVecaFeatureExtractor()
									A__							 =       AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'						)

									A__							 =       WavaVecaProcessor(__a  ,					__a						)

									# save in new folder
									processor.save_pretrained(__a						)

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

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

									A__							 =       AutoProcessor.from_pretrained(__a						)

					self.assertIsInstance(__a  ,					__a						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					with tempfile.TemporaryDirectory() as tmpdirname:
									A__							 =       WavaVecaFeatureExtractor()
									A__							 =       AutoTokenizer.from_pretrained('facebook/wav2vec2-base-960h'						)

									A__							 =       WavaVecaProcessor(__a  ,					__a						)

									# save in new folder
									processor.save_pretrained(__a						)

									# drop `processor_class` in feature extractor
									with open(os.path.join(__a  ,					__a						)  ,					'r'						) as f:
													A__							 =       json.load(__a						)
													config_dict.pop('processor_class'						)

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

									A__							 =       AutoProcessor.from_pretrained(__a						)

					self.assertIsInstance(__a  ,					__a						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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

									A__							 =       AutoProcessor.from_pretrained(__a						)

					self.assertIsInstance(__a  ,					__a						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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

					A__							 =       AutoProcessor.from_pretrained('hf-internal-testing/test_dynamic_processor'  ,					trust_remote_code=__a						)
					self.assertTrue(processor.special_attribute_present						)
					self.assertEqual(processor.__class__.__name__  ,					'NewProcessor'						)

					A__							 =       processor.feature_extractor
					self.assertTrue(feature_extractor.special_attribute_present						)
					self.assertEqual(feature_extractor.__class__.__name__  ,					'NewFeatureExtractor'						)

					A__							 =       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__							 =       AutoProcessor.from_pretrained(
									    'hf-internal-testing/test_dynamic_processor'  ,					trust_remote_code=__a  ,					use_fast=__a						)
									A__							 =       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       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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

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

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

									A__							 =       CustomProcessor(__a  ,					__a						)

									with tempfile.TemporaryDirectory() as tmp_dir:
													processor.save_pretrained(__a						)
													A__							 =       AutoProcessor.from_pretrained(__a						)
													self.assertIsInstance(__a  ,					__a						)

					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       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					class       snake_case_       (    _lowerCamelCase       ):



						"""simple docstring"""

						SCREAMING_SNAKE_CASE_:							Any   			=					False





					class       snake_case_       (    _lowerCamelCase       ):



						"""simple docstring"""

						SCREAMING_SNAKE_CASE_:							Union[str, Any]   			=					False





					class       snake_case_       (    _lowerCamelCase       ):



						"""simple docstring"""

						SCREAMING_SNAKE_CASE_:							Dict   			=					"""AutoFeatureExtractor"""
						SCREAMING_SNAKE_CASE_:							Union[str, Any]   			=					"""AutoTokenizer"""
						SCREAMING_SNAKE_CASE_:							List[str]   			=					False

					try:
									AutoConfig.register('custom'  ,					__a						)
									AutoFeatureExtractor.register(__a  ,					__a						)
									AutoTokenizer.register(__a  ,					slow_tokenizer_class=__a						)
									AutoProcessor.register(__a  ,					__a						)
									# If remote code is not set, the default is to use local classes.
									A__							 =       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__							 =       AutoProcessor.from_pretrained(
									    'hf-internal-testing/test_dynamic_processor'  ,					trust_remote_code=__a						)
									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__							 =       AutoProcessor.from_pretrained(
									    'hf-internal-testing/test_dynamic_processor'  ,					trust_remote_code=__a						)
									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       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       AutoProcessor.from_pretrained('hf-internal-testing/tiny-random-bert'						)
					self.assertEqual(processor.__class__.__name__  ,					'BertTokenizerFast'						)






	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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





@is_staging_test
class       snake_case_       (    unittest.TestCase       ):



	"""simple docstring"""

	SCREAMING_SNAKE_CASE_:							str   			=					["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]


	@classmethod
	def       _UpperCAmelCase						(      cls						):


					"""simple docstring"""

					A__							 =       TOKEN
					HfFolder.save_token(__a						)


	@classmethod
	def       _UpperCAmelCase						(      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       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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

									A__							 =       WavaVecaProcessor.from_pretrained(f'''{USER}/test-processor'''						)
									for k, v in processor.feature_extractor.__dict__.items():
													self.assertEqual(__a  ,					getattr(new_processor.feature_extractor  ,					__a						)						)
									self.assertDictEqual(new_processor.tokenizer.get_vocab()  ,					processor.tokenizer.get_vocab()						)


	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

					A__							 =       WavaVecaProcessor.from_pretrained(__a						)

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

									A__							 =       WavaVecaProcessor.from_pretrained('valid_org/test-processor-org'						)
									for k, v in processor.feature_extractor.__dict__.items():
													self.assertEqual(__a  ,					getattr(new_processor.feature_extractor  ,					__a						)						)
									self.assertDictEqual(new_processor.tokenizer.get_vocab()  ,					processor.tokenizer.get_vocab()						)






	def       _UpperCAmelCase						(      self						):


					"""simple docstring"""

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

					A__							 =       CustomFeatureExtractor.from_pretrained(__a						)

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

					A__							 =       CustomProcessor(__a  ,					__a						)

					with tempfile.TemporaryDirectory() as tmp_dir:
									create_repo(f'''{USER}/test-dynamic-processor'''  ,					token=self._token						)
									A__							 =       Repository(__a  ,					clone_from=f'''{USER}/test-dynamic-processor'''  ,					token=self._token						)
									processor.save_pretrained(__a						)

									# 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(__a  ,					'tokenizer_config.json'						)						) as f:
													A__							 =       json.load(__a						)
									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(__a  ,					'custom_feature_extraction.py'						)						)						)
									self.assertTrue(os.path.isfile(os.path.join(__a  ,					'custom_tokenization.py'						)						)						)
									self.assertTrue(os.path.isfile(os.path.join(__a  ,					'custom_processing.py'						)						)						)

									repo.push_to_hub()

					A__							 =       AutoProcessor.from_pretrained(f'''{USER}/test-dynamic-processor'''  ,					trust_remote_code=__a						)
					# 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'						)










"""simple docstring"""






from ...configuration_utils import PretrainedConfig
from ...utils import logging


SCREAMING_SNAKE_CASE						:						Any					   =       logging.get_logger(__name__)





class       snake_case_       (    _lowerCamelCase       ):



	"""simple docstring"""

	SCREAMING_SNAKE_CASE_:							Union[str, Any]   			=					"""timm_backbone"""


	def __init__(      self  ,					__a=None  ,					__a=3  ,					__a=True  ,					__a=True  ,					__a=None  ,					**__a  ,					):


					"""simple docstring"""

					super().__init__(**__a						)
					A__							 =       backbone
					A__							 =       num_channels
					A__							 =       features_only
					A__							 =       use_pretrained_backbone
					A__							 =       True
					A__							 =       out_indices if out_indices is not None else (-1,)


