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

import regex as re

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


if TYPE_CHECKING:
							from transformers.pipelines.conversational import Conversation

_lowerCamelCase    :      Any	 =						logging.get_logger(__name__)

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

_lowerCamelCase    :      Any	 =						{
    """vocab_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/vocab.json""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/vocab.json""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/vocab.json""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/vocab.json""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/vocab.json""",
    },
    """merges_file""": {
        """gpt2""": """https://huggingface.co/gpt2/resolve/main/merges.txt""",
        """gpt2-medium""": """https://huggingface.co/gpt2-medium/resolve/main/merges.txt""",
        """gpt2-large""": """https://huggingface.co/gpt2-large/resolve/main/merges.txt""",
        """gpt2-xl""": """https://huggingface.co/gpt2-xl/resolve/main/merges.txt""",
        """distilgpt2""": """https://huggingface.co/distilgpt2/resolve/main/merges.txt""",
    },
}

_lowerCamelCase    :      str	 =						{
    """gpt2""": 10_24,
    """gpt2-medium""": 10_24,
    """gpt2-large""": 10_24,
    """gpt2-xl""": 10_24,
    """distilgpt2""": 10_24,
}





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





def 		__lowerCamelCase							(UpperCAmelCase__   :							Dict     ):
					SCREAMING_SNAKE_CASE				=				set()
					SCREAMING_SNAKE_CASE				=				word[0]
					for char in word[1:]:
										pairs.add((prev_char, char)     )
										SCREAMING_SNAKE_CASE				=				char
					return pairs


