"""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


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

class lowerCamelCase		(unittest.TestCase      ):






					'''simple docstring'''


					a   	=		["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]



					def 			lowerCAmelCase_					(						self			:				Tuple		)							->       Union[str, Any]:
									SCREAMING_SNAKE_CASE__		  =				0



					def 			lowerCAmelCase_					(						self			:				Optional[int]		)							->       Optional[Any]:
									SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h"		)
									self.assertIsInstance(_snake_case   , _snake_case		)



					def 			lowerCAmelCase_					(						self			:				int		)							->       List[Any]:
									with tempfile.TemporaryDirectory() as tmpdirname:
													SCREAMING_SNAKE_CASE__		  =				WavaVecaConfig()
													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h"		)

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

													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(_snake_case		)

									self.assertIsInstance(_snake_case   , _snake_case		)



					def 			lowerCAmelCase_					(						self			:				str		)							->       int:
									with tempfile.TemporaryDirectory() as tmpdirname:
													# copy relevant files
													copyfile(_snake_case   , os.path.join(_snake_case   , _snake_case		)		)
													copyfile(_snake_case   , os.path.join(_snake_case   , "vocab.json"		)		)

													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(_snake_case		)

									self.assertIsInstance(_snake_case   , _snake_case		)



					def 			lowerCAmelCase_					(						self			:				Optional[int]		)							->       Optional[int]:
									with tempfile.TemporaryDirectory() as tmpdirname:
													SCREAMING_SNAKE_CASE__		  =				WavaVecaFeatureExtractor()
													SCREAMING_SNAKE_CASE__		  =				AutoTokenizer.from_pretrained("facebook/wav2vec2-base-960h"		)

													SCREAMING_SNAKE_CASE__		  =				WavaVecaProcessor(_snake_case   , _snake_case		)

													# save in new folder
													processor.save_pretrained(_snake_case		)

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

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

													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(_snake_case		)

									self.assertIsInstance(_snake_case   , _snake_case		)



					def 			lowerCAmelCase_					(						self			:				Union[str, Any]		)							->       Union[str, Any]:
									with tempfile.TemporaryDirectory() as tmpdirname:
													SCREAMING_SNAKE_CASE__		  =				WavaVecaFeatureExtractor()
													SCREAMING_SNAKE_CASE__		  =				AutoTokenizer.from_pretrained("facebook/wav2vec2-base-960h"		)

													SCREAMING_SNAKE_CASE__		  =				WavaVecaProcessor(_snake_case   , _snake_case		)

													# save in new folder
													processor.save_pretrained(_snake_case		)

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

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

													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(_snake_case		)

									self.assertIsInstance(_snake_case   , _snake_case		)



					def 			lowerCAmelCase_					(						self			:				List[str]		)							->       List[str]:
									with tempfile.TemporaryDirectory() as tmpdirname:
													SCREAMING_SNAKE_CASE__		  =				WavaVecaConfig(processor_class="Wav2Vec2Processor"		)
													model_config.save_pretrained(_snake_case		)
													# copy relevant files
													copyfile(_snake_case   , os.path.join(_snake_case   , "vocab.json"		)		)
													# create emtpy sample processor
													with open(os.path.join(_snake_case   , _snake_case		)   , "w"		) as f:
																	f.write("{}"		)

													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(_snake_case		)

									self.assertIsInstance(_snake_case   , _snake_case		)



					def 			lowerCAmelCase_					(						self			:				Tuple		)							->       List[str]:
									# If remote code is not set, we will time out when asking whether to load the model.
									with self.assertRaises(_snake_case		):
													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained("hf-internal-testing/test_dynamic_processor"		)
									# If remote code is disabled, we can't load this config.
									with self.assertRaises(_snake_case		):
													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(
													    "hf-internal-testing/test_dynamic_processor"   , trust_remote_code=_snake_case		)

									SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained("hf-internal-testing/test_dynamic_processor"   , trust_remote_code=_snake_case		)
									self.assertTrue(processor.special_attribute_present		)
									self.assertEqual(processor.__class__.__name__   , "NewProcessor"		)

									SCREAMING_SNAKE_CASE__		  =				processor.feature_extractor
									self.assertTrue(feature_extractor.special_attribute_present		)
									self.assertEqual(feature_extractor.__class__.__name__   , "NewFeatureExtractor"		)

									SCREAMING_SNAKE_CASE__		  =				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
													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(
													    "hf-internal-testing/test_dynamic_processor"   , trust_remote_code=_snake_case   , use_fast=_snake_case		)
													SCREAMING_SNAKE_CASE__		  =				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		)							->       List[Any]:
									try:
													AutoConfig.register("custom"   , _snake_case		)
													AutoFeatureExtractor.register(_snake_case   , _snake_case		)
													AutoTokenizer.register(_snake_case   , slow_tokenizer_class=_snake_case		)
													AutoProcessor.register(_snake_case   , _snake_case		)
													# Trying to register something existing in the Transformers library will raise an error
													with self.assertRaises(_snake_case		):
																	AutoProcessor.register(_snake_case   , _snake_case		)

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

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

													SCREAMING_SNAKE_CASE__		  =				CustomProcessor(_snake_case   , _snake_case		)

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

									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			:				List[str]		)							->       List[str]:

									class lowerCamelCase		(_SCREAMING_SNAKE_CASE      ):






														'''simple docstring'''


														a   	=		False



									class lowerCamelCase		(_SCREAMING_SNAKE_CASE      ):






														'''simple docstring'''


														a   	=		False



									class lowerCamelCase		(_SCREAMING_SNAKE_CASE      ):






														'''simple docstring'''


														a   	=		"AutoFeatureExtractor"
														a   	=		"AutoTokenizer"
														a   	=		False

									try:
													AutoConfig.register("custom"   , _snake_case		)
													AutoFeatureExtractor.register(_snake_case   , _snake_case		)
													AutoTokenizer.register(_snake_case   , slow_tokenizer_class=_snake_case		)
													AutoProcessor.register(_snake_case   , _snake_case		)
													# If remote code is not set, the default is to use local classes.
													SCREAMING_SNAKE_CASE__		  =				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.
													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(
													    "hf-internal-testing/test_dynamic_processor"   , trust_remote_code=_snake_case		)
													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.
													SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(
													    "hf-internal-testing/test_dynamic_processor"   , trust_remote_code=_snake_case		)
													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			:				str		)							->       Dict:
									SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained("hf-internal-testing/tiny-random-bert"		)
									self.assertEqual(processor.__class__.__name__   , "BertTokenizerFast"		)



					def 			lowerCAmelCase_					(						self			:				Union[str, Any]		)							->       str:
									SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained("hf-internal-testing/tiny-random-convnext"		)
									self.assertEqual(processor.__class__.__name__   , "ConvNextImageProcessor"		)



@is_staging_test
class lowerCamelCase		(unittest.TestCase      ):






					'''simple docstring'''


					a   	=		["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]



					@classmethod
					def 			lowerCAmelCase_					(						cls			:				Tuple		)							->       Tuple:
									SCREAMING_SNAKE_CASE__		  =				TOKEN
									HfFolder.save_token(_snake_case		)



					@classmethod
					def 			lowerCAmelCase_					(						cls			:				Dict		)							->       Optional[Any]:
									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			:				List[str]		)							->       Optional[Any]:
									SCREAMING_SNAKE_CASE__		  =				WavaVecaProcessor.from_pretrained(_snake_case		)
									with tempfile.TemporaryDirectory() as tmp_dir:
													processor.save_pretrained(
													    os.path.join(_snake_case   , "test-processor"		)   , push_to_hub=_snake_case   , use_auth_token=self._token		)

													SCREAMING_SNAKE_CASE__		  =				WavaVecaProcessor.from_pretrained(F"""{USER}/test-processor"""		)
													for k, v in processor.feature_extractor.__dict__.items():
																	self.assertEqual(_snake_case   , getattr(new_processor.feature_extractor   , _snake_case		)		)
													self.assertDictEqual(new_processor.tokenizer.get_vocab()   , processor.tokenizer.get_vocab()		)



					def 			lowerCAmelCase_					(						self			:				Optional[int]		)							->       str:
									SCREAMING_SNAKE_CASE__		  =				WavaVecaProcessor.from_pretrained(_snake_case		)

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

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



					def 			lowerCAmelCase_					(						self			:				List[Any]		)							->       Optional[int]:
									CustomFeatureExtractor.register_for_auto_class()
									CustomTokenizer.register_for_auto_class()
									CustomProcessor.register_for_auto_class()

									SCREAMING_SNAKE_CASE__		  =				CustomFeatureExtractor.from_pretrained(_snake_case		)

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

									SCREAMING_SNAKE_CASE__		  =				CustomProcessor(_snake_case   , _snake_case		)

									with tempfile.TemporaryDirectory() as tmp_dir:
													create_repo(F"""{USER}/test-dynamic-processor"""   , token=self._token		)
													SCREAMING_SNAKE_CASE__		  =				Repository(_snake_case   , clone_from=F"""{USER}/test-dynamic-processor"""   , token=self._token		)
													processor.save_pretrained(_snake_case		)

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

													repo.push_to_hub()

									SCREAMING_SNAKE_CASE__		  =				AutoProcessor.from_pretrained(F"""{USER}/test-dynamic-processor"""   , trust_remote_code=_snake_case		)
									# 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"""






