
import argparse
import json
import os
import re
import shutil

import torch

from transformers import BioGptConfig, BioGptForCausalLM
from transformers.models.biogpt.tokenization_biogpt import VOCAB_FILES_NAMES
from transformers.tokenization_utils_base import TOKENIZER_CONFIG_FILE
from transformers.utils import WEIGHTS_NAME, logging


logging.set_verbosity_warning()

_SCREAMING_SNAKE_CASE    :      List[str]          =			2







class  UpperCamelCase__		:







	'''simple docstring'''





	def __init__(			self				,       *,  # begin keyword-only arguments
	    UpperCamelCase__="<s>"				,       UpperCamelCase__="<pad>"				,       UpperCamelCase__="</s>"				,       UpperCamelCase__="<unk>"				,       UpperCamelCase__=None				,       ):
					A__       ,							A__       ,							A__       ,							A__  :	str         =							bos, unk, pad, eos
					A__  :	Optional[int]         =							[]
					A__  :	List[str]         =							[]
					A__  :	Any         =							{}
					A__  :	str         =							self.add_symbol(UpperCamelCase__		)
					A__  :	Optional[Any]         =							self.add_symbol(UpperCamelCase__		)
					A__  :	Optional[Any]         =							self.add_symbol(UpperCamelCase__		)
					A__  :	str         =							self.add_symbol(UpperCamelCase__		)
					if extra_special_symbols:
									for s in extra_special_symbols:
													self.add_symbol(UpperCamelCase__		)
					A__  :	Optional[int]         =							len(self.symbols		)





	def __eq__(			self				,       UpperCamelCase__		):
					return self.indices == other.indices





	def __getitem__(			self				,       UpperCamelCase__		):
					if idx < len(self.symbols		):
									return self.symbols[idx]
					return self.unk_word





	def __len__(			self		):
					return len(self.symbols		)





	def __contains__(			self				,       UpperCamelCase__		):
					return sym in self.indices





	@classmethod
	def   __snake_case				(			cls				,       UpperCamelCase__		):
					A__  :	int         =							cls()
					d.add_from_file(UpperCamelCase__		)
					return d





	def   __snake_case				(			self				,       UpperCamelCase__				,       UpperCamelCase__=1				,       UpperCamelCase__=False		):
					if word in self.indices and not overwrite:
									A__  :	Optional[Any]         =							self.indices[word]
									A__  :	Any         =							self.count[idx] + n
									return idx
					else:
									A__  :	Optional[Any]         =							len(self.symbols		)
									A__  :	List[str]         =							idx
									self.symbols.append(UpperCamelCase__		)
									self.count.append(UpperCamelCase__		)
									return idx





	def   __snake_case				(			self				,       UpperCamelCase__		):
					return 0





	def   __snake_case				(			self				,       UpperCamelCase__		):
					if isinstance(UpperCamelCase__				,       UpperCamelCase__		):
									try:
													with open(UpperCamelCase__				,       '''r'''				,       encoding='''utf-8'''		) as fd:
																	self.add_from_file(UpperCamelCase__		)
									except FileNotFoundError as fnfe:
													raise fnfe
									except UnicodeError:
													raise Exception('''Incorrect encoding detected in {}, please rebuild the dataset'''.format(UpperCamelCase__		)		)
									return

					A__  :	List[Any]         =							f.readlines()
					A__  :	Any         =							self._load_meta(UpperCamelCase__		)

					for line in lines[indices_start_line:]:
									try:
													A__       ,							A__  :	str         =							line.rstrip().rsplit(''' '''				,       1		)
													if field == "#fairseq:overwrite":
																	A__  :	List[Any]         =							True
																	A__       ,							A__  :	List[Any]         =							line.rsplit(''' '''				,       1		)
													else:
																	A__  :	Optional[int]         =							False
													A__  :	Dict         =							int(UpperCamelCase__		)
													A__  :	int         =							line
													if word in self and not overwrite:
																	raise RuntimeError(
																	    '''Duplicate word found when loading Dictionary: \'{}\'. '''
																	    '''Duplicate words can overwrite earlier ones by adding the '''
																	    '''#fairseq:overwrite flag at the end of the corresponding row '''
																	    '''in the dictionary file. If using the Camembert model, please '''
																	    '''download an updated copy of the model file.'''.format(UpperCamelCase__		)		)
													self.add_symbol(UpperCamelCase__				,       n=UpperCamelCase__				,       overwrite=UpperCamelCase__		)
									except ValueError:
													raise ValueError('''Incorrect dictionary format, expected \'<token> <cnt> [flags]\''''		)