class   lowercase			(   __A   ):
							lowercase__  :				Optional[int]									=							VOCAB_FILES_NAMES
							lowercase__  :				int									=							PRETRAINED_VOCAB_FILES_MAP
							lowercase__  :				int									=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							lowercase__  :				Dict									=							["""input_ids""", """attention_mask"""]




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



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				AddedToken(lowerCamelCase__		,							lstrip=lowerCamelCase__		,							rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__		,							lowerCamelCase__					) else bos_token
												SCREAMING_SNAKE_CASE				=				AddedToken(lowerCamelCase__		,							lstrip=lowerCamelCase__		,							rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__		,							lowerCamelCase__					) else eos_token
												SCREAMING_SNAKE_CASE				=				AddedToken(lowerCamelCase__		,							lstrip=lowerCamelCase__		,							rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__		,							lowerCamelCase__					) else unk_token
												SCREAMING_SNAKE_CASE				=				AddedToken(lowerCamelCase__		,							lstrip=lowerCamelCase__		,							rstrip=lowerCamelCase__					) if isinstance(lowerCamelCase__		,							lowerCamelCase__					) else pad_token
												super().__init__(
												    errors=lowerCamelCase__		,							unk_token=lowerCamelCase__		,							bos_token=lowerCamelCase__		,							eos_token=lowerCamelCase__		,							pad_token=lowerCamelCase__		,							add_prefix_space=lowerCamelCase__		,							add_bos_token=lowerCamelCase__		,							**lowerCamelCase__		,							)
												SCREAMING_SNAKE_CASE				=				add_bos_token

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

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




							@property
							def 		__snake_case(  self      :    Union[str, Any]					)    -> Dict:



												'''simple docstring'''

												return len(self.encoder					)




							def 		__snake_case(  self      :    List[Any]					)    -> Dict:



												'''simple docstring'''

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




							def 		__snake_case(  self      :    Dict		,							_UpperCamelCase      :    Any					)    -> Dict:



												'''simple docstring'''

												if token in self.cache:
																	return self.cache[token]
												SCREAMING_SNAKE_CASE				=				tuple(lowerCamelCase__					)
												SCREAMING_SNAKE_CASE				=				get_pairs(lowerCamelCase__					)

												if not pairs:
																	return token

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

																						if word[i] == first and i < len(lowerCamelCase__					) - 1 and word[i + 1] == second:
																											new_word.append(first + second					)
																											i += 2
																						else:
																											new_word.append(word[i]					)
																											i += 1
																	SCREAMING_SNAKE_CASE				=				tuple(lowerCamelCase__					)
																	SCREAMING_SNAKE_CASE				=				new_word
																	if len(lowerCamelCase__					) == 1:
																						break
																	else:
																						SCREAMING_SNAKE_CASE				=				get_pairs(lowerCamelCase__					)
												SCREAMING_SNAKE_CASE				=				" ".join(lowerCamelCase__					)
												SCREAMING_SNAKE_CASE				=				word
												return word




							def 		__snake_case(  self      :    Optional[int]		,							_UpperCamelCase      :    Dict		,							_UpperCamelCase      :    str=None					)    -> List[Any]:



												'''simple docstring'''

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

												SCREAMING_SNAKE_CASE				=				bos_token_ids + token_ids_a

												if token_ids_a is None:
																	return output

												return output + bos_token_ids + token_ids_a




							def 		__snake_case(  self      :    int		,							_UpperCamelCase      :    Union[str, Any]		,							_UpperCamelCase      :    Optional[Any] = None		,							_UpperCamelCase      :    Union[str, Any] = False					)    -> List[int]:



												'''simple docstring'''

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

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

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




							def 		__snake_case(  self      :    str		,							_UpperCamelCase      :    List[str]					)    -> str:



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				[]
												for token in re.findall(self.pat		,							lowerCamelCase__					):
																	SCREAMING_SNAKE_CASE				=				"".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(lowerCamelCase__					).split(" "					)					)
												return bpe_tokens




							def 		__snake_case(  self      :    List[str]		,							_UpperCamelCase      :    Tuple					)    -> Tuple:



												'''simple docstring'''

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




							def 		__snake_case(  self      :    Any		,							_UpperCamelCase      :    Any					)    -> Tuple:



												'''simple docstring'''

												return self.decoder.get(lowerCamelCase__					)




							def 		__snake_case(  self      :    str		,							_UpperCamelCase      :    Any					)    -> List[str]:



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				"".join(lowerCamelCase__					)
												SCREAMING_SNAKE_CASE				=				bytearray([self.byte_decoder[c] for c in text]					).decode("utf-8"		,							errors=self.errors					)
												return text




							def 		__snake_case(  self      :    Optional[Any]		,							_UpperCamelCase      :    Optional[Any]		,							_UpperCamelCase      :    List[Any] = None					)    -> Tuple[str]:



												'''simple docstring'''

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

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

												SCREAMING_SNAKE_CASE				=				0
												with open(lowerCamelCase__		,							"w"		,							encoding="utf-8"					) as writer:
																	writer.write("#version: 0.2\n"					)
																	for bpe_tokens, token_index in sorted(self.bpe_ranks.items()		,							key=lambda _UpperCamelCase					: 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!"					)
																											SCREAMING_SNAKE_CASE				=				token_index
																						writer.write(" ".join(lowerCamelCase__					) + "\n"					)
																						index += 1

												return vocab_file, merge_file




							def 		__snake_case(  self      :    int		,							_UpperCamelCase      :    Tuple		,							_UpperCamelCase      :    Optional[int]=False		,							**_UpperCamelCase      :    int					)    -> Optional[int]:



												'''simple docstring'''

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





							def 		__snake_case(  self      :    List[str]		,							_UpperCamelCase      :    Any					)    -> List[int]:



												'''simple docstring'''

												SCREAMING_SNAKE_CASE				=				[]
												for is_user, text in conversation.iter_texts():
																	input_ids.extend(self.encode(lowerCamelCase__		,							add_special_tokens=lowerCamelCase__					) + [self.eos_token_id]					)
												if len(lowerCamelCase__					) > self.model_max_length:
																	SCREAMING_SNAKE_CASE				=				input_ids[-self.model_max_length :]
												return input_ids










'''simple docstring'''
import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class        SCREAMING_SNAKE_CASE(				__A   ,      __A   ,      unittest.TestCase     ):
       snake_case_ :     str    =					StableDiffusionParadigmsPipeline
       snake_case_ :     Optional[Any]    =					TEXT_TO_IMAGE_PARAMS
       snake_case_ :     Tuple    =					TEXT_TO_IMAGE_BATCH_PARAMS
       snake_case_ :     Optional[Any]    =					TEXT_TO_IMAGE_IMAGE_PARAMS
       snake_case_ :     str    =					TEXT_TO_IMAGE_IMAGE_PARAMS

       def     snake_case__			(    self     )    ->					int:

           """simple docstring"""



           torch.manual_seed(0     )
           __lowercase							=						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					,		attention_head_dim=(2, 4)					,		use_linear_projection=lowerCamelCase__					,		)
           __lowercase							=						DDIMParallelScheduler(
               beta_start=0.0_00_85					,		beta_end=0.0_12					,		beta_schedule="""scaled_linear"""					,		clip_sample=lowerCamelCase__					,		set_alpha_to_one=lowerCamelCase__					,		)
           torch.manual_seed(0     )
           __lowercase							=						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					,		sample_size=128					,		)
           torch.manual_seed(0     )
           __lowercase							=						CLIPTextConfig(
               bos_token_id=0					,		eos_token_id=2					,		hidden_size=32					,		intermediate_size=37					,		layer_norm_eps=1E-0_5					,		num_attention_heads=4					,		num_hidden_layers=5					,		pad_token_id=1					,		vocab_size=1000					,		hidden_act="""gelu"""					,		projection_dim=512					,		)
           __lowercase							=						CLIPTextModel(lowerCamelCase__     )
           __lowercase							=						CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     )

           __lowercase							=						{
               """unet""": unet,
               """scheduler""": scheduler,
               """vae""": vae,
               """text_encoder""": text_encoder,
               """tokenizer""": tokenizer,
               """safety_checker""": None,
               """feature_extractor""": None,
           }
           return components

       def     snake_case__			(    self					,		lowerCamelCase__					,		lowerCamelCase__=0     )    ->					Any:

           """simple docstring"""



           if str(lowerCamelCase__     ).startswith("""mps"""     ):
               __lowercase							=						torch.manual_seed(lowerCamelCase__     )
           else:
               __lowercase							=						torch.Generator(device=lowerCamelCase__     ).manual_seed(lowerCamelCase__     )
           __lowercase							=						{
               """prompt""": """a photograph of an astronaut riding a horse""",
               """generator""": generator,
               """num_inference_steps""": 10,
               """guidance_scale""": 6.0,
               """output_type""": """numpy""",
               """parallel""": 3,
               """debug""": True,
           }
           return inputs

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

           """simple docstring"""



           __lowercase							=						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
           __lowercase							=						self.get_dummy_components()
           __lowercase							=						StableDiffusionParadigmsPipeline(**lowerCamelCase__     )
           __lowercase							=						sd_pipe.to(lowerCamelCase__     )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase__     )

           __lowercase							=						self.get_dummy_inputs(lowerCamelCase__     )
           __lowercase							=						sd_pipe(**lowerCamelCase__     ).images
           __lowercase							=						image[0, -3:, -3:, -1]
           assert image.shape == (1, 64, 64, 3)

           __lowercase							=						np.array([0.47_73, 0.54_17, 0.47_23, 0.49_25, 0.56_31, 0.47_52, 0.52_40, 0.49_35, 0.50_23]     )

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

       def     snake_case__			(    self     )    ->					str:

           """simple docstring"""



           __lowercase							=						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
           __lowercase							=						self.get_dummy_components()
           torch.manual_seed(0     )
           __lowercase							=						DDPMParallelScheduler()
           torch.manual_seed(0     )
           __lowercase							=						StableDiffusionParadigmsPipeline(**lowerCamelCase__     )
           __lowercase							=						sd_pipe.to(lowerCamelCase__     )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase__     )

           __lowercase							=						self.get_dummy_inputs(lowerCamelCase__     )
           __lowercase							=						sd_pipe(**lowerCamelCase__     ).images
           __lowercase							=						image[0, -3:, -3:, -1]
           assert image.shape == (1, 64, 64, 3)

           __lowercase							=						np.array([0.35_73, 0.44_20, 0.49_60, 0.47_99, 0.37_96, 0.38_79, 0.48_19, 0.43_65, 0.44_68]     )

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

       def     snake_case__			(    self     )    ->					Any:

           """simple docstring"""



           super().test_inference_batch_consistent(batch_sizes=[1, 2]     )

       def     snake_case__			(    self     )    ->					Optional[Any]:

           """simple docstring"""



           super().test_inference_batch_single_identical(batch_size=2					,		expected_max_diff=3E-3     )




       def     snake_case__			(    self     )    ->					str:

           """simple docstring"""



           __lowercase							=						"""cpu"""  # ensure determinism for the device-dependent torch.Generator
           __lowercase							=						self.get_dummy_components()
           __lowercase							=						StableDiffusionParadigmsPipeline(**lowerCamelCase__     )
           __lowercase							=						sd_pipe.to(lowerCamelCase__     )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase__     )

           __lowercase							=						self.get_dummy_inputs(lowerCamelCase__     )
           __lowercase							=						"""french fries"""
           __lowercase							=						sd_pipe(**lowerCamelCase__					,		negative_prompt=lowerCamelCase__     )
           __lowercase							=						output.images
           __lowercase							=						image[0, -3:, -3:, -1]

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

           __lowercase							=						np.array([0.47_71, 0.54_20, 0.46_83, 0.49_18, 0.56_36, 0.47_25, 0.52_30, 0.49_23, 0.50_15]     )

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






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

       def     snake_case__			(    self     )    ->					str:

           """simple docstring"""



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

       def     snake_case__			(    self					,		lowerCamelCase__=0     )    ->					int:

           """simple docstring"""



           __lowercase							=						torch.Generator(device=lowerCamelCase__     ).manual_seed(lowerCamelCase__     )
           __lowercase							=						{
               """prompt""": """a photograph of an astronaut riding a horse""",
               """generator""": generator,
               """num_inference_steps""": 10,
               """guidance_scale""": 7.5,
               """output_type""": """numpy""",
               """parallel""": 3,
               """debug""": True,
           }
           return inputs




       def     snake_case__			(    self     )    ->					Optional[Any]:

           """simple docstring"""



           __lowercase							=						"""stabilityai/stable-diffusion-2-base"""
           __lowercase							=						DDIMParallelScheduler.from_pretrained(lowerCamelCase__					,		subfolder="""scheduler"""     )
           __lowercase							=						StableDiffusionParadigmsPipeline.from_pretrained(lowerCamelCase__					,		scheduler=lowerCamelCase__					,		safety_checker=lowerCamelCase__     )
           pipe.to(lowerCamelCase__     )
           pipe.set_progress_bar_config(disable=lowerCamelCase__     )
           pipe.enable_attention_slicing()

           __lowercase							=						self.get_inputs()
           __lowercase							=						pipe(**lowerCamelCase__     ).images
           __lowercase							=						image[0, -3:, -3:, -1].flatten()

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

           __lowercase							=						np.array([0.96_22, 0.96_02, 0.97_48, 0.95_91, 0.96_30, 0.96_91, 0.96_61, 0.96_31, 0.97_41]     )

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




