"""simple docstring"""







import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

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


if TYPE_CHECKING:
     from transformers.pipelines.conversational import Conversation

_SCREAMING_SNAKE_CASE	:		int   =   logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE	:		str   =   {
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

_SCREAMING_SNAKE_CASE	:		Dict   =   {
    '''vocab_file''': {
        '''gpt2''': '''https://huggingface.co/gpt2/resolve/main/vocab.json''',
        '''gpt2-medium''': '''https://huggingface.co/gpt2-medium/resolve/main/vocab.json''',
        '''gpt2-large''': '''https://huggingface.co/gpt2-large/resolve/main/vocab.json''',
        '''gpt2-xl''': '''https://huggingface.co/gpt2-xl/resolve/main/vocab.json''',
        '''distilgpt2''': '''https://huggingface.co/distilgpt2/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''gpt2''': '''https://huggingface.co/gpt2/resolve/main/merges.txt''',
        '''gpt2-medium''': '''https://huggingface.co/gpt2-medium/resolve/main/merges.txt''',
        '''gpt2-large''': '''https://huggingface.co/gpt2-large/resolve/main/merges.txt''',
        '''gpt2-xl''': '''https://huggingface.co/gpt2-xl/resolve/main/merges.txt''',
        '''distilgpt2''': '''https://huggingface.co/distilgpt2/resolve/main/merges.txt''',
    },
}

_SCREAMING_SNAKE_CASE	:		int   =   {
    '''gpt2''': 1_0_2_4,
    '''gpt2-medium''': 1_0_2_4,
    '''gpt2-large''': 1_0_2_4,
    '''gpt2-xl''': 1_0_2_4,
    '''distilgpt2''': 1_0_2_4,
}



@lru_cache()
def       A__    (	)  ->							Optional[int]:
   __lowerCamelCase    		=					(
       list(range(ord('!'						)      , ord('~'						) + 1						)						) + list(range(ord('¡'						)      , ord('¬'						) + 1						)						) + list(range(ord('®'						)      , ord('ÿ'						) + 1						)						)
   )
   __lowerCamelCase    		=					bs[:]
   __lowerCamelCase    		=					0
   for b in range(2**8						):
      if b not in bs:
         bs.append(lowercase_						)
         cs.append(2**8 + n						)
         n += 1
   __lowerCamelCase    		=					[chr(lowercase_						) for n in cs]
   return dict(zip(lowercase_      , lowercase_						)						)



def       A__    (	lowercase_     :	int						)  ->							Optional[Any]:
   __lowerCamelCase    		=					set()
   __lowerCamelCase    		=					word[0]
   for char in word[1:]:
      pairs.add((prev_char, char)						)
      __lowerCamelCase    		=					char
   return pairs






class      _SCREAMING_SNAKE_CASE							(			UpperCAmelCase				):



   '''simple docstring'''



   __UpperCAmelCase	:							Tuple    =    VOCAB_FILES_NAMES
   __UpperCAmelCase	:							int    =    PRETRAINED_VOCAB_FILES_MAP
   __UpperCAmelCase	:							Union[str, Any]    =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   __UpperCAmelCase	:							Dict    =    ["input_ids", "attention_mask"]







   def __init__(	self ,				_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE="replace" ,				_SCREAMING_SNAKE_CASE="<|endoftext|>" ,				_SCREAMING_SNAKE_CASE="<|endoftext|>" ,				_SCREAMING_SNAKE_CASE="<|endoftext|>" ,				_SCREAMING_SNAKE_CASE=None ,				_SCREAMING_SNAKE_CASE=False ,				_SCREAMING_SNAKE_CASE=False ,				**_SCREAMING_SNAKE_CASE ,				):


      '''simple docstring'''

      __lowerCamelCase    		=					AddedToken(_SCREAMING_SNAKE_CASE ,				lstrip=_SCREAMING_SNAKE_CASE ,				rstrip=_SCREAMING_SNAKE_CASE	) if isinstance(_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE	) else bos_token
      __lowerCamelCase    		=					AddedToken(_SCREAMING_SNAKE_CASE ,				lstrip=_SCREAMING_SNAKE_CASE ,				rstrip=_SCREAMING_SNAKE_CASE	) if isinstance(_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE	) else eos_token
      __lowerCamelCase    		=					AddedToken(_SCREAMING_SNAKE_CASE ,				lstrip=_SCREAMING_SNAKE_CASE ,				rstrip=_SCREAMING_SNAKE_CASE	) if isinstance(_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE	) else unk_token
      __lowerCamelCase    		=					AddedToken(_SCREAMING_SNAKE_CASE ,				lstrip=_SCREAMING_SNAKE_CASE ,				rstrip=_SCREAMING_SNAKE_CASE	) if isinstance(_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE	) else pad_token
      super().__init__(
          errors=_SCREAMING_SNAKE_CASE ,				unk_token=_SCREAMING_SNAKE_CASE ,				bos_token=_SCREAMING_SNAKE_CASE ,				eos_token=_SCREAMING_SNAKE_CASE ,				pad_token=_SCREAMING_SNAKE_CASE ,				add_prefix_space=_SCREAMING_SNAKE_CASE ,				add_bos_token=_SCREAMING_SNAKE_CASE ,				**_SCREAMING_SNAKE_CASE ,				)
      __lowerCamelCase    		=					add_bos_token

      with open(_SCREAMING_SNAKE_CASE ,				encoding='utf-8'	) as vocab_handle:
         __lowerCamelCase    		=					json.load(_SCREAMING_SNAKE_CASE	)
      __lowerCamelCase    		=					{v: k for k, v in self.encoder.items()}
      __lowerCamelCase    		=					errors  # how to handle errors in decoding
      __lowerCamelCase    		=					bytes_to_unicode()
      __lowerCamelCase    		=					{v: k for k, v in self.byte_encoder.items()}
      with open(_SCREAMING_SNAKE_CASE ,				encoding='utf-8'	) as merges_handle:
         __lowerCamelCase    		=					merges_handle.read().split('\n'	)[1:-1]
      __lowerCamelCase    		=					[tuple(merge.split()	) for merge in bpe_merges]
      __lowerCamelCase    		=					dict(zip(_SCREAMING_SNAKE_CASE ,				range(len(_SCREAMING_SNAKE_CASE	)	)	)	)
      __lowerCamelCase    		=					{}
      __lowerCamelCase    		=					add_prefix_space

      # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
      __lowerCamelCase    		=					re.compile(r'\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'	)







   @property
   def    _A				(	self	):


      '''simple docstring'''

      return len(self.encoder	)







   def    _A				(	self	):


      '''simple docstring'''

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







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE	):


      '''simple docstring'''

      if token in self.cache:
         return self.cache[token]
      __lowerCamelCase    		=					tuple(_SCREAMING_SNAKE_CASE	)
      __lowerCamelCase    		=					get_pairs(_SCREAMING_SNAKE_CASE	)

      if not pairs:
         return token

      while True:
         __lowerCamelCase    		=					min(_SCREAMING_SNAKE_CASE ,				key=lambda _SCREAMING_SNAKE_CASE	: self.bpe_ranks.get(_SCREAMING_SNAKE_CASE ,				float('inf'	)	)	)
         if bigram not in self.bpe_ranks:
            break
         __lowerCamelCase		,__lowerCamelCase    		=					bigram
         __lowerCamelCase    		=					[]
         __lowerCamelCase    		=					0
         while i < len(_SCREAMING_SNAKE_CASE	):
            try:
               __lowerCamelCase    		=					word.index(_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE	)
            except ValueError:
               new_word.extend(word[i:]	)
               break
            else:
               new_word.extend(word[i:j]	)
               __lowerCamelCase    		=					j

            if word[i] == first and i < len(_SCREAMING_SNAKE_CASE	) - 1 and word[i + 1] == second:
               new_word.append(first + second	)
               i += 2
            else:
               new_word.append(word[i]	)
               i += 1
         __lowerCamelCase    		=					tuple(_SCREAMING_SNAKE_CASE	)
         __lowerCamelCase    		=					new_word
         if len(_SCREAMING_SNAKE_CASE	) == 1:
            break
         else:
            __lowerCamelCase    		=					get_pairs(_SCREAMING_SNAKE_CASE	)
      __lowerCamelCase    		=					' '.join(_SCREAMING_SNAKE_CASE	)
      __lowerCamelCase    		=					word
      return word







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE=None	):


      '''simple docstring'''

      if self.add_bos_token:
         __lowerCamelCase    		=					[self.bos_token_id]
      else:
         __lowerCamelCase    		=					[]

      __lowerCamelCase    		=					bos_token_ids + token_ids_a

      if token_ids_a is None:
         return output

      return output + bos_token_ids + token_ids_a







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE = None ,				_SCREAMING_SNAKE_CASE = False	):


      '''simple docstring'''

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

      if not self.add_bos_token:
         return super().get_special_tokens_mask(
             token_ids_a=_SCREAMING_SNAKE_CASE ,				token_ids_a=_SCREAMING_SNAKE_CASE ,				already_has_special_tokens=_SCREAMING_SNAKE_CASE	)

      if token_ids_a is None:
         return [1] + ([0] * len(_SCREAMING_SNAKE_CASE	))
      return [1] + ([0] * len(_SCREAMING_SNAKE_CASE	)) + [1] + ([0] * len(_SCREAMING_SNAKE_CASE	))







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE	):


      '''simple docstring'''

      __lowerCamelCase    		=					[]
      for token in re.findall(self.pat ,				_SCREAMING_SNAKE_CASE	):
         __lowerCamelCase    		=					''.join(
             self.byte_encoder[b] for b in token.encode('utf-8'	)	)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
         bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_SCREAMING_SNAKE_CASE	).split(' '	)	)
      return bpe_tokens







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE	):


      '''simple docstring'''

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







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE	):


      '''simple docstring'''

      return self.decoder.get(_SCREAMING_SNAKE_CASE	)







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE	):


      '''simple docstring'''

      __lowerCamelCase    		=					''.join(_SCREAMING_SNAKE_CASE	)
      __lowerCamelCase    		=					bytearray([self.byte_decoder[c] for c in text]	).decode('utf-8' ,				errors=self.errors	)
      return text







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE = None	):


      '''simple docstring'''

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

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

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

      return vocab_file, merge_file







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE ,				_SCREAMING_SNAKE_CASE=False ,				**_SCREAMING_SNAKE_CASE	):


      '''simple docstring'''

      __lowerCamelCase    		=					kwargs.pop('add_prefix_space' ,				self.add_prefix_space	)
      if is_split_into_words or add_prefix_space:
         __lowerCamelCase    		=					' ' + text
      return (text, kwargs)







   def    _A				(	self ,				_SCREAMING_SNAKE_CASE	):


      '''simple docstring'''

      __lowerCamelCase    		=					[]
      for is_user, text in conversation.iter_texts():
         input_ids.extend(self.encode(_SCREAMING_SNAKE_CASE ,				add_special_tokens=_SCREAMING_SNAKE_CASE	) + [self.eos_token_id]	)
      if len(_SCREAMING_SNAKE_CASE	) > self.model_max_length:
         __lowerCamelCase    		=					input_ids[-self.model_max_length :]
      return input_ids







