'''simple docstring'''

import gc
import os
import tempfile
import unittest

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, StableDiffusionPipeline, UNetaDConditionModel
from diffusers.loaders import AttnProcsLayers, LoraLoaderMixin
from diffusers.models.attention_processor import (
    Attention,
    AttnProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRAAttnProcessor2_0,
    LoRAXFormersAttnProcessor,
    XFormersAttnProcessor,
)
from diffusers.utils import TEXT_ENCODER_ATTN_MODULE, floats_tensor, torch_device



def  _lowerCAmelCase				(				__a      )  ->      Tuple:
     '''simple docstring'''







     _UpperCamelCase  :Optional[int]       ={}
     for name in unet.attn_processors.keys():
          _UpperCamelCase  :List[str]       =None if name.endswith("""attn1.processor"""      ) else unet.config.cross_attention_dim
          if name.startswith("""mid_block"""      ):
               _UpperCamelCase  :int       =unet.config.block_out_channels[-1]
          elif name.startswith("""up_blocks"""      ):
               _UpperCamelCase  :str       =int(name[len("""up_blocks."""      )]      )
               _UpperCamelCase  :List[str]       =list(reversed(unet.config.block_out_channels      )      )[block_id]
          elif name.startswith("""down_blocks"""      ):
               _UpperCamelCase  :Optional[Any]       =int(name[len("""down_blocks."""      )]      )
               _UpperCamelCase  :List[Any]       =unet.config.block_out_channels[block_id]
          _UpperCamelCase  :int       =(
              LoRAAttnProcessor2_0 if hasattr(lowerCAmelCase__			,      """scaled_dot_product_attention"""      ) else LoRAAttnProcessor
          )
          _UpperCamelCase  :str       =lora_attn_processor_class(
              hidden_size=lowerCAmelCase__			,      cross_attention_dim=lowerCAmelCase__      )
     _UpperCamelCase  :int       =AttnProcsLayers(lowerCAmelCase__      )
     return lora_attn_procs, unet_lora_layers



def  _lowerCAmelCase				(				__a      )  ->      int:
     '''simple docstring'''







     _UpperCamelCase  :List[str]       ={}
     _UpperCamelCase  :List[str]       =(
         LoRAAttnProcessor2_0 if hasattr(lowerCAmelCase__			,      """scaled_dot_product_attention"""      ) else LoRAAttnProcessor
     )
     for name, module in text_encoder.named_modules():
          if name.endswith(lowerCAmelCase__      ):
               _UpperCamelCase  :Optional[int]       =lora_attn_processor_class(
                   hidden_size=module.out_proj.out_features			,      cross_attention_dim=lowerCAmelCase__      )
     return text_lora_attn_procs



def  _lowerCAmelCase				(				__a      )  ->      str:
     '''simple docstring'''







     _UpperCamelCase  :str       =create_text_encoder_lora_attn_procs(lowerCAmelCase__      )
     _UpperCamelCase  :Optional[Any]       =AttnProcsLayers(lowerCAmelCase__      )
     return text_encoder_lora_layers



def  _lowerCAmelCase				(				__a			,      __a=False      )  ->      Dict:
     '''simple docstring'''







     for _, attn_proc in text_lora_attn_procs.items():
          # set up.weights
          for layer_name, layer_module in attn_proc.named_modules():
               if layer_name.endswith("""_lora"""      ):
                    _UpperCamelCase  :Tuple       =(
                        torch.randn_like(layer_module.up.weight      )
                        if randn_weight
                        else torch.zeros_like(layer_module.up.weight      )
                    )
                    _UpperCamelCase  :Optional[int]       =torch.nn.Parameter(lowerCAmelCase__      )