def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Union[str, Any]       ) ->       Union[str, Any]:

				"""simple docstring"""







				A__  :	Optional[int]         =							dict((re.sub(R'''@@$'''							,						''''''							,						__UpperCamelCase       ), v) if k.endswith('''@@'''       ) else (re.sub(R'''$'''							,						'''</w>'''							,						__UpperCamelCase       ), v) for k, v in d.items()       )
				A__  :	str         =							'''<s> <pad> </s> <unk>'''.split()
				# restore the special tokens
				for k in keep_keys:
								del da[F"{k}</w>"]
								A__  :	int         =							d[k]  # restore
				return da

def 				SCREAMING_SNAKE_CASE     (      __UpperCamelCase :						Tuple							,						__UpperCamelCase :						int       ) ->       Dict:

				"""simple docstring"""







				if not os.path.exists(__UpperCamelCase       ):
								raise ValueError(F"path {biogpt_checkpoint_path} does not exist!"       )
				os.makedirs(__UpperCamelCase							,						exist_ok=__UpperCamelCase       )
				print(F"Writing results to {pytorch_dump_folder_path}"       )

				# handle various types of models

				A__  :	Tuple         =							os.path.join(__UpperCamelCase							,						'''checkpoint.pt'''       )
				if not os.path.isfile(__UpperCamelCase       ):
								raise ValueError(F"path to the file {checkpoint_file} does not exist!"       )
				A__  :	Dict         =							torch.load(__UpperCamelCase							,						map_location='''cpu'''       )

				A__  :	Union[str, Any]         =							chkpt['''cfg''']['''model''']

				# dicts
				A__  :	Dict         =							os.path.join(__UpperCamelCase							,						'''dict.txt'''       )
				if not os.path.isfile(__UpperCamelCase       ):
								raise ValueError(F"path to the file {dict_file} does not exist!"       )
				A__  :	Union[str, Any]         =							Dictionary.load(__UpperCamelCase       )
				A__  :	Any         =							rewrite_dict_keys(src_dict.indices       )
				A__  :	Tuple         =							len(__UpperCamelCase       )
				A__  :	List[str]         =							os.path.join(__UpperCamelCase							,						VOCAB_FILES_NAMES['''vocab_file''']       )
				print(F"Generating {src_vocab_file} of {src_vocab_size} records"       )
				with open(__UpperCamelCase							,						'''w'''							,						encoding='''utf-8'''       ) as f:
								f.write(json.dumps(__UpperCamelCase							,						ensure_ascii=__UpperCamelCase							,						indent=__UpperCamelCase       )       )

				# merges_file (bpecodes)
				A__  :	List[Any]         =							os.path.join(__UpperCamelCase							,						'''bpecodes'''       )
				if not os.path.isfile(__UpperCamelCase       ):
								raise ValueError(F"path to the file {bpecodes_file} does not exist!"       )

				A__  :	Optional[int]         =							os.path.join(__UpperCamelCase							,						VOCAB_FILES_NAMES['''merges_file''']       )
				shutil.copyfile(__UpperCamelCase							,						__UpperCamelCase       )

				# model config
				A__  :	Optional[Any]         =							os.path.join(__UpperCamelCase							,						'''config.json'''       )

				A__  :	Optional[Any]         =							{
				    '''activation_dropout''': args['''activation_dropout'''],
				    '''architectures''': ['''BioGptForCausalLM'''],
				    '''attention_probs_dropout_prob''': args['''attention_dropout'''],
				    '''bos_token_id''': 0,
				    '''eos_token_id''': 2,
				    '''hidden_act''': args['''activation_fn'''],
				    '''hidden_dropout_prob''': args['''dropout'''],
				    '''hidden_size''': args['''decoder_embed_dim'''],
				    '''initializer_range''': 0.0_2,
				    '''intermediate_size''': args['''decoder_ffn_embed_dim'''],
				    '''layer_norm_eps''': 1e-12,
				    '''layerdrop''': args['''decoder_layerdrop'''],
				    '''max_position_embeddings''': args['''max_target_positions'''],
				    '''model_type''': '''biogpt''',
				    '''num_attention_heads''': args['''decoder_attention_heads'''],
				    '''num_hidden_layers''': args['''decoder_layers'''],
				    '''pad_token_id''': 1,
				    '''scale_embedding''': not args['''no_scale_embedding'''],
				    '''tie_word_embeddings''': args['''share_decoder_input_output_embed'''],
				    '''vocab_size''': src_vocab_size,
				}

				# good hparam defaults to start with

				print(F"Generating {biogpt_model_config_file}"       )
				with open(__UpperCamelCase							,						'''w'''							,						encoding='''utf-8'''       ) as f:
								f.write(json.dumps(__UpperCamelCase							,						ensure_ascii=__UpperCamelCase							,						indent=__UpperCamelCase       )       )

				# tokenizer config
				A__  :	Tuple         =							os.path.join(__UpperCamelCase							,						__UpperCamelCase       )

				A__  :	List[Any]         =							{
				    '''bos_token''': '''<s>''',
				    '''eos_token''': '''</s>''',
				    '''model_max_length''': 10_24,
				    '''pad_token''': '''<pad>''',
				    '''special_tokens_map_file''': None,
				    '''tokenizer_class''': '''BioGptTokenizer''',
				    '''unk_token''': '''<unk>''',
				}

				print(F"Generating {biogpt_tokenizer_config_file}"       )
				with open(__UpperCamelCase							,						'''w'''							,						encoding='''utf-8'''       ) as f:
								f.write(json.dumps(__UpperCamelCase							,						ensure_ascii=__UpperCamelCase							,						indent=__UpperCamelCase       )       )

				# model
				A__  :	Union[str, Any]         =							chkpt['''model''']

				# remove unneeded keys
				A__  :	str         =							[
				    '''decoder.version''',
				]
				for k in ignore_keys:
								model_state_dict.pop(__UpperCamelCase							,						__UpperCamelCase       )

				A__  :	Optional[int]         =							list(model_state_dict.keys()       )
				for layer_name in layer_names:
								if layer_name.endswith('''output_projection.weight'''       ):
												A__  :	int         =							model_state_dict.pop(__UpperCamelCase       )
								else:
												A__  :	Tuple         =							model_state_dict.pop(__UpperCamelCase       )

				A__  :	Optional[int]         =							BioGptConfig.from_pretrained(__UpperCamelCase       )
				A__  :	Dict         =							BioGptForCausalLM(__UpperCamelCase       )

				# check that it loads ok
				model_new.load_state_dict(__UpperCamelCase       )

				# save
				A__  :	Dict         =							os.path.join(__UpperCamelCase							,						__UpperCamelCase       )
				print(F"Generating {pytorch_weights_dump_path}"       )
				torch.save(__UpperCamelCase							,						__UpperCamelCase       )

				print('''Conversion is done!'''       )