import inspect
import json
import os
import tempfile
import unittest

from transformers.models.speech_to_text_a import SpeechaTextaTokenizer
from transformers.models.speech_to_text_a.tokenization_speech_to_text_a import VOCAB_FILES_NAMES

from ...test_tokenization_common import TokenizerTesterMixin







class       __snake_case (  lowerCamelCase__       , unittest.TestCase ):
 __lowerCamelCase     : Any 	=   SpeechaTextaTokenizer
 __lowerCamelCase     : Tuple 	=   False
 def 						UpperCAmelCase__							(	self    )						->				List[Any]:
       '''simple docstring'''





       super().setUp()

       UpperCAmelCase     :			Union[str, Any]     	='''<s> <pad> </s> <unk> here@@ a couple of@@ words for the he@@ re@@ vocab'''.split(''' '''    )
       UpperCAmelCase     :			int     	=['''he re</w> 123''', '''here a 1456''']
       UpperCAmelCase     :			int     	=dict(zip(snake_case__		, range(len(snake_case__    )    )    )    )

       UpperCAmelCase     :			Tuple     	={'''pad_token''': '''<pad>''', '''unk_token''': '''<unk>''', '''bos_token''': '''<s>''', '''eos_token''': '''</s>'''}

       UpperCAmelCase     :			Tuple     	=tempfile.mkdtemp()
       UpperCAmelCase     :			Optional[Any]     	=os.path.join(self.tmpdirname		, VOCAB_FILES_NAMES['''vocab_file''']    )
       UpperCAmelCase     :			Dict     	=os.path.join(self.tmpdirname		, VOCAB_FILES_NAMES['''merges_file''']    )

       with open(self.vocab_file		, '''w'''		, encoding='''utf-8'''    ) as fp:
             fp.write(json.dumps(snake_case__    ) + '''\n'''    )

       with open(self.merges_file		, '''w'''    ) as fp:
             fp.write('''\n'''.join(snake_case__    )    )
 def 						UpperCAmelCase__							(	self    )						->				Union[str, Any]:
       '''simple docstring'''





       UpperCAmelCase     :			Union[str, Any]     	=list(self.get_tokenizer().get_vocab().keys()    )

       self.assertEqual(vocab_keys[0]		, '''<s>'''    )
       self.assertEqual(vocab_keys[1]		, '''<pad>'''    )
       self.assertEqual(vocab_keys[-1]		, '''vocab'''    )
       self.assertEqual(len(snake_case__    )		, 14    )
 def 						UpperCAmelCase__							(	self    )						->				Union[str, Any]:
       '''simple docstring'''





       self.assertEqual(self.get_tokenizer().vocab_size		, 14    )
 def 						UpperCAmelCase__							(	self    )						->				List[Any]:
       '''simple docstring'''





       UpperCAmelCase     :			Union[str, Any]     	=SpeechaTextaTokenizer.from_pretrained(self.tmpdirname    )

       # make sure @@ is correctly concatenated
       UpperCAmelCase     :			str     	=[4, 6, 8, 7, 10]  # ["here@@", "couple", "words", "of@@", "the"]
       UpperCAmelCase     :			Optional[Any]     	=tokenizer.decode(snake_case__    )

       self.assertTrue(output_string == '''herecouple words ofthe'''    )
 def 						UpperCAmelCase__							(	self    )						->				Tuple:
       '''simple docstring'''





       UpperCAmelCase     :			Optional[Any]     	=SpeechaTextaTokenizer.from_pretrained(self.tmpdirname    )

       with tempfile.TemporaryDirectory() as tmp_dirname:
             tokenizer.save_pretrained(snake_case__    )
             os.remove(os.path.join(snake_case__		, '''merges.txt'''    )    )

             # load tokenizer without merges file should not throw an error
             UpperCAmelCase     :			Any     	=SpeechaTextaTokenizer.from_pretrained(snake_case__    )

       with tempfile.TemporaryDirectory() as tmp_dirname:
             # save tokenizer and load again
             tokenizer.save_pretrained(snake_case__    )
             UpperCAmelCase     :			List[str]     	=SpeechaTextaTokenizer.from_pretrained(snake_case__    )

       self.assertIsNotNone(snake_case__    )






 def 						UpperCAmelCase__							(	self    )						->				int:
       '''simple docstring'''





       if not self.test_slow_tokenizer:
             return

       UpperCAmelCase     :			List[str]     	=inspect.signature(self.tokenizer_class.__init__    )
       UpperCAmelCase     :			Union[str, Any]     	=self.get_tokenizer()

       for parameter_name, parameter in signature.parameters.items():
             if parameter.default != inspect.Parameter.empty and parameter_name != "merges_file":
                   self.assertIn(snake_case__		, tokenizer.init_kwargs    )
