



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


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

class 				snake_case__(unittest.TestCase      ):

					"""simple docstring"""






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







					def 			snake_case					(							self :     List[Any]					):
									lowercase__					:   Optional[int]													=					0







					def 			snake_case					(							self :     int					):
									lowercase__					:   Optional[Any]													=					AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h"					)
									self.assertIsInstance(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Optional[int]					):
									with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase__					:   Dict													=					WavaVecaConfig()
													lowercase__					:   int													=					AutoProcessor.from_pretrained("facebook/wav2vec2-base-960h"					)

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

													lowercase__					:   str													=					AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)

									self.assertIsInstance(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Tuple					):
									with tempfile.TemporaryDirectory() as tmpdirname:
													# copy relevant files
													copyfile(SCREAMING_SNAKE_CASE				,       os.path.join(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)					)
													copyfile(SCREAMING_SNAKE_CASE				,       os.path.join(SCREAMING_SNAKE_CASE				,       "vocab.json"					)					)

													lowercase__					:   Any													=					AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)

									self.assertIsInstance(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Any					):
									with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase__					:   int													=					WavaVecaFeatureExtractor()
													lowercase__					:   str													=					AutoTokenizer.from_pretrained("facebook/wav2vec2-base-960h"					)

													lowercase__					:   Union[str, Any]													=					WavaVecaProcessor(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)

													# save in new folder
													processor.save_pretrained(SCREAMING_SNAKE_CASE					)

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

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

													lowercase__					:   Any													=					AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)

									self.assertIsInstance(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Tuple					):
									with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase__					:   List[str]													=					WavaVecaFeatureExtractor()
													lowercase__					:   Optional[Any]													=					AutoTokenizer.from_pretrained("facebook/wav2vec2-base-960h"					)

													lowercase__					:   List[Any]													=					WavaVecaProcessor(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)

													# save in new folder
													processor.save_pretrained(SCREAMING_SNAKE_CASE					)

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

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

													lowercase__					:   Any													=					AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)

									self.assertIsInstance(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     Optional[Any]					):
									with tempfile.TemporaryDirectory() as tmpdirname:
													lowercase__					:   Optional[int]													=					WavaVecaConfig(processor_class="Wav2Vec2Processor"					)
													model_config.save_pretrained(SCREAMING_SNAKE_CASE					)
													# copy relevant files
													copyfile(SCREAMING_SNAKE_CASE				,       os.path.join(SCREAMING_SNAKE_CASE				,       "vocab.json"					)					)
													# create emtpy sample processor
													with open(os.path.join(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)				,       "w"					) as f:
																	f.write("{}"					)

													lowercase__					:   Union[str, Any]													=					AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)

									self.assertIsInstance(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)







					def 			snake_case					(							self :     str					):
									# If remote code is not set, we will time out when asking whether to load the model.
									with self.assertRaises(SCREAMING_SNAKE_CASE					):
													lowercase__					:   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(SCREAMING_SNAKE_CASE					):
													lowercase__					:   List[str]													=					AutoProcessor.from_pretrained(
													    "hf-internal-testing/test_dynamic_processor"				,       trust_remote_code=SCREAMING_SNAKE_CASE					)

									lowercase__					:   Optional[Any]													=					AutoProcessor.from_pretrained("hf-internal-testing/test_dynamic_processor"				,       trust_remote_code=SCREAMING_SNAKE_CASE					)
									self.assertTrue(processor.special_attribute_present					)
									self.assertEqual(processor.__class__.__name__				,       "NewProcessor"					)

									lowercase__					:   Any													=					processor.feature_extractor
									self.assertTrue(feature_extractor.special_attribute_present					)
									self.assertEqual(feature_extractor.__class__.__name__				,       "NewFeatureExtractor"					)

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

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

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

													lowercase__					:   Optional[int]													=					CustomProcessor(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)

													with tempfile.TemporaryDirectory() as tmp_dir:
																	processor.save_pretrained(SCREAMING_SNAKE_CASE					)
																	lowercase__					:   str													=					AutoProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)
																	self.assertIsInstance(SCREAMING_SNAKE_CASE				,       SCREAMING_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 			snake_case					(							self :     List[str]					):

									class 				snake_case__(_UpperCamelCase      ):

														"""simple docstring"""






														lowercase_								=		False






									class 				snake_case__(_UpperCamelCase      ):

														"""simple docstring"""






														lowercase_								=		False






									class 				snake_case__(_UpperCamelCase      ):

														"""simple docstring"""






														lowercase_								=		"""AutoFeatureExtractor"""
														lowercase_								=		"""AutoTokenizer"""
														lowercase_								=		False

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







					def 			snake_case					(							self :     Tuple					):
									lowercase__					:   Any													=					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"""






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







					@classmethod
					def 			snake_case					(							cls :     Optional[int]					):
									lowercase__					:   int													=					TOKEN
									HfFolder.save_token(SCREAMING_SNAKE_CASE					)







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

													lowercase__					:   List[str]													=					WavaVecaProcessor.from_pretrained(f"""{USER}/test-processor"""					)
													for k, v in processor.feature_extractor.__dict__.items():
																	self.assertEqual(SCREAMING_SNAKE_CASE				,       getattr(new_processor.feature_extractor				,       SCREAMING_SNAKE_CASE					)					)
													self.assertDictEqual(new_processor.tokenizer.get_vocab()				,       processor.tokenizer.get_vocab()					)







					def 			snake_case					(							self :     str					):
									lowercase__					:   List[Any]													=					WavaVecaProcessor.from_pretrained(SCREAMING_SNAKE_CASE					)

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

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







					def 			snake_case					(							self :     Tuple					):
									CustomFeatureExtractor.register_for_auto_class()
									CustomTokenizer.register_for_auto_class()
									CustomProcessor.register_for_auto_class()

									lowercase__					:   int													=					CustomFeatureExtractor.from_pretrained(SCREAMING_SNAKE_CASE					)

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

									lowercase__					:   Optional[Any]													=					CustomProcessor(SCREAMING_SNAKE_CASE				,       SCREAMING_SNAKE_CASE					)

									with tempfile.TemporaryDirectory() as tmp_dir:
													create_repo(f"""{USER}/test-dynamic-processor"""				,       token=self._token					)
													lowercase__					:   List[str]													=					Repository(SCREAMING_SNAKE_CASE				,       clone_from=f"""{USER}/test-dynamic-processor"""				,       token=self._token					)
													processor.save_pretrained(SCREAMING_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(SCREAMING_SNAKE_CASE				,       "tokenizer_config.json"					)					) as f:
																	lowercase__					:   Any													=					json.load(SCREAMING_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(SCREAMING_SNAKE_CASE				,       "custom_feature_extraction.py"					)					)					)
													self.assertTrue(os.path.isfile(os.path.join(SCREAMING_SNAKE_CASE				,       "custom_tokenization.py"					)					)					)
													self.assertTrue(os.path.isfile(os.path.join(SCREAMING_SNAKE_CASE				,       "custom_processing.py"					)					)					)

													repo.push_to_hub()

									lowercase__					:   str													=					AutoProcessor.from_pretrained(f"""{USER}/test-dynamic-processor"""				,       trust_remote_code=SCREAMING_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"					)









import unittest

from transformers import is_flax_available
from transformers.testing_utils import require_flax, require_sentencepiece, require_tokenizers, require_torch, slow


if is_flax_available():
				import optax
				from flax.training.common_utils import onehot

				from transformers import AutoTokenizer, FlaxMTaForConditionalGeneration
				from transformers.models.ta.modeling_flax_ta import shift_tokens_right

@require_torch
@require_sentencepiece
@require_tokenizers
@require_flax
class 				snake_case__(unittest.TestCase      ):

					"""simple docstring"""






					@slow
					def 			snake_case					(							self :     Optional[int]					):
									lowercase__					:   Dict													=					FlaxMTaForConditionalGeneration.from_pretrained("google/mt5-small"					)
									lowercase__					:   Dict													=					AutoTokenizer.from_pretrained("google/mt5-small"					)

									lowercase__					:   Optional[Any]													=					tokenizer("Hello there"				,       return_tensors="np"					).input_ids
									lowercase__					:   Optional[Any]													=					tokenizer("Hi I am"				,       return_tensors="np"					).input_ids

									lowercase__					:   int													=					shift_tokens_right(SCREAMING_SNAKE_CASE				,       model.config.pad_token_id				,       model.config.decoder_start_token_id					)

									lowercase__					:   Tuple													=					model(SCREAMING_SNAKE_CASE				,       decoder_input_ids=SCREAMING_SNAKE_CASE					).logits
									lowercase__					:   Dict													=					optax.softmax_cross_entropy(SCREAMING_SNAKE_CASE				,       onehot(SCREAMING_SNAKE_CASE				,       logits.shape[-1]					)					).mean()

									lowercase__					:   Union[str, Any]													=					-(labels.shape[-1] * loss.item())

									lowercase__					:   Union[str, Any]													=					-84.9_127
									self.assertTrue(abs(mtf_score - EXPECTED_SCORE					) < 1E-4					)




