


import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


lowerCAmelCase__   	=				logging.get_logger(__name__)

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

lowerCAmelCase__   	=				{
    'vocab_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/vocab.json'},
    'merges_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/merges.txt'},
}

lowerCAmelCase__   	=				{
    'openai-gpt': 5_12,
}
def 							_UpperCAmelCase       (UpperCamelCase__  :						List[str]		):
       _A   : Union[str, Any]       					=				text.strip()
       if not text:
              return []
       _A   : Tuple       					=				text.split()
       return tokens







class   lowerCAmelCase__ (       a):




     '''simple docstring'''


     def __init__(		self				,		__lowerCamelCase=True				,		__lowerCamelCase=None				,		__lowerCamelCase=True				,		__lowerCamelCase=None				,		__lowerCamelCase=True				,		)    ->       int:
            if never_split is None:
                   _A   : List[Any]       					=				[]
            _A   : Union[str, Any]       					=				do_lower_case
            _A   : Tuple       					=				set(__lowerCamelCase)
            _A   : Optional[int]       					=				tokenize_chinese_chars
            _A   : int       					=				strip_accents
            _A   : Any       					=				do_split_on_punc


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase				,		__lowerCamelCase=None)    ->       Dict:
            _A   : str       					=				self.never_split.union(set(__lowerCamelCase)) if never_split else self.never_split
            _A   : int       					=				self._clean_text(__lowerCamelCase)

            # This was added on November 1st, 2018 for the multilingual and Chinese
            # models. This is also applied to the English models now, but it doesn't
            # matter since the English models were not trained on any Chinese data
            # and generally don't have any Chinese data in them (there are Chinese
            # characters in the vocabulary because Wikipedia does have some Chinese
            # words in the English Wikipedia.).
            if self.tokenize_chinese_chars:
                   _A   : int       					=				self._tokenize_chinese_chars(__lowerCamelCase)
            # prevents treating the same character with different unicode codepoints as different characters
            _A   : Optional[int]       					=				unicodedata.normalize("NFC"				,		__lowerCamelCase)
            _A   : Optional[int]       					=				whitespace_tokenize(__lowerCamelCase)
            _A   : List[Any]       					=				[]
            for token in orig_tokens:
                   if token not in never_split:
                          if self.do_lower_case:
                                 _A   : str       					=				token.lower()
                                 if self.strip_accents is not False:
                                        _A   : List[str]       					=				self._run_strip_accents(__lowerCamelCase)
                          elif self.strip_accents:
                                 _A   : Any       					=				self._run_strip_accents(__lowerCamelCase)
                   split_tokens.extend(self._run_split_on_punc(__lowerCamelCase				,		__lowerCamelCase))

            _A   : List[Any]       					=				whitespace_tokenize(" ".join(__lowerCamelCase))
            return output_tokens


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       List[Any]:
            _A   : List[Any]       					=				unicodedata.normalize("NFD"				,		__lowerCamelCase)
            _A   : Dict       					=				[]
            for char in text:
                   _A   : Optional[int]       					=				unicodedata.category(__lowerCamelCase)
                   if cat == "Mn":
                          continue
                   output.append(__lowerCamelCase)
            return "".join(__lowerCamelCase)


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase				,		__lowerCamelCase=None)    ->       List[str]:
            if not self.do_split_on_punc or (never_split is not None and text in never_split):
                   return [text]
            _A   : List[Any]       					=				list(__lowerCamelCase)
            _A   : List[str]       					=				0
            _A   : List[Any]       					=				True
            _A   : Any       					=				[]
            while i < len(__lowerCamelCase):
                   _A   : Union[str, Any]       					=				chars[i]
                   if _is_punctuation(__lowerCamelCase):
                          output.append([char])
                          _A   : List[Any]       					=				True
                   else:
                          if start_new_word:
                                 output.append([])
                          _A   : Union[str, Any]       					=				False
                          output[-1].append(__lowerCamelCase)
                   i += 1

            return ["".join(__lowerCamelCase) for x in output]


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       List[Any]:
            _A   : Tuple       					=				[]
            for char in text:
                   _A   : List[str]       					=				ord(__lowerCamelCase)
                   if self._is_chinese_char(__lowerCamelCase):
                          output.append(" ")
                          output.append(__lowerCamelCase)
                          output.append(" ")
                   else:
                          output.append(__lowerCamelCase)
            return "".join(__lowerCamelCase)


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       List[str]:
            # This defines a "chinese character" as anything in the CJK Unicode block:
            #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
            #
            # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
            # despite its name. The modern Korean Hangul alphabet is a different block,
            # as is Japanese Hiragana and Katakana. Those alphabets are used to write
            # space-separated words, so they are not treated specially and handled
            # like the all of the other languages.
            if (
                (cp >= 0x4_e00 and cp <= 0x9_fff)
                or (cp >= 0x3_400 and cp <= 0x4_dbf)  #
                or (cp >= 0x20_000 and cp <= 0x2a_6df)  #
                or (cp >= 0x2a_700 and cp <= 0x2b_73f)  #
                or (cp >= 0x2b_740 and cp <= 0x2b_81f)  #
                or (cp >= 0x2b_820 and cp <= 0x2c_eaf)  #
                or (cp >= 0xf_900 and cp <= 0xf_aff)
                or (cp >= 0x2f_800 and cp <= 0x2f_a1f)  #
            ):  #
                   return True

            return False


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       int:
            _A   : Optional[int]       					=				[]
            for char in text:
                   _A   : List[str]       					=				ord(__lowerCamelCase)
                   if cp == 0 or cp == 0xf_ffd or _is_control(__lowerCamelCase):
                          continue
                   if _is_whitespace(__lowerCamelCase):
                          output.append(" ")
                   else:
                          output.append(__lowerCamelCase)
            return "".join(__lowerCamelCase)




