






"""simple docstring"""







import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

from ...tokenization_utils import BatchEncoding, PreTrainedTokenizer
from ...utils import logging


__a       :	Dict             = logging.get_logger(__name__)

__a       :	Dict             = '▁'

__a       :	int             = {
    'vocab_file': 'vocab.json',
    'spm_file': 'sentencepiece.bpe.model',
    'tokenizer_config_file': 'tokenizer_config.json',
}

__a       :	str             = {
    'vocab_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json',
    },
    'spm_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model',
    },
    'tokenizer_config_file': {
        'facebook/m2m100_418M': 'https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json',
        'facebook/m2m100_1.2B': 'https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json',
    },
}

__a       :	Any             = {
    'facebook/m2m100_418M': 1024,
}

# fmt: off
__a       :	Optional[Any]             = {
    'm2m100': ['af', 'am', 'ar', 'ast', 'az', 'ba', 'be', 'bg', 'bn', 'br', 'bs', 'ca', 'ceb', 'cs', 'cy', 'da', 'de', 'el', 'en', 'es', 'et', 'fa', 'ff', 'fi', 'fr', 'fy', 'ga', 'gd', 'gl', 'gu', 'ha', 'he', 'hi', 'hr', 'ht', 'hu', 'hy', 'id', 'ig', 'ilo', 'is', 'it', 'ja', 'jv', 'ka', 'kk', 'km', 'kn', 'ko', 'lb', 'lg', 'ln', 'lo', 'lt', 'lv', 'mg', 'mk', 'ml', 'mn', 'mr', 'ms', 'my', 'ne', 'nl', 'no', 'ns', 'oc', 'or', 'pa', 'pl', 'ps', 'pt', 'ro', 'ru', 'sd', 'si', 'sk', 'sl', 'so', 'sq', 'sr', 'ss', 'su', 'sv', 'sw', 'ta', 'th', 'tl', 'tn', 'tr', 'uk', 'ur', 'uz', 'vi', 'wo', 'xh', 'yi', 'yo', 'zh', 'zu'],
    'wmt21': ['en', 'ha', 'is', 'ja', 'cs', 'ru', 'zh', 'de']
}