class    lowerCamelCase__  (   unittest.TestCase     ):



   def  _UpperCamelCase		(    self     )				->     str:
        """simple docstring"""





        torch.manual_seed(0     )
        _UpperCamelCase  :Optional[int]       =UNetaDConditionModel(
            block_out_channels=(32, 64)					,      layers_per_block=2					,      sample_size=32					,      in_channels=4					,      out_channels=4					,      down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					,      up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")					,      cross_attention_dim=32					,      )
        _UpperCamelCase  :List[str]       =DDIMScheduler(
            beta_start=0.0_0085					,      beta_end=0.012					,      beta_schedule="""scaled_linear"""					,      clip_sample=lowerCAmelCase__					,      set_alpha_to_one=lowerCAmelCase__					,      steps_offset=1					,      )
        torch.manual_seed(0     )
        _UpperCamelCase  :List[str]       =AutoencoderKL(
            block_out_channels=[32, 64]					,      in_channels=3					,      out_channels=3					,      down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]					,      up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]					,      latent_channels=4					,      )
        _UpperCamelCase  :List[Any]       =CLIPTextConfig(
            bos_token_id=0					,      eos_token_id=2					,      hidden_size=32					,      intermediate_size=37					,      layer_norm_eps=1e-05					,      num_attention_heads=4					,      num_hidden_layers=5					,      pad_token_id=1					,      vocab_size=1_000					,      )
        _UpperCamelCase  :List[Any]       =CLIPTextModel(lowerCAmelCase__     )
        _UpperCamelCase  :Union[str, Any]       =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     )

        _UpperCamelCase       ,      _UpperCamelCase  :Any       =create_unet_lora_layers(lowerCAmelCase__     )
        _UpperCamelCase  :Union[str, Any]       =create_text_encoder_lora_layers(lowerCAmelCase__     )

        _UpperCamelCase  :Optional[Any]       ={
            """unet""": unet,
            """scheduler""": scheduler,
            """vae""": vae,
            """text_encoder""": text_encoder,
            """tokenizer""": tokenizer,
            """safety_checker""": None,
            """feature_extractor""": None,
        }
        _UpperCamelCase  :int       ={
            """unet_lora_layers""": unet_lora_layers,
            """text_encoder_lora_layers""": text_encoder_lora_layers,
            """unet_lora_attn_procs""": unet_lora_attn_procs,
        }
        return pipeline_components, lora_components



   def  _UpperCamelCase		(    self     )				->     List[str]:
        """simple docstring"""





        _UpperCamelCase  :List[Any]       =1
        _UpperCamelCase  :Union[str, Any]       =10
        _UpperCamelCase  :Union[str, Any]       =4
        _UpperCamelCase  :Dict       =(32, 32)

        _UpperCamelCase  :Union[str, Any]       =torch.manual_seed(0     )
        _UpperCamelCase  :int       =floats_tensor((batch_size, num_channels) + sizes     )
        _UpperCamelCase  :Optional[int]       =torch.randint(1					,      lowerCAmelCase__					,      size=(batch_size, sequence_length)					,      generator=lowerCAmelCase__     )

        _UpperCamelCase  :Dict       ={
            """prompt""": """A painting of a squirrel eating a burger""",
            """generator""": generator,
            """num_inference_steps""": 2,
            """guidance_scale""": 6.0,
            """output_type""": """numpy""",
        }

        return noise, input_ids, pipeline_inputs

        # copied from: https://colab.research.google.com/gist/sayakpaul/df2ef6e1ae6d8c10a49d859883b10860/scratchpad.ipynb



   def  _UpperCamelCase		(    self     )				->     int:
        """simple docstring"""





        _UpperCamelCase  :str       =77

        _UpperCamelCase  :Tuple       =torch.randint(2					,      56					,      size=(1, max_seq_length)					,      generator=torch.manual_seed(0     )     )

        _UpperCamelCase  :Dict       ={}
        _UpperCamelCase  :Dict       =inputs
        return prepared_inputs



   def  _UpperCamelCase		(    self					,      lowerCAmelCase__     )				->     Any:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :Optional[Any]       =self.get_dummy_components()
        LoraLoaderMixin.save_lora_weights(
            save_directory=lowerCAmelCase__					,      unet_lora_layers=lora_components["""unet_lora_layers"""]					,      text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]					,      )
        self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase__					,      """pytorch_lora_weights.bin"""     )     )     )



   def  _UpperCamelCase		(    self     )				->     int:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :Dict       =self.get_dummy_components()
        _UpperCamelCase  :Dict       =StableDiffusionPipeline(**lowerCAmelCase__     )
        _UpperCamelCase  :Optional[int]       =sd_pipe.to(lowerCAmelCase__     )
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__     )

        _UpperCamelCase       ,      _UpperCamelCase       ,      _UpperCamelCase  :Optional[Any]       =self.get_dummy_inputs()

        _UpperCamelCase  :Optional[Any]       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :Tuple       =original_images[0, -3:, -3:, -1]

        with tempfile.TemporaryDirectory() as tmpdirname:
             LoraLoaderMixin.save_lora_weights(
                 save_directory=lowerCAmelCase__					,      unet_lora_layers=lora_components["""unet_lora_layers"""]					,      text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]					,      )
             self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase__					,      """pytorch_lora_weights.bin"""     )     )     )
             sd_pipe.load_lora_weights(lowerCAmelCase__     )

        _UpperCamelCase  :List[str]       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :int       =lora_images[0, -3:, -3:, -1]

        # Outputs shouldn't match.
        self.assertFalse(torch.allclose(torch.from_numpy(lowerCAmelCase__     )					,      torch.from_numpy(lowerCAmelCase__     )     )     )



   def  _UpperCamelCase		(    self     )				->     List[Any]:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :Any       =self.get_dummy_components()
        _UpperCamelCase  :Dict       =StableDiffusionPipeline(**lowerCAmelCase__     )
        _UpperCamelCase  :Union[str, Any]       =sd_pipe.to(lowerCAmelCase__     )
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__     )

        _UpperCamelCase       ,      _UpperCamelCase       ,      _UpperCamelCase  :List[str]       =self.get_dummy_inputs()

        _UpperCamelCase  :Optional[int]       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :int       =original_images[0, -3:, -3:, -1]

        with tempfile.TemporaryDirectory() as tmpdirname:
             LoraLoaderMixin.save_lora_weights(
                 save_directory=lowerCAmelCase__					,      unet_lora_layers=lora_components["""unet_lora_layers"""]					,      text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]					,      safe_serialization=lowerCAmelCase__					,      )
             self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase__					,      """pytorch_lora_weights.safetensors"""     )     )     )
             sd_pipe.load_lora_weights(lowerCAmelCase__     )

        _UpperCamelCase  :Dict       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :Dict       =lora_images[0, -3:, -3:, -1]

        # Outputs shouldn't match.
        self.assertFalse(torch.allclose(torch.from_numpy(lowerCAmelCase__     )					,      torch.from_numpy(lowerCAmelCase__     )     )     )



   def  _UpperCamelCase		(    self     )				->     Any:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :Tuple       =self.get_dummy_components()
        _UpperCamelCase  :int       =lora_components["""unet_lora_attn_procs"""]
        _UpperCamelCase  :Tuple       =StableDiffusionPipeline(**lowerCAmelCase__     )
        _UpperCamelCase  :Dict       =sd_pipe.to(lowerCAmelCase__     )
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__     )

        _UpperCamelCase       ,      _UpperCamelCase       ,      _UpperCamelCase  :List[Any]       =self.get_dummy_inputs()

        _UpperCamelCase  :Tuple       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :Union[str, Any]       =original_images[0, -3:, -3:, -1]

        with tempfile.TemporaryDirectory() as tmpdirname:
             _UpperCamelCase  :Optional[Any]       =sd_pipe.unet
             unet.set_attn_processor(lowerCAmelCase__     )
             unet.save_attn_procs(lowerCAmelCase__     )
             self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase__					,      """pytorch_lora_weights.bin"""     )     )     )
             sd_pipe.load_lora_weights(lowerCAmelCase__     )

        _UpperCamelCase  :Tuple       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :Optional[int]       =lora_images[0, -3:, -3:, -1]

        # Outputs shouldn't match.
        self.assertFalse(torch.allclose(torch.from_numpy(lowerCAmelCase__     )					,      torch.from_numpy(lowerCAmelCase__     )     )     )



   def  _UpperCamelCase		(    self     )				->     Dict:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :Any       =self.get_dummy_components()
        _UpperCamelCase  :Dict       =StableDiffusionPipeline(**lowerCAmelCase__     )

        _UpperCamelCase  :Any       =self.get_dummy_tokens()

        # inference without lora
        _UpperCamelCase  :Any       =pipe.text_encoder(**lowerCAmelCase__     )[0]
        assert outputs_without_lora.shape == (1, 77, 32)

        # create lora_attn_procs with zeroed out up.weights
        _UpperCamelCase  :str       =create_text_encoder_lora_attn_procs(pipe.text_encoder     )
        set_lora_up_weights(lowerCAmelCase__					,      randn_weight=lowerCAmelCase__     )

        # monkey patch
        pipe._modify_text_encoder(lowerCAmelCase__     )

        # verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
        del text_attn_procs
        gc.collect()

        # inference with lora
        _UpperCamelCase  :Any       =pipe.text_encoder(**lowerCAmelCase__     )[0]
        assert outputs_with_lora.shape == (1, 77, 32)

        assert torch.allclose(
            lowerCAmelCase__					,      lowerCAmelCase__     ), "lora_up_weight are all zero, so the lora outputs should be the same to without lora outputs"

        # create lora_attn_procs with randn up.weights
        _UpperCamelCase  :List[str]       =create_text_encoder_lora_attn_procs(pipe.text_encoder     )
        set_lora_up_weights(lowerCAmelCase__					,      randn_weight=lowerCAmelCase__     )

        # monkey patch
        pipe._modify_text_encoder(lowerCAmelCase__     )

        # verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
        del text_attn_procs
        gc.collect()

        # inference with lora
        _UpperCamelCase  :int       =pipe.text_encoder(**lowerCAmelCase__     )[0]
        assert outputs_with_lora.shape == (1, 77, 32)

        assert not torch.allclose(
            lowerCAmelCase__					,      lowerCAmelCase__     ), "lora_up_weight are not zero, so the lora outputs should be different to without lora outputs"



   def  _UpperCamelCase		(    self     )				->     Optional[Any]:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :Optional[int]       =self.get_dummy_components()
        _UpperCamelCase  :Any       =StableDiffusionPipeline(**lowerCAmelCase__     )

        _UpperCamelCase  :Optional[Any]       =self.get_dummy_tokens()

        # inference without lora
        _UpperCamelCase  :Any       =pipe.text_encoder(**lowerCAmelCase__     )[0]
        assert outputs_without_lora.shape == (1, 77, 32)

        # create lora_attn_procs with randn up.weights
        _UpperCamelCase  :List[Any]       =create_text_encoder_lora_attn_procs(pipe.text_encoder     )
        set_lora_up_weights(lowerCAmelCase__					,      randn_weight=lowerCAmelCase__     )

        # monkey patch
        pipe._modify_text_encoder(lowerCAmelCase__     )

        # verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
        del text_attn_procs
        gc.collect()

        # inference with lora
        _UpperCamelCase  :str       =pipe.text_encoder(**lowerCAmelCase__     )[0]
        assert outputs_with_lora.shape == (1, 77, 32)

        assert not torch.allclose(
            lowerCAmelCase__					,      lowerCAmelCase__     ), "lora outputs should be different to without lora outputs"

        # remove monkey patch
        pipe._remove_text_encoder_monkey_patch()

        # inference with removed lora
        _UpperCamelCase  :Dict       =pipe.text_encoder(**lowerCAmelCase__     )[0]
        assert outputs_without_lora_removed.shape == (1, 77, 32)

        assert torch.allclose(
            lowerCAmelCase__					,      lowerCAmelCase__     ), "remove lora monkey patch should restore the original outputs"



   def  _UpperCamelCase		(    self     )				->     List[Any]:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :List[Any]       =self.get_dummy_components()
        _UpperCamelCase  :List[Any]       =StableDiffusionPipeline(**lowerCAmelCase__     )
        _UpperCamelCase  :int       =sd_pipe.to(lowerCAmelCase__     )
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__     )

        _UpperCamelCase       ,      _UpperCamelCase       ,      _UpperCamelCase  :Optional[Any]       =self.get_dummy_inputs()

        with tempfile.TemporaryDirectory() as tmpdirname:
             LoraLoaderMixin.save_lora_weights(
                 save_directory=lowerCAmelCase__					,      unet_lora_layers=lora_components["""unet_lora_layers"""]					,      text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]					,      )
             self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase__					,      """pytorch_lora_weights.bin"""     )     )     )
             sd_pipe.load_lora_weights(lowerCAmelCase__     )

        _UpperCamelCase  :int       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :int       =lora_images[0, -3:, -3:, -1]

        _UpperCamelCase  :Optional[int]       =sd_pipe(**lowerCAmelCase__					,      cross_attention_kwargs={"""scale""": 0.5}     ).images
        _UpperCamelCase  :Optional[Any]       =lora_images_with_scale[0, -3:, -3:, -1]

        # Outputs shouldn't match.
        self.assertFalse(
            torch.allclose(torch.from_numpy(lowerCAmelCase__     )					,      torch.from_numpy(lowerCAmelCase__     )     )     )



   def  _UpperCamelCase		(    self     )				->     Union[str, Any]:
        """simple docstring"""





        with tempfile.TemporaryDirectory() as tmpdirname:
             self.create_lora_weight_file(lowerCAmelCase__     )

             _UpperCamelCase       ,      _UpperCamelCase  :int       =self.get_dummy_components()
             _UpperCamelCase  :Tuple       =StableDiffusionPipeline(**lowerCAmelCase__     )
             _UpperCamelCase  :List[str]       =sd_pipe.to(lowerCAmelCase__     )
             sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__     )

             # check if vanilla attention processors are used
             for _, module in sd_pipe.unet.named_modules():
                  if isinstance(lowerCAmelCase__					,      lowerCAmelCase__     ):
                       self.assertIsInstance(module.processor					,      (AttnProcessor, AttnProcessor2_0)     )

            # load LoRA weight file
             sd_pipe.load_lora_weights(lowerCAmelCase__     )

             # check if lora attention processors are used
             for _, module in sd_pipe.unet.named_modules():
                  if isinstance(lowerCAmelCase__					,      lowerCAmelCase__     ):
                       _UpperCamelCase  :Any       =(
                           LoRAAttnProcessor2_0 if hasattr(lowerCAmelCase__					,      """scaled_dot_product_attention"""     ) else LoRAAttnProcessor
                       )
                       self.assertIsInstance(module.processor					,      lowerCAmelCase__     )



   @unittest.skipIf(torch_device != """cuda"""					,      """This test is supposed to run on GPU"""     )
   def  _UpperCamelCase		(    self     )				->     int:
        """simple docstring"""





        with tempfile.TemporaryDirectory() as tmpdirname:
             self.create_lora_weight_file(lowerCAmelCase__     )

             _UpperCamelCase       ,      _UpperCamelCase  :List[Any]       =self.get_dummy_components()
             _UpperCamelCase  :List[str]       =StableDiffusionPipeline(**lowerCAmelCase__     )
             _UpperCamelCase  :Dict       =sd_pipe.to(lowerCAmelCase__     )
             sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__     )

             # enable XFormers
             sd_pipe.enable_xformers_memory_efficient_attention()

             # check if xFormers attention processors are used
             for _, module in sd_pipe.unet.named_modules():
                  if isinstance(lowerCAmelCase__					,      lowerCAmelCase__     ):
                       self.assertIsInstance(module.processor					,      lowerCAmelCase__     )

            # load LoRA weight file
             sd_pipe.load_lora_weights(lowerCAmelCase__     )

             # check if lora attention processors are used
             for _, module in sd_pipe.unet.named_modules():
                  if isinstance(lowerCAmelCase__					,      lowerCAmelCase__     ):
                       self.assertIsInstance(module.processor					,      lowerCAmelCase__     )




   @unittest.skipIf(torch_device != """cuda"""					,      """This test is supposed to run on GPU"""     )
   def  _UpperCamelCase		(    self     )				->     Tuple:
        """simple docstring"""





        _UpperCamelCase       ,      _UpperCamelCase  :int       =self.get_dummy_components()
        _UpperCamelCase  :List[Any]       =StableDiffusionPipeline(**lowerCAmelCase__     )
        _UpperCamelCase  :int       =sd_pipe.to(lowerCAmelCase__     )
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__     )

        _UpperCamelCase       ,      _UpperCamelCase       ,      _UpperCamelCase  :List[Any]       =self.get_dummy_inputs()

        # enable XFormers
        sd_pipe.enable_xformers_memory_efficient_attention()

        _UpperCamelCase  :Tuple       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :List[Any]       =original_images[0, -3:, -3:, -1]

        with tempfile.TemporaryDirectory() as tmpdirname:
             LoraLoaderMixin.save_lora_weights(
                 save_directory=lowerCAmelCase__					,      unet_lora_layers=lora_components["""unet_lora_layers"""]					,      text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]					,      )
             self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase__					,      """pytorch_lora_weights.bin"""     )     )     )
             sd_pipe.load_lora_weights(lowerCAmelCase__     )

        _UpperCamelCase  :List[Any]       =sd_pipe(**lowerCAmelCase__     ).images
        _UpperCamelCase  :Union[str, Any]       =lora_images[0, -3:, -3:, -1]

        # Outputs shouldn't match.
        self.assertFalse(torch.allclose(torch.from_numpy(lowerCAmelCase__     )					,      torch.from_numpy(lowerCAmelCase__     )     )     )



