


import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLPipeline,
    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 			lowercase_				(lowercase__			,				lowercase__			,				unittest.TestCase      ):
				snake_case  						=StableDiffusionXLPipeline
				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)     ->   Union[str, Any]:
											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')					,     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=80					,     cross_attention_dim=64					,     )
											a__     						=EulerDiscreteScheduler(
											    beta_start=0.0_00_85					,     beta_end=0.0_12					,     steps_offset=1					,     beta_schedule='scaled_linear'					,     timestep_spacing='leading'					,     )
											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=128					,     )
											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=1000					,     hidden_act='gelu'					,     projection_dim=32					,     )
											a__     						=CLIPTextModel(lowercase_)
											a__     						=CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					,     local_files_only=lowercase_)

											a__     						=CLIPTextModelWithProjection(lowercase_)
											a__     						=CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'					,     local_files_only=lowercase_)

											a__     						={
											    '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 	__UpperCamelCase						(							self					,     lowercase_					,     lowercase_=0)     ->   int:
											if str(lowercase_).startswith('mps'):
																		a__     						=torch.manual_seed(lowercase_)
											else:
																		a__     						=torch.Generator(device=lowercase_).manual_seed(lowercase_)
											a__     						={
											    'prompt': 'A painting of a squirrel eating a burger',
											    'generator': generator,
											    'num_inference_steps': 2,
											    'guidance_scale': 5.0,
											    'output_type': 'numpy',
											}
											return inputs





				def 	__UpperCamelCase						(							self)     ->   Optional[Any]:
											a__     						='cpu'  # ensure determinism for the device-dependent torch.Generator
											a__     						=self.get_dummy_components()
											a__     						=StableDiffusionXLPipeline(**lowercase_)
											a__     						=sd_pipe.to(lowercase_)
											sd_pipe.set_progress_bar_config(disable=lowercase_)

											a__     						=self.get_dummy_inputs(lowercase_)
											a__     						=sd_pipe(**lowercase_).images
											a__     						=image[0, -3:, -3:, -1]

											assert image.shape == (1, 64, 64, 3)
											a__     						=np.array([0.58_73, 0.61_28, 0.47_97, 0.51_22, 0.56_74, 0.46_39, 0.52_27, 0.51_49, 0.47_47])

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





				def 	__UpperCamelCase						(							self)     ->   str:
											a__     						=self.get_dummy_components()
											a__     						=StableDiffusionXLPipeline(**lowercase_)
											a__     						=sd_pipe.to(lowercase_)
											a__     						=sd_pipe.to(lowercase_)
											sd_pipe.set_progress_bar_config(disable=lowercase_)

											# forward without prompt embeds
											a__     						=self.get_dummy_inputs(lowercase_)
											a__     						=3 * ['this is a negative prompt']
											a__     						=negative_prompt
											a__     						=3 * [inputs['prompt']]

											a__     						=sd_pipe(**lowercase_)
											a__     						=output.images[0, -3:, -3:, -1]

											# forward with prompt embeds
											a__     						=self.get_dummy_inputs(lowercase_)
											a__     						=3 * ['this is a negative prompt']
											a__     						=3 * [inputs.pop('prompt')]

											(
											    (
											    a__
											)						,    (
											    a__
											)						,    (
											    a__
											)						,    (
											    a__
											)						,    
											)     						=sd_pipe.encode_prompt(lowercase_					,     negative_prompt=lowercase_)

											a__     						=sd_pipe(
											    **lowercase_					,     prompt_embeds=lowercase_					,     negative_prompt_embeds=lowercase_					,     pooled_prompt_embeds=lowercase_					,     negative_pooled_prompt_embeds=lowercase_					,     )
											a__     						=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





				def 	__UpperCamelCase						(							self)     ->   str:
											super().test_attention_slicing_forward_pass(expected_max_diff=3e-3)





				def 	__UpperCamelCase						(							self)     ->   Union[str, Any]:
											super().test_inference_batch_single_identical(expected_max_diff=3e-3)







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





				def 	__UpperCamelCase						(							self)     ->   Dict:
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()





				def 	__UpperCamelCase						(							self					,     lowercase_					,     lowercase_="cpu"					,     lowercase_=torch.floataa					,     lowercase_=0)     ->   Dict:
											a__     						=torch.Generator(device=lowercase_).manual_seed(lowercase_)
											a__     						=np.random.RandomState(lowercase_).standard_normal((1, 4, 64, 64))
											a__     						=torch.from_numpy(lowercase_).to(device=lowercase_					,     dtype=lowercase_)
											a__     						={
											    '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 	__UpperCamelCase						(							self)     ->   List[str]:
											a__     						=DiffusionPipeline.from_pretrained('stabilityai/stable-diffusion-2-base')
											pipe.to(lowercase_)
											pipe.set_progress_bar_config(disable=lowercase_)

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

											assert image.shape == (1, 512, 512, 3)
											a__     						=np.array([0.4_94_93, 0.4_78_96, 0.4_07_98, 0.5_42_14, 0.5_32_12, 0.4_82_02, 0.4_76_56, 0.4_63_29, 0.4_85_06])
											assert np.abs(image_slice - expected_slice).max() < 7e-3





from typing import Dict, Optional, Tuple

import torch






def      _lowercase(				__a							:	torch.Tensor    ):
							a__     						=boundaries[1] - boundaries[0]
							a__     						=boundaries + step / 2
							a__     						=torch.cat([bin_centers, (bin_centers[-1] + step).unsqueeze(-1    )]						,    dim=0    )
							return bin_centers






def      _lowercase(				__a							:	torch.Tensor						,    __a							:	torch.Tensor						,    ):
							a__     						=_calculate_bin_centers(__a    )
							return (
							    torch.sum(aligned_distance_error_probs * bin_centers						,    dim=-1    ),
							    bin_centers[-1],
							)






def      _lowercase(				__a							:	torch.Tensor						,    __a							:	int = 31						,    __a							:	int = 64						,    **__a							:	List[str]						,    ):
							a__     						=torch.linspace(0						,    __a						,    steps=(no_bins - 1)						,    device=logits.device    )

							a__     						=torch.nn.functional.softmax(__a						,    dim=-1    )
							a__						,    a__     						=_calculate_expected_aligned_error(
							    alignment_confidence_breaks=__a						,    aligned_distance_error_probs=__a						,    )

							return {
							    "aligned_confidence_probs": aligned_confidence_probs,
							    "predicted_aligned_error": predicted_aligned_error,
							    "max_predicted_aligned_error": max_predicted_aligned_error,
							}







def      _lowercase(				__a							:	torch.Tensor						,    __a							:	Optional[torch.Tensor] = None						,    __a							:	int = 31						,    __a							:	int = 64						,    __a							:	float = 1e-8						,    **__a							:	str						,    ):
							if residue_weights is None:
														a__     						=logits.new_ones(logits.shape[-2]    )

							a__     						=torch.linspace(0						,    __a						,    steps=(no_bins - 1)						,    device=logits.device    )

							a__     						=_calculate_bin_centers(__a    )
							torch.sum(__a    )
							a__     						=logits.shape[-2]
							a__     						=max(__a						,    19    )

							a__     						=1.24 * (clipped_n - 15) ** (1.0 / 3) - 1.8

							a__     						=torch.nn.functional.softmax(__a						,    dim=-1    )

							a__     						=1.0 / (1 + (bin_centers**2) / (da**2))
							a__     						=torch.sum(probs * tm_per_bin						,    dim=-1    )

							a__     						=residue_weights / (eps + residue_weights.sum())
							a__     						=torch.sum(predicted_tm_term * normed_residue_mask						,    dim=-1    )

							a__     						=per_alignment * residue_weights

							a__     						=(weighted == torch.max(__a    )).nonzero()[0]
							return per_alignment[tuple(__a    )]