def 						SCREAMING_SNAKE_CASE					(  __UpperCAmelCase       )							->   int:

				if not isinstance(__UpperCAmelCase  ,					__UpperCAmelCase       ):
								raise ValueError("multiplicative_persistence() only accepts integral values"       )
				if num < 0:
								raise ValueError("multiplicative_persistence() does not accept negative values"       )

				SCREAMING_SNAKE_CASE__		  =				0
				SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

				while len(__UpperCAmelCase       ) != 1:
								SCREAMING_SNAKE_CASE__		  =				[int(__UpperCAmelCase       ) for i in num_string]

								SCREAMING_SNAKE_CASE__		  =				1
								for i in range(0  ,					len(__UpperCAmelCase       )       ):
												total *= numbers[i]

								SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

								steps += 1
				return steps




def 						SCREAMING_SNAKE_CASE					(  __UpperCAmelCase       )							->   int:

				if not isinstance(__UpperCAmelCase  ,					__UpperCAmelCase       ):
								raise ValueError("additive_persistence() only accepts integral values"       )
				if num < 0:
								raise ValueError("additive_persistence() does not accept negative values"       )

				SCREAMING_SNAKE_CASE__		  =				0
				SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

				while len(__UpperCAmelCase       ) != 1:
								SCREAMING_SNAKE_CASE__		  =				[int(__UpperCAmelCase       ) for i in num_string]

								SCREAMING_SNAKE_CASE__		  =				0
								for i in range(0  ,					len(__UpperCAmelCase       )       ):
												total += numbers[i]

								SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

								steps += 1
				return steps


if __name__ == "__main__":
						import doctest

						doctest.testmod()