class        _SCREAMING_SNAKE_CASE						(  __snake_case							):





							"""simple docstring"""





							_SCREAMING_SNAKE_CASE    =VOCAB_FILES_NAMES
							_SCREAMING_SNAKE_CASE    =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							_SCREAMING_SNAKE_CASE    =PRETRAINED_VOCAB_FILES_MAP
							_SCREAMING_SNAKE_CASE    =['input_ids', 'attention_mask']

							_SCREAMING_SNAKE_CASE    =[]
							_SCREAMING_SNAKE_CASE    =[]




							def __init__(   self:     List[Any]				,		__A:     Optional[Any]				,		__A:     List[Any]				,		__A:     Tuple=None				,		__A:     Tuple=None				,		__A:     Any="<s>"				,		__A:     str="</s>"				,		__A:     List[Any]="</s>"				,		__A:     Union[str, Any]="<pad>"				,		__A:     Optional[int]="<unk>"				,		__A:     Union[str, Any]="m2m100"				,		__A:     Optional[Dict[str, Any]] = None				,		__A:     Union[str, Any]=8				,		**__A:     List[str]				,		):

												'''simple docstring'''

												a__          =  {} if sp_model_kwargs is None else sp_model_kwargs

												a__          =  language_codes
												a__          =  FAIRSEQ_LANGUAGE_CODES[language_codes]
												a__          =  {lang_code: F'__{lang_code}__' for lang_code in fairseq_language_code}

												a__          =  kwargs.get('''additional_special_tokens'''				,		[]    )
												kwargs["additional_special_tokens"] += [
												    self.get_lang_token(__A    )
												    for lang_code in fairseq_language_code
												    if self.get_lang_token(__A    ) not in kwargs["additional_special_tokens"]
												]

												super().__init__(
												    src_lang=__A				,		tgt_lang=__A				,		bos_token=__A				,		eos_token=__A				,		sep_token=__A				,		unk_token=__A				,		pad_token=__A				,		language_codes=__A				,		sp_model_kwargs=self.sp_model_kwargs				,		num_madeup_words=__A				,		**__A				,		)

												a__          =  vocab_file
												a__          =  load_json(__A    )
												a__          =  {v: k for k, v in self.encoder.items()}
												a__          =  spm_file
												a__          =  load_spm(__A				,		self.sp_model_kwargs    )

												a__          =  len(self.encoder    )

												a__          =  {
												    self.get_lang_token(__A    ): self.encoder_size + i for i, lang_code in enumerate(__A    )
												}
												a__          =  {lang_code: self.encoder_size + i for i, lang_code in enumerate(__A    )}
												a__          =  {v: k for k, v in self.lang_token_to_id.items()}

												a__          =  src_lang if src_lang is not None else '''en'''
												a__          =  tgt_lang
												a__          =  self.get_lang_id(self._src_lang    )
												self.set_src_lang_special_tokens(self._src_lang    )

												a__          =  num_madeup_words




							@property
							def  lowercase		(   self:     Optional[Any]    ):

												'''simple docstring'''

												return len(self.encoder    ) + len(self.lang_token_to_id    )




							@property
							def  lowercase		(   self:     Optional[Any]    ):

												'''simple docstring'''

												return self._src_lang




							@src_lang.setter
							def  lowercase		(   self:     Any				,		__A:     str    ):

												'''simple docstring'''

												a__          =  new_src_lang
												self.set_src_lang_special_tokens(self._src_lang    )




							def  lowercase		(   self:     Dict				,		__A:     str    ):

												'''simple docstring'''

												return self.sp_model.encode(__A				,		out_type=__A    )




							def  lowercase		(   self:     Any				,		__A:     List[str]    ):

												'''simple docstring'''

												if token in self.lang_token_to_id:
																	return self.lang_token_to_id[token]
												return self.encoder.get(__A				,		self.encoder[self.unk_token]    )




							def  lowercase		(   self:     Optional[Any]				,		__A:     int    ):

												'''simple docstring'''

												if index in self.id_to_lang_token:
																	return self.id_to_lang_token[index]
												return self.decoder.get(__A				,		self.unk_token    )




							def  lowercase		(   self:     Optional[int]				,		__A:     str    ):

												'''simple docstring'''

												a__          =  []
												a__          =  ''''''
												for token in tokens:
																	# make sure that special tokens are not decoded using sentencepiece model
																	if token in self.all_special_tokens:
																						out_string += self.sp_model.decode(__A    ) + token
																						a__          =  []
																	else:
																						current_sub_tokens.append(__A    )
												out_string += self.sp_model.decode(__A    )
												return out_string.strip()




							def  lowercase		(   self:     List[str]				,		__A:     List[int]				,		__A:     Optional[List[int]] = None				,		__A:     bool = False    ):

												'''simple docstring'''

												if already_has_special_tokens:
																	return super().get_special_tokens_mask(
																	    token_ids_a=__A				,		token_ids_a=__A				,		already_has_special_tokens=__A    )

												a__          =  [1] * len(self.prefix_tokens    )
												a__          =  [1] * len(self.suffix_tokens    )
												if token_ids_a is None:
																	return prefix_ones + ([0] * len(__A    )) + suffix_ones
												return prefix_ones + ([0] * len(__A    )) + ([0] * len(__A    )) + suffix_ones




							def  lowercase		(   self:     Tuple				,		__A:     List[int]				,		__A:     Optional[List[int]] = None    ):

												'''simple docstring'''

												if token_ids_a is None:
																	return self.prefix_tokens + token_ids_a + self.suffix_tokens
												# We don't expect to process pairs, but leave the pair logic for API consistency
												return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens




							def  lowercase		(   self:     Any    ):

												'''simple docstring'''

												a__          =  {self.convert_ids_to_tokens(__A    ): i for i in range(self.vocab_size    )}
												vocab.update(self.added_tokens_encoder    )
												return vocab




							def __getstate__(   self:     Optional[Any]    ):

												'''simple docstring'''

												a__          =  self.__dict__.copy()
												a__          =  None
												return state




							def __setstate__(   self:     Dict				,		__A:     Dict    ):

												'''simple docstring'''

												a__          =  d

												# for backward compatibility
												if not hasattr(self				,		'''sp_model_kwargs'''    ):
																	a__          =  {}

												a__          =  load_spm(self.spm_file				,		self.sp_model_kwargs    )




							def  lowercase		(   self:     Optional[int]				,		__A:     str				,		__A:     Optional[str] = None    ):

												'''simple docstring'''

												a__          =  Path(__A    )
												if not save_dir.is_dir():
																	raise OSError(F'{save_directory} should be a directory'    )
												a__          =  save_dir / (
												    (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''vocab_file''']
												)
												a__          =  save_dir / (
												    (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''spm_file''']
												)

												save_json(self.encoder				,		__A    )

												if os.path.abspath(self.spm_file    ) != os.path.abspath(__A    ) and os.path.isfile(self.spm_file    ):
																	copyfile(self.spm_file				,		__A    )
												elif not os.path.isfile(self.spm_file    ):
																	with open(__A				,		'''wb'''    ) as fi:
																						a__          =  self.sp_model.serialized_model_proto()
																						fi.write(__A    )

												return (str(__A    ), str(__A    ))




							def  lowercase		(   self:     Optional[int]				,		__A:     List[str]				,		__A:     str = "en"				,		__A:     Optional[List[str]] = None				,		__A:     str = "ro"				,		**__A:     str				,		):

												'''simple docstring'''

												a__          =  src_lang
												a__          =  tgt_lang
												self.set_src_lang_special_tokens(self.src_lang    )
												return super().prepare_seqaseq_batch(__A				,		__A				,		**__A    )




							def  lowercase		(   self:     Union[str, Any]				,		__A:     Tuple				,		__A:     Optional[str]				,		__A:     Optional[str]				,		**__A:     List[str]    ):

												'''simple docstring'''

												if src_lang is None or tgt_lang is None:
																	raise ValueError('''Translation requires a `src_lang` and a `tgt_lang` for this model'''    )
												a__          =  src_lang
												a__          =  self(__A				,		add_special_tokens=__A				,		**__A    )
												a__          =  self.get_lang_id(__A    )
												a__          =  tgt_lang_id
												return inputs




							def  lowercase		(   self:     Dict    ):

												'''simple docstring'''

												self.set_src_lang_special_tokens(self.src_lang    )




							def  lowercase		(   self:     int    ):

												'''simple docstring'''

												self.set_tgt_lang_special_tokens(self.tgt_lang    )




							def  lowercase		(   self:     Any				,		__A:     str    ):

												'''simple docstring'''

												a__          =  self.get_lang_token(__A    )
												a__          =  self.lang_token_to_id[lang_token]
												a__          =  [self.cur_lang_id]
												a__          =  [self.eos_token_id]




							def  lowercase		(   self:     Tuple				,		__A:     str    ):

												'''simple docstring'''

												a__          =  self.get_lang_token(__A    )
												a__          =  self.lang_token_to_id[lang_token]
												a__          =  [self.cur_lang_id]
												a__          =  [self.eos_token_id]




							def  lowercase		(   self:     Any				,		__A:     str    ):

												'''simple docstring'''

												return self.lang_code_to_token[lang]





							def  lowercase		(   self:     List[Any]				,		__A:     str    ):

												'''simple docstring'''

												a__          =  self.get_lang_token(__A    )
												return self.lang_token_to_id[lang_token]





