"""simple docstring"""



import gc
import random
import tempfile
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.semantic_stable_diffusion import SemanticStableDiffusionPipeline as StableDiffusionPipeline
from diffusers.utils import floats_tensor, nightly, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()

class SCREAMING_SNAKE_CASE     (    unittest.TestCase	):







 """simple docstring"""


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




    '''simple docstring'''






    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()


 @property
 def        _lowerCAmelCase				(   self		:      str				)				->			str:




    '''simple docstring'''






    a__     				=		1
    a__     				=		3
    a__     				=		(32, 32)

    a__     				=		floats_tensor((batch_size, num_channels) + sizes					,   rng=random.Random(0				)				).to(_snake_case				)
    return image


 @property
 def        _lowerCAmelCase				(   self		:      int				)				->			List[str]:




    '''simple docstring'''






    torch.manual_seed(0				)
    a__     				=		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					,   )
    return model


 @property
 def        _lowerCAmelCase				(   self		:      Union[str, Any]				)				->			Tuple:




    '''simple docstring'''






    torch.manual_seed(0				)
    a__     				=		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					,   )
    return model


 @property
 def        _lowerCAmelCase				(   self		:      Union[str, Any]				)				->			int:




    '''simple docstring'''






    torch.manual_seed(0				)
    a__     				=		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=1000					,   )
    return CLIPTextModel(_snake_case				)


 @property
 def        _lowerCAmelCase				(   self		:      Union[str, Any]				)				->			List[Any]:




    '''simple docstring'''






    def extract(*_snake_case		:      List[str]					,   **_snake_case		:      int				):

       class SCREAMING_SNAKE_CASE     :







        """simple docstring"""


        def __init__(   self		:      Optional[Any]				)				->			List[Any]:




           '''simple docstring'''






           a__     				=		torch.ones([0]				)



        def        _lowerCAmelCase				(   self		:      Dict					,   _snake_case		:      Dict				)				->			List[str]:




           '''simple docstring'''






           self.pixel_values.to(_snake_case				)
           return self
       return Out()

    return extract


 def        _lowerCAmelCase				(   self		:      int				)				->			Dict:




    '''simple docstring'''






    a__     				=		'cpu'  # ensure determinism for the device-dependent torch.Generator
    a__     				=		self.dummy_cond_unet
    a__     				=		DDIMScheduler(
        beta_start=0.0_0085					,   beta_end=0.012					,   beta_schedule='scaled_linear'					,   clip_sample=_snake_case					,   set_alpha_to_one=_snake_case					,   )

    a__     				=		self.dummy_vae
    a__     				=		self.dummy_text_encoder
    a__     				=		CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				)

    # make sure here that pndm scheduler skips prk
    a__     				=		StableDiffusionPipeline(
        unet=_snake_case					,   scheduler=_snake_case					,   vae=_snake_case					,   text_encoder=_snake_case					,   tokenizer=_snake_case					,   safety_checker=_snake_case					,   feature_extractor=self.dummy_extractor					,   )
    a__     				=		sd_pipe.to(_snake_case				)
    sd_pipe.set_progress_bar_config(disable=_snake_case				)

    a__     				=		'A painting of a squirrel eating a burger'

    a__     				=		torch.Generator(device=_snake_case				).manual_seed(0				)
    a__     				=		sd_pipe([prompt]					,   generator=_snake_case					,   guidance_scale=6.0					,   num_inference_steps=2					,   output_type='np'				)
    a__     				=		output.images

    a__     				=		torch.Generator(device=_snake_case				).manual_seed(0				)
    a__     				=		sd_pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=6.0					,   num_inference_steps=2					,   output_type='np'					,   return_dict=_snake_case					,   )[0]

    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		image_from_tuple[0, -3:, -3:, -1]

    assert image.shape == (1, 64, 64, 3)
    a__     				=		np.array([0.5753, 0.6114, 0.5001, 0.5034, 0.5470, 0.4729, 0.4971, 0.4867, 0.4867]				)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2
    assert np.abs(image_from_tuple_slice.flatten() - expected_slice				).max() < 1E-2


 def        _lowerCAmelCase				(   self		:      List[str]				)				->			List[str]:




    '''simple docstring'''






    a__     				=		'cpu'  # ensure determinism for the device-dependent torch.Generator
    a__     				=		self.dummy_cond_unet
    a__     				=		PNDMScheduler(skip_prk_steps=_snake_case				)
    a__     				=		self.dummy_vae
    a__     				=		self.dummy_text_encoder
    a__     				=		CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				)

    # make sure here that pndm scheduler skips prk
    a__     				=		StableDiffusionPipeline(
        unet=_snake_case					,   scheduler=_snake_case					,   vae=_snake_case					,   text_encoder=_snake_case					,   tokenizer=_snake_case					,   safety_checker=_snake_case					,   feature_extractor=self.dummy_extractor					,   )
    a__     				=		sd_pipe.to(_snake_case				)
    sd_pipe.set_progress_bar_config(disable=_snake_case				)

    a__     				=		'A painting of a squirrel eating a burger'
    a__     				=		torch.Generator(device=_snake_case				).manual_seed(0				)
    a__     				=		sd_pipe([prompt]					,   generator=_snake_case					,   guidance_scale=6.0					,   num_inference_steps=2					,   output_type='np'				)

    a__     				=		output.images

    a__     				=		torch.Generator(device=_snake_case				).manual_seed(0				)
    a__     				=		sd_pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=6.0					,   num_inference_steps=2					,   output_type='np'					,   return_dict=_snake_case					,   )[0]

    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		image_from_tuple[0, -3:, -3:, -1]

    assert image.shape == (1, 64, 64, 3)
    a__     				=		np.array([0.5122, 0.5712, 0.4825, 0.5053, 0.5646, 0.4769, 0.5179, 0.4894, 0.4994]				)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2
    assert np.abs(image_from_tuple_slice.flatten() - expected_slice				).max() < 1E-2


 def        _lowerCAmelCase				(   self		:      Optional[int]				)				->			Tuple:




    '''simple docstring'''






    a__     				=		StableDiffusionPipeline.from_pretrained(
        'hf-internal-testing/tiny-stable-diffusion-lms-pipe'					,   safety_checker=_snake_case				)
    assert isinstance(_snake_case					,   _snake_case				)
    assert isinstance(pipe.scheduler					,   _snake_case				)
    assert pipe.safety_checker is None

    a__     				=		pipe('example prompt'					,   num_inference_steps=2				).images[0]
    assert image is not None

    # check that there's no error when saving a pipeline with one of the models being None
    with tempfile.TemporaryDirectory() as tmpdirname:
       pipe.save_pretrained(_snake_case				)
       a__     				=		StableDiffusionPipeline.from_pretrained(_snake_case				)

    # sanity check that the pipeline still works
    assert pipe.safety_checker is None
    a__     				=		pipe('example prompt'					,   num_inference_steps=2				).images[0]
    assert image is not None



 @unittest.skipIf(torch_device != 'cuda'					,   'This test requires a GPU'				)
 def        _lowerCAmelCase				(   self		:      Optional[Any]				)				->			Optional[int]:




    '''simple docstring'''






    a__     				=		self.dummy_cond_unet
    a__     				=		PNDMScheduler(skip_prk_steps=_snake_case				)
    a__     				=		self.dummy_vae
    a__     				=		self.dummy_text_encoder
    a__     				=		CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				)

    # put models in fp16
    a__     				=		unet.half()
    a__     				=		vae.half()
    a__     				=		bert.half()

    # make sure here that pndm scheduler skips prk
    a__     				=		StableDiffusionPipeline(
        unet=_snake_case					,   scheduler=_snake_case					,   vae=_snake_case					,   text_encoder=_snake_case					,   tokenizer=_snake_case					,   safety_checker=_snake_case					,   feature_extractor=self.dummy_extractor					,   )
    a__     				=		sd_pipe.to(_snake_case				)
    sd_pipe.set_progress_bar_config(disable=_snake_case				)

    a__     				=		'A painting of a squirrel eating a burger'
    a__     				=		sd_pipe([prompt]					,   num_inference_steps=2					,   output_type='np'				).images

    assert image.shape == (1, 64, 64, 3)

