




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							):







	'''simple docstring'''





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


def 						lowerCamelCase__   (		_A							):







	'''simple docstring'''





	snake_case_		   =					{}
	snake_case_		   =					(
	    LoRAAttnProcessor2_0 if hasattr(_A					,     "scaled_dot_product_attention"							) else LoRAAttnProcessor
	)
	for name, module in text_encoder.named_modules():
		if name.endswith(_A							):
			snake_case_		   =					lora_attn_processor_class(
			    hidden_size=module.out_proj.out_features					,     cross_attention_dim=_A							)
	return text_lora_attn_procs


def 						lowerCamelCase__   (		_A							):







	'''simple docstring'''





	snake_case_		   =					create_text_encoder_lora_attn_procs(_A							)
	snake_case_		   =					AttnProcsLayers(_A							)
	return text_encoder_lora_layers


def 						lowerCamelCase__   (		_A					,     _A=False							):







	'''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"							):
				snake_case_		   =					(
				    torch.randn_like(layer_module.up.weight							)
				    if randn_weight
				    else torch.zeros_like(layer_module.up.weight							)
				)
				snake_case_		   =					torch.nn.Parameter(_A							)






class 		UpperCAmelCase       (    unittest.TestCase      ):







						'''simple docstring'''





						def 	snake_case__						(						self							:					int						):






							"""simple docstring"""






							torch.manual_seed(0						)
							snake_case_		   =					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	,     )
							snake_case_		   =					DDIMScheduler(
							    beta_start=0.0_0085	,     beta_end=0.012	,     beta_schedule="scaled_linear"	,     clip_sample=__lowercase	,     set_alpha_to_one=__lowercase	,     steps_offset=1	,     )
							torch.manual_seed(0						)
							snake_case_		   =					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	,     )
							snake_case_		   =					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=10_00	,     )
							snake_case_		   =					CLIPTextModel(__lowercase						)
							snake_case_		   =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)

							snake_case_   ,				snake_case_		   =					create_unet_lora_layers(__lowercase						)
							snake_case_		   =					create_text_encoder_lora_layers(__lowercase						)

							snake_case_		   =					{
							    "unet": unet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
							snake_case_		   =					{
							    "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 	snake_case__						(						self							:					Optional[Any]						):






							"""simple docstring"""






							snake_case_		   =					1
							snake_case_		   =					10
							snake_case_		   =					4
							snake_case_		   =					(32, 32)

							snake_case_		   =					torch.manual_seed(0						)
							snake_case_		   =					floats_tensor((batch_size, num_channels) + sizes						)
							snake_case_		   =					torch.randint(1	,     __lowercase	,     size=(batch_size, sequence_length)	,     generator=__lowercase						)

							snake_case_		   =					{
							    "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 	snake_case__						(						self							:					Tuple						):






							"""simple docstring"""






							snake_case_		   =					77

							snake_case_		   =					torch.randint(2	,     56	,     size=(1, max_seq_length)	,     generator=torch.manual_seed(0						)						)

							snake_case_		   =					{}
							snake_case_		   =					inputs
							return prepared_inputs






						def 	snake_case__						(						self							:					List[Any]	,     __lowercase							:					Any						):






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							LoraLoaderMixin.save_lora_weights(
							    save_directory=__lowercase	,     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(__lowercase	,     "pytorch_lora_weights.bin"						)						)						)






						def 	snake_case__						(						self							:					Optional[Any]						):






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							snake_case_		   =					StableDiffusionPipeline(**__lowercase						)
							snake_case_		   =					sd_pipe.to(__lowercase						)
							sd_pipe.set_progress_bar_config(disable=__lowercase						)

							snake_case_   ,				snake_case_   ,				snake_case_		   =					self.get_dummy_inputs()

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					original_images[0, -3:, -3:, -1]

							with tempfile.TemporaryDirectory() as tmpdirname:
								LoraLoaderMixin.save_lora_weights(
								    save_directory=__lowercase	,     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(__lowercase	,     "pytorch_lora_weights.bin"						)						)						)
								sd_pipe.load_lora_weights(__lowercase						)

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					lora_images[0, -3:, -3:, -1]

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






						def 	snake_case__						(						self							:					List[str]						):






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							snake_case_		   =					StableDiffusionPipeline(**__lowercase						)
							snake_case_		   =					sd_pipe.to(__lowercase						)
							sd_pipe.set_progress_bar_config(disable=__lowercase						)

							snake_case_   ,				snake_case_   ,				snake_case_		   =					self.get_dummy_inputs()

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					original_images[0, -3:, -3:, -1]

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

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					lora_images[0, -3:, -3:, -1]

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






						def 	snake_case__						(						self							:					Any						):






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							snake_case_		   =					lora_components["unet_lora_attn_procs"]
							snake_case_		   =					StableDiffusionPipeline(**__lowercase						)
							snake_case_		   =					sd_pipe.to(__lowercase						)
							sd_pipe.set_progress_bar_config(disable=__lowercase						)

							snake_case_   ,				snake_case_   ,				snake_case_		   =					self.get_dummy_inputs()

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					original_images[0, -3:, -3:, -1]

							with tempfile.TemporaryDirectory() as tmpdirname:
								snake_case_		   =					sd_pipe.unet
								unet.set_attn_processor(__lowercase						)
								unet.save_attn_procs(__lowercase						)
								self.assertTrue(os.path.isfile(os.path.join(__lowercase	,     "pytorch_lora_weights.bin"						)						)						)
								sd_pipe.load_lora_weights(__lowercase						)

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					lora_images[0, -3:, -3:, -1]

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






						def 	snake_case__						(						self							:					Any						):






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							snake_case_		   =					StableDiffusionPipeline(**__lowercase						)

							snake_case_		   =					self.get_dummy_tokens()

							# inference without lora
							snake_case_		   =					pipe.text_encoder(**__lowercase						)[0]
							assert outputs_without_lora.shape == (1, 77, 32)

							# create lora_attn_procs with zeroed out up.weights
							snake_case_		   =					create_text_encoder_lora_attn_procs(pipe.text_encoder						)
							set_lora_up_weights(__lowercase	,     randn_weight=__lowercase						)

							# monkey patch
							pipe._modify_text_encoder(__lowercase						)

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

							# inference with lora
							snake_case_		   =					pipe.text_encoder(**__lowercase						)[0]
							assert outputs_with_lora.shape == (1, 77, 32)

							assert torch.allclose(
							    __lowercase	,     __lowercase						), "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
							snake_case_		   =					create_text_encoder_lora_attn_procs(pipe.text_encoder						)
							set_lora_up_weights(__lowercase	,     randn_weight=__lowercase						)

							# monkey patch
							pipe._modify_text_encoder(__lowercase						)

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

							# inference with lora
							snake_case_		   =					pipe.text_encoder(**__lowercase						)[0]
							assert outputs_with_lora.shape == (1, 77, 32)

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






						def 	snake_case__						(						self							:					Optional[Any]						):






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							snake_case_		   =					StableDiffusionPipeline(**__lowercase						)

							snake_case_		   =					self.get_dummy_tokens()

							# inference without lora
							snake_case_		   =					pipe.text_encoder(**__lowercase						)[0]
							assert outputs_without_lora.shape == (1, 77, 32)

							# create lora_attn_procs with randn up.weights
							snake_case_		   =					create_text_encoder_lora_attn_procs(pipe.text_encoder						)
							set_lora_up_weights(__lowercase	,     randn_weight=__lowercase						)

							# monkey patch
							pipe._modify_text_encoder(__lowercase						)

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

							# inference with lora
							snake_case_		   =					pipe.text_encoder(**__lowercase						)[0]
							assert outputs_with_lora.shape == (1, 77, 32)

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

							# remove monkey patch
							pipe._remove_text_encoder_monkey_patch()

							# inference with removed lora
							snake_case_		   =					pipe.text_encoder(**__lowercase						)[0]
							assert outputs_without_lora_removed.shape == (1, 77, 32)

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






						def 	snake_case__						(						self							:					Union[str, Any]						):






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							snake_case_		   =					StableDiffusionPipeline(**__lowercase						)
							snake_case_		   =					sd_pipe.to(__lowercase						)
							sd_pipe.set_progress_bar_config(disable=__lowercase						)

							snake_case_   ,				snake_case_   ,				snake_case_		   =					self.get_dummy_inputs()

							with tempfile.TemporaryDirectory() as tmpdirname:
								LoraLoaderMixin.save_lora_weights(
								    save_directory=__lowercase	,     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(__lowercase	,     "pytorch_lora_weights.bin"						)						)						)
								sd_pipe.load_lora_weights(__lowercase						)

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					lora_images[0, -3:, -3:, -1]

							snake_case_		   =					sd_pipe(**__lowercase	,     cross_attention_kwargs={"scale": 0.5}						).images
							snake_case_		   =					lora_images_with_scale[0, -3:, -3:, -1]

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






						def 	snake_case__						(						self							:					Any						):






							"""simple docstring"""






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

								snake_case_   ,				snake_case_		   =					self.get_dummy_components()
								snake_case_		   =					StableDiffusionPipeline(**__lowercase						)
								snake_case_		   =					sd_pipe.to(__lowercase						)
								sd_pipe.set_progress_bar_config(disable=__lowercase						)

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

            # load LoRA weight file
								sd_pipe.load_lora_weights(__lowercase						)

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






						@unittest.skipIf(torch_device != "cuda"	,     "This test is supposed to run on GPU"						)
						def 	snake_case__						(						self							:					str						):






							"""simple docstring"""






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

								snake_case_   ,				snake_case_		   =					self.get_dummy_components()
								snake_case_		   =					StableDiffusionPipeline(**__lowercase						)
								snake_case_		   =					sd_pipe.to(__lowercase						)
								sd_pipe.set_progress_bar_config(disable=__lowercase						)

								# 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(__lowercase	,     __lowercase						):
										self.assertIsInstance(module.processor	,     __lowercase						)

            # load LoRA weight file
								sd_pipe.load_lora_weights(__lowercase						)

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






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






							"""simple docstring"""






							snake_case_   ,				snake_case_		   =					self.get_dummy_components()
							snake_case_		   =					StableDiffusionPipeline(**__lowercase						)
							snake_case_		   =					sd_pipe.to(__lowercase						)
							sd_pipe.set_progress_bar_config(disable=__lowercase						)

							snake_case_   ,				snake_case_   ,				snake_case_		   =					self.get_dummy_inputs()

							# enable XFormers
							sd_pipe.enable_xformers_memory_efficient_attention()

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					original_images[0, -3:, -3:, -1]

							with tempfile.TemporaryDirectory() as tmpdirname:
								LoraLoaderMixin.save_lora_weights(
								    save_directory=__lowercase	,     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(__lowercase	,     "pytorch_lora_weights.bin"						)						)						)
								sd_pipe.load_lora_weights(__lowercase						)

							snake_case_		   =					sd_pipe(**__lowercase						).images
							snake_case_		   =					lora_images[0, -3:, -3:, -1]

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






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


lowercase__    :					List[str]					=						logging.get_logger(__name__)

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

lowercase__    :					Any					=						{
    "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"},
}

lowercase__    :					Tuple					=						{
    "openai-gpt": 512,
}


def 						lowerCamelCase__   (		_A							):







	'''simple docstring'''





	snake_case_		   =					text.strip()
	if not text:
		return []
	snake_case_		   =					text.split()
	return tokens
class 		UpperCAmelCase       (    UpperCAmelCase__      ):







						'''simple docstring'''





						def __init__(						self							:					Dict	,     __lowercase							:					str=True	,     __lowercase							:					Union[str, Any]=None	,     __lowercase							:					str=True	,     __lowercase							:					Dict=None	,     __lowercase							:					str=True	,     ):






							"""simple docstring"""






							if never_split is None:
								snake_case_		   =					[]
							snake_case_		   =					do_lower_case
							snake_case_		   =					set(__lowercase						)
							snake_case_		   =					tokenize_chinese_chars
							snake_case_		   =					strip_accents
							snake_case_		   =					do_split_on_punc






						def 	snake_case__						(						self							:					Any	,     __lowercase							:					Any	,     __lowercase							:					Optional[int]=None						):






							"""simple docstring"""






							snake_case_		   =					self.never_split.union(set(__lowercase						)						) if never_split else self.never_split
							snake_case_		   =					self._clean_text(__lowercase						)

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

							snake_case_		   =					whitespace_tokenize(" ".join(__lowercase						)						)
							return output_tokens






						def 	snake_case__						(						self							:					str	,     __lowercase							:					List[str]						):






							"""simple docstring"""






							snake_case_		   =					unicodedata.normalize("NFD"	,     __lowercase						)
							snake_case_		   =					[]
							for char in text:
								snake_case_		   =					unicodedata.category(__lowercase						)
								if cat == "Mn":
									continue
								output.append(__lowercase						)
							return "".join(__lowercase						)






						def 	snake_case__						(						self							:					Tuple	,     __lowercase							:					Union[str, Any]	,     __lowercase							:					Optional[int]=None						):






							"""simple docstring"""






							if not self.do_split_on_punc or (never_split is not None and text in never_split):
								return [text]
							snake_case_		   =					list(__lowercase						)
							snake_case_		   =					0
							snake_case_		   =					True
							snake_case_		   =					[]
							while i < len(__lowercase						):
								snake_case_		   =					chars[i]
								if _is_punctuation(__lowercase						):
									output.append([char]						)
									snake_case_		   =					True
								else:
									if start_new_word:
										output.append([]						)
									snake_case_		   =					False
									output[-1].append(__lowercase						)
								i += 1

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






						def 	snake_case__						(						self							:					Optional[int]	,     __lowercase							:					str						):






							"""simple docstring"""






							snake_case_		   =					[]
							for char in text:
								snake_case_		   =					ord(__lowercase						)
								if self._is_chinese_char(__lowercase						):
									output.append(" "						)
									output.append(__lowercase						)
									output.append(" "						)
								else:
									output.append(__lowercase						)
							return "".join(__lowercase						)






						def 	snake_case__						(						self							:					Union[str, Any]	,     __lowercase							:					str						):






							"""simple docstring"""






							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 	snake_case__						(						self							:					int	,     __lowercase							:					Dict						):






							"""simple docstring"""






							snake_case_		   =					[]
							for char in text:
								snake_case_		   =					ord(__lowercase						)
								if cp == 0 or cp == 0Xf_ffd or _is_control(__lowercase						):
									continue
								if _is_whitespace(__lowercase						):
									output.append(" "						)
								else:
									output.append(__lowercase						)
							return "".join(__lowercase						)







def 						lowerCamelCase__   (		_A							):







	'''simple docstring'''





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


def 						lowerCamelCase__   (		_A							):







	'''simple docstring'''





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






class 		UpperCAmelCase       (    UpperCAmelCase__      ):







						'''simple docstring'''





						lowerCAmelCase_							=       VOCAB_FILES_NAMES
						lowerCAmelCase_							=       PRETRAINED_VOCAB_FILES_MAP
						lowerCAmelCase_							=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						lowerCAmelCase_							=       ['''input_ids''', '''attention_mask''']






						def __init__(						self							:					Dict	,     __lowercase							:					Union[str, Any]	,     __lowercase							:					Tuple	,     __lowercase							:					Any="<unk>"	,     **__lowercase							:					List[Any]						):






							"""simple docstring"""






							super().__init__(unk_token=__lowercase	,     **__lowercase						)

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

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

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






						@property
						def 	snake_case__						(						self							:					List[Any]						):






							"""simple docstring"""






							return True






						@property
						def 	snake_case__						(						self							:					Union[str, Any]						):






							"""simple docstring"""






							return len(self.encoder						)






						def 	snake_case__						(						self							:					str						):






							"""simple docstring"""






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






						def 	snake_case__						(						self							:					Optional[Any]	,     __lowercase							:					int						):






							"""simple docstring"""






							snake_case_		   =					tuple(token[:-1]						) + (token[-1] + "</w>",)
							if token in self.cache:
								return self.cache[token]
							snake_case_		   =					get_pairs(__lowercase						)

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

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

									if word[i] == first and i < len(__lowercase						) - 1 and word[i + 1] == second:
										new_word.append(first + second						)
										i += 2
									else:
										new_word.append(word[i]						)
										i += 1
								snake_case_		   =					tuple(__lowercase						)
								snake_case_		   =					new_word
								if len(__lowercase						) == 1:
									break
								else:
									snake_case_		   =					get_pairs(__lowercase						)
							snake_case_		   =					" ".join(__lowercase						)
							if word == "\n  </w>":
								snake_case_		   =					"\n</w>"
							snake_case_		   =					word
							return word






						def 	snake_case__						(						self							:					List[str]	,     __lowercase							:					Tuple						):






							"""simple docstring"""






							snake_case_		   =					[]
							if self.fix_text is None:
								# Using BERT's BasicTokenizer
								snake_case_		   =					self.nlp.tokenize(__lowercase						)
								for token in text:
									split_tokens.extend(list(self.bpe(__lowercase						).split(" "						)						)						)
							else:
								# Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
								snake_case_		   =					self.nlp(text_standardize(self.fix_text(__lowercase						)						)						)
								for token in text:
									split_tokens.extend(list(self.bpe(token.text.lower()						).split(" "						)						)						)
							return split_tokens






						def 	snake_case__						(						self							:					int	,     __lowercase							:					Tuple						):






							"""simple docstring"""






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






						def 	snake_case__						(						self							:					Tuple	,     __lowercase							:					Any						):






							"""simple docstring"""






							return self.decoder.get(__lowercase	,     self.unk_token						)






						def 	snake_case__						(						self							:					str	,     __lowercase							:					List[Any]						):






							"""simple docstring"""






							snake_case_		   =					"".join(__lowercase						).replace("</w>"	,     " "						).strip()
							return out_string






						def 	snake_case__						(						self							:					int	,     __lowercase							:					str	,     __lowercase							:					Optional[str] = None						):






							"""simple docstring"""






							if not os.path.isdir(__lowercase						):
								logger.error(f"Vocabulary path ({save_directory}) should be a directory"						)
								return
							snake_case_		   =					os.path.join(
							    __lowercase	,     (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]						)
							snake_case_		   =					os.path.join(
							    __lowercase	,     (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]						)

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

							snake_case_		   =					0
							with open(__lowercase	,     "w"	,     encoding="utf-8"						) as writer:
								writer.write("#version: 0.2\n"						)
								for bpe_tokens, token_index in sorted(self.bpe_ranks.items()	,     key=lambda __lowercase						: kv[1]						):
									if index != token_index:
										logger.warning(
										    f"Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."
										    " Please check that the tokenizer is not corrupted!"						)
										snake_case_		   =					token_index
									writer.write(" ".join(__lowercase						) + "\n"						)
									index += 1

							return vocab_file, merge_file