if __name__ == "__main__":
					_SCREAMING_SNAKE_CASE    :      Union[str, Any]          =			argparse.ArgumentParser()
					# Required parameters
					parser.add_argument(
					    '--biogpt_checkpoint_path',
					    default=None,
					    type=str,
					    required=True,
					    help=(
					        'Path to the official PyTorch checkpoint file which is expected to reside in the dump dir with dicts,'
					        ' bpecodes, etc.'
					    ),
					)
					parser.add_argument(
					    '--pytorch_dump_folder_path', default=None, type=str, required=True, help='Path to the output PyTorch model.'
					)
					_SCREAMING_SNAKE_CASE    :      str          =			parser.parse_args()
					convert_biogpt_checkpoint_to_pytorch(args.biogpt_checkpoint_path, args.pytorch_dump_folder_path)

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    :      List[Any]          =			get_tests_dir('fixtures/dummy_feature_extractor_config.json')
_SCREAMING_SNAKE_CASE    :      int          =			get_tests_dir('fixtures/vocab.json')
_SCREAMING_SNAKE_CASE    :      Tuple          =			get_tests_dir('fixtures')







class  UpperCamelCase__		(    unittest.TestCase   ):







	'''simple docstring'''
	_lowerCAmelCase							       =				["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]





	def   __snake_case				(			self		):
					A__  :	List[Any]         =							0





	def   __snake_case				(			self		):
					A__  :	Dict         =							AutoProcessor.from_pretrained('''facebook/wav2vec2-base-960h'''		)
					self.assertIsInstance(UpperCamelCase__				,       UpperCamelCase__		)





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

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

									A__  :	Any         =							AutoProcessor.from_pretrained(UpperCamelCase__		)

					self.assertIsInstance(UpperCamelCase__				,       UpperCamelCase__		)





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

									A__  :	List[Any]         =							AutoProcessor.from_pretrained(UpperCamelCase__		)

					self.assertIsInstance(UpperCamelCase__				,       UpperCamelCase__		)





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

									A__  :	Optional[int]         =							WavaVecaProcessor(UpperCamelCase__				,       UpperCamelCase__		)

									# save in new folder
									processor.save_pretrained(UpperCamelCase__		)

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

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

									A__  :	Optional[int]         =							AutoProcessor.from_pretrained(UpperCamelCase__		)

					self.assertIsInstance(UpperCamelCase__				,       UpperCamelCase__		)





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

									A__  :	str         =							WavaVecaProcessor(UpperCamelCase__				,       UpperCamelCase__		)

									# save in new folder
									processor.save_pretrained(UpperCamelCase__		)

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

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

									A__  :	List[Any]         =							AutoProcessor.from_pretrained(UpperCamelCase__		)

					self.assertIsInstance(UpperCamelCase__				,       UpperCamelCase__		)





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

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

					self.assertIsInstance(UpperCamelCase__				,       UpperCamelCase__		)





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

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

					A__  :	List[Any]         =							processor.feature_extractor
					self.assertTrue(feature_extractor.special_attribute_present		)
					self.assertEqual(feature_extractor.__class__.__name__				,       '''NewFeatureExtractor'''		)

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

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

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

									A__  :	Optional[Any]         =							CustomProcessor(UpperCamelCase__				,       UpperCamelCase__		)

									with tempfile.TemporaryDirectory() as tmp_dir:
													processor.save_pretrained(UpperCamelCase__		)
													A__  :	Union[str, Any]         =							AutoProcessor.from_pretrained(UpperCamelCase__		)
													self.assertIsInstance(UpperCamelCase__				,       UpperCamelCase__		)

					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		):







					class  UpperCamelCase__		(    SCREAMING_SNAKE_CASE_   ):







						'''simple docstring'''
						_lowerCAmelCase							       =				False







					class  UpperCamelCase__		(    SCREAMING_SNAKE_CASE_   ):







						'''simple docstring'''
						_lowerCAmelCase							       =				False







					class  UpperCamelCase__		(    SCREAMING_SNAKE_CASE_   ):







						'''simple docstring'''
						_lowerCAmelCase							       =				"AutoFeatureExtractor"
						_lowerCAmelCase							       =				"AutoTokenizer"
						_lowerCAmelCase							       =				False

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





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