@nightly
@require_torch_gpu
class SCREAMING_SNAKE_CASE     (    unittest.TestCase	):







 """simple docstring"""


 def        _lowerCAmelCase				(   self		:      List[Any]				)				->			int:




    '''simple docstring'''






    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()


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




    '''simple docstring'''






    a__     				=		'cuda'
    a__     				=		StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'				)
    a__     				=		pipe.to(_snake_case				)
    pipe.set_progress_bar_config(disable=_snake_case				)

    a__     				=		'a photo of a cat'
    a__     				=		{
        'editing_prompt': ['sunglasses'],
        'reverse_editing_direction': [False],
        'edit_warmup_steps': 10,
        'edit_guidance_scale': 6,
        'edit_threshold': 0.95,
        'edit_momentum_scale': 0.5,
        'edit_mom_beta': 0.6,
    }

    a__     				=		3
    a__     				=		7

    # no sega enabled
    a__     				=		torch.Generator(_snake_case				)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.3467_3113,
        0.3849_2733,
        0.3759_7352,
        0.3408_6335,
        0.3565_0748,
        0.3557_9205,
        0.338_4763,
        0.3434_0236,
        0.357_3271,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2

    # with sega enabled
    # generator = torch.manual_seed(seed)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   **_snake_case					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.4188_7826,
        0.3772_8766,
        0.3013_8272,
        0.4141_6335,
        0.4166_4985,
        0.3628_3392,
        0.3619_1246,
        0.4336_4465,
        0.4300_1732,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2


 def        _lowerCAmelCase				(   self		:      int				)				->			Dict:




    '''simple docstring'''






    a__     				=		'cuda'
    a__     				=		StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'				)
    a__     				=		pipe.to(_snake_case				)
    pipe.set_progress_bar_config(disable=_snake_case				)

    a__     				=		'an image of a crowded boulevard, realistic, 4k'
    a__     				=		{
        'editing_prompt': 'crowd, crowded, people',
        'reverse_editing_direction': True,
        'edit_warmup_steps': 10,
        'edit_guidance_scale': 8.3,
        'edit_threshold': 0.9,
        'edit_momentum_scale': 0.5,
        'edit_mom_beta': 0.6,
    }

    a__     				=		9
    a__     				=		7

    # no sega enabled
    a__     				=		torch.Generator(_snake_case				)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.4349_7998,
        0.9181_4065,
        0.754_0739,
        0.5558_0205,
        0.846_7265,
        0.538_9691,
        0.6257_4506,
        0.5889_7763,
        0.5092_6757,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2

    # with sega enabled
    # generator = torch.manual_seed(seed)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   **_snake_case					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.308_9719,
        0.3050_0144,
        0.2901_6042,
        0.3063_0964,
        0.32_5687,
        0.2941_9225,
        0.290_8091,
        0.2872_3598,
        0.2769_6294,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2


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




    '''simple docstring'''






    a__     				=		'cuda'
    a__     				=		StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'				)
    a__     				=		pipe.to(_snake_case				)
    pipe.set_progress_bar_config(disable=_snake_case				)

    a__     				=		'a castle next to a river'
    a__     				=		{
        'editing_prompt': ['boat on a river, boat', 'monet, impression, sunrise'],
        'reverse_editing_direction': False,
        'edit_warmup_steps': [15, 18],
        'edit_guidance_scale': 6,
        'edit_threshold': [0.9, 0.8],
        'edit_momentum_scale': 0.5,
        'edit_mom_beta': 0.6,
    }

    a__     				=		48
    a__     				=		7

    # no sega enabled
    a__     				=		torch.Generator(_snake_case				)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.7516_3555,
        0.7603_7145,
        0.6_1785,
        0.918_9673,
        0.862_7701,
        0.8518_9694,
        0.851_2813,
        0.8701_2076,
        0.831_2857,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2

    # with sega enabled
    # generator = torch.manual_seed(seed)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   **_snake_case					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.7355_3365,
        0.753_7271,
        0.7434_1905,
        0.6648_0356,
        0.647_2925,
        0.6303_9416,
        0.6481_2905,
        0.674_9717,
        0.651_7102,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2



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




    '''simple docstring'''






    a__     				=		'cuda'
    a__     				=		StableDiffusionPipeline.from_pretrained('runwayml/stable-diffusion-v1-5'					,   torch_dtype=torch.floataa				)
    a__     				=		pipe.to(_snake_case				)
    pipe.set_progress_bar_config(disable=_snake_case				)

    a__     				=		'a photo of a cat'
    a__     				=		{
        'editing_prompt': ['sunglasses'],
        'reverse_editing_direction': [False],
        'edit_warmup_steps': 10,
        'edit_guidance_scale': 6,
        'edit_threshold': 0.95,
        'edit_momentum_scale': 0.5,
        'edit_mom_beta': 0.6,
    }

    a__     				=		3
    a__     				=		7

    # no sega enabled
    a__     				=		torch.Generator(_snake_case				)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.3488_7695,
        0.387_6953,
        0.375,
        0.3442_3828,
        0.358_1543,
        0.3571_7773,
        0.338_3789,
        0.3457_0312,
        0.35_9375,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2

    # with sega enabled
    # generator = torch.manual_seed(seed)
    generator.manual_seed(_snake_case				)
    a__     				=		pipe(
        [prompt]					,   generator=_snake_case					,   guidance_scale=_snake_case					,   num_inference_steps=50					,   output_type='np'					,   width=512					,   height=512					,   **_snake_case					,   )

    a__     				=		output.images
    a__     				=		image[0, -3:, -3:, -1]
    a__     				=		[
        0.4228_5156,
        0.3691_4062,
        0.2907_7148,
        0.4204_1016,
        0.4191_8945,
        0.3549_8047,
        0.361_8164,
        0.442_3828,
        0.4311_5234,
    ]

    assert image.shape == (1, 512, 512, 3)

    assert np.abs(image_slice.flatten() - expected_slice				).max() < 1E-2



