
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   (						__SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    unittest.TestCase						):




	'''simple docstring'''





	snake_case_      					=  StableDiffusionParadigmsPipeline
	snake_case_      					=  TEXT_TO_IMAGE_PARAMS
	snake_case_      					=  TEXT_TO_IMAGE_BATCH_PARAMS
	snake_case_      					=  TEXT_TO_IMAGE_IMAGE_PARAMS
	snake_case_      					=  TEXT_TO_IMAGE_IMAGE_PARAMS






	def        UpperCamelCase_						( self      :				Tuple				):
					torch.manual_seed(0				)
					__A         =				UNetaDConditionModel(
					    block_out_channels=(32, 64)							,layers_per_block=2							,sample_size=32							,in_channels=4							,out_channels=4							,down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")							,up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")							,cross_attention_dim=32							,attention_head_dim=(2, 4)							,use_linear_projection=A							,)
					__A         =				DDIMParallelScheduler(
					    beta_start=0.0_00_85							,beta_end=0.0_12							,beta_schedule="scaled_linear"							,clip_sample=A							,set_alpha_to_one=A							,)
					torch.manual_seed(0				)
					__A         =				AutoencoderKL(
					    block_out_channels=[32, 64]							,in_channels=3							,out_channels=3							,down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]							,up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]							,latent_channels=4							,sample_size=1_28							,)
					torch.manual_seed(0				)
					__A         =				CLIPTextConfig(
					    bos_token_id=0							,eos_token_id=2							,hidden_size=32							,intermediate_size=37							,layer_norm_eps=1E-05							,num_attention_heads=4							,num_hidden_layers=5							,pad_token_id=1							,vocab_size=10_00							,hidden_act="gelu"							,projection_dim=5_12							,)
					__A         =				CLIPTextModel(A				)
					__A         =				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"				)

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






	def        UpperCamelCase_						( self      :				List[str]							,A      :				Dict							,A      :				List[Any]=0				):
					if str(A				).startswith("mps"				):
									__A         =				torch.manual_seed(A				)
					else:
									__A         =				torch.Generator(device=A				).manual_seed(A				)
					__A         =				{
					    "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        UpperCamelCase_						( self      :				Tuple				):
					__A         =				"cpu"  # ensure determinism for the device-dependent torch.Generator
					__A         =				self.get_dummy_components()
					__A         =				StableDiffusionParadigmsPipeline(**A				)
					__A         =				sd_pipe.to(A				)
					sd_pipe.set_progress_bar_config(disable=A				)

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

					__A         =				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        UpperCamelCase_						( self      :				Dict				):
					__A         =				"cpu"  # ensure determinism for the device-dependent torch.Generator
					__A         =				self.get_dummy_components()
					torch.manual_seed(0				)
					__A         =				DDPMParallelScheduler()
					torch.manual_seed(0				)
					__A         =				StableDiffusionParadigmsPipeline(**A				)
					__A         =				sd_pipe.to(A				)
					sd_pipe.set_progress_bar_config(disable=A				)

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

					__A         =				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        UpperCamelCase_						( self      :				Optional[Any]				):
					super().test_inference_batch_consistent(batch_sizes=[1, 2]				)






	def        UpperCamelCase_						( self      :				str				):
					super().test_inference_batch_single_identical(batch_size=2							,expected_max_diff=3E-3				)







	def        UpperCamelCase_						( self      :				Dict				):
					__A         =				"cpu"  # ensure determinism for the device-dependent torch.Generator
					__A         =				self.get_dummy_components()
					__A         =				StableDiffusionParadigmsPipeline(**A				)
					__A         =				sd_pipe.to(A				)
					sd_pipe.set_progress_bar_config(disable=A				)

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

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

					__A         =				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        UpperCamelCase_						( self      :				Tuple				):
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()






	def        UpperCamelCase_						( self      :				Optional[Any]							,A      :				Tuple=0				):
					__A         =				torch.Generator(device=A				).manual_seed(A				)
					__A         =				{
					    "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        UpperCamelCase_						( self      :				Dict				):
					__A         =				"stabilityai/stable-diffusion-2-base"
					__A         =				DDIMParallelScheduler.from_pretrained(A							,subfolder="scheduler"				)
					__A         =				StableDiffusionParadigmsPipeline.from_pretrained(A							,scheduler=A							,safety_checker=A				)
					pipe.to(A				)
					pipe.set_progress_bar_config(disable=A				)
					pipe.enable_attention_slicing()

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

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

					__A         =				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


from ...utils import is_transformers_available
from .pipeline_latent_diffusion_superresolution import LDMSuperResolutionPipeline


if is_transformers_available():
					from .pipeline_latent_diffusion import LDMBertModel, LDMTextToImagePipeline