'''simple docstring'''






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


a_       :     Tuple 							=     logging.get_logger(__name__)

a_       :     Optional[Any] 							=     {
    "vocab_file": "vocab.json",
    "merges_file": "merges.txt",
}

a_       :     str 							=     {
    "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"},
}

a_       :     Union[str, Any] 							=     {
    "openai-gpt": 5_1_2,
}






def 						_A					(lowerCAmelCase__	:Any		) ->	int:




     '''simple docstring'''



     _a         =				text.strip()
     if not text:
          return []
     _a         =				text.split()
     return tokens




class 			a  (  _SCREAMING_SNAKE_CASE							):
 def __init__(			self   ,					__magic_name__=True   ,					__magic_name__=None   ,					__magic_name__=True   ,					__magic_name__=None   ,					__magic_name__=True   ,					)       ->				Optional[Any]:
      if never_split is None:
           _a         =				[]
      _a         =				do_lower_case
      _a         =				set(__magic_name__    )
      _a         =				tokenize_chinese_chars
      _a         =				strip_accents
      _a         =				do_split_on_punc
 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__=None    )       ->				List[str]:
      _a         =				self.never_split.union(set(__magic_name__    )    ) if never_split else self.never_split
      _a         =				self._clean_text(__magic_name__    )

      # 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         =				self._tokenize_chinese_chars(__magic_name__    )
      # prevents treating the same character with different unicode codepoints as different characters
      _a         =				unicodedata.normalize('NFC'   ,					__magic_name__    )
      _a         =				whitespace_tokenize(__magic_name__    )
      _a         =				[]
      for token in orig_tokens:
           if token not in never_split:
                if self.do_lower_case:
                     _a         =				token.lower()
                     if self.strip_accents is not False:
                          _a         =				self._run_strip_accents(__magic_name__    )
                elif self.strip_accents:
                     _a         =				self._run_strip_accents(__magic_name__    )
           split_tokens.extend(self._run_split_on_punc(__magic_name__   ,					__magic_name__    )    )

      _a         =				whitespace_tokenize(' '.join(__magic_name__    )    )
      return output_tokens
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				List[str]:
      _a         =				unicodedata.normalize('NFD'   ,					__magic_name__    )
      _a         =				[]
      for char in text:
           _a         =				unicodedata.category(__magic_name__    )
           if cat == "Mn":
                continue
           output.append(__magic_name__    )
      return "".join(__magic_name__    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__=None    )       ->				Dict:
      if not self.do_split_on_punc or (never_split is not None and text in never_split):
           return [text]
      _a         =				list(__magic_name__    )
      _a         =				0
      _a         =				True
      _a         =				[]
      while i < len(__magic_name__    ):
           _a         =				chars[i]
           if _is_punctuation(__magic_name__    ):
                output.append([char]    )
                _a         =				True
           else:
                if start_new_word:
                     output.append([]    )
                _a         =				False
                output[-1].append(__magic_name__    )
           i += 1

      return ["".join(__magic_name__    ) for x in output]
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Tuple:
      _a         =				[]
      for char in text:
           _a         =				ord(__magic_name__    )
           if self._is_chinese_char(__magic_name__    ):
                output.append(' '    )
                output.append(__magic_name__    )
                output.append(' '    )
           else:
                output.append(__magic_name__    )
      return "".join(__magic_name__    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Union[str, Any]:
      # 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 >= 0x4E00 and cp <= 0x9FFF)
          or (cp >= 0x3400 and cp <= 0x4DBF)  #
          or (cp >= 0x2_0000 and cp <= 0x2_A6DF)  #
          or (cp >= 0x2_A700 and cp <= 0x2_B73F)  #
          or (cp >= 0x2_B740 and cp <= 0x2_B81F)  #
          or (cp >= 0x2_B820 and cp <= 0x2_CEAF)  #
          or (cp >= 0xF900 and cp <= 0xFAFF)
          or (cp >= 0x2_F800 and cp <= 0x2_FA1F)  #
      ):  #
           return True

      return False




 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      _a         =				[]
      for char in text:
           _a         =				ord(__magic_name__    )
           if cp == 0 or cp == 0xFFFD or _is_control(__magic_name__    ):
                continue
           if _is_whitespace(__magic_name__    ):
                output.append(' '    )
           else:
                output.append(__magic_name__    )
      return "".join(__magic_name__    )






