"""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








# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
      import tensorflow as tf






class     a_       (			TensorFormatter[Mapping, 'tf.Tensor', Mapping]							):



       """simple docstring"""






       def __init__(		self				,       _lowerCamelCase=None				,       **_lowerCamelCase   )					->int:
           super().__init__(features=_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				Any				      = tf_tensor_kwargs
           import tensorflow as tf  # noqa: import tf at initialization






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase   )					->Tuple:
           import tensorflow as tf

           if isinstance(_lowerCamelCase				,       _lowerCamelCase   ) and column:
               if all(
                   isinstance(_lowerCamelCase				,       tf.Tensor   ) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column   ):
                   return tf.stack(_lowerCamelCase   )
               elif all(
                   isinstance(_lowerCamelCase				,       (tf.Tensor, tf.RaggedTensor)   ) and x.ndim == 1 and x.dtype == column[0].dtype
                   for x in column   ):
                   # only rag 1-D tensors, otherwise some dimensions become ragged even though they were consolidated
                   return tf.ragged.stack(_lowerCamelCase   )

           return column






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase   )					->Union[str, Any]:
           import tensorflow as tf

           if value is None:
               return value

           SCREAMING_SNAKE_CASE				:				int				      = {}

           if isinstance(_lowerCamelCase				,       (np.number, np.ndarray)   ) and np.issubdtype(value.dtype				,       np.integer   ):
               SCREAMING_SNAKE_CASE				:				List[Any]				      = {'''dtype''': tf.intaa}
           elif isinstance(_lowerCamelCase				,       (np.number, np.ndarray)   ) and np.issubdtype(value.dtype				,       np.floating   ):
               SCREAMING_SNAKE_CASE				:				Optional[int]				      = {'''dtype''': tf.floataa}
           elif config.PIL_AVAILABLE and "PIL" in sys.modules:
               import PIL.Image

               if isinstance(_lowerCamelCase				,       PIL.Image.Image   ):
                   SCREAMING_SNAKE_CASE				:				Any				      = np.asarray(_lowerCamelCase   )

           return tf.convert_to_tensor(_lowerCamelCase				,       **{**default_dtype, **self.tf_tensor_kwargs}   )






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase   )					->Tuple:
           import tensorflow as tf

           # support for torch, tf, jax etc.
           if config.TORCH_AVAILABLE and "torch" in sys.modules:
               import torch

               if isinstance(_lowerCamelCase				,       torch.Tensor   ):
                   return self._tensorize(data_struct.detach().cpu().numpy()[()]   )
           if hasattr(_lowerCamelCase				,       '''__array__'''   ) and not isinstance(_lowerCamelCase				,       tf.Tensor   ):
               SCREAMING_SNAKE_CASE				:				int				      = data_struct.__array__()
           # support for nested types like struct of list of struct
           if isinstance(_lowerCamelCase				,       np.ndarray   ):
               if data_struct.dtype == object:  # tf tensors cannot be instantied from an array of objects
                   return self._consolidate([self.recursive_tensorize(_lowerCamelCase   ) for substruct in data_struct]   )
           elif isinstance(_lowerCamelCase				,       (list, tuple)   ):
               return self._consolidate([self.recursive_tensorize(_lowerCamelCase   ) for substruct in data_struct]   )
           return self._tensorize(_lowerCamelCase   )






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase   )					->Optional[int]:
           return map_nested(self._recursive_tensorize				,       _lowerCamelCase				,       map_list=_lowerCamelCase   )






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase   )					->Mapping:
           SCREAMING_SNAKE_CASE				:				str				      = self.numpy_arrow_extractor().extract_row(_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				List[str]				      = self.python_features_decoder.decode_row(_lowerCamelCase   )
           return self.recursive_tensorize(_lowerCamelCase   )






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase   )					->"tf.Tensor":
           SCREAMING_SNAKE_CASE				:				Optional[int]				      = self.numpy_arrow_extractor().extract_column(_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				str				      = self.python_features_decoder.decode_column(_lowerCamelCase				,       pa_table.column_names[0]   )
           SCREAMING_SNAKE_CASE				:				str				      = self.recursive_tensorize(_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				int				      = self._consolidate(_lowerCamelCase   )
           return column






       def 	__lowerCAmelCase     (		self				,       _lowerCamelCase   )					->Mapping:
           SCREAMING_SNAKE_CASE				:				Tuple				      = self.numpy_arrow_extractor().extract_batch(_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				Dict				      = self.python_features_decoder.decode_batch(_lowerCamelCase   )
           SCREAMING_SNAKE_CASE				:				int				      = self.recursive_tensorize(_lowerCamelCase   )
           for column_name in batch:
               SCREAMING_SNAKE_CASE				:				Optional[Any]				      = self._consolidate(batch[column_name]   )
           return batch
