





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

lowerCAmelCase_   =   logging.get_logger(__name__)

lowerCAmelCase_   =   {
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
}

lowerCAmelCase_   =   {
    "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",
    },
}

lowerCAmelCase_   =   {
    "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_			(  )			->						Tuple:
       _snake_case      :							Union[str, Any]     				=						(
           list(range(ord('''!'''  )      ,		ord('''~'''  ) + 1  )  ) + list(range(ord('''¡'''  )      ,		ord('''¬'''  ) + 1  )  ) + list(range(ord('''®'''  )      ,		ord('''ÿ'''  ) + 1  )  )
       )
       _snake_case      :							Any     				=						bs[:]
       _snake_case      :							Dict     				=						0
       for b in range(2**8  ):
              if b not in bs:
                     bs.append(lowercase_  )
                     cs.append(2**8 + n  )
                     n += 1
       _snake_case      :							Optional[int]     				=						[chr(lowercase_  ) for n in cs]
       return dict(zip(lowercase_      ,		lowercase_  )  )




def 						A_			(  lowercase_  )			->						Dict:
       _snake_case      :							Tuple     				=						set()
       _snake_case      :							int     				=						word[0]
       for char in word[1:]:
              pairs.add((prev_char, char)  )
              _snake_case      :							str     				=						char
       return pairs

class 				A    (__UpperCAmelCase				):
      _SCREAMING_SNAKE_CASE					=							VOCAB_FILES_NAMES
      _SCREAMING_SNAKE_CASE					=							PRETRAINED_VOCAB_FILES_MAP
      _SCREAMING_SNAKE_CASE					=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _SCREAMING_SNAKE_CASE					=							["""input_ids""", """attention_mask"""]



      def __init__(					self     ,  lowercase_     ,  lowercase_     ,  lowercase_="replace"     ,  lowercase_="<|endoftext|>"     ,  lowercase_="<|endoftext|>"     ,  lowercase_="<|endoftext|>"     ,  lowercase_=None     ,  lowercase_=False     ,  lowercase_=False     ,  **lowercase_     ,  )      ->   Optional[Any]:



             '''simple docstring'''

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

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

             # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
             _snake_case      :							Union[str, 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    __a (					self					)      ->   Optional[Any]:



             '''simple docstring'''

             return len(self.encoder					)



      def    __a (					self					)      ->   Dict:



             '''simple docstring'''

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



      def    __a (					self     ,  lowercase_					)      ->   Union[str, Any]:



             '''simple docstring'''

             if token in self.cache:
                    return self.cache[token]
             _snake_case      :							List[str]     				=						tuple(lowercase_					)
             _snake_case      :							Union[str, Any]     				=						get_pairs(lowercase_					)

             if not pairs:
                    return token

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

                           if word[i] == first and i < len(lowercase_					) - 1 and word[i + 1] == second:
                                  new_word.append(first + second					)
                                  i += 2
                           else:
                                  new_word.append(word[i]					)
                                  i += 1
                    _snake_case      :							Any     				=						tuple(lowercase_					)
                    _snake_case      :							Optional[int]     				=						new_word
                    if len(lowercase_					) == 1:
                           break
                    else:
                           _snake_case      :							Dict     				=						get_pairs(lowercase_					)
             _snake_case      :							int     				=						''' '''.join(lowercase_					)
             _snake_case      :							str     				=						word
             return word



      def    __a (					self     ,  lowercase_     ,  lowercase_=None					)      ->   int:



             '''simple docstring'''

             if self.add_bos_token:
                    _snake_case      :							Dict     				=						[self.bos_token_id]
             else:
                    _snake_case      :							Optional[Any]     				=						[]

             _snake_case      :							Dict     				=						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     ,  lowercase_     ,  lowercase_ = None     ,  lowercase_ = False					)      ->   List[int]:



             '''simple docstring'''

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

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

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



      def    __a (					self     ,  lowercase_					)      ->   int:



             '''simple docstring'''

             _snake_case      :							List[str]     				=						[]
             for token in re.findall(self.pat     ,  lowercase_					):
                    _snake_case      :							List[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(lowercase_					).split(''' '''					)					)
             return bpe_tokens



      def    __a (					self     ,  lowercase_					)      ->   Tuple:



             '''simple docstring'''

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



      def    __a (					self     ,  lowercase_					)      ->   Optional[int]:



             '''simple docstring'''

             return self.decoder.get(lowercase_					)



      def    __a (					self     ,  lowercase_					)      ->   List[str]:



             '''simple docstring'''

             _snake_case      :							int     				=						''''''.join(lowercase_					)
             _snake_case      :							List[str]     				=						bytearray([self.byte_decoder[c] for c in text]					).decode('''utf-8'''     ,  errors=self.errors					)
             return text



      def    __a (					self     ,  lowercase_     ,  lowercase_ = None					)      ->   Tuple[str]:



             '''simple docstring'''

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

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

             _snake_case      :							Optional[int]     				=						0
             with open(lowercase_     ,  '''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 lowercase_					: 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!'''					)
                                  _snake_case      :							Optional[Any]     				=						token_index
                           writer.write(''' '''.join(lowercase_					) + '''\n'''					)
                           index += 1

             return vocab_file, merge_file



      def    __a (					self     ,  lowercase_     ,  lowercase_=False     ,  **lowercase_					)      ->   Union[str, Any]:



             '''simple docstring'''

             _snake_case      :							List[str]     				=						kwargs.pop('''add_prefix_space'''     ,  self.add_prefix_space					)
             if is_split_into_words or add_prefix_space:
                    _snake_case      :							int     				=						''' ''' + text
             return (text, kwargs)



      def    __a (					self     ,  lowercase_					)      ->   List[int]:



             '''simple docstring'''

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













import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
   import torch

if is_vision_available():
   from PIL import Image

   from transformers import ConvNextImageProcessor

class 				A    (unittest.TestCase				):



      def __init__(					self     ,  lowercase_     ,  lowercase_=7     ,  lowercase_=3     ,  lowercase_=18     ,  lowercase_=30     ,  lowercase_=400     ,  lowercase_=True     ,  lowercase_=None     ,  lowercase_=0.875     ,  lowercase_=True     ,  lowercase_=[0.5, 0.5, 0.5]     ,  lowercase_=[0.5, 0.5, 0.5]     ,  )      ->   List[str]:



             '''simple docstring'''

             _snake_case      :							Tuple     				=						size if size is not None else {'''shortest_edge''': 20}
             _snake_case      :							List[Any]     				=						parent
             _snake_case      :							Optional[int]     				=						batch_size
             _snake_case      :							str     				=						num_channels
             _snake_case      :							Union[str, Any]     				=						image_size
             _snake_case      :							Optional[Any]     				=						min_resolution
             _snake_case      :							Union[str, Any]     				=						max_resolution
             _snake_case      :							Tuple     				=						do_resize
             _snake_case      :							List[Any]     				=						size
             _snake_case      :							Optional[int]     				=						crop_pct
             _snake_case      :							Any     				=						do_normalize
             _snake_case      :							Optional[int]     				=						image_mean
             _snake_case      :							Union[str, Any]     				=						image_std



      def    __a (					self					)      ->   str:



             '''simple docstring'''

             return {
                 "image_mean": self.image_mean,
                 "image_std": self.image_std,
                 "do_normalize": self.do_normalize,
                 "do_resize": self.do_resize,
                 "size": self.size,
                 "crop_pct": self.crop_pct,
             }


@require_torch
@require_vision
class 				A    (__UpperCAmelCase       ,unittest.TestCase				):
      _SCREAMING_SNAKE_CASE					=							ConvNextImageProcessor if is_vision_available() else None



      def    __a (					self					)      ->   Dict:



             '''simple docstring'''

             _snake_case      :							List[Any]     				=						ConvNextImageProcessingTester(self					)



      @property
      def    __a (					self					)      ->   Dict:



             '''simple docstring'''

             return self.image_processor_tester.prepare_image_processor_dict()



      def    __a (					self					)      ->   str:



             '''simple docstring'''

             _snake_case      :							List[Any]     				=						self.image_processing_class(**self.image_processor_dict					)
             self.assertTrue(hasattr(lowercase_     ,  '''do_resize'''					)					)
             self.assertTrue(hasattr(lowercase_     ,  '''size'''					)					)
             self.assertTrue(hasattr(lowercase_     ,  '''crop_pct'''					)					)
             self.assertTrue(hasattr(lowercase_     ,  '''do_normalize'''					)					)
             self.assertTrue(hasattr(lowercase_     ,  '''image_mean'''					)					)
             self.assertTrue(hasattr(lowercase_     ,  '''image_std'''					)					)



      def    __a (					self					)      ->   List[Any]:



             '''simple docstring'''

             _snake_case      :							int     				=						self.image_processing_class.from_dict(self.image_processor_dict					)
             self.assertEqual(image_processor.size     ,  {'''shortest_edge''': 20}					)

             _snake_case      :							Tuple     				=						self.image_processing_class.from_dict(self.image_processor_dict     ,  size=42					)
             self.assertEqual(image_processor.size     ,  {'''shortest_edge''': 42}					)



      def    __a (					self					)      ->   List[Any]:



             '''simple docstring'''

             pass



      def    __a (					self					)      ->   int:



             '''simple docstring'''

             _snake_case      :							Tuple     				=						self.image_processing_class(**self.image_processor_dict					)
             # create random PIL images
             _snake_case      :							List[str]     				=						prepare_image_inputs(self.image_processor_tester     ,  equal_resolution=lowercase_					)
             for image in image_inputs:
                    self.assertIsInstance(lowercase_     ,  Image.Image					)

             # Test not batched input
             _snake_case      :							List[str]     				=						image_processing(image_inputs[0]     ,  return_tensors='''pt'''					).pixel_values
             self.assertEqual(
                 encoded_images.shape     ,  (
                     1,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.size['''shortest_edge'''],
                     self.image_processor_tester.size['''shortest_edge'''],
                 )     ,  )

             # Test batched
             _snake_case      :							int     				=						image_processing(lowercase_     ,  return_tensors='''pt'''					).pixel_values
             self.assertEqual(
                 encoded_images.shape     ,  (
                     self.image_processor_tester.batch_size,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.size['''shortest_edge'''],
                     self.image_processor_tester.size['''shortest_edge'''],
                 )     ,  )



      def    __a (					self					)      ->   Dict:



             '''simple docstring'''

             _snake_case      :							Tuple     				=						self.image_processing_class(**self.image_processor_dict					)
             # create random numpy tensors
             _snake_case      :							Tuple     				=						prepare_image_inputs(self.image_processor_tester     ,  equal_resolution=lowercase_     ,  numpify=lowercase_					)
             for image in image_inputs:
                    self.assertIsInstance(lowercase_     ,  np.ndarray					)

             # Test not batched input
             _snake_case      :							List[str]     				=						image_processing(image_inputs[0]     ,  return_tensors='''pt'''					).pixel_values
             self.assertEqual(
                 encoded_images.shape     ,  (
                     1,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.size['''shortest_edge'''],
                     self.image_processor_tester.size['''shortest_edge'''],
                 )     ,  )

             # Test batched
             _snake_case      :							Union[str, Any]     				=						image_processing(lowercase_     ,  return_tensors='''pt'''					).pixel_values
             self.assertEqual(
                 encoded_images.shape     ,  (
                     self.image_processor_tester.batch_size,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.size['''shortest_edge'''],
                     self.image_processor_tester.size['''shortest_edge'''],
                 )     ,  )



      def    __a (					self					)      ->   Union[str, Any]:



             '''simple docstring'''

             _snake_case      :							Union[str, Any]     				=						self.image_processing_class(**self.image_processor_dict					)
             # create random PyTorch tensors
             _snake_case      :							Dict     				=						prepare_image_inputs(self.image_processor_tester     ,  equal_resolution=lowercase_     ,  torchify=lowercase_					)
             for image in image_inputs:
                    self.assertIsInstance(lowercase_     ,  torch.Tensor					)

             # Test not batched input
             _snake_case      :							Union[str, Any]     				=						image_processing(image_inputs[0]     ,  return_tensors='''pt'''					).pixel_values
             self.assertEqual(
                 encoded_images.shape     ,  (
                     1,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.size['''shortest_edge'''],
                     self.image_processor_tester.size['''shortest_edge'''],
                 )     ,  )

             # Test batched
             _snake_case      :							List[Any]     				=						image_processing(lowercase_     ,  return_tensors='''pt'''					).pixel_values
             self.assertEqual(
                 encoded_images.shape     ,  (
                     self.image_processor_tester.batch_size,
                     self.image_processor_tester.num_channels,
                     self.image_processor_tester.size['''shortest_edge'''],
                     self.image_processor_tester.size['''shortest_edge'''],
                 )     ,  )






