




'''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 			UpperCamelCase_   (a__,					a__,					unittest.TestCase   ):







						"""simple docstring"""


						_lowerCAmelCase			  =    StableDiffusionParadigmsPipeline
						_lowerCAmelCase			  =    TEXT_TO_IMAGE_PARAMS
						_lowerCAmelCase			  =    TEXT_TO_IMAGE_BATCH_PARAMS
						_lowerCAmelCase			  =    TEXT_TO_IMAGE_IMAGE_PARAMS
						_lowerCAmelCase			  =    TEXT_TO_IMAGE_IMAGE_PARAMS







						def 					_a			(   self					:				Any			):





											"""simple docstring"""







											torch.manual_seed(0			)
											A_       :   Any       =							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=lowercase__							,       )
											A_       :   Tuple       =							DDIMParallelScheduler(
											    beta_start=0.0_00_85							,       beta_end=0.0_12							,       beta_schedule='''scaled_linear'''							,       clip_sample=lowercase__							,       set_alpha_to_one=lowercase__							,       )
											torch.manual_seed(0			)
											A_       :   Optional[int]       =							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			)
											A_       :   str       =							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							,       hidden_act='''gelu'''							,       projection_dim=512							,       )
											A_       :   List[str]       =							CLIPTextModel(lowercase__			)
											A_       :   Optional[int]       =							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''			)

											A_       :   Optional[int]       =							{
											    '''unet''': unet,
											    '''scheduler''': scheduler,
											    '''vae''': vae,
											    '''text_encoder''': text_encoder,
											    '''tokenizer''': tokenizer,
											    '''safety_checker''': None,
											    '''feature_extractor''': None,
											}
											return components







						def 					_a			(   self					:				List[Any]							,       _lowerCamelCase					:				Any							,       _lowerCamelCase					:				Optional[Any]=0			):





											"""simple docstring"""







											if str(lowercase__			).startswith('''mps'''			):
																A_       :   Optional[int]       =							torch.manual_seed(lowercase__			)
											else:
																A_       :   int       =							torch.Generator(device=lowercase__			).manual_seed(lowercase__			)
											A_       :   Any       =							{
											    '''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 					_a			(   self					:				List[Any]			):





											"""simple docstring"""







											A_       :   List[str]       =							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											A_       :   Dict       =							self.get_dummy_components()
											A_       :   str       =							StableDiffusionParadigmsPipeline(**lowercase__			)
											A_       :   Union[str, Any]       =							sd_pipe.to(lowercase__			)
											sd_pipe.set_progress_bar_config(disable=lowercase__			)

											A_       :   int       =							self.get_dummy_inputs(lowercase__			)
											A_       :   List[Any]       =							sd_pipe(**lowercase__			).images
											A_       :   Optional[Any]       =							image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											A_       :   Optional[int]       =							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 					_a			(   self					:				int			):





											"""simple docstring"""







											A_       :   List[str]       =							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											A_       :   int       =							self.get_dummy_components()
											torch.manual_seed(0			)
											A_       :   int       =							DDPMParallelScheduler()
											torch.manual_seed(0			)
											A_       :   List[Any]       =							StableDiffusionParadigmsPipeline(**lowercase__			)
											A_       :   str       =							sd_pipe.to(lowercase__			)
											sd_pipe.set_progress_bar_config(disable=lowercase__			)

											A_       :   Dict       =							self.get_dummy_inputs(lowercase__			)
											A_       :   List[Any]       =							sd_pipe(**lowercase__			).images
											A_       :   List[str]       =							image[0, -3:, -3:, -1]
											assert image.shape == (1, 64, 64, 3)

											A_       :   int       =							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 					_a			(   self					:				Optional[int]			):





											"""simple docstring"""







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







						def 					_a			(   self					:				Optional[int]			):





											"""simple docstring"""







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







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





											"""simple docstring"""







											A_       :   List[str]       =							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											A_       :   Tuple       =							self.get_dummy_components()
											A_       :   Any       =							StableDiffusionParadigmsPipeline(**lowercase__			)
											A_       :   Tuple       =							sd_pipe.to(lowercase__			)
											sd_pipe.set_progress_bar_config(disable=lowercase__			)

											A_       :   str       =							self.get_dummy_inputs(lowercase__			)
											A_       :   Tuple       =							'''french fries'''
											A_       :   Union[str, Any]       =							sd_pipe(**lowercase__							,       negative_prompt=lowercase__			)
											A_       :   Tuple       =							output.images
											A_       :   Dict       =							image[0, -3:, -3:, -1]

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

											A_       :   Optional[Any]       =							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 			UpperCamelCase_   (unittest.TestCase   ):







						"""simple docstring"""


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





											"""simple docstring"""







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







						def 					_a			(   self					:				List[Any]							,       _lowerCamelCase					:				int=0			):





											"""simple docstring"""







											A_       :   Union[str, Any]       =							torch.Generator(device=lowercase__			).manual_seed(lowercase__			)
											A_       :   List[str]       =							{
											    '''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 					_a			(   self					:				Optional[Any]			):





											"""simple docstring"""







											A_       :   Union[str, Any]       =							'''stabilityai/stable-diffusion-2-base'''
											A_       :   Dict       =							DDIMParallelScheduler.from_pretrained(lowercase__							,       subfolder='''scheduler'''			)
											A_       :   List[str]       =							StableDiffusionParadigmsPipeline.from_pretrained(lowercase__							,       scheduler=lowercase__							,       safety_checker=lowercase__			)
											pipe.to(lowercase__			)
											pipe.set_progress_bar_config(disable=lowercase__			)
											pipe.enable_attention_slicing()

											A_       :   List[Any]       =							self.get_inputs()
											A_       :   int       =							pipe(**lowercase__			).images
											A_       :   int       =							image[0, -3:, -3:, -1].flatten()

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

											A_       :   Any       =							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





"""simple docstring"""




# Lint as: python3

import hashlib
from typing import Union

def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_			):
					if isinstance(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			):
										# Data already in bytes, returns as it as
										return hash_data
					elif isinstance(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			):
										# We keep the data as it as for it ot be later encoded to UTF-8
										# However replace `\\` with `/` for Windows compatibility
										SCREAMING_SNAKE_CASE				=	hash_data.replace('\\',     '/'			)
					elif isinstance(SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			):
										SCREAMING_SNAKE_CASE				=	str(SCREAMING_SNAKE_CASE_			)
					else:
										# If data is not of the required type, raise error
										raise InvalidKeyError(SCREAMING_SNAKE_CASE_			)

					return hash_data.encode('utf-8'			)




class      UpperCamelCase	(       __magic_name__      ):

				"""simple docstring"""



				def __init__(				self		,	lowercase__			)    ->    Dict:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	'\nFAILURE TO GENERATE DATASET: Invalid key type detected'
									SCREAMING_SNAKE_CASE				=	f'''\nFound Key {hash_data} of type {type(lowercase__			)}'''
									SCREAMING_SNAKE_CASE				=	'\nKeys should be either str, int or bytes type'
									super().__init__(f'''{self.prefix}{self.err_msg}{self.suffix}'''			)






class      UpperCamelCase	(       __magic_name__      ):

				"""simple docstring"""



				def __init__(				self		,	lowercase__		,	lowercase__		,	lowercase__=""			)    ->    Tuple:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	key
									SCREAMING_SNAKE_CASE				=	duplicate_key_indices
									SCREAMING_SNAKE_CASE				=	fix_msg
									SCREAMING_SNAKE_CASE				=	'Found multiple examples generated with the same key'
									if len(lowercase__			) <= 20:
														SCREAMING_SNAKE_CASE				=	f'''\nThe examples at index {', '.join(lowercase__			)} have the key {key}'''
									else:
														SCREAMING_SNAKE_CASE				=	f'''\nThe examples at index {', '.join(duplicate_key_indices[:20]			)}... ({len(lowercase__			) - 20} more) have the key {key}'''
									SCREAMING_SNAKE_CASE				=	'\n' + fix_msg if fix_msg else ''
									super().__init__(f'''{self.prefix}{self.err_msg}{self.suffix}'''			)






class      UpperCamelCase	:

				"""simple docstring"""



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







									SCREAMING_SNAKE_CASE				=	hashlib.mda(_as_bytes(lowercase__			)			)




				def 							A					(				self		,	lowercase__			)    ->    int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	self._split_mda.copy()
									SCREAMING_SNAKE_CASE				=	_as_bytes(lowercase__			)
									mda.update(lowercase__			)
									# Convert to integer with hexadecimal conversion
									return int(mda.hexdigest()		,	16			)
