






"""simple docstring"""
import json
import os
import re
import unicodedata
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase__							= logging.get_logger(__name__)

lowerCAmelCase__							= {
    'src_vocab_file': 'vocab-src.json',
    'tgt_vocab_file': 'vocab-tgt.json',
    'merges_file': 'merges.txt',
}

lowerCAmelCase__							= {
    'src_vocab_file': {
        'stas/tiny-wmt19-en-de': 'https://huggingface.co/stas/tiny-wmt19-en-de/resolve/main/vocab-src.json'
    },
    'tgt_vocab_file': {
        'stas/tiny-wmt19-en-de': 'https://huggingface.co/stas/tiny-wmt19-en-de/resolve/main/vocab-tgt.json'
    },
    'merges_file': {'stas/tiny-wmt19-en-de': 'https://huggingface.co/stas/tiny-wmt19-en-de/resolve/main/merges.txt'},
}

lowerCAmelCase__							= {'stas/tiny-wmt19-en-de': 1_024}
lowerCAmelCase__							= {
    'stas/tiny-wmt19-en-de': {
        'langs': ['en', 'de'],
        'model_max_length': 1_024,
        'special_tokens_map_file': None,
        'full_tokenizer_file': None,
    }
}





def 							lowercase__		(				lowerCamelCase					):
				_SCREAMING_SNAKE_CASE      :		List[str]							=						set()
				_SCREAMING_SNAKE_CASE      :		List[str]							=						word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)					)
								_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						char
				return pairs





def 							lowercase__		(				lowerCamelCase					):
				_SCREAMING_SNAKE_CASE      :		Optional[int]							=						text.replace('，',	','					)
				_SCREAMING_SNAKE_CASE      :		Any							=						re.sub(R'。\s*',	'. ',	lowerCamelCase					)
				_SCREAMING_SNAKE_CASE      :		str							=						text.replace('、',	','					)
				_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						text.replace('”',	'"'					)
				_SCREAMING_SNAKE_CASE      :		int							=						text.replace('“',	'"'					)
				_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						text.replace('∶',	':'					)
				_SCREAMING_SNAKE_CASE      :		str							=						text.replace('：',	':'					)
				_SCREAMING_SNAKE_CASE      :		Optional[int]							=						text.replace('？',	'?'					)
				_SCREAMING_SNAKE_CASE      :		Any							=						text.replace('《',	'"'					)
				_SCREAMING_SNAKE_CASE      :		List[Any]							=						text.replace('》',	'"'					)
				_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						text.replace('）',	')'					)
				_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						text.replace('！',	'!'					)
				_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						text.replace('（',	'('					)
				_SCREAMING_SNAKE_CASE      :		Optional[int]							=						text.replace('；',	';'					)
				_SCREAMING_SNAKE_CASE      :		Dict							=						text.replace('１',	'1'					)
				_SCREAMING_SNAKE_CASE      :		List[str]							=						text.replace('」',	'"'					)
				_SCREAMING_SNAKE_CASE      :		Dict							=						text.replace('「',	'"'					)
				_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						text.replace('０',	'0'					)
				_SCREAMING_SNAKE_CASE      :		Tuple							=						text.replace('３',	'3'					)
				_SCREAMING_SNAKE_CASE      :		List[str]							=						text.replace('２',	'2'					)
				_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						text.replace('５',	'5'					)
				_SCREAMING_SNAKE_CASE      :		Tuple							=						text.replace('６',	'6'					)
				_SCREAMING_SNAKE_CASE      :		List[str]							=						text.replace('９',	'9'					)
				_SCREAMING_SNAKE_CASE      :		Optional[int]							=						text.replace('７',	'7'					)
				_SCREAMING_SNAKE_CASE      :		Tuple							=						text.replace('８',	'8'					)
				_SCREAMING_SNAKE_CASE      :		int							=						text.replace('４',	'4'					)
				_SCREAMING_SNAKE_CASE      :		List[Any]							=						re.sub(R'．\s*',	'. ',	lowerCamelCase					)
				_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						text.replace('～',	'~'					)
				_SCREAMING_SNAKE_CASE      :		str							=						text.replace('’',	'\''					)
				_SCREAMING_SNAKE_CASE      :		Dict							=						text.replace('…',	'...'					)
				_SCREAMING_SNAKE_CASE      :		Any							=						text.replace('━',	'-'					)
				_SCREAMING_SNAKE_CASE      :		int							=						text.replace('〈',	'<'					)
				_SCREAMING_SNAKE_CASE      :		List[str]							=						text.replace('〉',	'>'					)
				_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						text.replace('【',	'['					)
				_SCREAMING_SNAKE_CASE      :		List[str]							=						text.replace('】',	']'					)
				_SCREAMING_SNAKE_CASE      :		Dict							=						text.replace('％',	'%'					)
				return text