def 						_A					(lowerCAmelCase__	:str		) ->	Tuple:




     '''simple docstring'''



     _a         =				set()
     _a         =				word[0]
     for char in word[1:]:
          pairs.add((prev_char, char)		)
          _a         =				char
     return pairs






def 						_A					(lowerCAmelCase__	:Tuple		) ->	Union[str, Any]:




     '''simple docstring'''



     _a         =				text.replace('—'					,    '-'		)
     _a         =				text.replace('–'					,    '-'		)
     _a         =				text.replace('―'					,    '-'		)
     _a         =				text.replace('…'					,    '...'		)
     _a         =				text.replace('´'					,    '\''		)
     _a         =				re.sub(r'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'					,    r' \1 '					,    lowerCAmelCase__		)
     _a         =				re.sub(r'\s*\n\s*'					,    ' \n '					,    lowerCAmelCase__		)
     _a         =				re.sub(r'[^\S\n]+'					,    ' '					,    lowerCAmelCase__		)
     return text.strip()






class 			a  (  _SCREAMING_SNAKE_CASE							):
 _lowerCAmelCase			       =						VOCAB_FILES_NAMES
 _lowerCAmelCase			       =						PRETRAINED_VOCAB_FILES_MAP
 _lowerCAmelCase			       =						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _lowerCAmelCase			       =						["""input_ids""", """attention_mask"""]
 def __init__(			self   ,					__magic_name__   ,					__magic_name__   ,					__magic_name__="<unk>"   ,					**__magic_name__    )       ->				Any:
      super().__init__(unk_token=__magic_name__   ,					**__magic_name__    )

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

           _a         =				English()
           _a         =				_nlp.tokenizer
           _a         =				ftfy.fix_text
      except ImportError:
           logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.'    )
           _a         =				BasicTokenizer(do_lower_case=__magic_name__    )
           _a         =				None

      with open(__magic_name__   ,					encoding='utf-8'    ) as vocab_handle:
           _a         =				json.load(__magic_name__    )
      _a         =				{v: k for k, v in self.encoder.items()}
      with open(__magic_name__   ,					encoding='utf-8'    ) as merges_handle:
           _a         =				merges_handle.read().split('\n'    )[1:-1]
      _a         =				[tuple(merge.split()    ) for merge in merges]
      _a         =				dict(zip(__magic_name__   ,					range(len(__magic_name__    )    )    )    )
      _a         =				{}
 @property
 def 				__UpperCAmelCase		(			self    )       ->				Optional[int]:
      return True
 @property
 def 				__UpperCAmelCase		(			self    )       ->				int:
      return len(self.encoder    )
 def 				__UpperCAmelCase		(			self    )       ->				List[str]:
      return dict(self.encoder   ,					**self.added_tokens_encoder    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      _a         =				tuple(token[:-1]    ) + (token[-1] + '</w>',)
      if token in self.cache:
           return self.cache[token]
      _a         =				get_pairs(__magic_name__    )

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

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

                if word[i] == first and i < len(__magic_name__    ) - 1 and word[i + 1] == second:
                     new_word.append(first + second    )
                     i += 2
                else:
                     new_word.append(word[i]    )
                     i += 1
           _a         =				tuple(__magic_name__    )
           _a         =				new_word
           if len(__magic_name__    ) == 1:
                break
           else:
                _a         =				get_pairs(__magic_name__    )
      _a         =				' '.join(__magic_name__    )
      if word == "\n  </w>":
           _a         =				'\n</w>'
      _a         =				word
      return word
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Union[str, Any]:
      _a         =				[]
      if self.fix_text is None:
           # Using BERT's BasicTokenizer
           _a         =				self.nlp.tokenize(__magic_name__    )
           for token in text:
                split_tokens.extend(list(self.bpe(__magic_name__    ).split(' '    )    )    )
      else:
           # Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
           _a         =				self.nlp(text_standardize(self.fix_text(__magic_name__    )    )    )
           for token in text:
                split_tokens.extend(list(self.bpe(token.text.lower()    ).split(' '    )    )    )
      return split_tokens
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      return self.encoder.get(__magic_name__   ,					self.encoder.get(self.unk_token    )    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Any:
      return self.decoder.get(__magic_name__   ,					self.unk_token    )
 def 				__UpperCAmelCase		(			self   ,					__magic_name__    )       ->				Dict:
      _a         =				''.join(__magic_name__    ).replace('</w>'   ,					' '    ).strip()
      return out_string




 def 				__UpperCAmelCase		(			self   ,					__magic_name__   ,					__magic_name__ = None    )       ->				Tuple[str]:
      if not os.path.isdir(__magic_name__    ):
           logger.error(f'Vocabulary path ({save_directory}) should be a directory'    )
           return
      _a         =				os.path.join(
          __magic_name__   ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']    )
      _a         =				os.path.join(
          __magic_name__   ,					(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']    )

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

      _a         =				0
      with open(__magic_name__   ,					'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 __magic_name__    : 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         =				token_index
                writer.write(' '.join(__magic_name__    ) + '\n'    )
                index += 1

      return vocab_file, merge_file

