

"""simple docstring"""






import unittest

from transformers import AutoTokenizer, FalconConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
 import torch

 from transformers import (
     FalconForCausalLM,
     FalconForQuestionAnswering,
     FalconForSequenceClassification,
     FalconForTokenClassification,
     FalconModel,
 )







class 	__UpperCamelCase    :

  def __init__(		self    ,_A    ,_A=3    ,_A=7    ,_A=True    ,_A=True    ,_A=False    ,_A=True    ,_A=99    ,_A=32    ,_A=5    ,_A=4    ,_A=37    ,_A="gelu"    ,_A=0.1    ,_A=0.1    ,_A=512    ,_A=16    ,_A=2    ,_A=0.0_2    ,_A=3    ,_A=4    ,_A=None    ,):
      '''simple docstring'''

      _lowerCAmelCase       :   str  =    parent
      _lowerCAmelCase       :   Any  =    batch_size
      _lowerCAmelCase       :   Any  =    seq_length
      _lowerCAmelCase       :   Optional[int]  =    is_training
      _lowerCAmelCase       :   List[str]  =    use_input_mask
      _lowerCAmelCase       :   Union[str, Any]  =    use_token_type_ids
      _lowerCAmelCase       :   List[str]  =    use_labels
      _lowerCAmelCase       :   List[str]  =    vocab_size
      _lowerCAmelCase       :   int  =    hidden_size
      _lowerCAmelCase       :   Union[str, Any]  =    num_hidden_layers
      _lowerCAmelCase       :   int  =    num_attention_heads
      _lowerCAmelCase       :   List[str]  =    intermediate_size
      _lowerCAmelCase       :   Tuple  =    hidden_act
      _lowerCAmelCase       :   Optional[int]  =    hidden_dropout_prob
      _lowerCAmelCase       :   Any  =    attention_probs_dropout_prob
      _lowerCAmelCase       :   Tuple  =    max_position_embeddings
      _lowerCAmelCase       :   Tuple  =    type_vocab_size
      _lowerCAmelCase       :   Dict  =    type_sequence_label_size
      _lowerCAmelCase       :   str  =    initializer_range
      _lowerCAmelCase       :   str  =    num_labels
      _lowerCAmelCase       :   Any  =    num_choices
      _lowerCAmelCase       :   Dict  =    scope

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   List[str]  =    ids_tensor([self.batch_size, self.seq_length]    ,self.vocab_size      )

      _lowerCAmelCase       :   List[Any]  =    None
      if self.use_input_mask:
          _lowerCAmelCase       :   List[Any]  =    random_attention_mask([self.batch_size, self.seq_length]      )

      _lowerCAmelCase       :   Dict  =    None

      _lowerCAmelCase       :   Union[str, Any]  =    None
      _lowerCAmelCase       :   Optional[int]  =    None
      _lowerCAmelCase       :   List[Any]  =    None
      if self.use_labels:
          _lowerCAmelCase       :   str  =    ids_tensor([self.batch_size]    ,self.type_sequence_label_size      )
          _lowerCAmelCase       :   Optional[int]  =    ids_tensor([self.batch_size, self.seq_length]    ,self.num_labels      )
          _lowerCAmelCase       :   List[str]  =    ids_tensor([self.batch_size]    ,self.num_choices      )

      _lowerCAmelCase       :   Optional[int]  =    self.get_config()

      return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return FalconConfig(
          vocab_size=self.vocab_size    ,hidden_size=self.hidden_size    ,num_hidden_layers=self.num_hidden_layers    ,num_attention_heads=self.num_attention_heads    ,intermediate_size=self.intermediate_size    ,hidden_act=self.hidden_act    ,hidden_dropout_prob=self.hidden_dropout_prob    ,attention_probs_dropout_prob=self.attention_probs_dropout_prob    ,max_position_embeddings=self.max_position_embeddings    ,type_vocab_size=self.type_vocab_size    ,is_decoder=_A    ,initializer_range=self.initializer_range    ,pad_token_id=1    ,new_decoder_architecture=_A    ,)

  def __lowerCamelCase		(		self    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A      ):
      '''simple docstring'''

      _lowerCAmelCase       :   List[Any]  =    FalconModel(config=_A      )
      model.to(_A      )
      model.eval()
      _lowerCAmelCase       :   Tuple  =    model(_A    ,attention_mask=_A      )
      _lowerCAmelCase       :   Dict  =    model(_A      )
      self.parent.assertEqual(result.last_hidden_state.shape    ,(self.batch_size, self.seq_length, self.hidden_size)      )

  def __lowerCamelCase		(		self    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,):
      '''simple docstring'''

      _lowerCAmelCase       :   List[str]  =    True
      _lowerCAmelCase       :   str  =    FalconModel(_A      )
      model.to(_A      )
      model.eval()
      _lowerCAmelCase       :   Optional[Any]  =    model(
          _A    ,attention_mask=_A    ,encoder_hidden_states=_A    ,encoder_attention_mask=_A    ,)
      _lowerCAmelCase       :   Union[str, Any]  =    model(
          _A    ,attention_mask=_A    ,encoder_hidden_states=_A    ,)
      _lowerCAmelCase       :   Optional[Any]  =    model(_A    ,attention_mask=_A      )
      self.parent.assertEqual(result.last_hidden_state.shape    ,(self.batch_size, self.seq_length, self.hidden_size)      )

  def __lowerCamelCase		(		self    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,):
      '''simple docstring'''

      _lowerCAmelCase       :   List[str]  =    FalconForCausalLM(config=_A      )
      model.to(_A      )
      model.eval()
      _lowerCAmelCase       :   Tuple  =    model(_A    ,attention_mask=_A    ,labels=_A      )
      self.parent.assertEqual(result.logits.shape    ,(self.batch_size, self.seq_length, self.vocab_size)      )

  def __lowerCamelCase		(		self    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A    ,):
      '''simple docstring'''

      _lowerCAmelCase       :   int  =    True
      _lowerCAmelCase       :   Union[str, Any]  =    True
      _lowerCAmelCase       :   int  =    FalconForCausalLM(config=_A      )
      model.to(_A      )
      model.eval()

      # first forward pass
      _lowerCAmelCase       :   Optional[int]  =    model(
          _A    ,attention_mask=_A    ,encoder_hidden_states=_A    ,encoder_attention_mask=_A    ,use_cache=_A    ,)
      _lowerCAmelCase       :   int  =    outputs.past_key_values

      # create hypothetical multiple next token and extent to next_input_ids
      _lowerCAmelCase       :   List[Any]  =    ids_tensor((self.batch_size, 3)    ,config.vocab_size      )
      _lowerCAmelCase       :   Optional[Any]  =    ids_tensor((self.batch_size, 3)    ,vocab_size=2      )

      # append to next input_ids and
      _lowerCAmelCase       :   Dict  =    torch.cat([input_ids, next_tokens]    ,dim=-1      )
      _lowerCAmelCase       :   Union[str, Any]  =    torch.cat([input_mask, next_mask]    ,dim=-1      )

      _lowerCAmelCase       :   str  =    model(
          _A    ,attention_mask=_A    ,encoder_hidden_states=_A    ,encoder_attention_mask=_A    ,output_hidden_states=_A    ,)['hidden_states'][0]
      _lowerCAmelCase       :   Optional[Any]  =    model(
          _A    ,attention_mask=_A    ,encoder_hidden_states=_A    ,encoder_attention_mask=_A    ,past_key_values=_A    ,output_hidden_states=_A    ,)['hidden_states'][0]

      # select random slice
      _lowerCAmelCase       :   List[Any]  =    ids_tensor((1,)    ,output_from_past.shape[-1]      ).item()
      _lowerCAmelCase       :   Any  =    output_from_no_past[:, -3:, random_slice_idx].detach()
      _lowerCAmelCase       :   Any  =    output_from_past[:, :, random_slice_idx].detach()

      self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]      )

      # test that outputs are equal for slice
      self.parent.assertTrue(torch.allclose(_A    ,_A    ,atol=1E-3      )      )







  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   Tuple  =    self.prepare_config_and_inputs()
      (
          (
          _lowerCAmelCase
      ),					(
          _lowerCAmelCase
      ),					(
          _lowerCAmelCase
      ),					(
          _lowerCAmelCase
      ),					(
          _lowerCAmelCase
      ),					(
          _lowerCAmelCase
      ),					(
          _lowerCAmelCase
      ),					
      )       :   int  =    config_and_inputs
      _lowerCAmelCase       :   Tuple  =    {'input_ids': input_ids, 'attention_mask': input_mask}
      return config, inputs_dict







