






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			:	Optional[Any]								=get_tests_dir('fixtures/dummy_feature_extractor_config.json')
_UpperCamelCase			:	Optional[Any]								=get_tests_dir('fixtures/vocab.json')
_UpperCamelCase			:	Union[str, Any]								=get_tests_dir('fixtures')






class 						UpperCAmelCase__ (							unittest.TestCase							):
		__snake_case      :       List[str]     =   ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]







		def  A__      (					self		):
						_A	:				Optional[Any]   =      0







		def  A__      (					self		):
						_A	:				Union[str, Any]   =      AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''		)
						self.assertIsInstance(A__       ,A__		)







		def  A__      (					self		):
						with tempfile.TemporaryDirectory() as tmpdirname:
										_A	:				Optional[int]   =      WavaVecaConfig()
										_A	:				Any   =      AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''		)

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

										_A	:				Union[str, Any]   =      AutoProcessor.from_pretrained(A__		)

						self.assertIsInstance(A__       ,A__		)







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

										_A	:				Union[str, Any]   =      AutoProcessor.from_pretrained(A__		)

						self.assertIsInstance(A__       ,A__		)







		def  A__      (					self		):
						with tempfile.TemporaryDirectory() as tmpdirname:
										_A	:				List[Any]   =      WavaVecaFeatureExtractor()
										_A	:				Optional[int]   =      AutoTokenizer.from_pretrained('''facebook/wav2vec2-base-960h'''		)

										_A	:				List[str]   =      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	:				int   =      json.load(A__		)
														config_dict.pop('''processor_class'''		)

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

										_A	:				Tuple   =      AutoProcessor.from_pretrained(A__		)

						self.assertIsInstance(A__       ,A__		)







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

										_A	:				List[Any]   =      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	:				Any   =      json.load(A__		)
														config_dict.pop('''processor_class'''		)

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

										_A	:				Optional[Any]   =      AutoProcessor.from_pretrained(A__		)

						self.assertIsInstance(A__       ,A__		)







		def  A__      (					self		):
						with tempfile.TemporaryDirectory() as tmpdirname:
										_A	:				int   =      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	:				Tuple   =      AutoProcessor.from_pretrained(A__		)

						self.assertIsInstance(A__       ,A__		)







		def  A__      (					self		):
						# If remote code is not set, we will time out when asking whether to load the model.
						with self.assertRaises(A__		):
										_A	:				Optional[int]   =      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	:				Dict   =      AutoProcessor.from_pretrained(
										    '''hf-internal-testing/test_dynamic_processor'''       ,trust_remote_code=A__		)

						_A	:				Optional[int]   =      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	:				Any   =      processor.feature_extractor
						self.assertTrue(feature_extractor.special_attribute_present		)
						self.assertEqual(feature_extractor.__class__.__name__       ,'''NewFeatureExtractor'''		)

						_A	:				Optional[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	:				int   =      AutoProcessor.from_pretrained(
										    '''hf-internal-testing/test_dynamic_processor'''       ,trust_remote_code=A__       ,use_fast=A__		)
										_A	:				Optional[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  A__      (					self		):
						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	:				Union[str, Any]   =      CustomFeatureExtractor.from_pretrained(A__		)

										with tempfile.TemporaryDirectory() as tmp_dir:
														_A	:				List[str]   =      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	:				Optional[int]   =      CustomTokenizer(A__		)

										_A	:				Dict   =      CustomProcessor(A__       ,A__		)

										with tempfile.TemporaryDirectory() as tmp_dir:
														processor.save_pretrained(A__		)
														_A	:				Tuple   =      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  A__      (					self		):






						class 						UpperCAmelCase__ (							__snake_case							):
								__snake_case      :       List[Any]     =   False






						class 						UpperCAmelCase__ (							__snake_case							):
								__snake_case      :       List[str]     =   False






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







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






@is_staging_test
class 						UpperCAmelCase__ (							unittest.TestCase							):
		__snake_case      :       int     =   ["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]







		@classmethod
		def  A__      (					cls		):
						_A	:				List[Any]   =      TOKEN
						HfFolder.save_token(A__		)







		@classmethod
		def  A__      (					cls		):
						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  A__      (					self		):
						_A	:				int   =      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	:				Dict   =      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  A__      (					self		):
						_A	:				Optional[int]   =      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	:				Optional[int]   =      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  A__      (					self		):
						CustomFeatureExtractor.register_for_auto_class()
						CustomTokenizer.register_for_auto_class()
						CustomProcessor.register_for_auto_class()

						_A	:				Tuple   =      CustomFeatureExtractor.from_pretrained(A__		)

						with tempfile.TemporaryDirectory() as tmp_dir:
										_A	:				Optional[int]   =      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	:				Optional[int]   =      CustomTokenizer(A__		)

						_A	:				int   =      CustomProcessor(A__       ,A__		)

						with tempfile.TemporaryDirectory() as tmp_dir:
										create_repo(f"""{USER}/test-dynamic-processor"""       ,token=self._token		)
										_A	:				Dict   =      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	:				Tuple   =      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	:				Any   =      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'''		)











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


_UpperCamelCase			:	Optional[int]								=logging.get_logger(__name__)

_UpperCamelCase			:	Tuple								={
    'google/canine-s': 'https://huggingface.co/google/canine-s/resolve/main/config.json',
    # See all CANINE models at https://huggingface.co/models?filter=canine
}






class 						UpperCAmelCase__ (							__snake_case							):
		__snake_case      :       Any     =   "canine"







		def __init__(					self       ,A__=768       ,A__=12       ,A__=12       ,A__=3072       ,A__="gelu"       ,A__=0.1       ,A__=0.1       ,A__=16384       ,A__=16       ,A__=0.02       ,A__=1E-12       ,A__=0       ,A__=0xE_000       ,A__=0xE_001       ,A__=4       ,A__=4       ,A__=8       ,A__=16384       ,A__=128       ,**A__       ,):
						super().__init__(pad_token_id=A__       ,bos_token_id=A__       ,eos_token_id=A__       ,**A__		)

						_A	:				int   =      max_position_embeddings
						_A	:				Dict   =      hidden_size
						_A	:				List[str]   =      num_hidden_layers
						_A	:				Optional[Any]   =      num_attention_heads
						_A	:				Optional[Any]   =      intermediate_size
						_A	:				Optional[Any]   =      hidden_act
						_A	:				str   =      hidden_dropout_prob
						_A	:				Optional[int]   =      attention_probs_dropout_prob
						_A	:				Union[str, Any]   =      initializer_range
						_A	:				Dict   =      type_vocab_size
						_A	:				Optional[Any]   =      layer_norm_eps

						# Character config:
						_A	:				Union[str, Any]   =      downsampling_rate
						_A	:				List[str]   =      upsampling_kernel_size
						_A	:				str   =      num_hash_functions
						_A	:				Optional[Any]   =      num_hash_buckets
						_A	:				Optional[Any]   =      local_transformer_stride



