


from __future__ import annotations

import unittest

import numpy as np

from transformers import OPTConfig, is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
 import tensorflow as tf

 from transformers import GPTaTokenizer, TFOPTForCausalLM, TFOPTModel




def        snake_case_					(  snake_case      ,							snake_case      ,							snake_case=None      ,							snake_case=None					)  ->    int:
       if attention_mask is None:
              lowercase__:       Tuple        =       tf.cast(tf.math.not_equal(snake_case      ,							config.pad_token_id					)      ,							tf.inta					)
       return {"input_ids": input_ids, "attention_mask": attention_mask}





@require_tf
class   __a      :
 __lowercase			:							Optional[Any]       					=			OPTConfig
 __lowercase			:							str       					=			{}
 __lowercase			:							Any       					=			'gelu'
 def __init__(       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__=13    ,			lowerCAmelCase__=7    ,			lowerCAmelCase__=True    ,			lowerCAmelCase__=False    ,			lowerCAmelCase__=99    ,			lowerCAmelCase__=16    ,			lowerCAmelCase__=2    ,			lowerCAmelCase__=4    ,			lowerCAmelCase__=4    ,			lowerCAmelCase__="gelu"    ,			lowerCAmelCase__=0.1    ,			lowerCAmelCase__=0.1    ,			lowerCAmelCase__=20    ,			lowerCAmelCase__=2    ,			lowerCAmelCase__=1    ,			lowerCAmelCase__=0    ,			lowerCAmelCase__=16    ,			lowerCAmelCase__=16    ,			) ->    Any:




        '''simple docstring'''
        lowercase__:       List[str]        =       parent
        lowercase__:       Union[str, Any]        =       batch_size
        lowercase__:       List[Any]        =       seq_length
        lowercase__:       Union[str, Any]        =       is_training
        lowercase__:       Optional[Any]        =       use_labels
        lowercase__:       Optional[int]        =       vocab_size
        lowercase__:       Any        =       hidden_size
        lowercase__:       Dict        =       num_hidden_layers
        lowercase__:       str        =       num_attention_heads
        lowercase__:       str        =       intermediate_size
        lowercase__:       Optional[Any]        =       hidden_act
        lowercase__:       Tuple        =       hidden_dropout_prob
        lowercase__:       Optional[Any]        =       attention_probs_dropout_prob
        lowercase__:       Tuple        =       max_position_embeddings
        lowercase__:       Any        =       eos_token_id
        lowercase__:       Union[str, Any]        =       pad_token_id
        lowercase__:       List[Any]        =       bos_token_id
        lowercase__:       int        =       embed_dim
        lowercase__:       Optional[Any]        =       word_embed_proj_dim
        lowercase__:       Tuple        =       False
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Optional[Any]:




        '''simple docstring'''
        lowercase__:       Tuple        =       ids_tensor([self.batch_size, self.seq_length - 1]    ,			self.vocab_size			)
        lowercase__:       str        =       tf.expand_dims(tf.constant([self.eos_token_id] * self.batch_size			)    ,			1			)
        lowercase__:       Union[str, Any]        =       tf.concat([input_ids, eos_tensor]    ,			axis=1			)

        lowercase__:       Optional[int]        =       self.config_cls(
            vocab_size=self.vocab_size    ,			hidden_size=self.hidden_size    ,			num_hidden_layers=self.num_hidden_layers    ,			num_attention_heads=self.num_attention_heads    ,			ffn_dim=self.intermediate_size    ,			dropout=self.hidden_dropout_prob    ,			attention_dropout=self.attention_probs_dropout_prob    ,			max_position_embeddings=self.max_position_embeddings    ,			eos_token_id=self.eos_token_id    ,			bos_token_id=self.bos_token_id    ,			pad_token_id=self.pad_token_id    ,			embed_dim=self.embed_dim    ,			word_embed_proj_dim=self.word_embed_proj_dim    ,			is_encoder_decoder=lowerCAmelCase__    ,			**self.config_updates    ,			)
        lowercase__:       int        =       prepare_opt_inputs_dict(lowerCAmelCase__    ,			lowerCAmelCase__			)
        return config, inputs_dict
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__			) ->    int:




        '''simple docstring'''
        lowercase__:       int        =       TFOPTModel(config=lowerCAmelCase__			)
        lowercase__:       str        =       inputs_dict['input_ids']

        lowercase__:       Any        =       input_ids[:1, :]
        lowercase__:       List[str]        =       inputs_dict['attention_mask'][:1, :]
        lowercase__:       Any        =       1

        # first forward pass
        lowercase__:       List[str]        =       model(lowerCAmelCase__    ,			attention_mask=lowerCAmelCase__    ,			use_cache=lowerCAmelCase__			)

        lowercase__      ,						lowercase__:       Optional[Any]        =       outputs.to_tuple()

        # create hypothetical next token and extent to next_input_ids
        lowercase__:       Optional[int]        =       ids_tensor((self.batch_size, 3)    ,			config.vocab_size			)
        lowercase__:       Tuple        =       tf.cast(ids_tensor((self.batch_size, 3)    ,			2			)    ,			tf.inta			)

        # append to next input_ids and
        lowercase__:       Union[str, Any]        =       tf.concat([input_ids, next_tokens]    ,			axis=-1			)
        lowercase__:       Any        =       tf.concat([attention_mask, next_attn_mask]    ,			axis=-1			)

        lowercase__:       List[Any]        =       model(lowerCAmelCase__    ,			attention_mask=lowerCAmelCase__			)[0]
        lowercase__:       Dict        =       model(lowerCAmelCase__    ,			attention_mask=lowerCAmelCase__    ,			past_key_values=lowerCAmelCase__			)[0]

        self.parent.assertEqual(next_tokens.shape[1]    ,			output_from_past.shape[1]			)

        # select random slice
        lowercase__:       List[str]        =       int(ids_tensor((1,)    ,			output_from_past.shape[-1]			)			)
        lowercase__:       int        =       output_from_no_past[:, -3:, random_slice_idx]
        lowercase__:       int        =       output_from_past[:, :, random_slice_idx]

        # test that outputs are equal for slice
        tf.debugging.assert_near(lowerCAmelCase__    ,			lowerCAmelCase__    ,			rtol=1E-3			)