@require_torch
class 	__UpperCamelCase    (    a__							,							a__							,							a__							,							unittest.TestCase						):
  _UpperCAmelCase          =		(
      (
          FalconModel,
          FalconForCausalLM,
          FalconForSequenceClassification,
          FalconForTokenClassification,
          FalconForQuestionAnswering,
      )
      if is_torch_available()
      else ()
  )
  _UpperCAmelCase          =		(FalconForCausalLM,) if is_torch_available() else ()
  _UpperCAmelCase          =		(
      {
          "feature-extraction": FalconModel,
          "text-classification": FalconForSequenceClassification,
          "text-generation": FalconForCausalLM,
          "question-answering": FalconForQuestionAnswering,
          "token-classification": FalconForTokenClassification,
          "zero-shot": FalconForSequenceClassification,
      }
      if is_torch_available()
      else {}
  )
  _UpperCAmelCase          =		False
  _UpperCAmelCase          =		False

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   str  =    FalconModelTester(self      )
      _lowerCAmelCase       :   int  =    ConfigTester(self    ,config_class=_A    ,hidden_size=37      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      self.config_tester.run_common_tests()

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   List[Any]  =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*_A      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase,					*_lowerCAmelCase       :   List[Any]  =    self.model_tester.prepare_config_and_inputs()
      for alibi in [True, False]:
          _lowerCAmelCase       :   Optional[int]  =    alibi
          self.model_tester.create_and_check_model(_A    ,*_A      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase,					_lowerCAmelCase       :   Optional[int]  =    self.model_tester.prepare_config_and_inputs_for_common()
      _lowerCAmelCase       :   List[Any]  =    3
      _lowerCAmelCase       :   Optional[Any]  =    input_dict['input_ids']
      _lowerCAmelCase       :   Union[str, Any]  =    input_ids.ne(1      ).to(_A      )
      _lowerCAmelCase       :   Tuple  =    ids_tensor([self.model_tester.batch_size]    ,self.model_tester.type_sequence_label_size      )
      _lowerCAmelCase       :   Optional[int]  =    FalconForSequenceClassification(_A      )
      model.to(_A      )
      model.eval()
      _lowerCAmelCase       :   List[Any]  =    model(_A    ,attention_mask=_A    ,labels=_A      )
      self.assertEqual(result.logits.shape    ,(self.model_tester.batch_size, self.model_tester.num_labels)      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase,					_lowerCAmelCase       :   Optional[int]  =    self.model_tester.prepare_config_and_inputs_for_common()
      _lowerCAmelCase       :   Any  =    3
      _lowerCAmelCase       :   Dict  =    'single_label_classification'
      _lowerCAmelCase       :   Union[str, Any]  =    input_dict['input_ids']
      _lowerCAmelCase       :   Optional[Any]  =    input_ids.ne(1      ).to(_A      )
      _lowerCAmelCase       :   str  =    ids_tensor([self.model_tester.batch_size]    ,self.model_tester.type_sequence_label_size      )
      _lowerCAmelCase       :   Optional[int]  =    FalconForSequenceClassification(_A      )
      model.to(_A      )
      model.eval()
      _lowerCAmelCase       :   int  =    model(_A    ,attention_mask=_A    ,labels=_A      )
      self.assertEqual(result.logits.shape    ,(self.model_tester.batch_size, self.model_tester.num_labels)      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase,					_lowerCAmelCase       :   Any  =    self.model_tester.prepare_config_and_inputs_for_common()
      _lowerCAmelCase       :   List[str]  =    input_dict['input_ids']
      _lowerCAmelCase       :   int  =    FalconForCausalLM(_A      )
      model.to(_A      )
      model.eval()
      _lowerCAmelCase       :   Optional[Any]  =    model(_A    ,use_cache=_A      )
      _lowerCAmelCase       :   str  =    input_ids.shape[0]
      _lowerCAmelCase       :   Optional[Any]  =    model._convert_to_rw_cache(result.past_key_values      )
      _lowerCAmelCase       :   int  =    model._convert_cache_to_standard_format(_A    ,_A      )
      for layer in range(len(_A      )      ):
          for tensor_idx in range(2      ):
              self.assertTrue(rw_cache[layer][tensor_idx].ndim == 3      )
              self.assertTrue(result.past_key_values[layer][tensor_idx].ndim == 4      )
              self.assertTrue(
                  torch.all(result.past_key_values[layer][tensor_idx] == standard_cache[layer][tensor_idx]      )      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase,					_lowerCAmelCase       :   Any  =    self.model_tester.prepare_config_and_inputs_for_common()
      _lowerCAmelCase       :   Union[str, Any]  =    3
      _lowerCAmelCase       :   Optional[Any]  =    'multi_label_classification'
      _lowerCAmelCase       :   List[str]  =    input_dict['input_ids']
      _lowerCAmelCase       :   Optional[Any]  =    input_ids.ne(1      ).to(_A      )
      _lowerCAmelCase       :   int  =    ids_tensor(
          [self.model_tester.batch_size, config.num_labels]    ,self.model_tester.type_sequence_label_size      ).to(torch.float      )
      _lowerCAmelCase       :   Tuple  =    FalconForSequenceClassification(_A      )
      model.to(_A      )
      model.eval()
      _lowerCAmelCase       :   int  =    model(_A    ,attention_mask=_A    ,labels=_A      )
      self.assertEqual(result.logits.shape    ,(self.model_tester.batch_size, self.model_tester.num_labels)      )







  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      for model_class in self.all_generative_model_classes:
          _lowerCAmelCase,					_lowerCAmelCase       :   List[Any]  =    self.model_tester.prepare_config_and_inputs_for_common()

          # If it doesn't support cache, pass the test
          if not hasattr(_A    ,'use_cache'      ):
              return

          _lowerCAmelCase       :   Dict  =    model_class(_A      ).to(_A      )
          if "use_cache" not in inputs:
              _lowerCAmelCase       :   Tuple  =    True
          _lowerCAmelCase       :   Any  =    model(**_A      )

          # If "past_key_values" is not returned, pass the test (e.g. RWKV uses a different cache name and format)
          if "past_key_values" not in outputs:
              return

          _lowerCAmelCase       :   List[Any]  =    (
              getattr(_A    ,'decoder_layers'    ,_A      )
              or getattr(_A    ,'num_decoder_layers'    ,_A      )
              or config.num_hidden_layers
          )
          _lowerCAmelCase       :   Dict  =    getattr(_A    ,'num_kv_heads'    ,config.num_attention_heads      )
          _lowerCAmelCase       :   Any  =    getattr(_A    ,'d_model'    ,config.hidden_size      )
          _lowerCAmelCase       :   Dict  =    embed_dim // num_attention_heads

          _lowerCAmelCase       :   Optional[Any]  =    outputs['past_key_values']
          self.assertEqual(len(_A      )    ,_A      )

          _lowerCAmelCase,					_lowerCAmelCase       :   Any  =    inputs['input_ids'].shape
          for i in range(_A      ):
              if config.new_decoder_architecture:
                  _lowerCAmelCase       :   int  =    config.num_attention_heads
              elif config.multi_query:
                  _lowerCAmelCase       :   Tuple  =    1
              self.assertEqual(len(past_kv[0]      )    ,2      )  # K V for the decoder = 2
              self.assertEqual(
                  past_kv[i][0].shape    ,(batch_size, num_attention_heads, seq_length, per_head_embed_dim)      )
              self.assertEqual(
                  past_kv[i][1].shape    ,(batch_size, num_attention_heads, seq_length, per_head_embed_dim)      )







@require_torch
class 	__UpperCamelCase    (    unittest.TestCase						):

  @slow
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   Tuple  =    AutoTokenizer.from_pretrained('Rocketknight1/falcon-rw-1b'      )
      _lowerCAmelCase       :   int  =    FalconForCausalLM.from_pretrained('Rocketknight1/falcon-rw-1b'      )
      model.eval()
      model.to(_A      )
      _lowerCAmelCase       :   str  =    tokenizer('My favorite food is'    ,return_tensors='pt'      ).to(_A      )

      _lowerCAmelCase       :   int  =    (
          'My favorite food is pizza. I love it so much that I have a pizza party every year for my birthday.'
      )

      _lowerCAmelCase       :   Optional[int]  =    model.generate(**_A    ,do_sample=_A    ,max_new_tokens=19      )
      _lowerCAmelCase       :   List[Any]  =    tokenizer.batch_decode(_A      )[0]

      self.assertEqual(_A    ,_A      )

  @slow
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      for repo in ["Rocketknight1/tiny-random-falcon-7b", "Rocketknight1/tiny-random-falcon-40b"]:
          _lowerCAmelCase       :   List[str]  =    AutoTokenizer.from_pretrained(_A      )
          _lowerCAmelCase       :   Optional[Any]  =    FalconForCausalLM.from_pretrained(_A      )
          model.eval()
          model.to(_A      )
          _lowerCAmelCase       :   Optional[Any]  =    tokenizer('My favorite food is'    ,return_tensors='pt'      ).to(_A      )

          # We just test that these run without errors - the models are randomly initialized
          # and so the actual text outputs will be garbage
          model.generate(**_A    ,do_sample=_A    ,max_new_tokens=4      )
          model.generate(**_A    ,do_sample=_A    ,max_new_tokens=4      )
          model.generate(**_A    ,num_beams=2    ,max_new_tokens=4      )







  @slow
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      with torch.no_grad():
          for repo in [
              "Rocketknight1/falcon-rw-1b",
              "Rocketknight1/tiny-random-falcon-7b",
              "Rocketknight1/tiny-random-falcon-40b",
          ]:
              _lowerCAmelCase       :   Union[str, Any]  =    AutoTokenizer.from_pretrained(_A      )
              _lowerCAmelCase       :   List[Any]  =    FalconForCausalLM.from_pretrained(_A      )
              model.eval()
              model.to(device=_A      )
              _lowerCAmelCase       :   Optional[Any]  =    tokenizer('My favorite food is'    ,return_tensors='pt'      ).to(_A      )

              # Test results are the same with and without cache
              _lowerCAmelCase       :   Optional[int]  =    model.generate(**_A    ,do_sample=_A    ,max_new_tokens=20    ,use_cache=_A      )
              _lowerCAmelCase       :   Union[str, Any]  =    model.generate(**_A    ,do_sample=_A    ,max_new_tokens=20    ,use_cache=_A      )
              self.assertTrue((outputs_cache - outputs_no_cache).sum().item() == 0      )








"""simple docstring"""






import json
import os
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


_lowerCAmelCase						=		logging.get_logger(__name__)


_lowerCAmelCase						=		{
    """vocab_file""": """vocab.json""",
    """tokenizer_config_file""": """tokenizer_config.json""",
    """merges_file""": """merges.txt""",
}

_lowerCAmelCase						=		{
    """vocab_file""": {
        """facebook/s2t-wav2vec2-large-en-de""": (
            """https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/vocab.json"""
        ),
    },
    """tokenizer_config_file""": {
        """facebook/s2t-wav2vec2-large-en-de""": (
            """https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/tokenizer_config.json"""
        ),
    },
    """merges_file""": {
        """facebook/s2t-wav2vec2-large-en-de""": (
            """https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/merges.txt"""
        ),
    },
}

_lowerCAmelCase						=		"""</w>"""
_lowerCAmelCase						=		"""@@ """

def 		lowerCamelCase__		(					_lowerCamelCase   ):

    '''simple docstring'''



    _lowerCAmelCase       :   List[str]  =    set()
    _lowerCAmelCase       :   Dict  =    word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)   )
        _lowerCAmelCase       :   Any  =    char
    return pairs


# Speech2Text2 has no max input length
_lowerCAmelCase						=		{"""facebook/s2t-wav2vec2-large-en-de""": 1_0_2_4}







class 	__UpperCamelCase    (    a__						):
  _UpperCAmelCase          =		VOCAB_FILES_NAMES
  _UpperCAmelCase          =		PRETRAINED_VOCAB_FILES_MAP
  _UpperCAmelCase          =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  _UpperCAmelCase          =		["input_ids", "attention_mask"]

  def __init__(		self    ,_A    ,_A="<s>"    ,_A="<pad>"    ,_A="</s>"    ,_A="<unk>"    ,_A=False    ,_A=None    ,**_A    ,):
      '''simple docstring'''

      super().__init__(
          unk_token=_A    ,bos_token=_A    ,eos_token=_A    ,pad_token=_A    ,do_lower_case=_A    ,**_A    ,)

      _lowerCAmelCase       :   List[Any]  =    do_lower_case

      with open(_A    ,encoding='utf-8'      ) as vocab_handle:
          _lowerCAmelCase       :   Optional[int]  =    json.load(_A      )
      _lowerCAmelCase       :   Tuple  =    {v: k for k, v in self.encoder.items()}

      if merges_file is None:
          logger.info(F"""No merges files provided. {self.__class__.__name__} can only be used for decoding."""      )

          _lowerCAmelCase       :   Optional[Any]  =    None
          _lowerCAmelCase       :   Tuple  =    None
      else:
          with open(_A    ,encoding='utf-8'      ) as merges_handle:
              _lowerCAmelCase       :   Optional[Any]  =    merges_handle.read().split('\n'      )[:-1]

          _lowerCAmelCase       :   List[str]  =    [tuple(merge.split()[:2]      ) for merge in merges]
          _lowerCAmelCase       :   List[Any]  =    dict(zip(_A    ,range(len(_A      )      )      )      )
          _lowerCAmelCase       :   Union[str, Any]  =    {}

  @property
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return len(self.decoder      )

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return dict(self.encoder    ,**self.added_tokens_encoder      )

  def __lowerCamelCase		(		self    ,_A      ):
      '''simple docstring'''

      _lowerCAmelCase       :   str  =    tuple(token[:-1]      ) + (token[-1] + BPE_TOKEN_MERGES,)
      if token in self.cache:
          return self.cache[token]
      _lowerCAmelCase       :   str  =    get_pairs(_A      )

      if not pairs:
          return token

      while True:
          _lowerCAmelCase       :   List[str]  =    min(_A    ,key=lambda _A      : self.bpe_ranks.get(_A    ,float('inf'      )      )      )
          if bigram not in self.bpe_ranks:
              break
          _lowerCAmelCase,					_lowerCAmelCase       :   Optional[int]  =    bigram
          _lowerCAmelCase       :   Union[str, Any]  =    []
          _lowerCAmelCase       :   Dict  =    0
          while i < len(_A      ):
              try:
                  _lowerCAmelCase       :   Dict  =    word.index(_A    ,_A      )
              except ValueError:
                  new_word.extend(word[i:]      )
                  break
              else:
                  new_word.extend(word[i:j]      )
                  _lowerCAmelCase       :   Optional[Any]  =    j

              if word[i] == first and i < len(_A      ) - 1 and word[i + 1] == second:
                  new_word.append(first + second      )
                  i += 2
              else:
                  new_word.append(word[i]      )
                  i += 1
          _lowerCAmelCase       :   Optional[Any]  =    tuple(_A      )
          _lowerCAmelCase       :   List[str]  =    new_word
          if len(_A      ) == 1:
              break
          else:
              _lowerCAmelCase       :   List[str]  =    get_pairs(_A      )
      _lowerCAmelCase       :   Any  =    ' '.join(_A      )
      if word == "\n  " + BPE_TOKEN_MERGES:
          _lowerCAmelCase       :   str  =    '\n' + BPE_TOKEN_MERGES

      if word.endswith(_A      ):
          _lowerCAmelCase       :   Dict  =    word.replace(_A    ,''      )

      _lowerCAmelCase       :   str  =    word.replace(' '    ,_A      )
      _lowerCAmelCase       :   str  =    word
      return word

  def __lowerCamelCase		(		self    ,_A      ):
      '''simple docstring'''

      if self.bpe_ranks is None:
          raise ValueError(
              'This tokenizer was instantiated without a `merges.txt` file, so'
              ' that it can only be used for decoding, not for encoding.'
              'Make sure to provide `merges.txt` file at instantiation to enable '
              'encoding.'      )

      if self.do_lower_case:
          _lowerCAmelCase       :   Optional[Any]  =    text.lower()

      _lowerCAmelCase       :   Tuple  =    text.split()

      _lowerCAmelCase       :   Union[str, Any]  =    []
      for token in text:
          if token:
              split_tokens.extend(list(self.bpe(_A      ).split(' '      )      )      )

      return split_tokens

  def __lowerCamelCase		(		self    ,_A      ):
      '''simple docstring'''

      return self.encoder.get(_A    ,self.encoder.get(self.unk_token      )      )

  def __lowerCamelCase		(		self    ,_A      ):
      '''simple docstring'''

      _lowerCAmelCase       :   int  =    self.decoder.get(_A    ,self.unk_token      )
      return result

  def __lowerCamelCase		(		self    ,_A      ):
      '''simple docstring'''

      _lowerCAmelCase       :   Optional[Any]  =    ' '.join(_A      )

      # make sure @@ tokens are concatenated
      _lowerCAmelCase       :   int  =    ''.join(string.split(_A      )      )

      return string







  def __lowerCamelCase		(		self    ,_A    ,_A = None      ):
      '''simple docstring'''

      if not os.path.isdir(_A      ):
          logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""      )
          return
      _lowerCAmelCase       :   List[Any]  =    os.path.join(
          _A    ,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']      )
      _lowerCAmelCase       :   str  =    os.path.join(
          _A    ,(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']      )

      with open(_A    ,'w'    ,encoding='utf-8'      ) as f:
          f.write(json.dumps(self.encoder    ,indent=2    ,sort_keys=_A    ,ensure_ascii=_A      ) + '\n'      )

      _lowerCAmelCase       :   str  =    0
      if self.bpe_ranks is None:
          return (vocab_file,)

      with open(_A    ,'w'    ,encoding='utf-8'      ) as writer:
          for bpe_tokens, token_index in sorted(self.bpe_ranks.items()    ,key=lambda _A      : kv[1]      ):
              if index != token_index:
                  logger.warning(
                      F"""Saving vocabulary to {merges_file}: BPE merge indices are not consecutive."""
                      ' Please check that the tokenizer is not corrupted!'      )
                  _lowerCAmelCase       :   Dict  =    token_index
              writer.write(' '.join(_A      ) + '\n'      )
              index += 1

      return (vocab_file, merges_file)