def        SCREAMING_SNAKE_CASE       (      lowerCamelCase_ ,     lowerCamelCase_):
					a__          =  sentencepiece.SentencePieceProcessor(**lowerCamelCase_)
					spm.Load(str(lowerCamelCase_))
					return spm





def        SCREAMING_SNAKE_CASE       (      lowerCamelCase_):
					with open(lowerCamelCase_ ,     '''r''') as f:
										return json.load(lowerCamelCase_)





def        SCREAMING_SNAKE_CASE       (      lowerCamelCase_ ,     lowerCamelCase_):
					with open(lowerCamelCase_ ,     '''w''') as f:
										json.dump(lowerCamelCase_ ,     lowerCamelCase_ ,     indent=2)










"""simple docstring"""







import json
import os
import shutil
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoConfig, BertConfig, GPTaConfig
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_configuration import CustomConfig  # noqa E402


__a       :	Dict             = {
    'return_dict': False,
    'output_hidden_states': True,
    'output_attentions': True,
    'torchscript': True,
    'torch_dtype': 'float16',
    'use_bfloat16': True,
    'tf_legacy_loss': True,
    'pruned_heads': {'a': 1},
    'tie_word_embeddings': False,
    'is_decoder': True,
    'cross_attention_hidden_size': 128,
    'add_cross_attention': True,
    'tie_encoder_decoder': True,
    'max_length': 50,
    'min_length': 3,
    'do_sample': True,
    'early_stopping': True,
    'num_beams': 3,
    'num_beam_groups': 3,
    'diversity_penalty': 0.5,
    'temperature': 2.0,
    'top_k': 10,
    'top_p': 0.7,
    'typical_p': 0.2,
    'repetition_penalty': 0.8,
    'length_penalty': 0.8,
    'no_repeat_ngram_size': 5,
    'encoder_no_repeat_ngram_size': 5,
    'bad_words_ids': [1, 2, 3],
    'num_return_sequences': 3,
    'chunk_size_feed_forward': 5,
    'output_scores': True,
    'return_dict_in_generate': True,
    'forced_bos_token_id': 2,
    'forced_eos_token_id': 3,
    'remove_invalid_values': True,
    'architectures': ['BertModel'],
    'finetuning_task': 'translation',
    'id2label': {0: 'label'},
    'label2id': {'label': '0'},
    'tokenizer_class': 'BertTokenizerFast',
    'prefix': 'prefix',
    'bos_token_id': 6,
    'pad_token_id': 7,
    'eos_token_id': 8,
    'sep_token_id': 9,
    'decoder_start_token_id': 10,
    'exponential_decay_length_penalty': (5, 1.01),
    'suppress_tokens': [0, 1],
    'begin_suppress_tokens': 2,
    'task_specific_params': {'translation': 'some_params'},
    'problem_type': 'regression',
}







