

import gc
import random
import unittest

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

import diffusers
from diffusers import (
    AutoencoderKL,
    EulerDiscreteScheduler,
    StableDiffusionLatentUpscalePipeline,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.schedulers import KarrasDiffusionSchedulers
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


def   __SCREAMING_SNAKE_CASE					(SCREAMING_SNAKE_CASE__       ):
						snake_case_   =	[tensor.shape for tensor in tensor_list]
						return all(shape == shapes[0] for shape in shapes[1:]       )





class 	snake_case_   (		__A	,       __A	,       __A	,       unittest.TestCase					):







			'''simple docstring'''

			SCREAMING_SNAKE_CASE     :   Tuple									=				StableDiffusionLatentUpscalePipeline
			SCREAMING_SNAKE_CASE     :   Tuple									=				TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {
			    "height",
			    "width",
			    "cross_attention_kwargs",
			    "negative_prompt_embeds",
			    "prompt_embeds",
			}
			SCREAMING_SNAKE_CASE     :   int									=				PipelineTesterMixin.required_optional_params - {"num_images_per_prompt"}
			SCREAMING_SNAKE_CASE     :   Optional[Any]									=				TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
			SCREAMING_SNAKE_CASE     :   Optional[int]									=				frozenset(
			    []					)  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
			SCREAMING_SNAKE_CASE     :   str									=				frozenset([]					)

			SCREAMING_SNAKE_CASE     :   List[Any]									=				True







			@property
			def 				snake_case__(     self :				str							)      ->int:
									snake_case_   =	1
									snake_case_   =	4
									snake_case_   =	(1_6, 1_6)

									snake_case_   =	floats_tensor((batch_size, num_channels) + sizes							,   rng=random.Random(0							)							).to(_UpperCamelCase							)
									return image







			def 				snake_case__(     self :				Optional[Any]							)      ->Dict:
									torch.manual_seed(0							)
									snake_case_   =	UNetaDConditionModel(
									    act_fn='''gelu'''							,   attention_head_dim=8							,   norm_num_groups=_UpperCamelCase							,   block_out_channels=[3_2, 3_2, 6_4, 6_4]							,   time_cond_proj_dim=1_6_0							,   conv_in_kernel=1							,   conv_out_kernel=1							,   cross_attention_dim=3_2							,   down_block_types=(
									        '''KDownBlock2D''',
									        '''KCrossAttnDownBlock2D''',
									        '''KCrossAttnDownBlock2D''',
									        '''KCrossAttnDownBlock2D''',
									    )							,   in_channels=8							,   mid_block_type=_UpperCamelCase							,   only_cross_attention=_UpperCamelCase							,   out_channels=5							,   resnet_time_scale_shift='''scale_shift'''							,   time_embedding_type='''fourier'''							,   timestep_post_act='''gelu'''							,   up_block_types=('''KCrossAttnUpBlock2D''', '''KCrossAttnUpBlock2D''', '''KCrossAttnUpBlock2D''', '''KUpBlock2D''')							,   )
									snake_case_   =	AutoencoderKL(
									    block_out_channels=[3_2, 3_2, 6_4, 6_4]							,   in_channels=3							,   out_channels=3							,   down_block_types=[
									        '''DownEncoderBlock2D''',
									        '''DownEncoderBlock2D''',
									        '''DownEncoderBlock2D''',
									        '''DownEncoderBlock2D''',
									    ]							,   up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''', '''UpDecoderBlock2D''', '''UpDecoderBlock2D''']							,   latent_channels=4							,   )
									snake_case_   =	EulerDiscreteScheduler(prediction_type='''sample'''							)
									snake_case_   =	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='''quick_gelu'''							,   projection_dim=5_1_2							,   )
									snake_case_   =	CLIPTextModel(_UpperCamelCase							)
									snake_case_   =	CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''							)

									snake_case_   =	{
									    '''unet''': model.eval(),
									    '''vae''': vae.eval(),
									    '''scheduler''': scheduler,
									    '''text_encoder''': text_encoder,
									    '''tokenizer''': tokenizer,
									}

									return components







			def 				snake_case__(     self :				Union[str, Any]							,   _UpperCamelCase :				Union[str, Any]							,   _UpperCamelCase :				Tuple=0							)      ->Optional[Any]:
									if str(_UpperCamelCase							).startswith('''mps'''							):
															snake_case_   =	torch.manual_seed(_UpperCamelCase							)
									else:
															snake_case_   =	torch.Generator(device=_UpperCamelCase							).manual_seed(_UpperCamelCase							)
									snake_case_   =	{
									    '''prompt''': '''A painting of a squirrel eating a burger''',
									    '''image''': self.dummy_image.cpu(),
									    '''generator''': generator,
									    '''num_inference_steps''': 2,
									    '''output_type''': '''numpy''',
									}
									return inputs







			def 				snake_case__(     self :				Dict							)      ->int:
									snake_case_   =	'''cpu'''

									snake_case_   =	self.get_dummy_components()
									snake_case_   =	self.pipeline_class(**_UpperCamelCase							)
									pipe.to(_UpperCamelCase							)
									pipe.set_progress_bar_config(disable=_UpperCamelCase							)

									snake_case_   =	self.get_dummy_inputs(_UpperCamelCase							)
									snake_case_   =	pipe(**_UpperCamelCase							).images
									snake_case_   =	image[0, -3:, -3:, -1]

									self.assertEqual(image.shape							,   (1, 2_5_6, 2_5_6, 3)							)
									snake_case_   =	np.array(
									    [0.47222412, 0.41921633, 0.44717434, 0.46874192, 0.42588258, 0.46150726, 0.4677534, 0.45583832, 0.48579055]							)
									snake_case_   =	np.abs(image_slice.flatten() - expected_slice							).max()
									self.assertLessEqual(_UpperCamelCase							,   1e-3							)







			def 				snake_case__(     self :				Optional[Any]							)      ->Tuple:
									super().test_attention_slicing_forward_pass(expected_max_diff=7e-3							)







			def 				snake_case__(     self :				Union[str, Any]							)      ->Tuple:
									super().test_cpu_offload_forward_pass(expected_max_diff=3e-3							)







			def 				snake_case__(     self :				Optional[Any]							)      ->Tuple:
									super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3							)







			def 				snake_case__(     self :				Any							)      ->Dict:
									super().test_inference_batch_single_identical(expected_max_diff=7e-3							)







			def 				snake_case__(     self :				List[str]							)      ->List[Any]:
									super().test_pt_np_pil_outputs_equivalent(expected_max_diff=3e-3							)







			def 				snake_case__(     self :				Any							)      ->List[str]:
									super().test_save_load_local(expected_max_difference=3e-3							)







			def 				snake_case__(     self :				List[str]							)      ->List[str]:
									super().test_save_load_optional_components(expected_max_difference=3e-3							)







			def 				snake_case__(     self :				str							)      ->Optional[Any]:
									snake_case_   =	[
									    '''DDIMScheduler''',
									    '''DDPMScheduler''',
									    '''PNDMScheduler''',
									    '''HeunDiscreteScheduler''',
									    '''EulerAncestralDiscreteScheduler''',
									    '''KDPM2DiscreteScheduler''',
									    '''KDPM2AncestralDiscreteScheduler''',
									    '''DPMSolverSDEScheduler''',
									]
									snake_case_   =	self.get_dummy_components()
									snake_case_   =	self.pipeline_class(**_UpperCamelCase							)

									# make sure that PNDM does not need warm-up
									pipe.scheduler.register_to_config(skip_prk_steps=_UpperCamelCase							)

									pipe.to(_UpperCamelCase							)
									pipe.set_progress_bar_config(disable=_UpperCamelCase							)
									snake_case_   =	self.get_dummy_inputs(_UpperCamelCase							)
									snake_case_   =	2

									snake_case_   =	[]
									for scheduler_enum in KarrasDiffusionSchedulers:
															if scheduler_enum.name in skip_schedulers:
																					# no sigma schedulers are not supported
																					# no schedulers
																					continue

															snake_case_   =	getattr(_UpperCamelCase							,   scheduler_enum.name							)
															snake_case_   =	scheduler_cls.from_config(pipe.scheduler.config							)
															snake_case_   =	pipe(**_UpperCamelCase							)[0]
															outputs.append(_UpperCamelCase							)

									assert check_same_shape(_UpperCamelCase							)






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







			'''simple docstring'''

			def 				snake_case__(     self :				Any							)      ->int:
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()







			def 				snake_case__(     self :				List[Any]							)      ->Any:
									snake_case_   =	torch.manual_seed(3_3							)

									snake_case_   =	StableDiffusionPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''							,   torch_dtype=torch.floataa							)
									pipe.to('''cuda'''							)

									snake_case_   =	StableDiffusionLatentUpscalePipeline.from_pretrained(
									    '''stabilityai/sd-x2-latent-upscaler'''							,   torch_dtype=torch.floataa							)
									upscaler.to('''cuda'''							)

									snake_case_   =	'''a photo of an astronaut high resolution, unreal engine, ultra realistic'''

									snake_case_   =	pipe(_UpperCamelCase							,   generator=_UpperCamelCase							,   output_type='''latent'''							).images

									snake_case_   =	upscaler(
									    prompt=_UpperCamelCase							,   image=_UpperCamelCase							,   num_inference_steps=2_0							,   guidance_scale=0							,   generator=_UpperCamelCase							,   output_type='''np'''							,   ).images[0]

									snake_case_   =	load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/astronaut_1024.npy'''							)
									assert np.abs((expected_image - image).mean()							) < 5e-2







			def 				snake_case__(     self :				Optional[int]							)      ->Optional[Any]:
									snake_case_   =	torch.manual_seed(3_3							)

									snake_case_   =	StableDiffusionLatentUpscalePipeline.from_pretrained(
									    '''stabilityai/sd-x2-latent-upscaler'''							,   torch_dtype=torch.floataa							)
									upscaler.to('''cuda'''							)

									snake_case_   =	'''the temple of fire by Ross Tran and Gerardo Dottori, oil on canvas'''

									snake_case_   =	load_image(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_512.png'''							)

									snake_case_   =	upscaler(
									    prompt=_UpperCamelCase							,   image=_UpperCamelCase							,   num_inference_steps=2_0							,   guidance_scale=0							,   generator=_UpperCamelCase							,   output_type='''np'''							,   ).images[0]

									snake_case_   =	load_numpy(
									    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/latent-upscaler/fire_temple_1024.npy'''							)
									assert np.abs((expected_image - image).max()							) < 5e-2


from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_sentencepiece_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


lowerCAmelCase_       	=	{'''configuration_xglm''': ['''XGLM_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''XGLMConfig''']}

try:
					if not is_sentencepiece_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					lowerCAmelCase_       	=	['''XGLMTokenizer''']

try:
					if not is_tokenizers_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					lowerCAmelCase_       	=	['''XGLMTokenizerFast''']

try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					lowerCAmelCase_       	=	[
					    '''XGLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
					    '''XGLMForCausalLM''',
					    '''XGLMModel''',
					    '''XGLMPreTrainedModel''',
					]


try:
					if not is_flax_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					lowerCAmelCase_       	=	[
					    '''FlaxXGLMForCausalLM''',
					    '''FlaxXGLMModel''',
					    '''FlaxXGLMPreTrainedModel''',
					]


try:
					if not is_tf_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					lowerCAmelCase_       	=	[
					    '''TF_XGLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
					    '''TFXGLMForCausalLM''',
					    '''TFXGLMModel''',
					    '''TFXGLMPreTrainedModel''',
					]


if TYPE_CHECKING:
					from .configuration_xglm import XGLM_PRETRAINED_CONFIG_ARCHIVE_MAP, XGLMConfig

					try:
										if not is_sentencepiece_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .tokenization_xglm import XGLMTokenizer

					try:
										if not is_tokenizers_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .tokenization_xglm_fast import XGLMTokenizerFast

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_xglm import XGLM_PRETRAINED_MODEL_ARCHIVE_LIST, XGLMForCausalLM, XGLMModel, XGLMPreTrainedModel

					try:
										if not is_flax_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_flax_xglm import FlaxXGLMForCausalLM, FlaxXGLMModel, FlaxXGLMPreTrainedModel

					try:
										if not is_tf_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_tf_xglm import (
										    TF_XGLM_PRETRAINED_MODEL_ARCHIVE_LIST,
										    TFXGLMForCausalLM,
										    TFXGLMModel,
										    TFXGLMPreTrainedModel,
										)


else:
					import sys

					lowerCAmelCase_       	=	_LazyModule(__name__, globals()['''__file__'''], _import_structure)