@is_staging_test
class  UpperCamelCase__		(    unittest.TestCase   ):







	'''simple docstring'''
	_lowerCAmelCase							       =				["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]





	@classmethod
	def   __snake_case				(			cls		):
					A__  :	List[str]         =							TOKEN
					HfFolder.save_token(UpperCamelCase__		)





	@classmethod
	def   __snake_case				(			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   __snake_case				(			self		):
					A__  :	Optional[Any]         =							WavaVecaProcessor.from_pretrained(UpperCamelCase__		)
					with tempfile.TemporaryDirectory() as tmp_dir:
									processor.save_pretrained(
									    os.path.join(UpperCamelCase__				,       '''test-processor'''		)				,       push_to_hub=UpperCamelCase__				,       use_auth_token=self._token		)

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





	def   __snake_case				(			self		):
					A__  :	int         =							WavaVecaProcessor.from_pretrained(UpperCamelCase__		)

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

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





	def   __snake_case				(			self		):
					CustomFeatureExtractor.register_for_auto_class()
					CustomTokenizer.register_for_auto_class()
					CustomProcessor.register_for_auto_class()

					A__  :	Optional[Any]         =							CustomFeatureExtractor.from_pretrained(UpperCamelCase__		)

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

					A__  :	List[Any]         =							CustomProcessor(UpperCamelCase__				,       UpperCamelCase__		)

					with tempfile.TemporaryDirectory() as tmp_dir:
									create_repo(F"{USER}/test-dynamic-processor"				,       token=self._token		)
									A__  :	Union[str, Any]         =							Repository(UpperCamelCase__				,       clone_from=F"{USER}/test-dynamic-processor"				,       token=self._token		)
									processor.save_pretrained(UpperCamelCase__		)

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

									repo.push_to_hub()

					A__  :	Tuple         =							AutoProcessor.from_pretrained(F"{USER}/test-dynamic-processor"				,       trust_remote_code=UpperCamelCase__		)
					# 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'''		)