@is_staging_test
class        _SCREAMING_SNAKE_CASE						(  unittest.TestCase							):





							"""simple docstring"""





							@classmethod
							def  lowercase		(   cls:     Union[str, Any]    ):

												'''simple docstring'''

												a__          =  TOKEN
												HfFolder.save_token(__A    )




							@classmethod
							def  lowercase		(   cls:     Any    ):

												'''simple docstring'''

												try:
																	delete_repo(token=cls._token				,		repo_id='''test-config'''    )
												except HTTPError:
																	pass

												try:
																	delete_repo(token=cls._token				,		repo_id='''valid_org/test-config-org'''    )
												except HTTPError:
																	pass

												try:
																	delete_repo(token=cls._token				,		repo_id='''test-dynamic-config'''    )
												except HTTPError:
																	pass




							def  lowercase		(   self:     Union[str, Any]    ):

												'''simple docstring'''

												a__          =  BertConfig(
												    vocab_size=99				,		hidden_size=32				,		num_hidden_layers=5				,		num_attention_heads=4				,		intermediate_size=37    )
												config.push_to_hub('''test-config'''				,		use_auth_token=self._token    )

												a__          =  BertConfig.from_pretrained(F'{USER}/test-config'    )
												for k, v in config.to_dict().items():
																	if k != "transformers_version":
																						self.assertEqual(__A				,		getattr(__A				,		__A    )    )

        # Reset repo
												delete_repo(token=self._token				,		repo_id='''test-config'''    )

												# Push to hub via save_pretrained
												with tempfile.TemporaryDirectory() as tmp_dir:
																	config.save_pretrained(__A				,		repo_id='''test-config'''				,		push_to_hub=__A				,		use_auth_token=self._token    )

												a__          =  BertConfig.from_pretrained(F'{USER}/test-config'    )
												for k, v in config.to_dict().items():
																	if k != "transformers_version":
																						self.assertEqual(__A				,		getattr(__A				,		__A    )    )




							def  lowercase		(   self:     Optional[Any]    ):

												'''simple docstring'''

												a__          =  BertConfig(
												    vocab_size=99				,		hidden_size=32				,		num_hidden_layers=5				,		num_attention_heads=4				,		intermediate_size=37    )
												config.push_to_hub('''valid_org/test-config-org'''				,		use_auth_token=self._token    )

												a__          =  BertConfig.from_pretrained('''valid_org/test-config-org'''    )
												for k, v in config.to_dict().items():
																	if k != "transformers_version":
																						self.assertEqual(__A				,		getattr(__A				,		__A    )    )

        # Reset repo
												delete_repo(token=self._token				,		repo_id='''valid_org/test-config-org'''    )

												# Push to hub via save_pretrained
												with tempfile.TemporaryDirectory() as tmp_dir:
																	config.save_pretrained(
																	    __A				,		repo_id='''valid_org/test-config-org'''				,		push_to_hub=__A				,		use_auth_token=self._token    )

												a__          =  BertConfig.from_pretrained('''valid_org/test-config-org'''    )
												for k, v in config.to_dict().items():
																	if k != "transformers_version":
																						self.assertEqual(__A				,		getattr(__A				,		__A    )    )





							def  lowercase		(   self:     List[str]    ):

												'''simple docstring'''

												CustomConfig.register_for_auto_class()
												a__          =  CustomConfig(attribute=42    )

												config.push_to_hub('''test-dynamic-config'''				,		use_auth_token=self._token    )

												# This has added the proper auto_map field to the config
												self.assertDictEqual(config.auto_map				,		{'''AutoConfig''': '''custom_configuration.CustomConfig'''}    )

												a__          =  AutoConfig.from_pretrained(F'{USER}/test-dynamic-config'				,		trust_remote_code=__A    )
												# Can't make an isinstance check because the new_config is from the FakeConfig class of a dynamic module
												self.assertEqual(new_config.__class__.__name__				,		'''CustomConfig'''    )
												self.assertEqual(new_config.attribute				,		42    )