@require_tf
class   __a      (   __UpperCamelCase	,	__UpperCamelCase	,	unittest.TestCase				):
 __lowercase			:							Union[str, Any]       					=			(TFOPTModel, TFOPTForCausalLM) if is_tf_available() else ()
 __lowercase			:							str       					=			(TFOPTForCausalLM,) if is_tf_available() else ()
 __lowercase			:							Tuple       					=			(
     {'feature-extraction': TFOPTModel, 'text-generation': TFOPTForCausalLM} if is_tf_available() else {}
 )
 __lowercase			:							int       					=			False
 __lowercase			:							List[str]       					=			False
 __lowercase			:							Tuple       					=			False
 __lowercase			:							List[str]       					=			10
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        lowercase__:       Optional[int]        =       TFOPTModelTester(self			)
        lowercase__:       List[str]        =       ConfigTester(self    ,			config_class=lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        self.config_tester.run_common_tests()
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    int:




        '''simple docstring'''
        lowercase__:       Any        =       self.model_tester.prepare_config_and_inputs_for_common()
        self.model_tester.check_decoder_model_past_large_inputs(*lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Optional[Any]:




        '''simple docstring'''
        lowercase__      ,						lowercase__:       Union[str, Any]        =       self.model_tester.prepare_config_and_inputs_for_common()

        def _get_word_embedding_weight(lowerCAmelCase__    ,			lowerCAmelCase__			):
               if hasattr(lowerCAmelCase__    ,			'weight'			):
                      return embedding_layer.weight
               else:
                      # Here we build the word embeddings weights if not exists.
                      # And then we retry to get the attribute once built.
                      model.build()
                      if hasattr(lowerCAmelCase__    ,			'weight'			):
                             return embedding_layer.weight
                      else:
                             return None

        for model_class in self.all_model_classes:
               for size in [config.vocab_size - 10, config.vocab_size + 10]:
                      # build the embeddings
                      lowercase__:       Optional[Any]        =       model_class(config=lowerCAmelCase__			)
                      lowercase__:       List[Any]        =       _get_word_embedding_weight(lowerCAmelCase__    ,			model.get_input_embeddings()			)
                      lowercase__:       List[str]        =       _get_word_embedding_weight(lowerCAmelCase__    ,			model.get_output_embeddings()			)

                      # reshape the embeddings
                      model.resize_token_embeddings(lowerCAmelCase__			)
                      lowercase__:       Union[str, Any]        =       _get_word_embedding_weight(lowerCAmelCase__    ,			model.get_input_embeddings()			)
                      lowercase__:       Union[str, Any]        =       _get_word_embedding_weight(lowerCAmelCase__    ,			model.get_output_embeddings()			)

                      # check that the resized embeddings size matches the desired size.
                      lowercase__:       int        =       size if size is not None else config.vocab_size

                      self.assertEqual(new_input_embeddings.shape[0]    ,			lowerCAmelCase__			)

                      # check that weights remain the same after resizing
                      lowercase__:       Any        =       True
                      for pa, pa in zip(old_input_embeddings.value()    ,			new_input_embeddings.value()			):
                             if tf.math.reduce_sum(tf.math.abs(pa - pa			)			) > 0:
                                    lowercase__:       Dict        =       False
                      self.assertTrue(lowerCAmelCase__			)

                      if old_output_embeddings is not None and new_output_embeddings is not None:
                             self.assertEqual(new_output_embeddings.shape[0]    ,			lowerCAmelCase__			)

                             lowercase__:       int        =       True
                             for pa, pa in zip(old_output_embeddings.value()    ,			new_output_embeddings.value()			):
                                    if tf.math.reduce_sum(tf.math.abs(pa - pa			)			) > 0:
                                           lowercase__:       Optional[Any]        =       False
                             self.assertTrue(lowerCAmelCase__			)





def        snake_case_					(  snake_case					)  ->    Optional[int]:
       return tf.constant(snake_case      ,							dtype=tf.intaa					)





@require_tf
class   __a      (   unittest.TestCase				):
 __lowercase			:							List[Any]       					=			99
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    str:




        '''simple docstring'''
        lowercase__:       List[str]        =       tf.ones((4, 1)    ,			dtype=tf.intaa			) * 2
        lowercase__:       str        =       tf.concat([ids_tensor((4, 6)    ,			self.vocab_size - 3			) + 3, eos_column_vector]    ,			axis=1			)
        lowercase__:       Optional[Any]        =       input_ids.shape[0]
        lowercase__:       Union[str, Any]        =       OPTConfig(
            vocab_size=self.vocab_size    ,			hidden_size=24    ,			num_hidden_layers=2    ,			num_attention_heads=2    ,			ffn_dim=32    ,			max_position_embeddings=48    ,			eos_token_id=2    ,			pad_token_id=1    ,			bos_token_id=0    ,			)
        return config, input_ids, batch_size





@require_sentencepiece
@require_tf
class   __a      (   unittest.TestCase				):
 @slow
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Optional[Any]:




        '''simple docstring'''
        lowercase__:       List[Any]        =       TFOPTModel.from_pretrained('facebook/opt-350m'			)
        lowercase__:       Optional[int]        =       _long_tensor([[0, 31_414, 232, 328, 740, 1_140, 12_695, 69, 46_078, 1_588, 2]]			)
        lowercase__:       Optional[int]        =       tf.not_equal(lowerCAmelCase__    ,			model.config.pad_token_id			)
        with tf.GradientTape():
               lowercase__:       Optional[int]        =       model(input_ids=lowerCAmelCase__    ,			attention_mask=lowerCAmelCase__			).last_hidden_state
        lowercase__:       List[str]        =       (1, 11, 512)
        self.assertEqual(output.shape    ,			lowerCAmelCase__			)
        lowercase__:       List[str]        =       tf.constant(
            [[-0.2_8_7_3, -1.9_2_1_8, -0.3_0_3_3], [-1.2_7_1_0, -0.1_3_3_8, -0.1_9_0_2], [0.4_0_9_5, 0.1_2_1_4, -1.3_1_2_1]]			)
        self.assertTrue(np.allclose(output[:, :3, :3]    ,			lowerCAmelCase__    ,			atol=4E-3			)			)

        lowercase__:       List[Any]        =       tf.function(lowerCAmelCase__    ,			jit_compile=lowerCAmelCase__			)
        lowercase__:       List[Any]        =       xla_generate(lowerCAmelCase__    ,			lowerCAmelCase__			)[0]
        self.assertTrue(np.allclose(output[:, :3, :3]    ,			lowerCAmelCase__    ,			atol=4E-2			)			)





@require_tf
@slow
class   __a      (   unittest.TestCase				):
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        super().setUp()
        lowercase__:       int        =       'facebook/opt-350m'
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        lowercase__:       Union[str, Any]        =       TFOPTForCausalLM.from_pretrained(self.path_model			)
        lowercase__:       Optional[Any]        =       GPTaTokenizer.from_pretrained(self.path_model			)

        lowercase__:       Dict        =       [
            'Today is a beautiful day and I want to',
            'In the city of',
            'Paris is the capital of France and',
            'Computers and mobile phones have taken',
        ]
        # verify that prompt without BOS token is identical to Metaseq -> add_special_tokens=False
        lowercase__:       Tuple        =       tokenizer(lowerCAmelCase__    ,			return_tensors='tf'    ,			padding=lowerCAmelCase__    ,			add_special_tokens=lowerCAmelCase__			)
        lowercase__:       List[Any]        =       tf.math.reduce_mean(model(inputs.input_ids    ,			attention_mask=inputs.attention_mask			)[0]    ,			axis=-1			)
        lowercase__:       Dict        =       tf.constant(
            [
                [1.3_8_5_1, -1_3.8_9_2_3, -1_0.5_2_2_9, -1_0.7_5_3_3, -0.2_3_0_9, -1_0.2_3_8_4, -0.5_3_6_5, -9.0_9_4_7, -5.1_6_7_0],
                [-4.7_0_7_3, -1_0.6_2_7_6, -3.9_4_1_5, -2_1.5_2_4_2, -0.2_8_2_2, -0.2_8_2_2, -0.2_8_2_2, -0.2_8_2_2, -0.2_8_2_2],
                [0.6_2_4_7, -3.4_2_2_9, -8.9_1_7_9, -1.4_2_9_7, -1_4.1_6_5_0, 1.4_1_4_6, -9.0_2_1_8, -0.2_7_0_3, -0.2_7_0_3],
                [6.4_7_8_3, -1.9_9_1_3, -1_0.7_9_2_6, -2.3_3_3_6, 1.5_0_9_2, -0.9_9_7_4, -6.8_2_1_3, 1.3_4_7_7, 1.3_4_7_7],
            ]			)
        self.assertTrue(np.allclose(lowerCAmelCase__    ,			lowerCAmelCase__    ,			atol=1E-4			)			)

        lowercase__:       Union[str, Any]        =       tf.function(lowerCAmelCase__    ,			jit_compile=lowerCAmelCase__			)
        lowercase__:       Dict        =       tf.math.reduce_mean(xla_generate(inputs.input_ids    ,			attention_mask=inputs.attention_mask			)[0]    ,			axis=-1			)
        self.assertTrue(np.allclose(lowerCAmelCase__    ,			lowerCAmelCase__    ,			atol=1E-4			)			)





@require_tf
@slow
class   __a      (   unittest.TestCase				):
 @property
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    List[Any]:




        '''simple docstring'''
        return [
            "Today is a beautiful day and I want",
            "In the city of",
            "Paris is the capital of France and",
            "Computers and mobile phones have taken",
        ]
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        lowercase__:       List[Any]        =       'facebook/opt-125m'

        lowercase__:       str        =       [
            'Today is a beautiful day and I want to',
            'In the city of New York, the city',
            'Paris is the capital of France and the capital',
            'Computers and mobile phones have taken over the',
        ]

        lowercase__:       str        =       []
        lowercase__:       str        =       GPTaTokenizer.from_pretrained(lowerCAmelCase__			)
        lowercase__:       int        =       TFOPTForCausalLM.from_pretrained(lowerCAmelCase__			)

        for prompt in self.prompts:
               lowercase__:       Optional[int]        =       tokenizer(lowerCAmelCase__    ,			return_tensors='tf'			).input_ids

               lowercase__:       List[str]        =       model.generate(lowerCAmelCase__    ,			max_length=10			)

               lowercase__:       Optional[int]        =       tokenizer.batch_decode(lowerCAmelCase__    ,			skip_special_tokens=lowerCAmelCase__			)
               predicted_outputs += generated_string

        self.assertListEqual(lowerCAmelCase__    ,			lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        lowercase__:       List[str]        =       'facebook/opt-350m'

        lowercase__:       Optional[Any]        =       GPTaTokenizer.from_pretrained(lowerCAmelCase__			)
        lowercase__:       List[str]        =       TFOPTForCausalLM.from_pretrained(lowerCAmelCase__			)

        lowercase__:       Union[str, Any]        =       'left'

        # use different length sentences to test batching
        lowercase__:       Optional[int]        =       [
            'Hello, my dog is a little',
            'Today, I',
        ]

        lowercase__:       Tuple        =       tokenizer(lowerCAmelCase__    ,			return_tensors='tf'    ,			padding=lowerCAmelCase__			)
        lowercase__:       str        =       inputs['input_ids']

        lowercase__:       List[str]        =       model.generate(input_ids=lowerCAmelCase__    ,			attention_mask=inputs['attention_mask']			)

        lowercase__:       Any        =       tokenizer(sentences[0]    ,			return_tensors='tf'			).input_ids
        lowercase__:       Tuple        =       model.generate(input_ids=lowerCAmelCase__			)

        lowercase__:       Tuple        =       inputs_non_padded.shape[-1] - tf.math.reduce_sum(
            tf.cast(inputs['attention_mask'][-1]    ,			tf.intaa			)			)
        lowercase__:       Tuple        =       tokenizer(sentences[1]    ,			return_tensors='tf'			).input_ids
        lowercase__:       Tuple        =       model.generate(input_ids=lowerCAmelCase__    ,			max_length=model.config.max_length - num_paddings			)

        lowercase__:       Optional[Any]        =       tokenizer.batch_decode(lowerCAmelCase__    ,			skip_special_tokens=lowerCAmelCase__			)
        lowercase__:       Optional[Any]        =       tokenizer.decode(output_non_padded[0]    ,			skip_special_tokens=lowerCAmelCase__			)
        lowercase__:       Optional[Any]        =       tokenizer.decode(output_padded[0]    ,			skip_special_tokens=lowerCAmelCase__			)

        lowercase__:       List[Any]        =       [
            'Hello, my dog is a little bit of a dork.\nI\'m a little bit',
            'Today, I was in the middle of a conversation with a friend about the',
        ]
        self.assertListEqual(lowerCAmelCase__    ,			lowerCAmelCase__			)
        self.assertListEqual(lowerCAmelCase__    ,			[non_padded_sentence, padded_sentence]			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Union[str, Any]:




        '''simple docstring'''
        lowercase__:       List[Any]        =       'facebook/opt-350m'

        lowercase__:       Optional[Any]        =       [
            'Today is a beautiful day and I want to',
            'In the city of San Francisco, the city',
            'Paris is the capital of France and the capital',
            'Computers and mobile phones have taken over the',
        ]

        lowercase__:       List[str]        =       []
        lowercase__:       Optional[Any]        =       GPTaTokenizer.from_pretrained(lowerCAmelCase__			)
        lowercase__:       Dict        =       TFOPTForCausalLM.from_pretrained(lowerCAmelCase__			)

        for prompt in self.prompts:
               lowercase__:       Optional[Any]        =       tokenizer(lowerCAmelCase__    ,			return_tensors='tf'			).input_ids

               lowercase__:       List[str]        =       model.generate(lowerCAmelCase__    ,			max_length=10			)

               lowercase__:       List[Any]        =       tokenizer.batch_decode(lowerCAmelCase__    ,			skip_special_tokens=lowerCAmelCase__			)
               predicted_outputs += generated_string

        self.assertListEqual(lowerCAmelCase__    ,			lowerCAmelCase__			)






import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


__lowerCAmelCase			=							logging.get_logger(__name__)

__lowerCAmelCase			=							'''▁'''

__lowerCAmelCase			=							{'''vocab_file''': '''sentencepiece.bpe.model''', '''monolingual_vocab_file''': '''dict.txt'''}

__lowerCAmelCase			=							{
    '''vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/sentencepiece.bpe.model''',
    },
    '''monolingual_vocab_file''': {
        '''vinai/bartpho-syllable''': '''https://huggingface.co/vinai/bartpho-syllable/resolve/main/dict.txt''',
    },
}

__lowerCAmelCase			=							{'''vinai/bartpho-syllable''': 10_24}





class   __a      (   __UpperCamelCase				):
 __lowercase			:							int       					=			VOCAB_FILES_NAMES
 __lowercase			:							str       					=			PRETRAINED_VOCAB_FILES_MAP
 __lowercase			:							str       					=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __lowercase			:							Optional[int]       					=			['input_ids', 'attention_mask']
 def __init__(       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__="<s>"    ,			lowerCAmelCase__="</s>"    ,			lowerCAmelCase__="</s>"    ,			lowerCAmelCase__="<s>"    ,			lowerCAmelCase__="<unk>"    ,			lowerCAmelCase__="<pad>"    ,			lowerCAmelCase__="<mask>"    ,			lowerCAmelCase__ = None    ,			**lowerCAmelCase__    ,			) ->    None:




        '''simple docstring'''
        # Mask token behave like a normal word, i.e. include the space before it
        lowercase__:       List[str]        =       AddedToken(lowerCAmelCase__    ,			lstrip=lowerCAmelCase__    ,			rstrip=lowerCAmelCase__			) if isinstance(lowerCAmelCase__    ,			lowerCAmelCase__			) else mask_token

        lowercase__:       Any        =       {} if sp_model_kwargs is None else sp_model_kwargs

        super().__init__(
            bos_token=lowerCAmelCase__    ,			eos_token=lowerCAmelCase__    ,			unk_token=lowerCAmelCase__    ,			sep_token=lowerCAmelCase__    ,			cls_token=lowerCAmelCase__    ,			pad_token=lowerCAmelCase__    ,			mask_token=lowerCAmelCase__    ,			sp_model_kwargs=self.sp_model_kwargs    ,			**lowerCAmelCase__    ,			)

        lowercase__:       Dict        =       vocab_file
        lowercase__:       str        =       monolingual_vocab_file
        lowercase__:       int        =       spm.SentencePieceProcessor(**self.sp_model_kwargs			)
        self.sp_model.Load(str(lowerCAmelCase__			)			)

        # Load the reduced vocab

        # Keep order of special tokens for backward compatibility
        lowercase__:       List[Any]        =       {}
        lowercase__:       Optional[int]        =       0
        for token in [bos_token, pad_token, eos_token, unk_token, sep_token, cls_token]:
               if str(lowerCAmelCase__			) not in self.fairseq_tokens_to_ids:
                      lowercase__:       str        =       cnt
                      cnt += 1
        with open(lowerCAmelCase__    ,			'r'    ,			encoding='utf-8'			) as f:
               for line in f.readlines():
                      lowercase__:       Optional[Any]        =       line.strip().split()[0]
                      lowercase__:       Optional[Any]        =       len(self.fairseq_tokens_to_ids			)
        if str(lowerCAmelCase__			) not in self.fairseq_tokens_to_ids:
               lowercase__:       Optional[int]        =       len(self.fairseq_tokens_to_ids			)

        lowercase__:       Optional[Any]        =       {v: k for k, v in self.fairseq_tokens_to_ids.items()}
 def __getstate__(       self			) ->    Optional[int]:




        '''simple docstring'''
        lowercase__:       Tuple        =       self.__dict__.copy()
        lowercase__:       Tuple        =       None
        lowercase__:       Optional[Any]        =       self.sp_model.serialized_model_proto()
        return state
 def __setstate__(       self    ,			lowerCAmelCase__			) ->    Optional[Any]:




        '''simple docstring'''
        lowercase__:       Union[str, Any]        =       d

        # for backward compatibility
        if not hasattr(self    ,			'sp_model_kwargs'			):
               lowercase__:       Union[str, Any]        =       {}

        lowercase__:       List[Any]        =       spm.SentencePieceProcessor(**self.sp_model_kwargs			)
        self.sp_model.LoadFromSerializedProto(self.sp_model_proto			)
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__ = None			) ->    List[int]:




        '''simple docstring'''

        if token_ids_a is None:
               return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
        lowercase__:       Optional[int]        =       [self.cls_token_id]
        lowercase__:       Union[str, Any]        =       [self.sep_token_id]
        return cls + token_ids_a + sep + sep + token_ids_a + sep
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__ = None    ,			lowerCAmelCase__ = False			) ->    List[int]:




        '''simple docstring'''

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

        if token_ids_a is None:
               return [1] + ([0] * len(lowerCAmelCase__			)) + [1]
        return [1] + ([0] * len(lowerCAmelCase__			)) + [1, 1] + ([0] * len(lowerCAmelCase__			)) + [1]
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__ = None			) ->    List[int]:




        '''simple docstring'''
        lowercase__:       Dict        =       [self.sep_token_id]
        lowercase__:       Dict        =       [self.cls_token_id]

        if token_ids_a is None:
               return len(cls + token_ids_a + sep			) * [0]
        return len(cls + token_ids_a + sep + sep + token_ids_a + sep			) * [0]
 @property
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        return len(self.fairseq_ids_to_tokens			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Union[str, Any]:




        '''simple docstring'''
        lowercase__:       Union[str, Any]        =       {self.convert_ids_to_tokens(lowerCAmelCase__			): i for i in range(self.vocab_size			)}
        vocab.update(self.added_tokens_encoder			)
        return vocab
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__			) ->    List[str]:




        '''simple docstring'''
        return self.sp_model.encode(lowerCAmelCase__    ,			out_type=lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__			) ->    List[Any]:




        '''simple docstring'''
        if token in self.fairseq_tokens_to_ids:
               return self.fairseq_tokens_to_ids[token]
        else:
               return self.unk_token_id
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__			) ->    Optional[Any]:




        '''simple docstring'''
        return self.fairseq_ids_to_tokens[index]
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__			) ->    Optional[Any]:




        '''simple docstring'''
        lowercase__:       Optional[Any]        =       ''.join(lowerCAmelCase__			).replace(lowerCAmelCase__    ,			' '			).strip()
        return out_string
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__ = None			) ->    Tuple[str]:




        '''simple docstring'''
        if not os.path.isdir(lowerCAmelCase__			):
               logger.error(F'Vocabulary path ({save_directory}) should be a directory'			)
               return
        lowercase__:       int        =       os.path.join(
            lowerCAmelCase__    ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']			)
        lowercase__:       List[str]        =       os.path.join(
            lowerCAmelCase__    ,			(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['monolingual_vocab_file']    ,			)

        if os.path.abspath(self.vocab_file			) != os.path.abspath(lowerCAmelCase__			) and os.path.isfile(self.vocab_file			):
               copyfile(self.vocab_file    ,			lowerCAmelCase__			)
        elif not os.path.isfile(self.vocab_file			):
               with open(lowerCAmelCase__    ,			'wb'			) as fi:
                      lowercase__:       Optional[Any]        =       self.sp_model.serialized_model_proto()
                      fi.write(lowerCAmelCase__			)

        if os.path.abspath(self.monolingual_vocab_file			) != os.path.abspath(
            lowerCAmelCase__			) and os.path.isfile(self.monolingual_vocab_file			):
               copyfile(self.monolingual_vocab_file    ,			lowerCAmelCase__			)
        elif not os.path.isfile(self.monolingual_vocab_file			):
               with open(lowerCAmelCase__    ,			'w'    ,			encoding='utf-8'			) as fp:
                      for token in self.fairseq_tokens_to_ids:
                             if token not in self.all_special_tokens:
                                    fp.write(F'{str(lowerCAmelCase__			)} \n'			)

        return out_vocab_file, out_monolingual_vocab_file