def 							_UpperCAmelCase       (UpperCamelCase__  :						Any		):
       _A   : Tuple       					=				set()
       _A   : List[str]       					=				word[0]
       for char in word[1:]:
              pairs.add((prev_char, char)		)
              _A   : Dict       					=				char
       return pairs
def 							_UpperCAmelCase       (UpperCamelCase__  :						int		):
       _A   : Optional[Any]       					=				text.replace("—"       ,							"-"		)
       _A   : Optional[int]       					=				text.replace("–"       ,							"-"		)
       _A   : Tuple       					=				text.replace("―"       ,							"-"		)
       _A   : List[str]       					=				text.replace("…"       ,							"..."		)
       _A   : Union[str, Any]       					=				text.replace("´"       ,							"'"		)
       _A   : Any       					=				re.sub(r"(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)"       ,							r" \1 "       ,							UpperCamelCase__		)
       _A   : Optional[Any]       					=				re.sub(r"\s*\n\s*"       ,							" \n "       ,							UpperCamelCase__		)
       _A   : List[str]       					=				re.sub(r"[^\S\n]+"       ,							" "       ,							UpperCamelCase__		)
       return text.strip()







class   lowerCAmelCase__ (       a):




     '''simple docstring'''
     __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				,		__lowerCamelCase				,		__lowerCamelCase				,		__lowerCamelCase="<unk>"				,		**__lowerCamelCase)    ->       Dict:
            super().__init__(unk_token=__lowerCamelCase				,		**__lowerCamelCase)

            try:
                   import ftfy
                   from spacy.lang.en import English

                   _A   : int       					=				English()
                   _A   : Dict       					=				_nlp.tokenizer
                   _A   : Dict       					=				ftfy.fix_text
            except ImportError:
                   logger.warning("ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.")
                   _A   : Optional[int]       					=				BasicTokenizer(do_lower_case=__lowerCamelCase)
                   _A   : str       					=				None

            with open(__lowerCamelCase				,		encoding="utf-8") as vocab_handle:
                   _A   : Tuple       					=				json.load(__lowerCamelCase)
            _A   : Any       					=				{v: k for k, v in self.encoder.items()}
            with open(__lowerCamelCase				,		encoding="utf-8") as merges_handle:
                   _A   : int       					=				merges_handle.read().split("\n")[1:-1]
            _A   : Tuple       					=				[tuple(merge.split()) for merge in merges]
            _A   : str       					=				dict(zip(__lowerCamelCase				,		range(len(__lowerCamelCase))))
            _A   : List[Any]       					=				{}


     @property
     def 				_lowerCamelCase				(		self)    ->       str:
            return True


     @property
     def 				_lowerCamelCase				(		self)    ->       str:
            return len(self.encoder)


     def 				_lowerCamelCase				(		self)    ->       Union[str, Any]:
            return dict(self.encoder				,		**self.added_tokens_encoder)


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Tuple:
            _A   : int       					=				tuple(token[:-1]) + (token[-1] + "</w>",)
            if token in self.cache:
                   return self.cache[token]
            _A   : List[str]       					=				get_pairs(__lowerCamelCase)

            if not pairs:
                   return token + "</w>"

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

                          if word[i] == first and i < len(__lowerCamelCase) - 1 and word[i + 1] == second:
                                 new_word.append(first + second)
                                 i += 2
                          else:
                                 new_word.append(word[i])
                                 i += 1
                   _A   : Optional[int]       					=				tuple(__lowerCamelCase)
                   _A   : List[Any]       					=				new_word
                   if len(__lowerCamelCase) == 1:
                          break
                   else:
                          _A   : Dict       					=				get_pairs(__lowerCamelCase)
            _A   : str       					=				" ".join(__lowerCamelCase)
            if word == "\n  </w>":
                   _A   : int       					=				"\n</w>"
            _A   : Dict       					=				word
            return word


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Any:
            _A   : Tuple       					=				[]
            if self.fix_text is None:
                   # Using BERT's BasicTokenizer
                   _A   : Union[str, Any]       					=				self.nlp.tokenize(__lowerCamelCase)
                   for token in text:
                          split_tokens.extend(list(self.bpe(__lowerCamelCase).split(" ")))
            else:
                   # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
                   _A   : Union[str, Any]       					=				self.nlp(text_standardize(self.fix_text(__lowerCamelCase)))
                   for token in text:
                          split_tokens.extend(list(self.bpe(token.text.lower()).split(" ")))
            return split_tokens


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       str:
            return self.encoder.get(__lowerCamelCase				,		self.encoder.get(self.unk_token))


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Tuple:
            return self.decoder.get(__lowerCamelCase				,		self.unk_token)


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase)    ->       Union[str, Any]:
            _A   : Dict       					=				"".join(__lowerCamelCase).replace("</w>"				,		" ").strip()
            return out_string


     def 				_lowerCamelCase				(		self				,		__lowerCamelCase				,		__lowerCamelCase = None)    ->       Tuple[str]:
            if not os.path.isdir(__lowerCamelCase):
                   logger.error(F"Vocabulary path ({save_directory}) should be a directory")
                   return
            _A   : Any       					=				os.path.join(
                __lowerCamelCase				,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"])
            _A   : str       					=				os.path.join(
                __lowerCamelCase				,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"])

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

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

            return vocab_file, merge_file




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 MobileNetVaImageProcessor







class   lowerCAmelCase__ (       unittest.TestCase):




     '''simple docstring'''


     def __init__(		self				,		__lowerCamelCase				,		__lowerCamelCase=7				,		__lowerCamelCase=3				,		__lowerCamelCase=1_8				,		__lowerCamelCase=3_0				,		__lowerCamelCase=4_0_0				,		__lowerCamelCase=True				,		__lowerCamelCase=None				,		__lowerCamelCase=True				,		__lowerCamelCase=None				,		)    ->       Tuple:
            _A   : int       					=				size if size is not None else {"shortest_edge": 2_0}
            _A   : Tuple       					=				crop_size if crop_size is not None else {"height": 1_8, "width": 1_8}
            _A   : str       					=				parent
            _A   : Any       					=				batch_size
            _A   : Dict       					=				num_channels
            _A   : int       					=				image_size
            _A   : Union[str, Any]       					=				min_resolution
            _A   : Tuple       					=				max_resolution
            _A   : Any       					=				do_resize
            _A   : Optional[int]       					=				size
            _A   : Tuple       					=				do_center_crop
            _A   : List[str]       					=				crop_size


     def 				_lowerCamelCase				(		self)    ->       str:
            return {
                "do_resize": self.do_resize,
                "size": self.size,
                "do_center_crop": self.do_center_crop,
                "crop_size": self.crop_size,
            }







@require_torch
@require_vision
class   lowerCAmelCase__ (       a  ,					unittest.TestCase):




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE						    =				MobileNetVaImageProcessor if is_vision_available() else None


     def 				_lowerCamelCase				(		self)    ->       Any:
            _A   : Dict       					=				MobileNetVaImageProcessingTester(self)


     @property
     def 				_lowerCamelCase				(		self)    ->       Dict:
            return self.image_processor_tester.prepare_image_processor_dict()


     def 				_lowerCamelCase				(		self)    ->       Tuple:
            _A   : List[Any]       					=				self.image_processing_class(**self.image_processor_dict)
            self.assertTrue(hasattr(__lowerCamelCase				,		"do_resize"))
            self.assertTrue(hasattr(__lowerCamelCase				,		"size"))
            self.assertTrue(hasattr(__lowerCamelCase				,		"do_center_crop"))
            self.assertTrue(hasattr(__lowerCamelCase				,		"center_crop"))


     def 				_lowerCamelCase				(		self)    ->       Union[str, Any]:
            _A   : int       					=				self.image_processing_class.from_dict(self.image_processor_dict)
            self.assertEqual(image_processor.size				,		{"shortest_edge": 2_0})
            self.assertEqual(image_processor.crop_size				,		{"height": 1_8, "width": 1_8})

            _A   : Any       					=				self.image_processing_class.from_dict(self.image_processor_dict				,		size=4_2				,		crop_size=8_4)
            self.assertEqual(image_processor.size				,		{"shortest_edge": 4_2})
            self.assertEqual(image_processor.crop_size				,		{"height": 8_4, "width": 8_4})


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


     def 				_lowerCamelCase				(		self)    ->       List[Any]:
            # Initialize image_processing
            _A   : Optional[int]       					=				self.image_processing_class(**self.image_processor_dict)
            # create random PIL images
            _A   : List[Any]       					=				prepare_image_inputs(self.image_processor_tester				,		equal_resolution=__lowerCamelCase)
            for image in image_inputs:
                   self.assertIsInstance(__lowerCamelCase				,		Image.Image)

            # Test not batched input
            _A   : 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.crop_size["height"],
                    self.image_processor_tester.crop_size["width"],
                )				,		)

            # Test batched
            _A   : Dict       					=				image_processing(__lowerCamelCase				,		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.crop_size["height"],
                    self.image_processor_tester.crop_size["width"],
                )				,		)


     def 				_lowerCamelCase				(		self)    ->       str:
            # Initialize image_processing
            _A   : List[Any]       					=				self.image_processing_class(**self.image_processor_dict)
            # create random numpy tensors
            _A   : int       					=				prepare_image_inputs(self.image_processor_tester				,		equal_resolution=__lowerCamelCase				,		numpify=__lowerCamelCase)
            for image in image_inputs:
                   self.assertIsInstance(__lowerCamelCase				,		np.ndarray)

            # Test not batched input
            _A   : Tuple       					=				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.crop_size["height"],
                    self.image_processor_tester.crop_size["width"],
                )				,		)

            # Test batched
            _A   : str       					=				image_processing(__lowerCamelCase				,		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.crop_size["height"],
                    self.image_processor_tester.crop_size["width"],
                )				,		)


     def 				_lowerCamelCase				(		self)    ->       Tuple:
            # Initialize image_processing
            _A   : str       					=				self.image_processing_class(**self.image_processor_dict)
            # create random PyTorch tensors
            _A   : List[str]       					=				prepare_image_inputs(self.image_processor_tester				,		equal_resolution=__lowerCamelCase				,		torchify=__lowerCamelCase)
            for image in image_inputs:
                   self.assertIsInstance(__lowerCamelCase				,		torch.Tensor)

            # Test not batched input
            _A   : 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.crop_size["height"],
                    self.image_processor_tester.crop_size["width"],
                )				,		)

            # Test batched
            _A   : Any       					=				image_processing(__lowerCamelCase				,		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.crop_size["height"],
                    self.image_processor_tester.crop_size["width"],
                )				,		)
