

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

__UpperCAmelCase     					=  logging.get_logger(__name__)

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

__UpperCAmelCase     					=  {
    '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',
    },
}

__UpperCAmelCase     					=  {
    '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 						_snake_case		(     )							->   Union[str, Any]:

     """simple docstring"""







     _lowerCAmelCase :       List[str]	   =     (
         list(range(ord("!"					)		,				ord("~"					) + 1					)					) + list(range(ord("¡"					)		,				ord("¬"					) + 1					)					) + list(range(ord("®"					)		,				ord("ÿ"					) + 1					)					)
     )
     _lowerCAmelCase :       Any	   =     bs[:]
     _lowerCAmelCase :       Union[str, Any]	   =     0
     for b in range(2**8					):
          if b not in bs:
               bs.append(lowerCamelCase_					)
               cs.append(2**8 + n					)
               n += 1
     _lowerCAmelCase :       Union[str, Any]	   =     [chr(lowerCamelCase_					) for n in cs]
     return dict(zip(lowerCamelCase_		,				lowerCamelCase_					)					)







def 						_snake_case		(     SCREAMING_SNAKE_CASE					)							->   str:

     """simple docstring"""







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





class A__     (			_lowercase			):



 """simple docstring"""







 _lowercase  :       List[str]  			=    VOCAB_FILES_NAMES
 _lowercase  :       List[str]  			=    PRETRAINED_VOCAB_FILES_MAP
 _lowercase  :       Any  			=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowercase  :       Any  			=    ['''input_ids''', '''attention_mask''']

 def __init__(		self			:						Any  ,    A_			:						int  ,    A_			:						Dict  ,    A_			:						Dict="replace"  ,    A_			:						List[Any]="<|endoftext|>"  ,    A_			:						List[Any]="<|endoftext|>"  ,    A_			:						Any="<|endoftext|>"  ,    A_			:						List[str]=None  ,    A_			:						str=False  ,    A_			:						Tuple=False  ,    **A_			:						List[str]  ,    ):






      '''simple docstring'''




      _lowerCAmelCase :       Union[str, Any]	   =     AddedToken(snake_case__  ,    lstrip=snake_case__  ,    rstrip=snake_case__							) if isinstance(snake_case__  ,    snake_case__							) else bos_token
      _lowerCAmelCase :       Union[str, Any]	   =     AddedToken(snake_case__  ,    lstrip=snake_case__  ,    rstrip=snake_case__							) if isinstance(snake_case__  ,    snake_case__							) else eos_token
      _lowerCAmelCase :       str	   =     AddedToken(snake_case__  ,    lstrip=snake_case__  ,    rstrip=snake_case__							) if isinstance(snake_case__  ,    snake_case__							) else unk_token
      _lowerCAmelCase :       Optional[Any]	   =     AddedToken(snake_case__  ,    lstrip=snake_case__  ,    rstrip=snake_case__							) if isinstance(snake_case__  ,    snake_case__							) else pad_token
      super().__init__(
          errors=snake_case__  ,    unk_token=snake_case__  ,    bos_token=snake_case__  ,    eos_token=snake_case__  ,    pad_token=snake_case__  ,    add_prefix_space=snake_case__  ,    add_bos_token=snake_case__  ,    **snake_case__  ,    )
      _lowerCAmelCase :       int	   =     add_bos_token

      with open(snake_case__  ,    encoding="utf-8"							) as vocab_handle:
           _lowerCAmelCase :       Optional[int]	   =     json.load(snake_case__							)
      _lowerCAmelCase :       Dict	   =     {v: k for k, v in self.encoder.items()}
      _lowerCAmelCase :       List[Any]	   =     errors  # how to handle errors in decoding
      _lowerCAmelCase :       Optional[Any]	   =     bytes_to_unicode()
      _lowerCAmelCase :       Tuple	   =     {v: k for k, v in self.byte_encoder.items()}
      with open(snake_case__  ,    encoding="utf-8"							) as merges_handle:
           _lowerCAmelCase :       List[Any]	   =     merges_handle.read().split("\n"							)[1:-1]
      _lowerCAmelCase :       int	   =     [tuple(merge.split()							) for merge in bpe_merges]
      _lowerCAmelCase :       List[str]	   =     dict(zip(snake_case__  ,    range(len(snake_case__							)							)							)							)
      _lowerCAmelCase :       List[str]	   =     {}
      _lowerCAmelCase :       List[str]	   =     add_prefix_space

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

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






      '''simple docstring'''




      return len(self.encoder							)

 def  __magic_name__							(		self			:						Tuple							):






      '''simple docstring'''




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

 def  __magic_name__							(		self			:						Optional[int]  ,    A_			:						Optional[Any]							):






      '''simple docstring'''




      if token in self.cache:
           return self.cache[token]
      _lowerCAmelCase :       Union[str, Any]	   =     tuple(snake_case__							)
      _lowerCAmelCase :       Optional[Any]	   =     get_pairs(snake_case__							)

      if not pairs:
           return token

      while True:
           _lowerCAmelCase :       str	   =     min(snake_case__  ,    key=lambda A_							: self.bpe_ranks.get(snake_case__  ,    float("inf"							)							)							)
           if bigram not in self.bpe_ranks:
                break
           _lowerCAmelCase :       List[str]	   =     bigram
           _lowerCAmelCase :       int	   =     []
           _lowerCAmelCase :       Optional[int]	   =     0
           while i < len(snake_case__							):
                try:
                     _lowerCAmelCase :       Union[str, Any]	   =     word.index(snake_case__  ,    snake_case__							)
                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(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	   =     tuple(snake_case__							)
           _lowerCAmelCase :       List[Any]	   =     new_word
           if len(snake_case__							) == 1:
                break
           else:
                _lowerCAmelCase :       Optional[int]	   =     get_pairs(snake_case__							)
      _lowerCAmelCase :       List[Any]	   =     """ """.join(snake_case__							)
      _lowerCAmelCase :       Any	   =     word
      return word

 def  __magic_name__							(		self			:						List[Any]  ,    A_			:						int  ,    A_			:						Tuple=None							):






      '''simple docstring'''




      if self.add_bos_token:
           _lowerCAmelCase :       str	   =     [self.bos_token_id]
      else:
           _lowerCAmelCase :       List[Any]	   =     []

      _lowerCAmelCase :       Tuple	   =     bos_token_ids + token_ids_a

      if token_ids_a is None:
           return output

      return output + bos_token_ids + token_ids_a

 def  __magic_name__							(		self			:						int  ,    A_			:						List[int]  ,    A_			:						Optional[List[int]] = None  ,    A_			:						bool = False							):






      '''simple docstring'''




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

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

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

 def  __magic_name__							(		self			:						Any  ,    A_			:						Dict							):






      '''simple docstring'''




      _lowerCAmelCase :       Union[str, Any]	   =     []
      for token in re.findall(self.pat  ,    snake_case__							):
           _lowerCAmelCase :       Any	   =     """""".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(snake_case__							).split(" "							)							)
      return bpe_tokens

 def  __magic_name__							(		self			:						Tuple  ,    A_			:						Optional[int]							):






      '''simple docstring'''




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

 def  __magic_name__							(		self			:						Optional[Any]  ,    A_			:						List[Any]							):






      '''simple docstring'''




      return self.decoder.get(snake_case__							)

 def  __magic_name__							(		self			:						Tuple  ,    A_			:						List[str]							):






      '''simple docstring'''




      _lowerCAmelCase :       Any	   =     """""".join(snake_case__							)
      _lowerCAmelCase :       int	   =     bytearray([self.byte_decoder[c] for c in text]							).decode("utf-8"  ,    errors=self.errors							)
      return text

 def  __magic_name__							(		self			:						int  ,    A_			:						str  ,    A_			:						Optional[str] = None							):






      '''simple docstring'''




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

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

      _lowerCAmelCase :       Optional[int]	   =     0
      with open(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 A_							: 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 :       Dict	   =     token_index
                writer.write(" ".join(snake_case__							) + "\n"							)
                index += 1

      return vocab_file, merge_file

 def  __magic_name__							(		self			:						Dict  ,    A_			:						Tuple  ,    A_			:						Tuple=False  ,    **A_			:						Optional[Any]							):






      '''simple docstring'''




      _lowerCAmelCase :       Optional[int]	   =     kwargs.pop("add_prefix_space"  ,    self.add_prefix_space							)
      if is_split_into_words or add_prefix_space:
           _lowerCAmelCase :       Tuple	   =     """ """ + text
      return (text, kwargs)


 def  __magic_name__							(		self			:						Optional[int]  ,    A_			:						"Conversation"							):






      '''simple docstring'''




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








"""simple docstring"""
import os
import shutil
import unittest
from unittest.mock import patch

from transformers.testing_utils import CaptureStd, is_pt_tf_cross_test, require_torch




class  _snake_case	(    unittest.TestCase							):






							@patch("""sys.argv"""    ,			["""fakeprogrampath""", """env"""]     )
							def 				__lowercase	(       self    :      Tuple     ):






												"""simple docstring"""



												import transformers.commands.transformers_cli

												with CaptureStd() as cs:
																	transformers.commands.transformers_cli.main()
												self.assertIn("""Python version"""    ,			cs.out     )
												self.assertIn("""Platform"""    ,			cs.out     )
												self.assertIn("""Using distributed or parallel set-up in script?"""    ,			cs.out     )






							@is_pt_tf_cross_test
							@patch(
							    """sys.argv"""    ,			["""fakeprogrampath""", """pt-to-tf""", """--model-name""", """hf-internal-testing/tiny-random-gptj""", """--no-pr"""]     )
							def 				__lowercase	(       self    :      List[str]     ):






												"""simple docstring"""



												import transformers.commands.transformers_cli

												shutil.rmtree("""/tmp/hf-internal-testing/tiny-random-gptj"""    ,			ignore_errors=snake_case__     )  # cleans potential past runs
												transformers.commands.transformers_cli.main()

												# The original repo has no TF weights -- if they exist, they were created by the CLI
												self.assertTrue(os.path.exists("""/tmp/hf-internal-testing/tiny-random-gptj/tf_model.h5"""     )     )






							@require_torch
							@patch("""sys.argv"""    ,			["""fakeprogrampath""", """download""", """hf-internal-testing/tiny-random-gptj""", """--cache-dir""", """/tmp"""]     )
							def 				__lowercase	(       self    :      Optional[Any]     ):






												"""simple docstring"""



												import transformers.commands.transformers_cli

												# # remove any previously downloaded model to start clean
												shutil.rmtree("""/tmp/models--hf-internal-testing--tiny-random-gptj"""    ,			ignore_errors=snake_case__     )

												# run the command
												transformers.commands.transformers_cli.main()

												# check if the model files are downloaded correctly on /tmp/models--hf-internal-testing--tiny-random-gptj
												self.assertTrue(os.path.exists("""/tmp/models--hf-internal-testing--tiny-random-gptj/blobs"""     )     )
												self.assertTrue(os.path.exists("""/tmp/models--hf-internal-testing--tiny-random-gptj/refs"""     )     )
												self.assertTrue(os.path.exists("""/tmp/models--hf-internal-testing--tiny-random-gptj/snapshots"""     )     )






							@require_torch
							@patch(
							    """sys.argv"""    ,			[
							        """fakeprogrampath""",
							        """download""",
							        """hf-internal-testing/test_dynamic_model_with_tokenizer""",
							        """--trust-remote-code""",
							        """--cache-dir""",
							        """/tmp""",
							    ]    ,			)
							def 				__lowercase	(       self    :      Union[str, Any]     ):






												"""simple docstring"""



												import transformers.commands.transformers_cli

												# # remove any previously downloaded model to start clean
												shutil.rmtree("""/tmp/models--hf-internal-testing--test_dynamic_model_with_tokenizer"""    ,			ignore_errors=snake_case__     )

												# run the command
												transformers.commands.transformers_cli.main()

												# check if the model files are downloaded correctly on /tmp/models--hf-internal-testing--test_dynamic_model_with_tokenizer
												self.assertTrue(os.path.exists("""/tmp/models--hf-internal-testing--test_dynamic_model_with_tokenizer/blobs"""     )     )
												self.assertTrue(os.path.exists("""/tmp/models--hf-internal-testing--test_dynamic_model_with_tokenizer/refs"""     )     )
												self.assertTrue(
												    os.path.exists("""/tmp/models--hf-internal-testing--test_dynamic_model_with_tokenizer/snapshots"""     )     )