"""simple docstring"""



import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
     from transformers.pipelines.conversational import Conversation

__magic_name__    				=					logging.get_logger(__name__)

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

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

__magic_name__    				=					{
    "gpt2": 1_024,
    "gpt2-medium": 1_024,
    "gpt2-large": 1_024,
    "gpt2-xl": 1_024,
    "distilgpt2": 1_024,
}





@lru_cache()
def   _lowerCamelCase ( )							->     int:


   '''simple docstring'''


   a__     				=		(
       list(range(ord('!'					),ord('~'					) + 1					)					) + list(range(ord('¡'					),ord('¬'					) + 1					)					) + list(range(ord('®'					),ord('ÿ'					) + 1					)					)
   )
   a__     				=		bs[:]
   a__     				=		0
   for b in range(2**8					):
      if b not in bs:
         bs.append(UpperCAmelCase__					)
         cs.append(2**8 + n					)
         n += 1
   a__     				=		[chr(UpperCAmelCase__					) for n in cs]
   return dict(zip(UpperCAmelCase__,UpperCAmelCase__					)					)





def   _lowerCamelCase ( UpperCAmelCase__					)							->     List[str]:


   '''simple docstring'''


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




class SCREAMING_SNAKE_CASE     (    a	):







 """simple docstring"""


 a_       :	Union[str, Any]  			=VOCAB_FILES_NAMES
 a_       :	Optional[Any]  			=PRETRAINED_VOCAB_FILES_MAP
 a_       :	Dict  			=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 a_       :	str  			=["input_ids", "attention_mask"]


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




    '''simple docstring'''






    a__     				=		AddedToken(_snake_case					,   lstrip=_snake_case					,   rstrip=_snake_case				) if isinstance(_snake_case					,   _snake_case				) else bos_token
    a__     				=		AddedToken(_snake_case					,   lstrip=_snake_case					,   rstrip=_snake_case				) if isinstance(_snake_case					,   _snake_case				) else eos_token
    a__     				=		AddedToken(_snake_case					,   lstrip=_snake_case					,   rstrip=_snake_case				) if isinstance(_snake_case					,   _snake_case				) else unk_token
    a__     				=		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					,   )
    a__     				=		add_bos_token

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

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


 @property
 def        _lowerCAmelCase				(   self		:      Tuple				)				->			Optional[int]:




    '''simple docstring'''






    return len(self.encoder				)


 def        _lowerCAmelCase				(   self		:      Tuple				)				->			Any:




    '''simple docstring'''






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


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




    '''simple docstring'''






    if token in self.cache:
       return self.cache[token]
    a__     				=		tuple(_snake_case				)
    a__     				=		get_pairs(_snake_case				)

    if not pairs:
       return token

    while True:
       a__     				=		min(_snake_case					,   key=lambda _snake_case				: self.bpe_ranks.get(_snake_case					,   float('inf'				)				)				)
       if bigram not in self.bpe_ranks:
          break
       a__				,       a__     				=		bigram
       a__     				=		[]
       a__     				=		0
       while i < len(_snake_case				):
          try:
             a__     				=		word.index(_snake_case					,   _snake_case				)
          except ValueError:
             new_word.extend(word[i:]				)
             break
          else:
             new_word.extend(word[i:j]				)
             a__     				=		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
       a__     				=		tuple(_snake_case				)
       a__     				=		new_word
       if len(_snake_case				) == 1:
          break
       else:
          a__     				=		get_pairs(_snake_case				)
    a__     				=		' '.join(_snake_case				)
    a__     				=		word
    return word


 def        _lowerCAmelCase				(   self		:      List[Any]					,   _snake_case		:      Dict					,   _snake_case		:      List[Any]=None				)				->			List[Any]:




    '''simple docstring'''






    if self.add_bos_token:
       a__     				=		[self.bos_token_id]
    else:
       a__     				=		[]

    a__     				=		bos_token_ids + token_ids_a

    if token_ids_a is None:
       return output

    return output + bos_token_ids + token_ids_a


 def        _lowerCAmelCase				(   self		:      Any					,   _snake_case		:      List[int]					,   _snake_case		:      Optional[List[int]] = None					,   _snake_case		:      bool = False				)				->			List[int]:




    '''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        _lowerCAmelCase				(   self		:      Dict					,   _snake_case		:      Optional[Any]				)				->			Tuple:




    '''simple docstring'''






    a__     				=		[]
    for token in re.findall(self.pat					,   _snake_case				):
       a__     				=		''.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        _lowerCAmelCase				(   self		:      Optional[Any]					,   _snake_case		:      Tuple				)				->			int:




    '''simple docstring'''






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


 def        _lowerCAmelCase				(   self		:      List[Any]					,   _snake_case		:      int				)				->			List[str]:




    '''simple docstring'''






    return self.decoder.get(_snake_case				)


 def        _lowerCAmelCase				(   self		:      Tuple					,   _snake_case		:      Any				)				->			Optional[Any]:




    '''simple docstring'''






    a__     				=		''.join(_snake_case				)
    a__     				=		bytearray([self.byte_decoder[c] for c in text]				).decode('utf-8'					,   errors=self.errors				)
    return text


 def        _lowerCAmelCase				(   self		:      Optional[int]					,   _snake_case		:      str					,   _snake_case		:      Optional[str] = None				)				->			Tuple[str]:




    '''simple docstring'''






    if not os.path.isdir(_snake_case				):
       logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''				)
       return
    a__     				=		os.path.join(
        _snake_case					,   (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']				)
    a__     				=		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'				)

    a__     				=		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 _snake_case				: kv[1]				):
          if index != token_index:
             logger.warning(
                 F'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
                 ' Please check that the tokenizer is not corrupted!'				)
             a__     				=		token_index
          writer.write(' '.join(_snake_case				) + '\n'				)
          index += 1

    return vocab_file, merge_file


 def        _lowerCAmelCase				(   self		:      int					,   _snake_case		:      Union[str, Any]					,   _snake_case		:      List[Any]=False					,   **_snake_case		:      Optional[int]				)				->			Union[str, Any]:




    '''simple docstring'''






    a__     				=		kwargs.pop('add_prefix_space'					,   self.add_prefix_space				)
    if is_split_into_words or add_prefix_space:
       a__     				=		' ' + text
    return (text, kwargs)



 def        _lowerCAmelCase				(   self		:      Dict					,   _snake_case		:      "Conversation"				)				->			List[int]:




    '''simple docstring'''






    a__     				=		[]
    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:
       a__     				=		input_ids[-self.model_max_length :]
    return input_ids


