



"""simple docstring"""


# DISCLAIMER: This code is strongly influenced by https://github.com/pesser/pytorch_diffusion
# and https://github.com/hojonathanho/diffusion
import math
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.schedulers.scheduling_utils import SchedulerMixin
from diffusers.utils import BaseOutput, deprecate







@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->DDIM
class 				A_      (					SCREAMING_SNAKE_CASE_    ):



						"""simple docstring"""
						__UpperCamelCase		    =							42
						__UpperCamelCase		    =							None


def 			_lowerCAmelCase				(       lowercase_		,     lowercase_=0.9_9_9		,     lowercase_="cosine"		,     ):
						if alpha_transform_type == "cosine":

												def alpha_bar_fn(lowercase_						):
																		return math.cos((t + 0.0_0_8) / 1.0_0_8 * math.pi / 2						) ** 2

						elif alpha_transform_type == "exp":

												def alpha_bar_fn(lowercase_						):
																		return math.exp(t * -1_2.0						)

						else:
												raise ValueError(F"""Unsupported alpha_tranform_type: {alpha_transform_type}"""						)

						UpperCAmelCase					=			[]
						for i in range(lowercase_						):
												UpperCAmelCase					=			i / num_diffusion_timesteps
												UpperCAmelCase					=			(i + 1) / num_diffusion_timesteps
												betas.append(min(1 - alpha_bar_fn(lowercase_						) / alpha_bar_fn(lowercase_						)		,     lowercase_						)						)
						return torch.tensor(lowercase_		,     dtype=torch.floataa						)







