


import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()




class 							lowerCamelCase   (  _UpperCamelCase							,						_UpperCamelCase							,						unittest.TestCase ):
	_lowerCAmelCase     :		Optional[Any]										=						StableDiffusionXLImgaImgPipeline
	_lowerCAmelCase     :		Optional[Any]										=						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
	_lowerCAmelCase     :		Union[str, Any]										=						PipelineTesterMixin.required_optional_params - {'''latents'''}
	_lowerCAmelCase     :		int										=						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	_lowerCAmelCase     :		Optional[int]										=						IMAGE_TO_IMAGE_IMAGE_PARAMS
	_lowerCAmelCase     :		List[Any]										=						IMAGE_TO_IMAGE_IMAGE_PARAMS


	def  A(    self):
							torch.manual_seed(0)
							__UpperCAmelCase			:							Optional[Any]	    =	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''')				,     attention_head_dim=(2, 4)				,     use_linear_projection=lowercase__				,     addition_embed_type='''text_time'''				,     addition_time_embed_dim=8				,     transformer_layers_per_block=(1, 2)				,     projection_class_embeddings_input_dim=8_0				,     cross_attention_dim=6_4				,     )
							__UpperCAmelCase			:							Union[str, Any]	    =	EulerDiscreteScheduler(
							    beta_start=0.0_0_0_8_5				,     beta_end=0.0_1_2				,     steps_offset=1				,     beta_schedule='''scaled_linear'''				,     timestep_spacing='''leading'''				,     )
							torch.manual_seed(0)
							__UpperCAmelCase			:							Union[str, 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)
							__UpperCAmelCase			:							Dict	    =	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=3_2				,     )
							__UpperCAmelCase			:							str	    =	CLIPTextModel(lowercase__)
							__UpperCAmelCase			:							str	    =	CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				,     local_files_only=lowercase__)

							__UpperCAmelCase			:							Tuple	    =	CLIPTextModelWithProjection(lowercase__)
							__UpperCAmelCase			:							str	    =	CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				,     local_files_only=lowercase__)

							__UpperCAmelCase			:							Dict	    =	{
							    '''unet''': unet,
							    '''scheduler''': scheduler,
							    '''vae''': vae,
							    '''text_encoder''': text_encoder,
							    '''tokenizer''': tokenizer,
							    '''text_encoder_2''': text_encoder_a,
							    '''tokenizer_2''': tokenizer_a,
							    # "safety_checker": None,
							    # "feature_extractor": None,
							}
							return components


	def  A(    self				,     lowercase__				,     lowercase__=0):
							__UpperCAmelCase			:							str	    =	floats_tensor((1, 3, 3_2, 3_2)				,     rng=random.Random(lowercase__)).to(lowercase__)
							__UpperCAmelCase			:							List[Any]	    =	image / 2 + 0.5
							if str(lowercase__).startswith('''mps'''):
													__UpperCAmelCase			:							List[str]	    =	torch.manual_seed(lowercase__)
							else:
													__UpperCAmelCase			:							List[str]	    =	torch.Generator(device=lowercase__).manual_seed(lowercase__)
							__UpperCAmelCase			:							Tuple	    =	{
							    '''prompt''': '''A painting of a squirrel eating a burger''',
							    '''image''': image,
							    '''generator''': generator,
							    '''num_inference_steps''': 2,
							    '''guidance_scale''': 5.0,
							    '''output_type''': '''numpy''',
							    '''strength''': 0.7_5,
							}
							return inputs


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

							__UpperCAmelCase			:							Any	    =	self.get_dummy_inputs(lowercase__)
							__UpperCAmelCase			:							List[Any]	    =	sd_pipe(**lowercase__).images
							__UpperCAmelCase			:							List[str]	    =	image[0, -3:, -3:, -1]

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

							__UpperCAmelCase			:							Any	    =	np.array([0.4_6_5_6, 0.4_8_4_0, 0.4_4_3_9, 0.6_6_9_8, 0.5_5_7_4, 0.4_5_2_4, 0.5_7_9_9, 0.5_9_4_3, 0.5_1_6_5])

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


	def  A(    self):
							super().test_attention_slicing_forward_pass(expected_max_diff=3e-3)


	def  A(    self):
							super().test_inference_batch_single_identical(expected_max_diff=3e-3)


	def  A(    self):
							pass


	def  A(    self):
							__UpperCAmelCase			:							int	    =	self.get_dummy_components()
							__UpperCAmelCase			:							List[str]	    =	StableDiffusionXLImgaImgPipeline(**lowercase__)
							__UpperCAmelCase			:							List[Any]	    =	sd_pipe.to(lowercase__)
							__UpperCAmelCase			:							Dict	    =	sd_pipe.to(lowercase__)
							sd_pipe.set_progress_bar_config(disable=lowercase__)

							# forward without prompt embeds
							__UpperCAmelCase			:							Any	    =	self.get_dummy_inputs(lowercase__)
							__UpperCAmelCase			:							Optional[Any]	    =	3 * ['''this is a negative prompt''']
							__UpperCAmelCase			:							Optional[Any]	    =	negative_prompt
							__UpperCAmelCase			:							str	    =	3 * [inputs['''prompt''']]

							__UpperCAmelCase			:							Optional[Any]	    =	sd_pipe(**lowercase__)
							__UpperCAmelCase			:							Union[str, Any]	    =	output.images[0, -3:, -3:, -1]

							# forward with prompt embeds
							__UpperCAmelCase			:							List[Any]	    =	self.get_dummy_inputs(lowercase__)
							__UpperCAmelCase			:							str	    =	3 * ['''this is a negative prompt''']
							__UpperCAmelCase			:							int	    =	3 * [inputs.pop('''prompt''')]

							(
							    (
							    __UpperCAmelCase
							)    ,      (
							    __UpperCAmelCase
							)    ,      (
							    __UpperCAmelCase
							)    ,      (
							    __UpperCAmelCase
							)    ,      
							)			:							int	    =	sd_pipe.encode_prompt(lowercase__				,     negative_prompt=lowercase__)

							__UpperCAmelCase			:							Optional[Any]	    =	sd_pipe(
							    **lowercase__				,     prompt_embeds=lowercase__				,     negative_prompt_embeds=lowercase__				,     pooled_prompt_embeds=lowercase__				,     negative_pooled_prompt_embeds=lowercase__				,     )
							__UpperCAmelCase			:							Optional[int]	    =	output.images[0, -3:, -3:, -1]

							# make sure that it's equal
							assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()).max() < 1e-4






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


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


	def  A(    self				,     lowercase__				,     lowercase__="cpu"				,     lowercase__=torch.floataa				,     lowercase__=0):
							__UpperCAmelCase			:							List[Any]	    =	torch.Generator(device=lowercase__).manual_seed(lowercase__)
							__UpperCAmelCase			:							Optional[Any]	    =	np.random.RandomState(lowercase__).standard_normal((1, 4, 6_4, 6_4))
							__UpperCAmelCase			:							Any	    =	torch.from_numpy(lowercase__).to(device=lowercase__				,     dtype=lowercase__)
							__UpperCAmelCase			:							Tuple	    =	{
							    '''prompt''': '''a photograph of an astronaut riding a horse''',
							    '''latents''': latents,
							    '''generator''': generator,
							    '''num_inference_steps''': 3,
							    '''guidance_scale''': 7.5,
							    '''output_type''': '''numpy''',
							}
							return inputs


	def  A(    self):
							__UpperCAmelCase			:							Tuple	    =	DiffusionPipeline.from_pretrained('''stabilityai/stable-diffusion-2-base''')
							pipe.to(lowercase__)
							pipe.set_progress_bar_config(disable=lowercase__)

							__UpperCAmelCase			:							Union[str, Any]	    =	self.get_inputs(lowercase__)
							__UpperCAmelCase			:							Optional[Any]	    =	pipe(**lowercase__).images
							__UpperCAmelCase			:							List[str]	    =	image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 5_1_2, 5_1_2, 3)
							__UpperCAmelCase			:							str	    =	np.array([0.4_9_4_9_3, 0.4_7_8_9_6, 0.4_0_7_9_8, 0.5_4_2_1_4, 0.5_3_2_1_2, 0.4_8_2_0_2, 0.4_7_6_5_6, 0.4_6_3_2_9, 0.4_8_5_0_6])
							assert np.abs(image_slice - expected_slice).max() < 7e-3





import importlib

import torch
import yaml
from omegaconf import OmegaConf
from taming.models.vqgan import VQModel




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_=False      )		->     Optional[Any]:




						'''simple docstring'''







						__UpperCAmelCase			:							Optional[Any]	    =	OmegaConf.load(lowercase_      )
						if display:
												print(yaml.dump(OmegaConf.to_container(lowercase_      )      )      )
						return config




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_=None      ,       lowercase_=None      )		->     Union[str, Any]:




						'''simple docstring'''







						if conf_path is None:
												__UpperCAmelCase			:							Optional[int]	    =	'''./model_checkpoints/vqgan_only.yaml'''
						__UpperCAmelCase			:							Dict	    =	load_config(lowercase_      ,       display=lowercase_      )
						__UpperCAmelCase			:							Tuple	    =	VQModel(**config.model.params      )
						if ckpt_path is None:
												__UpperCAmelCase			:							Dict	    =	'''./model_checkpoints/vqgan_only.pt'''
						__UpperCAmelCase			:							int	    =	torch.load(lowercase_      ,       map_location=lowercase_      )
						if ".ckpt" in ckpt_path:
												__UpperCAmelCase			:							Union[str, Any]	    =	sd['''state_dict''']
						model.load_state_dict(lowercase_      ,       strict=lowercase_      )
						model.to(lowercase_      )
						del sd
						return model




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_      )		->     List[Any]:




						'''simple docstring'''







						__UpperCAmelCase    ,      __UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[Any]	    =	model.encode(lowercase_      )
						print(f"VQGAN --- {model.__class__.__name__}: latent shape: {z.shape[2:]}"      )
						__UpperCAmelCase			:							List[str]	    =	model.decode(lowercase_      )
						return xrec




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_=False      )		->     int:




						'''simple docstring'''







						__UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[int]	    =	string.rsplit('''.'''      ,       1      )
						if reload:
												__UpperCAmelCase			:							Optional[int]	    =	importlib.import_module(lowercase_      )
												importlib.reload(lowercase_      )
						return getattr(importlib.import_module(lowercase_      ,       package=lowercase_      )      ,       cls      )




def     __SCREAMING_SNAKE_CASE							(       lowercase_      )		->     Union[str, Any]:




						'''simple docstring'''







						if "target" not in config:
												raise KeyError('''Expected key `target` to instantiate.'''      )
						return get_obj_from_str(config['''target''']      )(**config.get('''params'''      ,       {}      )      )




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_      ,       lowercase_=True      ,       lowercase_=True      )		->     int:




						'''simple docstring'''







						__UpperCAmelCase			:							Dict	    =	instantiate_from_config(lowercase_      )
						if sd is not None:
												model.load_state_dict(lowercase_      )
						if gpu:
												model.cuda()
						if eval_mode:
												model.eval()
						return {"model": model}





def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_      ,       lowercase_      ,       lowercase_      )		->     Optional[Any]:




						'''simple docstring'''







						if ckpt:
												__UpperCAmelCase			:							Optional[int]	    =	torch.load(lowercase_      ,       map_location='''cpu'''      )
												__UpperCAmelCase			:							Union[str, Any]	    =	pl_sd['''global_step''']
												print(f"loaded model from global step {global_step}."      )
						else:
												__UpperCAmelCase			:							Union[str, Any]	    =	{'''state_dict''': None}
												__UpperCAmelCase			:							Optional[int]	    =	None
						__UpperCAmelCase			:							str	    =	load_model_from_config(config.model      ,       pl_sd['''state_dict''']      ,       gpu=lowercase_      ,       eval_mode=lowercase_      )['''model''']
						return model, global_step

