

import gc
import random
import unittest

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

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableDiffusionUpscalePipeline, UNetaDConditionModel
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


enable_full_determinism()



class lowerCamelCase			(  unittest.TestCase						):



			'''simple docstring'''






			def lowercase__						(				self	:						str			)      ->		str:





									'''simple docstring'''
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()


			@property
			def lowercase__						(				self	:						Optional[Any]			)      ->		Union[str, Any]:





									'''simple docstring'''
									A__					:       Union[str, Any]     =1
									A__					:       Optional[Any]     =3
									A__					:       int     =(32, 32)

									A__					:       Union[str, Any]     =floats_tensor((batch_size, num_channels) + sizes		,      rng=random.Random(0			)			).to(lowerCAmelCase_			)
									return image


			@property
			def lowercase__						(				self	:						Optional[Any]			)      ->		Union[str, Any]:





									'''simple docstring'''
									torch.manual_seed(0			)
									A__					:       Union[str, Any]     =UNetaDConditionModel(
									    block_out_channels=(32, 32, 64)		,      layers_per_block=2		,      sample_size=32		,      in_channels=7		,      out_channels=4		,      down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""", """CrossAttnDownBlock2D""")		,      up_block_types=("""CrossAttnUpBlock2D""", """CrossAttnUpBlock2D""", """UpBlock2D""")		,      cross_attention_dim=32		,      attention_head_dim=8		,      use_linear_projection=lowerCAmelCase_		,      only_cross_attention=(True, True, False)		,      num_class_embeds=1_00		,      )
									return model


			@property
			def lowercase__						(				self	:						Union[str, Any]			)      ->		str:





									'''simple docstring'''
									torch.manual_seed(0			)
									A__					:       int     =AutoencoderKL(
									    block_out_channels=[32, 32, 64]		,      in_channels=3		,      out_channels=3		,      down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D""", """DownEncoderBlock2D"""]		,      up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D""", """UpDecoderBlock2D"""]		,      latent_channels=4		,      )
									return model


			@property
			def lowercase__						(				self	:						Tuple			)      ->		Dict:





									'''simple docstring'''
									torch.manual_seed(0			)
									A__					:       Any     =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		,      )
									return CLIPTextModel(lowerCAmelCase_			)


			def lowercase__						(				self	:						Optional[int]			)      ->		Optional[Any]:





									'''simple docstring'''
									A__					:       Optional[int]     ="""cpu"""  # ensure determinism for the device-dependent torch.Generator
									A__					:       Dict     =self.dummy_cond_unet_upscale
									A__					:       List[str]     =DDPMScheduler()
									A__					:       Union[str, Any]     =DDIMScheduler(prediction_type="""v_prediction"""			)
									A__					:       Union[str, Any]     =self.dummy_vae
									A__					:       Any     =self.dummy_text_encoder
									A__					:       Tuple     =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

									A__					:       List[Any]     =self.dummy_image.cpu().permute(0		,      2		,      3		,      1			)[0]
									A__					:       Tuple     =Image.fromarray(np.uinta(lowerCAmelCase_			)			).convert("""RGB"""			).resize((64, 64)			)

									# make sure here that pndm scheduler skips prk
									A__					:       int     =StableDiffusionUpscalePipeline(
									    unet=lowerCAmelCase_		,      low_res_scheduler=lowerCAmelCase_		,      scheduler=lowerCAmelCase_		,      vae=lowerCAmelCase_		,      text_encoder=lowerCAmelCase_		,      tokenizer=lowerCAmelCase_		,      max_noise_level=3_50		,      )
									A__					:       int     =sd_pipe.to(lowerCAmelCase_			)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_			)

									A__					:       int     ="""A painting of a squirrel eating a burger"""
									A__					:       List[str]     =torch.Generator(device=lowerCAmelCase_			).manual_seed(0			)
									A__					:       Union[str, Any]     =sd_pipe(
									    [prompt]		,      image=lowerCAmelCase_		,      generator=lowerCAmelCase_		,      guidance_scale=6.0		,      noise_level=20		,      num_inference_steps=2		,      output_type="""np"""		,      )

									A__					:       List[Any]     =output.images

									A__					:       Dict     =torch.Generator(device=lowerCAmelCase_			).manual_seed(0			)
									A__					:       Union[str, Any]     =sd_pipe(
									    [prompt]		,      image=lowerCAmelCase_		,      generator=lowerCAmelCase_		,      guidance_scale=6.0		,      noise_level=20		,      num_inference_steps=2		,      output_type="""np"""		,      return_dict=lowerCAmelCase_		,      )[0]

									A__					:       int     =image[0, -3:, -3:, -1]
									A__					:       Dict     =image_from_tuple[0, -3:, -3:, -1]

									A__					:       Optional[int]     =low_res_image.size[0] * 4
									assert image.shape == (1, expected_height_width, expected_height_width, 3)
									A__					:       int     =np.array([0.3113, 0.3910, 0.4272, 0.4859, 0.5061, 0.4652, 0.5362, 0.5715, 0.5661]			)

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


			def lowercase__						(				self	:						int			)      ->		List[str]:





									'''simple docstring'''
									A__					:       Optional[Any]     ="""cpu"""  # ensure determinism for the device-dependent torch.Generator
									A__					:       Optional[int]     =self.dummy_cond_unet_upscale
									A__					:       Optional[int]     =DDPMScheduler()
									A__					:       List[Any]     =DDIMScheduler(prediction_type="""v_prediction"""			)
									A__					:       List[Any]     =self.dummy_vae
									A__					:       Dict     =self.dummy_text_encoder
									A__					:       Union[str, Any]     =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

									A__					:       List[str]     =self.dummy_image.cpu().permute(0		,      2		,      3		,      1			)[0]
									A__					:       List[str]     =Image.fromarray(np.uinta(lowerCAmelCase_			)			).convert("""RGB"""			).resize((64, 64)			)

									# make sure here that pndm scheduler skips prk
									A__					:       Union[str, Any]     =StableDiffusionUpscalePipeline(
									    unet=lowerCAmelCase_		,      low_res_scheduler=lowerCAmelCase_		,      scheduler=lowerCAmelCase_		,      vae=lowerCAmelCase_		,      text_encoder=lowerCAmelCase_		,      tokenizer=lowerCAmelCase_		,      max_noise_level=3_50		,      )
									A__					:       Tuple     =sd_pipe.to(lowerCAmelCase_			)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_			)

									A__					:       int     ="""A painting of a squirrel eating a burger"""
									A__					:       Optional[Any]     =sd_pipe(
									    2 * [prompt]		,      image=2 * [low_res_image]		,      guidance_scale=6.0		,      noise_level=20		,      num_inference_steps=2		,      output_type="""np"""		,      )
									A__					:       List[Any]     =output.images
									assert image.shape[0] == 2

									A__					:       Optional[int]     =torch.Generator(device=lowerCAmelCase_			).manual_seed(0			)
									A__					:       Optional[int]     =sd_pipe(
									    [prompt]		,      image=lowerCAmelCase_		,      generator=lowerCAmelCase_		,      num_images_per_prompt=2		,      guidance_scale=6.0		,      noise_level=20		,      num_inference_steps=2		,      output_type="""np"""		,      )
									A__					:       Union[str, Any]     =output.images
									assert image.shape[0] == 2



			@unittest.skipIf(torch_device != """cuda"""		,      """This test requires a GPU"""			)
			def lowercase__						(				self	:						Tuple			)      ->		Optional[Any]:





									'''simple docstring'''
									A__					:       Any     =self.dummy_cond_unet_upscale
									A__					:       Any     =DDPMScheduler()
									A__					:       Optional[Any]     =DDIMScheduler(prediction_type="""v_prediction"""			)
									A__					:       Tuple     =self.dummy_vae
									A__					:       str     =self.dummy_text_encoder
									A__					:       List[str]     =CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

									A__					:       Optional[Any]     =self.dummy_image.cpu().permute(0		,      2		,      3		,      1			)[0]
									A__					:       str     =Image.fromarray(np.uinta(lowerCAmelCase_			)			).convert("""RGB"""			).resize((64, 64)			)

									# put models in fp16, except vae as it overflows in fp16
									A__					:       List[str]     =unet.half()
									A__					:       List[str]     =text_encoder.half()

									# make sure here that pndm scheduler skips prk
									A__					:       List[Any]     =StableDiffusionUpscalePipeline(
									    unet=lowerCAmelCase_		,      low_res_scheduler=lowerCAmelCase_		,      scheduler=lowerCAmelCase_		,      vae=lowerCAmelCase_		,      text_encoder=lowerCAmelCase_		,      tokenizer=lowerCAmelCase_		,      max_noise_level=3_50		,      )
									A__					:       int     =sd_pipe.to(lowerCAmelCase_			)
									sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_			)

									A__					:       Any     ="""A painting of a squirrel eating a burger"""
									A__					:       Tuple     =torch.manual_seed(0			)
									A__					:       Tuple     =sd_pipe(
									    [prompt]		,      image=lowerCAmelCase_		,      generator=lowerCAmelCase_		,      num_inference_steps=2		,      output_type="""np"""		,      ).images

									A__					:       int     =low_res_image.size[0] * 4
									assert image.shape == (1, expected_height_width, expected_height_width, 3)



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



			'''simple docstring'''






			def lowercase__						(				self	:						Any			)      ->		List[str]:





									'''simple docstring'''
									# clean up the VRAM after each test
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()


			def lowercase__						(				self	:						Optional[int]			)      ->		List[Any]:





									'''simple docstring'''
									A__					:       str     =load_image(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
									    """/sd2-upscale/low_res_cat.png"""			)
									A__					:       List[str]     =load_numpy(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale"""
									    """/upsampled_cat.npy"""			)

									A__					:       Optional[Any]     ="""stabilityai/stable-diffusion-x4-upscaler"""
									A__					:       Tuple     =StableDiffusionUpscalePipeline.from_pretrained(lowerCAmelCase_			)
									pipe.to(lowerCAmelCase_			)
									pipe.set_progress_bar_config(disable=lowerCAmelCase_			)
									pipe.enable_attention_slicing()

									A__					:       Optional[int]     ="""a cat sitting on a park bench"""

									A__					:       Optional[int]     =torch.manual_seed(0			)
									A__					:       int     =pipe(
									    prompt=lowerCAmelCase_		,      image=lowerCAmelCase_		,      generator=lowerCAmelCase_		,      output_type="""np"""		,      )
									A__					:       Any     =output.images[0]

									assert image.shape == (5_12, 5_12, 3)
									assert np.abs(expected_image - image			).max() < 1e-3


			def lowercase__						(				self	:						Optional[int]			)      ->		Optional[int]:





									'''simple docstring'''
									A__					:       Tuple     =load_image(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
									    """/sd2-upscale/low_res_cat.png"""			)
									A__					:       Tuple     =load_numpy(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale"""
									    """/upsampled_cat_fp16.npy"""			)

									A__					:       List[Any]     ="""stabilityai/stable-diffusion-x4-upscaler"""
									A__					:       Any     =StableDiffusionUpscalePipeline.from_pretrained(
									    lowerCAmelCase_		,      torch_dtype=torch.floataa		,      )
									pipe.to(lowerCAmelCase_			)
									pipe.set_progress_bar_config(disable=lowerCAmelCase_			)
									pipe.enable_attention_slicing()

									A__					:       Dict     ="""a cat sitting on a park bench"""

									A__					:       Union[str, Any]     =torch.manual_seed(0			)
									A__					:       Union[str, Any]     =pipe(
									    prompt=lowerCAmelCase_		,      image=lowerCAmelCase_		,      generator=lowerCAmelCase_		,      output_type="""np"""		,      )
									A__					:       str     =output.images[0]

									assert image.shape == (5_12, 5_12, 3)
									assert np.abs(expected_image - image			).max() < 5e-1



			def lowercase__						(				self	:						int			)      ->		str:





									'''simple docstring'''
									torch.cuda.empty_cache()
									torch.cuda.reset_max_memory_allocated()
									torch.cuda.reset_peak_memory_stats()

									A__					:       int     =load_image(
									    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
									    """/sd2-upscale/low_res_cat.png"""			)

									A__					:       Union[str, Any]     ="""stabilityai/stable-diffusion-x4-upscaler"""
									A__					:       Optional[int]     =StableDiffusionUpscalePipeline.from_pretrained(
									    lowerCAmelCase_		,      torch_dtype=torch.floataa		,      )
									pipe.to(lowerCAmelCase_			)
									pipe.set_progress_bar_config(disable=lowerCAmelCase_			)
									pipe.enable_attention_slicing(1			)
									pipe.enable_sequential_cpu_offload()

									A__					:       Dict     ="""a cat sitting on a park bench"""

									A__					:       int     =torch.manual_seed(0			)
									A__					:       Optional[Any]     =pipe(
									    prompt=lowerCAmelCase_		,      image=lowerCAmelCase_		,      generator=lowerCAmelCase_		,      num_inference_steps=5		,      output_type="""np"""		,      )

									A__					:       Any     =torch.cuda.max_memory_allocated()
									# make sure that less than 2.9 GB is allocated
									assert mem_bytes < 2.9 * 10**9




'''simple docstring'''



from typing import Dict

from .base import GenericTensor, Pipeline



class lowerCamelCase			(  lowercase_						):



			'''simple docstring'''






			def lowercase__						(				self	:						List[str]		,      lowerCAmelCase_	:						Tuple=None		,      lowerCAmelCase_	:						Dict=None		,      lowerCAmelCase_	:						Any=None		,      **lowerCAmelCase_	:						Union[str, Any]			)      ->		Optional[Any]:





									'''simple docstring'''
									if tokenize_kwargs is None:
															A__					:       List[Any]     ={}

									if truncation is not None:
															if "truncation" in tokenize_kwargs:
																					raise ValueError(
																					    """truncation parameter defined twice (given as keyword argument as well as in tokenize_kwargs)"""			)
															A__					:       str     =truncation

									A__					:       Optional[int]     =tokenize_kwargs

									A__					:       List[Any]     ={}
									if return_tensors is not None:
															A__					:       Any     =return_tensors

									return preprocess_params, {}, postprocess_params


			def lowercase__						(				self	:						int		,      lowerCAmelCase_	:						Union[str, Any]		,      **lowerCAmelCase_	:						Dict			)      ->		Dict[str, GenericTensor]:





									'''simple docstring'''
									A__					:       List[str]     =self.framework
									A__					:       Union[str, Any]     =self.tokenizer(lowerCAmelCase_		,      return_tensors=lowerCAmelCase_		,      **lowerCAmelCase_			)
									return model_inputs


			def lowercase__						(				self	:						Optional[int]		,      lowerCAmelCase_	:						Dict			)      ->		Optional[Any]:





									'''simple docstring'''
									A__					:       Union[str, Any]     =self.model(**lowerCAmelCase_			)
									return model_outputs


			def lowercase__						(				self	:						Tuple		,      lowerCAmelCase_	:						Tuple		,      lowerCAmelCase_	:						Any=False			)      ->		List[Any]:





									'''simple docstring'''
									# [0] is the first available tensor, logits or last_hidden_state.
									if return_tensors:
															return model_outputs[0]
									if self.framework == "pt":
															return model_outputs[0].tolist()
									elif self.framework == "tf":
															return model_outputs[0].numpy().tolist()



			def __call__(				self	:						int		,      *lowerCAmelCase_	:						Any		,      **lowerCAmelCase_	:						Union[str, Any]			)      ->		List[str]:





									'''simple docstring'''
									return super().__call__(*lowerCAmelCase_		,      **lowerCAmelCase_			)