class 				A_      (					SCREAMING_SNAKE_CASE_     ,       SCREAMING_SNAKE_CASE_    ):



						"""simple docstring"""
						__UpperCamelCase		    =							1






						@register_to_config
						def __init__( self			:Dict       ,  lowercase_			:int = 10_00       ,  lowercase_			:float = 0.0001       ,  lowercase_			:float = 0.02       ,  lowercase_			:str = "linear"       ,  lowercase_			:Optional[Union[np.ndarray, List[float]]] = None       ,  lowercase_			:bool = True       ,  lowercase_			:bool = True       ,  lowercase_			:int = 0       ,  lowercase_			:str = "epsilon"       ,  lowercase_			:float = 1.0       ,  **lowercase_			:Dict       ,  )    ->      List[str]:
												if kwargs.get('set_alpha_to_one'       ,  lowercase_ ) is not None:
																		UpperCAmelCase					=			(
																		    'The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead.'
																		)
																		deprecate('set_alpha_to_one'       ,  '1.0.0'       ,  lowercase_       ,  standard_warn=lowercase_ )
																		UpperCAmelCase					=			kwargs['set_alpha_to_one']
												if trained_betas is not None:
																		UpperCAmelCase					=			torch.tensor(lowercase_       ,  dtype=torch.floataa )
												elif beta_schedule == "linear":
																		UpperCAmelCase					=			torch.linspace(lowercase_       ,  lowercase_       ,  lowercase_       ,  dtype=torch.floataa )
												elif beta_schedule == "scaled_linear":
																		# this schedule is very specific to the latent diffusion model.
																		UpperCAmelCase					=			(
																		    torch.linspace(beta_start**0.5       ,  beta_end**0.5       ,  lowercase_       ,  dtype=torch.floataa ) ** 2
																		)
												elif beta_schedule == "squaredcos_cap_v2":
																		# Glide cosine schedule
																		UpperCAmelCase					=			betas_for_alpha_bar(lowercase_ )
												else:
																		raise NotImplementedError(f"""{beta_schedule} does is not implemented for {self.__class__}""" )

												UpperCAmelCase					=			1.0 - self.betas
												UpperCAmelCase					=			torch.cumprod(self.alphas       ,  dim=0 )

												# At every step in inverted ddim, we are looking into the next alphas_cumprod
												# For the final step, there is no next alphas_cumprod, and the index is out of bounds
												# `set_alpha_to_zero` decides whether we set this parameter simply to zero
												# in this case, self.step() just output the predicted noise
												# or whether we use the final alpha of the "non-previous" one.
												UpperCAmelCase					=			torch.tensor(0.0 ) if set_alpha_to_zero else self.alphas_cumprod[-1]

												# standard deviation of the initial noise distribution
												UpperCAmelCase					=			1.0

												# setable values
												UpperCAmelCase					=			None
												UpperCAmelCase					=			torch.from_numpy(np.arange(0       ,  lowercase_ ).copy().astype(np.intaa ) )






						def      UpperCAmelCase__					( self			:Any       ,  lowercase_			:torch.FloatTensor       ,  lowercase_			:Optional[int] = None )    ->      torch.FloatTensor:
												return sample






						def      UpperCAmelCase__					( self			:Tuple       ,  lowercase_			:int       ,  lowercase_			:Union[str, torch.device] = None )    ->      int:

												if num_inference_steps > self.config.num_train_timesteps:
																		raise ValueError(
																		    f"""`num_inference_steps`: {num_inference_steps} cannot be larger than `self.config.train_timesteps`:"""
																		    f""" {self.config.num_train_timesteps} as the unet model trained with this scheduler can only handle"""
																		    f""" maximal {self.config.num_train_timesteps} timesteps.""" )

												UpperCAmelCase					=			num_inference_steps
												UpperCAmelCase					=			self.config.num_train_timesteps // self.num_inference_steps
												# creates integer timesteps by multiplying by ratio
												# casting to int to avoid issues when num_inference_step is power of 3
												UpperCAmelCase					=			(np.arange(0       ,  lowercase_ ) * step_ratio).round().copy().astype(np.intaa )
												UpperCAmelCase					=			torch.from_numpy(lowercase_ ).to(lowercase_ )
												self.timesteps += self.config.steps_offset






						def      UpperCAmelCase__					( self			:Optional[Any]       ,  lowercase_			:torch.FloatTensor       ,  lowercase_			:int       ,  lowercase_			:torch.FloatTensor       ,  lowercase_			:float = 0.0       ,  lowercase_			:bool = False       ,  lowercase_			:Optional[torch.FloatTensor] = None       ,  lowercase_			:bool = True       ,  )    ->      Union[DDIMSchedulerOutput, Tuple]:
												# 1. get previous step value (=t+1)
												UpperCAmelCase					=			timestep + self.config.num_train_timesteps // self.num_inference_steps

												# 2. compute alphas, betas
												# change original implementation to exactly match noise levels for analogous forward process
												UpperCAmelCase					=			self.alphas_cumprod[timestep]
												UpperCAmelCase					=			(
												    self.alphas_cumprod[prev_timestep]
												    if prev_timestep < self.config.num_train_timesteps
												    else self.final_alpha_cumprod
												)

												UpperCAmelCase					=			1 - alpha_prod_t

												# 3. compute predicted original sample from predicted noise also called
												# "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
												if self.config.prediction_type == "epsilon":
																		UpperCAmelCase					=			(sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
																		UpperCAmelCase					=			model_output
												elif self.config.prediction_type == "sample":
																		UpperCAmelCase					=			model_output
																		UpperCAmelCase					=			(sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
												elif self.config.prediction_type == "v_prediction":
																		UpperCAmelCase					=			(alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
																		UpperCAmelCase					=			(alpha_prod_t**0.5) * model_output + (beta_prod_t**0.5) * sample
												else:
																		raise ValueError(
																		    f"""prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample`, or"""
																		    ' `v_prediction`' )

												# 4. Clip or threshold "predicted x_0"
												if self.config.clip_sample:
																		UpperCAmelCase					=			pred_original_sample.clamp(
																		    -self.config.clip_sample_range       ,  self.config.clip_sample_range )

												# 5. compute "direction pointing to x_t" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
												UpperCAmelCase					=			(1 - alpha_prod_t_prev) ** 0.5 * pred_epsilon

												# 6. compute x_t without "random noise" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
												UpperCAmelCase					=			alpha_prod_t_prev ** 0.5 * pred_original_sample + pred_sample_direction

												if not return_dict:
																		return (prev_sample, pred_original_sample)
												return DDIMSchedulerOutput(prev_sample=lowercase_       ,  pred_original_sample=lowercase_ )






						def __len__( self			:int )    ->      Optional[int]:
												return self.config.num_train_timesteps










"""simple docstring"""


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



						"""simple docstring"""






						def      UpperCAmelCase__					( self			:Tuple )    ->      Union[str, Any]:
												# clean up the VRAM after each test
												super().tearDown()
												gc.collect()
												torch.cuda.empty_cache()






						@property
						def      UpperCAmelCase__					( self			:int )    ->      Union[str, Any]:
												UpperCAmelCase					=			1
												UpperCAmelCase					=			3
												UpperCAmelCase					=			(32, 32)

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






						@property
						def      UpperCAmelCase__					( self			:List[Any] )    ->      Tuple:
												torch.manual_seed(0 )
												UpperCAmelCase					=			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=lowercase_       ,  only_cross_attention=(True, True, False)       ,  num_class_embeds=1_00       ,  )
												return model






						@property
						def      UpperCAmelCase__					( self			:str )    ->      Any:
												torch.manual_seed(0 )
												UpperCAmelCase					=			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      UpperCAmelCase__					( self			:Optional[int] )    ->      List[str]:
												torch.manual_seed(0 )
												UpperCAmelCase					=			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(lowercase_ )






						def      UpperCAmelCase__					( self			:str )    ->      Union[str, Any]:
												UpperCAmelCase					=			'cpu'  # ensure determinism for the device-dependent torch.Generator
												UpperCAmelCase					=			self.dummy_cond_unet_upscale
												UpperCAmelCase					=			DDPMScheduler()
												UpperCAmelCase					=			DDIMScheduler(prediction_type='v_prediction' )
												UpperCAmelCase					=			self.dummy_vae
												UpperCAmelCase					=			self.dummy_text_encoder
												UpperCAmelCase					=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

												UpperCAmelCase					=			self.dummy_image.cpu().permute(0       ,  2       ,  3       ,  1 )[0]
												UpperCAmelCase					=			Image.fromarray(np.uinta(lowercase_ ) ).convert('RGB' ).resize((64, 64) )

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

												UpperCAmelCase					=			'A painting of a squirrel eating a burger'
												UpperCAmelCase					=			torch.Generator(device=lowercase_ ).manual_seed(0 )
												UpperCAmelCase					=			sd_pipe(
												    [prompt]       ,  image=lowercase_       ,  generator=lowercase_       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type='np'       ,  )

												UpperCAmelCase					=			output.images

												UpperCAmelCase					=			torch.Generator(device=lowercase_ ).manual_seed(0 )
												UpperCAmelCase					=			sd_pipe(
												    [prompt]       ,  image=lowercase_       ,  generator=lowercase_       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type='np'       ,  return_dict=lowercase_       ,  )[0]

												UpperCAmelCase					=			image[0, -3:, -3:, -1]
												UpperCAmelCase					=			image_from_tuple[0, -3:, -3:, -1]

												UpperCAmelCase					=			low_res_image.size[0] * 4
												assert image.shape == (1, expected_height_width, expected_height_width, 3)
												UpperCAmelCase					=			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      UpperCAmelCase__					( self			:Optional[Any] )    ->      Any:
												UpperCAmelCase					=			'cpu'  # ensure determinism for the device-dependent torch.Generator
												UpperCAmelCase					=			self.dummy_cond_unet_upscale
												UpperCAmelCase					=			DDPMScheduler()
												UpperCAmelCase					=			DDIMScheduler(prediction_type='v_prediction' )
												UpperCAmelCase					=			self.dummy_vae
												UpperCAmelCase					=			self.dummy_text_encoder
												UpperCAmelCase					=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

												UpperCAmelCase					=			self.dummy_image.cpu().permute(0       ,  2       ,  3       ,  1 )[0]
												UpperCAmelCase					=			Image.fromarray(np.uinta(lowercase_ ) ).convert('RGB' ).resize((64, 64) )

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

												UpperCAmelCase					=			'A painting of a squirrel eating a burger'
												UpperCAmelCase					=			sd_pipe(
												    2 * [prompt]       ,  image=2 * [low_res_image]       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type='np'       ,  )
												UpperCAmelCase					=			output.images
												assert image.shape[0] == 2

												UpperCAmelCase					=			torch.Generator(device=lowercase_ ).manual_seed(0 )
												UpperCAmelCase					=			sd_pipe(
												    [prompt]       ,  image=lowercase_       ,  generator=lowercase_       ,  num_images_per_prompt=2       ,  guidance_scale=6.0       ,  noise_level=20       ,  num_inference_steps=2       ,  output_type='np'       ,  )
												UpperCAmelCase					=			output.images
												assert image.shape[0] == 2






						@unittest.skipIf(torch_device != 'cuda'       ,  'This test requires a GPU' )
						def      UpperCAmelCase__					( self			:Optional[int] )    ->      Optional[Any]:
												UpperCAmelCase					=			self.dummy_cond_unet_upscale
												UpperCAmelCase					=			DDPMScheduler()
												UpperCAmelCase					=			DDIMScheduler(prediction_type='v_prediction' )
												UpperCAmelCase					=			self.dummy_vae
												UpperCAmelCase					=			self.dummy_text_encoder
												UpperCAmelCase					=			CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

												UpperCAmelCase					=			self.dummy_image.cpu().permute(0       ,  2       ,  3       ,  1 )[0]
												UpperCAmelCase					=			Image.fromarray(np.uinta(lowercase_ ) ).convert('RGB' ).resize((64, 64) )

												# put models in fp16, except vae as it overflows in fp16
												UpperCAmelCase					=			unet.half()
												UpperCAmelCase					=			text_encoder.half()

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

												UpperCAmelCase					=			'A painting of a squirrel eating a burger'
												UpperCAmelCase					=			torch.manual_seed(0 )
												UpperCAmelCase					=			sd_pipe(
												    [prompt]       ,  image=lowercase_       ,  generator=lowercase_       ,  num_inference_steps=2       ,  output_type='np'       ,  ).images

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







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



						"""simple docstring"""






						def      UpperCAmelCase__					( self			:Dict )    ->      str:
												# clean up the VRAM after each test
												super().tearDown()
												gc.collect()
												torch.cuda.empty_cache()






						def      UpperCAmelCase__					( self			:Optional[int] )    ->      int:
												UpperCAmelCase					=			load_image(
												    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
												    '/sd2-upscale/low_res_cat.png' )
												UpperCAmelCase					=			load_numpy(
												    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale'
												    '/upsampled_cat.npy' )

												UpperCAmelCase					=			'stabilityai/stable-diffusion-x4-upscaler'
												UpperCAmelCase					=			StableDiffusionUpscalePipeline.from_pretrained(lowercase_ )
												pipe.to(lowercase_ )
												pipe.set_progress_bar_config(disable=lowercase_ )
												pipe.enable_attention_slicing()

												UpperCAmelCase					=			'a cat sitting on a park bench'

												UpperCAmelCase					=			torch.manual_seed(0 )
												UpperCAmelCase					=			pipe(
												    prompt=lowercase_       ,  image=lowercase_       ,  generator=lowercase_       ,  output_type='np'       ,  )
												UpperCAmelCase					=			output.images[0]

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






						def      UpperCAmelCase__					( self			:Union[str, Any] )    ->      Tuple:
												UpperCAmelCase					=			load_image(
												    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
												    '/sd2-upscale/low_res_cat.png' )
												UpperCAmelCase					=			load_numpy(
												    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale'
												    '/upsampled_cat_fp16.npy' )

												UpperCAmelCase					=			'stabilityai/stable-diffusion-x4-upscaler'
												UpperCAmelCase					=			StableDiffusionUpscalePipeline.from_pretrained(
												    lowercase_       ,  torch_dtype=torch.floataa       ,  )
												pipe.to(lowercase_ )
												pipe.set_progress_bar_config(disable=lowercase_ )
												pipe.enable_attention_slicing()

												UpperCAmelCase					=			'a cat sitting on a park bench'

												UpperCAmelCase					=			torch.manual_seed(0 )
												UpperCAmelCase					=			pipe(
												    prompt=lowercase_       ,  image=lowercase_       ,  generator=lowercase_       ,  output_type='np'       ,  )
												UpperCAmelCase					=			output.images[0]

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






						def      UpperCAmelCase__					( self			:Any )    ->      List[Any]:
												torch.cuda.empty_cache()
												torch.cuda.reset_max_memory_allocated()
												torch.cuda.reset_peak_memory_stats()

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

												UpperCAmelCase					=			'stabilityai/stable-diffusion-x4-upscaler'
												UpperCAmelCase					=			StableDiffusionUpscalePipeline.from_pretrained(
												    lowercase_       ,  torch_dtype=torch.floataa       ,  )
												pipe.to(lowercase_ )
												pipe.set_progress_bar_config(disable=lowercase_ )
												pipe.enable_attention_slicing(1 )
												pipe.enable_sequential_cpu_offload()

												UpperCAmelCase					=			'a cat sitting on a park bench'

												UpperCAmelCase					=			torch.manual_seed(0 )
												UpperCAmelCase					=			pipe(
												    prompt=lowercase_       ,  image=lowercase_       ,  generator=lowercase_       ,  num_inference_steps=5       ,  output_type='np'       ,  )

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