def 							lowercase__		(				lowerCamelCase					):
				_SCREAMING_SNAKE_CASE      :		Dict							=						[]
				for char in text:
								_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						unicodedata.category(lowerCamelCase					)
								if cat.startswith('C'					):
												continue
								output.append(lowerCamelCase					)
				return "".join(lowerCamelCase					)






class _lowerCAmelCase	(						__UpperCAmelCase					):
						SCREAMING_SNAKE_CASE_:				List[str]   				=							VOCAB_FILES_NAMES
						SCREAMING_SNAKE_CASE_:				List[str]   				=							PRETRAINED_VOCAB_FILES_MAP
						SCREAMING_SNAKE_CASE_:				List[Any]   				=							PRETRAINED_INIT_CONFIGURATION
						SCREAMING_SNAKE_CASE_:				Optional[Any]   				=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						SCREAMING_SNAKE_CASE_:				Any   				=							['input_ids', 'attention_mask']







						def __init__(						self      ,							lowerCAmelCase_=None      ,							lowerCAmelCase_=None      ,							lowerCAmelCase_=None      ,							lowerCAmelCase_=None      ,							lowerCAmelCase_=False      ,							lowerCAmelCase_="<unk>"      ,							lowerCAmelCase_="<s>"      ,							lowerCAmelCase_="</s>"      ,							lowerCAmelCase_="<pad>"      ,							**lowerCAmelCase_      ,							)		->       str:
										super().__init__(
										    langs=lowerCAmelCase_      ,							src_vocab_file=lowerCAmelCase_      ,							tgt_vocab_file=lowerCAmelCase_      ,							merges_file=lowerCAmelCase_      ,							do_lower_case=lowerCAmelCase_      ,							unk_token=lowerCAmelCase_      ,							bos_token=lowerCAmelCase_      ,							sep_token=lowerCAmelCase_      ,							pad_token=lowerCAmelCase_      ,							**lowerCAmelCase_      ,							)

										try:
														import sacremoses
										except ImportError:
														raise ImportError(
														    'You need to install sacremoses to use XLMTokenizer. '
														    'See https://pypi.org/project/sacremoses/ for installation.'       )

										_SCREAMING_SNAKE_CASE      :		Any							=						sacremoses

										_SCREAMING_SNAKE_CASE      :		Dict							=						src_vocab_file
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						tgt_vocab_file
										_SCREAMING_SNAKE_CASE      :		Tuple							=						merges_file
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						do_lower_case

										# cache of sm.MosesPunctNormalizer instance
										_SCREAMING_SNAKE_CASE      :		Tuple							=						{}
										# cache of sm.MosesTokenizer instance
										_SCREAMING_SNAKE_CASE      :		str							=						{}
										_SCREAMING_SNAKE_CASE      :		int							=						{}

										if langs and len(lowerCAmelCase_       ) == 2:
														_SCREAMING_SNAKE_CASE		,	_SCREAMING_SNAKE_CASE      :		Dict							=						langs
										else:
														raise ValueError(
														    F"""arg `langs` needs to be a list of 2 langs, e.g. ['en', 'ru'], but got {langs}. """
														    'Usually that means that tokenizer can\'t find a mapping for the given model path '
														    'in PRETRAINED_VOCAB_FILES_MAP, and other maps of this tokenizer.'       )

										with open(lowerCAmelCase_      ,							encoding='utf-8'       ) as src_vocab_handle:
														_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						json.load(lowerCAmelCase_       )
										with open(lowerCAmelCase_      ,							encoding='utf-8'       ) as tgt_vocab_handle:
														_SCREAMING_SNAKE_CASE      :		str							=						json.load(lowerCAmelCase_       )
														_SCREAMING_SNAKE_CASE      :		Any							=						{v: k for k, v in tgt_vocab.items()}
										with open(lowerCAmelCase_      ,							encoding='utf-8'       ) as merges_handle:
														_SCREAMING_SNAKE_CASE      :		str							=						merges_handle.read().split('\n'       )[:-1]
										_SCREAMING_SNAKE_CASE      :		int							=						[tuple(merge.split()[:2]       ) for merge in merges]
										_SCREAMING_SNAKE_CASE      :		Any							=						dict(zip(lowerCAmelCase_      ,							range(len(lowerCAmelCase_       )       )       )       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						{}







						def  A			(						self       )		->       Dict[str, int]:
										return self.get_src_vocab()







						@property
						def  A			(						self       )		->       int:
										return self.src_vocab_size







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_       )		->       Optional[int]:
										if lang not in self.cache_moses_punct_normalizer:
														_SCREAMING_SNAKE_CASE      :		Dict							=						self.sm.MosesPunctNormalizer(lang=lowerCAmelCase_       )
														_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						punct_normalizer
										return self.cache_moses_punct_normalizer[lang].normalize(lowerCAmelCase_       )







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_       )		->       Dict:
										if lang not in self.cache_moses_tokenizer:
														_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						self.sm.MosesTokenizer(lang=lowerCAmelCase_       )
														_SCREAMING_SNAKE_CASE      :		int							=						moses_tokenizer
										return self.cache_moses_tokenizer[lang].tokenize(
										    lowerCAmelCase_      ,							aggressive_dash_splits=lowerCAmelCase_      ,							return_str=lowerCAmelCase_      ,							escape=lowerCAmelCase_       )







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_       )		->       Tuple:
										if lang not in self.cache_moses_detokenizer:
														_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.sm.MosesDetokenizer(lang=lowerCAmelCase_       )
														_SCREAMING_SNAKE_CASE      :		Dict							=						moses_detokenizer
										return self.cache_moses_detokenizer[lang].detokenize(lowerCAmelCase_       )







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_       )		->       Dict:
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						replace_unicode_punct(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		int							=						self.moses_punct_norm(lowerCAmelCase_      ,							lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		str							=						remove_non_printing_char(lowerCAmelCase_       )
										return text







						@property
						def  A			(						self       )		->       Optional[int]:
										return len(self.encoder       )







						@property
						def  A			(						self       )		->       str:
										return len(self.decoder       )







						def  A			(						self       )		->       Optional[int]:
										return dict(self.encoder      ,							**self.added_tokens_encoder       )







						def  A			(						self       )		->       Tuple:
										return dict(self.decoder      ,							**self.added_tokens_decoder       )







						def  A			(						self      ,							lowerCAmelCase_       )		->       List[Any]:
										_SCREAMING_SNAKE_CASE      :		Dict							=						tuple(token[:-1]       ) + (token[-1] + '</w>',)
										if token in self.cache:
														return self.cache[token]
										_SCREAMING_SNAKE_CASE      :		Dict							=						get_pairs(lowerCAmelCase_       )

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

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

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







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_="en"      ,							lowerCAmelCase_=False       )		->       Union[str, Any]:
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						self.src_lang

										if self.do_lower_case:
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						text.lower()

										if bypass_tokenizer:
														_SCREAMING_SNAKE_CASE      :		str							=						text.split()
										else:
														_SCREAMING_SNAKE_CASE      :		str							=						self.moses_pipeline(lowerCAmelCase_      ,							lang=lowerCAmelCase_       )
														_SCREAMING_SNAKE_CASE      :		Dict							=						self.moses_tokenize(lowerCAmelCase_      ,							lang=lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						[]
										for token in text:
														if token:
																		split_tokens.extend(list(self.bpe(lowerCAmelCase_       ).split(' '       )       )       )

										return split_tokens







						def  A			(						self      ,							lowerCAmelCase_       )		->       Union[str, Any]:
										return self.encoder.get(lowerCAmelCase_      ,							self.encoder.get(self.unk_token       )       )







						def  A			(						self      ,							lowerCAmelCase_       )		->       Optional[int]:
										return self.decoder.get(lowerCAmelCase_      ,							self.unk_token       )







						def  A			(						self      ,							lowerCAmelCase_       )		->       Any:
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						[t.replace(' '      ,							''       ).replace('</w>'      ,							' '       ) for t in tokens]
										_SCREAMING_SNAKE_CASE      :		Dict							=						''.join(lowerCAmelCase_       ).split()
										# detokenize
										_SCREAMING_SNAKE_CASE      :		str							=						self.moses_detokenize(lowerCAmelCase_      ,							self.tgt_lang       )
										return text







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = None       )		->       List[int]:
										_SCREAMING_SNAKE_CASE      :		List[str]							=						[self.sep_token_id]

										# no bos used in fairseq
										if token_ids_a is None:
														return token_ids_a + sep
										return token_ids_a + sep + token_ids_a + sep







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = False       )		->       List[int]:

										if already_has_special_tokens:
														return super().get_special_tokens_mask(
														    token_ids_a=lowerCAmelCase_      ,							token_ids_a=lowerCAmelCase_      ,							already_has_special_tokens=lowerCAmelCase_       )
										# no bos used in fairseq
										if token_ids_a is not None:
														return ([0] * len(lowerCAmelCase_       )) + [1] + ([0] * len(lowerCAmelCase_       )) + [1]
										return ([0] * len(lowerCAmelCase_       )) + [1]







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = None       )		->       List[int]:
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						[self.sep_token_id]

										# no bos used in fairseq
										if token_ids_a is None:
														return len(token_ids_a + sep       ) * [0]
										return len(token_ids_a + sep       ) * [0] + len(token_ids_a + sep       ) * [1]







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = None       )		->       Tuple[str]:
										if not os.path.isdir(lowerCAmelCase_       ):
														logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
														return

										_SCREAMING_SNAKE_CASE      :		Any							=						os.path.join(
										    lowerCAmelCase_      ,							(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['src_vocab_file']       )
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						os.path.join(
										    lowerCAmelCase_      ,							(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['tgt_vocab_file']       )
										_SCREAMING_SNAKE_CASE      :		Any							=						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'       )

										with open(lowerCAmelCase_      ,							'w'      ,							encoding='utf-8'       ) as f:
														_SCREAMING_SNAKE_CASE      :		Optional[int]							=						{v: k for k, v in self.decoder.items()}
														f.write(json.dumps(lowerCAmelCase_      ,							indent=2      ,							sort_keys=lowerCAmelCase_      ,							ensure_ascii=lowerCAmelCase_       ) + '\n'       )

										_SCREAMING_SNAKE_CASE      :		List[str]							=						0
										with open(lowerCAmelCase_      ,							'w'      ,							encoding='utf-8'       ) as writer:
														for bpe_tokens, token_index in sorted(self.bpe_ranks.items()      ,							key=lambda lowerCAmelCase_       : kv[1]       ):
																		if index != token_index:
																						logger.warning(
																						    F"""Saving vocabulary to {merges_file}: BPE merge indices are not consecutive."""
																						    ' Please check that the tokenizer is not corrupted!'       )
																						_SCREAMING_SNAKE_CASE      :		List[Any]							=						token_index
																		writer.write(' '.join(lowerCAmelCase_       ) + '\n'       )
																		index += 1

										return src_vocab_file, tgt_vocab_file, merges_file







						def __getstate__(						self       )		->       Optional[int]:
										_SCREAMING_SNAKE_CASE      :		List[str]							=						self.__dict__.copy()
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						None
										return state







						def __setstate__(						self      ,							lowerCAmelCase_       )		->       Any:
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						d

										try:
														import sacremoses
										except ImportError:
														raise ImportError(
														    'You need to install sacremoses to use XLMTokenizer. '
														    'See https://pypi.org/project/sacremoses/ for installation.'       )

										_SCREAMING_SNAKE_CASE      :		List[str]							=						sacremoses









"""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 _lowerCAmelCase	(						__UpperCAmelCase      ,     __UpperCAmelCase      ,     unittest.TestCase					):
						SCREAMING_SNAKE_CASE_:				int   				=							StableDiffusionParadigmsPipeline
						SCREAMING_SNAKE_CASE_:				str   				=							TEXT_TO_IMAGE_PARAMS
						SCREAMING_SNAKE_CASE_:				Any   				=							TEXT_TO_IMAGE_BATCH_PARAMS
						SCREAMING_SNAKE_CASE_:				str   				=							TEXT_TO_IMAGE_IMAGE_PARAMS
						SCREAMING_SNAKE_CASE_:				List[Any]   				=							TEXT_TO_IMAGE_IMAGE_PARAMS







						def  A			(						self       )		->       int:
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		Dict							=						UNetaDConditionModel(
										    block_out_channels=(3_2, 6_4)      ,							layers_per_block=2      ,							sample_size=3_2      ,							in_channels=4      ,							out_channels=4      ,							down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')      ,							up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')      ,							cross_attention_dim=3_2      ,							attention_head_dim=(2, 4)      ,							use_linear_projection=lowerCAmelCase_      ,							)
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						DDIMParallelScheduler(
										    beta_start=0.00_085      ,							beta_end=0.012      ,							beta_schedule='scaled_linear'      ,							clip_sample=lowerCAmelCase_      ,							set_alpha_to_one=lowerCAmelCase_      ,							)
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						AutoencoderKL(
										    block_out_channels=[3_2, 6_4]      ,							in_channels=3      ,							out_channels=3      ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']      ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']      ,							latent_channels=4      ,							sample_size=1_2_8      ,							)
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						CLIPTextConfig(
										    bos_token_id=0      ,							eos_token_id=2      ,							hidden_size=3_2      ,							intermediate_size=3_7      ,							layer_norm_eps=1e-05      ,							num_attention_heads=4      ,							num_hidden_layers=5      ,							pad_token_id=1      ,							vocab_size=1_0_0_0      ,							hidden_act='gelu'      ,							projection_dim=5_1_2      ,							)
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						CLIPTextModel(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       )

										_SCREAMING_SNAKE_CASE      :		List[Any]							=						{
										    'unet': unet,
										    'scheduler': scheduler,
										    'vae': vae,
										    'text_encoder': text_encoder,
										    'tokenizer': tokenizer,
										    'safety_checker': None,
										    'feature_extractor': None,
										}
										return components







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_=0       )		->       Tuple:
										if str(lowerCAmelCase_       ).startswith('mps'       ):
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						torch.manual_seed(lowerCAmelCase_       )
										else:
														_SCREAMING_SNAKE_CASE      :		int							=						torch.Generator(device=lowerCAmelCase_       ).manual_seed(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		str							=						{
										    'prompt': 'a photograph of an astronaut riding a horse',
										    'generator': generator,
										    'num_inference_steps': 1_0,
										    'guidance_scale': 6.0,
										    'output_type': 'numpy',
										    'parallel': 3,
										    'debug': True,
										}
										return inputs







						def  A			(						self       )		->       int:
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.get_dummy_components()
										_SCREAMING_SNAKE_CASE      :		Any							=						StableDiffusionParadigmsPipeline(**lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						sd_pipe.to(lowerCAmelCase_       )
										sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.get_dummy_inputs(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Dict							=						sd_pipe(**lowerCAmelCase_       ).images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1]
										assert image.shape == (1, 6_4, 6_4, 3)

										_SCREAMING_SNAKE_CASE      :		int							=						np.array([0.4_773, 0.5_417, 0.4_723, 0.4_925, 0.5_631, 0.4_752, 0.5_240, 0.4_935, 0.5_023]       )

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







						def  A			(						self       )		->       Tuple:
										_SCREAMING_SNAKE_CASE      :		Tuple							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
										_SCREAMING_SNAKE_CASE      :		str							=						self.get_dummy_components()
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						DDPMParallelScheduler()
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						StableDiffusionParadigmsPipeline(**lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		int							=						sd_pipe.to(lowerCAmelCase_       )
										sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		Tuple							=						self.get_dummy_inputs(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						sd_pipe(**lowerCAmelCase_       ).images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1]
										assert image.shape == (1, 6_4, 6_4, 3)

										_SCREAMING_SNAKE_CASE      :		Dict							=						np.array([0.3_573, 0.4_420, 0.4_960, 0.4_799, 0.3_796, 0.3_879, 0.4_819, 0.4_365, 0.4_468]       )

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







						def  A			(						self       )		->       Optional[Any]:
										super().test_inference_batch_consistent(batch_sizes=[1, 2]       )







						def  A			(						self       )		->       Tuple:
										super().test_inference_batch_single_identical(batch_size=2      ,							expected_max_diff=3e-3       )







						def  A			(						self       )		->       List[str]:
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						self.get_dummy_components()
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						StableDiffusionParadigmsPipeline(**lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						sd_pipe.to(lowerCAmelCase_       )
										sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		int							=						self.get_dummy_inputs(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Tuple							=						'french fries'
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						sd_pipe(**lowerCAmelCase_      ,							negative_prompt=lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						output.images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1]

										assert image.shape == (1, 6_4, 6_4, 3)

										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						np.array([0.4_771, 0.5_420, 0.4_683, 0.4_918, 0.5_636, 0.4_725, 0.5_230, 0.4_923, 0.5_015]       )

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







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







						def  A			(						self       )		->       Tuple:
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()







						def  A			(						self      ,							lowerCAmelCase_=0       )		->       int:
										_SCREAMING_SNAKE_CASE      :		str							=						torch.Generator(device=lowerCAmelCase_       ).manual_seed(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						{
										    'prompt': 'a photograph of an astronaut riding a horse',
										    'generator': generator,
										    'num_inference_steps': 1_0,
										    'guidance_scale': 7.5,
										    'output_type': 'numpy',
										    'parallel': 3,
										    'debug': True,
										}
										return inputs







						def  A			(						self       )		->       List[str]:
										_SCREAMING_SNAKE_CASE      :		Any							=						'stabilityai/stable-diffusion-2-base'
										_SCREAMING_SNAKE_CASE      :		Dict							=						DDIMParallelScheduler.from_pretrained(lowerCAmelCase_      ,							subfolder='scheduler'       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						StableDiffusionParadigmsPipeline.from_pretrained(lowerCAmelCase_      ,							scheduler=lowerCAmelCase_      ,							safety_checker=lowerCAmelCase_       )
										pipe.to(lowerCAmelCase_       )
										pipe.set_progress_bar_config(disable=lowerCAmelCase_       )
										pipe.enable_attention_slicing()

										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.get_inputs()
										_SCREAMING_SNAKE_CASE      :		str							=						pipe(**lowerCAmelCase_       ).images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1].flatten()

										assert image.shape == (1, 5_1_2, 5_1_2, 3)

										_SCREAMING_SNAKE_CASE      :		Any							=						np.array([0.9_622, 0.9_602, 0.9_748, 0.9_591, 0.9_630, 0.9_691, 0.9_661, 0.9_631, 0.9_741]       )

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