class        _SCREAMING_SNAKE_CASE						(  unittest.TestCase							):





							"""simple docstring"""





							def  lowercase		(   self:     int    ):

												'''simple docstring'''

												a__          =  GPTaConfig()

												# attempt to modify each of int/float/bool/str config records and verify they were updated
												a__          =  c.n_embd + 1  # int
												a__          =  c.resid_pdrop + 1.0  # float
												a__          =  not c.scale_attn_weights  # bool
												a__          =  c.summary_type + '''foo'''  # str
												c.update_from_string(
												    F'n_embd={n_embd},resid_pdrop={resid_pdrop},scale_attn_weights={scale_attn_weights},summary_type={summary_type}'    )
												self.assertEqual(__A				,		c.n_embd				,		'''mismatch for key: n_embd'''    )
												self.assertEqual(__A				,		c.resid_pdrop				,		'''mismatch for key: resid_pdrop'''    )
												self.assertEqual(__A				,		c.scale_attn_weights				,		'''mismatch for key: scale_attn_weights'''    )
												self.assertEqual(__A				,		c.summary_type				,		'''mismatch for key: summary_type'''    )




							def  lowercase		(   self:     List[str]    ):

												'''simple docstring'''

												a__          =  PretrainedConfig()
												a__          =  [key for key in base_config.__dict__ if key not in config_common_kwargs]
												# If this part of the test fails, you have arguments to addin config_common_kwargs above.
												self.assertListEqual(
												    __A				,		['''is_encoder_decoder''', '''_name_or_path''', '''_commit_hash''', '''transformers_version''']    )
												a__          =  [key for key, value in config_common_kwargs.items() if value == getattr(__A				,		__A    )]
												if len(__A    ) > 0:
																	raise ValueError(
																	    '''The following keys are set with the default values in'''
																	    ''' `test_configuration_common.config_common_kwargs` pick another value for them:'''
																	    F' {", ".join(__A    )}.'    )




							def  lowercase		(   self:     str    ):

												'''simple docstring'''

												with self.assertRaises(__A    ):
																	# config is in subfolder, the following should not work without specifying the subfolder
																	a__          =  BertConfig.from_pretrained('''hf-internal-testing/tiny-random-bert-subfolder'''    )

												a__          =  BertConfig.from_pretrained('''hf-internal-testing/tiny-random-bert-subfolder'''				,		subfolder='''bert'''    )

												self.assertIsNotNone(__A    )




							def  lowercase		(   self:     Dict    ):

												'''simple docstring'''

												a__          =  mock.Mock()
												a__          =  500
												a__          =  {}
												a__          =  HTTPError
												a__          =  {}

												# Download this model to make sure it's in the cache.
												a__          =  BertConfig.from_pretrained('''hf-internal-testing/tiny-random-bert'''    )

												# Under the mock environment we get a 500 error when trying to reach the model.
												with mock.patch('''requests.Session.request'''				,		return_value=__A    ) as mock_head:
																	a__          =  BertConfig.from_pretrained('''hf-internal-testing/tiny-random-bert'''    )
																	# This check we did call the fake head request
																	mock_head.assert_called()




							def  lowercase		(   self:     Union[str, Any]    ):

												'''simple docstring'''

												a__          =  BertConfig.from_pretrained(
												    '''https://huggingface.co/hf-internal-testing/tiny-random-bert/resolve/main/config.json'''    )




							def  lowercase		(   self:     Optional[Any]    ):

												'''simple docstring'''

												a__          =  AutoConfig.from_pretrained('''bert-base-cased'''    )
												a__          =  ['''config.4.0.0.json''']

												with tempfile.TemporaryDirectory() as tmp_dir:
																	configuration.save_pretrained(__A    )
																	a__          =  2
																	json.dump(configuration.to_dict()				,		open(os.path.join(__A				,		'''config.4.0.0.json'''    )				,		'''w'''    )    )

																	# This should pick the new configuration file as the version of Transformers is > 4.0.0
																	a__          =  AutoConfig.from_pretrained(__A    )
																	self.assertEqual(new_configuration.hidden_size				,		2    )

																	# Will need to be adjusted if we reach v42 and this test is still here.
																	# Should pick the old configuration file as the version of Transformers is < 4.42.0
																	a__          =  ['''config.42.0.0.json''']
																	a__          =  768
																	configuration.save_pretrained(__A    )
																	shutil.move(os.path.join(__A				,		'''config.4.0.0.json'''    )				,		os.path.join(__A				,		'''config.42.0.0.json'''    )    )
																	a__          =  AutoConfig.from_pretrained(__A    )
																	self.assertEqual(new_configuration.hidden_size				,		768    )





							def  lowercase		(   self:     Optional[Any]    ):

												'''simple docstring'''

												a__          =  '''hf-internal-testing/test-two-configs'''

												import transformers as new_transformers

												a__          =  '''v4.0.0'''
												a__       ,a__          =  new_transformers.models.auto.AutoConfig.from_pretrained(
												    __A				,		return_unused_kwargs=__A    )
												self.assertEqual(new_configuration.hidden_size				,		2    )
												# This checks `_configuration_file` ia not kept in the kwargs by mistake.
												self.assertDictEqual(__A				,		{}    )

												# Testing an older version by monkey-patching the version in the module it's used.
												import transformers as old_transformers

												a__          =  '''v3.0.0'''
												a__          =  old_transformers.models.auto.AutoConfig.from_pretrained(__A    )
												self.assertEqual(old_configuration.hidden_size				,		768    )


