




import math
from dataclasses import dataclass
from typing import Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, randn_tensor
from .scheduling_utils import SchedulerMixin


@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->UnCLIP
class 							__UpperCamelCase			( SCREAMING_SNAKE_CASE__  ):





				'''simple docstring'''



				__a							: int										=4_2
				__a							: Union[str, Any]										=None




def       UpperCAmelCase       (     _snake_case						, _snake_case=0.999						, _snake_case="cosine"						, ):






							if alpha_transform_type == "cosine":

														def alpha_bar_fn(_snake_case							):
																					return math.cos((t + 0.008) / 1.008 * math.pi / 2							) ** 2

							elif alpha_transform_type == "exp":

														def alpha_bar_fn(_snake_case							):
																					return math.exp(t * -12.0							)

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

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


class 							__UpperCamelCase			( SCREAMING_SNAKE_CASE__     ,					SCREAMING_SNAKE_CASE__  ):





				'''simple docstring'''



				@register_to_config
				def __init__( self       , UpperCAmelCase_ = 10_00       , UpperCAmelCase_ = "fixed_small_log"       , UpperCAmelCase_ = True       , UpperCAmelCase_ = 1.0       , UpperCAmelCase_ = "epsilon"       , UpperCAmelCase_ = "squaredcos_cap_v2"       , ):

											if beta_schedule != "squaredcos_cap_v2":
																		raise ValueError('''UnCLIPScheduler only supports `beta_schedule`: \'squaredcos_cap_v2\''''						)

											lowerCAmelCase   =		betas_for_alpha_bar(_lowercase						)

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

											# standard deviation of the initial noise distribution
											lowerCAmelCase   =		1.0

											# setable values
											lowerCAmelCase   =		None
											lowerCAmelCase   =		torch.from_numpy(np.arange(0       , _lowercase						)[::-1].copy()						)

											lowerCAmelCase   =		variance_type



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_ = None						):
											return sample



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_ = None						):
											lowerCAmelCase   =		num_inference_steps
											lowerCAmelCase   =		(self.config.num_train_timesteps - 1) / (self.num_inference_steps - 1)
											lowerCAmelCase   =		(np.arange(0       , _lowercase						) * step_ratio).round()[::-1].copy().astype(np.intaa						)
											lowerCAmelCase   =		torch.from_numpy(_lowercase						).to(_lowercase						)



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_=None       , UpperCAmelCase_=None       , UpperCAmelCase_=None						):

											if prev_timestep is None:
																		lowerCAmelCase   =		t - 1

											lowerCAmelCase   =		self.alphas_cumprod[t]
											lowerCAmelCase   =		self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
											lowerCAmelCase   =		1 - alpha_prod_t
											lowerCAmelCase   =		1 - alpha_prod_t_prev

											if prev_timestep == t - 1:
																		lowerCAmelCase   =		self.betas[t]
											else:
																		lowerCAmelCase   =		1 - alpha_prod_t / alpha_prod_t_prev

											# For t > 0, compute predicted variance βt (see formula (6) and (7) from https://arxiv.org/pdf/2006.11239.pdf)
											# and sample from it to get previous sample
											# x_{t-1} ~ N(pred_prev_sample, variance) == add variance to pred_sample
											lowerCAmelCase   =		beta_prod_t_prev / beta_prod_t * beta

											if variance_type is None:
																		lowerCAmelCase   =		self.config.variance_type

											# hacks - were probably added for training stability
											if variance_type == "fixed_small_log":
																		lowerCAmelCase   =		torch.log(torch.clamp(_lowercase       , min=1E-2_0						)						)
																		lowerCAmelCase   =		torch.exp(0.5 * variance						)
											elif variance_type == "learned_range":
																		# NOTE difference with DDPM scheduler
																		lowerCAmelCase   =		variance.log()
																		lowerCAmelCase   =		beta.log()

																		lowerCAmelCase   =		(predicted_variance + 1) / 2
																		lowerCAmelCase   =		frac * max_log + (1 - frac) * min_log

											return variance



				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_       , UpperCAmelCase_       , UpperCAmelCase_ = None       , UpperCAmelCase_=None       , UpperCAmelCase_ = True       , ):
											lowerCAmelCase   =		timestep

											if model_output.shape[1] == sample.shape[1] * 2 and self.variance_type == "learned_range":
																		lowerCAmelCase   =		torch.split(_lowercase       , sample.shape[1]       , dim=1						)
											else:
																		lowerCAmelCase   =		None

											# 1. compute alphas, betas
											if prev_timestep is None:
																		lowerCAmelCase   =		t - 1

											lowerCAmelCase   =		self.alphas_cumprod[t]
											lowerCAmelCase   =		self.alphas_cumprod[prev_timestep] if prev_timestep >= 0 else self.one
											lowerCAmelCase   =		1 - alpha_prod_t
											lowerCAmelCase   =		1 - alpha_prod_t_prev

											if prev_timestep == t - 1:
																		lowerCAmelCase   =		self.betas[t]
																		lowerCAmelCase   =		self.alphas[t]
											else:
																		lowerCAmelCase   =		1 - alpha_prod_t / alpha_prod_t_prev
																		lowerCAmelCase   =		1 - beta

											# 2. compute predicted original sample from predicted noise also called
											# "predicted x_0" of formula (15) from https://arxiv.org/pdf/2006.11239.pdf
											if self.config.prediction_type == "epsilon":
																		lowerCAmelCase   =		(sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
											elif self.config.prediction_type == "sample":
																		lowerCAmelCase   =		model_output
											else:
																		raise ValueError(
																		    F"""prediction_type given as {self.config.prediction_type} must be one of `epsilon` or `sample`"""
																		    ''' for the UnCLIPScheduler.'''						)

											# 3. Clip "predicted x_0"
											if self.config.clip_sample:
																		lowerCAmelCase   =		torch.clamp(
																		    _lowercase       , -self.config.clip_sample_range       , self.config.clip_sample_range						)

											# 4. Compute coefficients for pred_original_sample x_0 and current sample x_t
											# See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
											lowerCAmelCase   =		(alpha_prod_t_prev ** 0.5 * beta) / beta_prod_t
											lowerCAmelCase   =		alpha ** 0.5 * beta_prod_t_prev / beta_prod_t

											# 5. Compute predicted previous sample µ_t
											# See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
											lowerCAmelCase   =		pred_original_sample_coeff * pred_original_sample + current_sample_coeff * sample

											# 6. Add noise
											lowerCAmelCase   =		0
											if t > 0:
																		lowerCAmelCase   =		randn_tensor(
																		    model_output.shape       , dtype=model_output.dtype       , generator=_lowercase       , device=model_output.device						)

																		lowerCAmelCase   =		self._get_variance(
																		    _lowercase       , predicted_variance=_lowercase       , prev_timestep=_lowercase       , )

																		if self.variance_type == "fixed_small_log":
																									lowerCAmelCase   =		variance
																		elif self.variance_type == "learned_range":
																									lowerCAmelCase   =		(0.5 * variance).exp()
																		else:
																									raise ValueError(
																									    F"""variance_type given as {self.variance_type} must be one of `fixed_small_log` or `learned_range`"""
																									    ''' for the UnCLIPScheduler.'''						)

																		lowerCAmelCase   =		variance * variance_noise

											lowerCAmelCase   =		pred_prev_sample + variance

											if not return_dict:
																		return (pred_prev_sample,)

											return UnCLIPSchedulerOutput(prev_sample=_lowercase       , pred_original_sample=_lowercase						)






				def       __snake_case		( self       , UpperCAmelCase_       , UpperCAmelCase_       , UpperCAmelCase_       , ):
											lowerCAmelCase   =		self.alphas_cumprod.to(device=original_samples.device       , dtype=original_samples.dtype						)
											lowerCAmelCase   =		timesteps.to(original_samples.device						)

											lowerCAmelCase   =		alphas_cumprod[timesteps] ** 0.5
											lowerCAmelCase   =		sqrt_alpha_prod.flatten()
											while len(sqrt_alpha_prod.shape						) < len(original_samples.shape						):
																		lowerCAmelCase   =		sqrt_alpha_prod.unsqueeze(-1						)

											lowerCAmelCase   =		(1 - alphas_cumprod[timesteps]) ** 0.5
											lowerCAmelCase   =		sqrt_one_minus_alpha_prod.flatten()
											while len(sqrt_one_minus_alpha_prod.shape						) < len(original_samples.shape						):
																		lowerCAmelCase   =		sqrt_one_minus_alpha_prod.unsqueeze(-1						)

											lowerCAmelCase   =		sqrt_alpha_prod * original_samples + sqrt_one_minus_alpha_prod * noise
											return noisy_samples









import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
		import onnxruntime as ort


class 							__UpperCamelCase			( __UpperCAmelCase     ,					unittest.TestCase  ):





				'''simple docstring'''



				__a							: Any										="""hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"""



				def       __snake_case		( self       , UpperCAmelCase_=0						):
											lowerCAmelCase   =		floats_tensor((1, 3, 1_28, 1_28)       , rng=random.Random(UpperCAmelCase_						)						)
											lowerCAmelCase   =		np.random.RandomState(UpperCAmelCase_						)
											lowerCAmelCase   =		{
											    '''prompt''': '''A painting of a squirrel eating a burger''',
											    '''image''': image,
											    '''generator''': generator,
											    '''num_inference_steps''': 3,
											    '''strength''': 0.75,
											    '''guidance_scale''': 7.5,
											    '''output_type''': '''numpy''',
											}
											return inputs



				def       __snake_case		( self						):
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       , provider='''CPUExecutionProvider'''						)
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

											lowerCAmelCase   =		self.get_dummy_inputs()
											lowerCAmelCase   =		pipe(**UpperCAmelCase_						).images
											lowerCAmelCase   =		image[0, -3:, -3:, -1].flatten()

											assert image.shape == (1, 1_28, 1_28, 3)
											lowerCAmelCase   =		np.array([0.69643, 0.58484, 0.50314, 0.58760, 0.55368, 0.59643, 0.51529, 0.41217, 0.49087]						)
											assert np.abs(image_slice - expected_slice						).max() < 1E-1



				def       __snake_case		( self						):
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       , provider='''CPUExecutionProvider'''						)
											lowerCAmelCase   =		PNDMScheduler.from_config(pipe.scheduler.config       , skip_prk_steps=UpperCAmelCase_						)
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

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

											assert image.shape == (1, 1_28, 1_28, 3)
											lowerCAmelCase   =		np.array([0.61737, 0.54642, 0.53183, 0.54465, 0.52742, 0.60525, 0.49969, 0.40655, 0.48154]						)

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



				def       __snake_case		( self						):
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       , provider='''CPUExecutionProvider'''						)
											lowerCAmelCase   =		LMSDiscreteScheduler.from_config(pipe.scheduler.config						)
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

											# warmup pass to apply optimizations
											lowerCAmelCase   =		pipe(**self.get_dummy_inputs()						)

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

											assert image.shape == (1, 1_28, 1_28, 3)
											lowerCAmelCase   =		np.array([0.52761, 0.59977, 0.49033, 0.49619, 0.54282, 0.50311, 0.47600, 0.40918, 0.45203]						)

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



				def       __snake_case		( self						):
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       , provider='''CPUExecutionProvider'''						)
											lowerCAmelCase   =		EulerDiscreteScheduler.from_config(pipe.scheduler.config						)
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

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

											assert image.shape == (1, 1_28, 1_28, 3)
											lowerCAmelCase   =		np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304]						)

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



				def       __snake_case		( self						):
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       , provider='''CPUExecutionProvider'''						)
											lowerCAmelCase   =		EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config						)
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

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

											assert image.shape == (1, 1_28, 1_28, 3)
											lowerCAmelCase   =		np.array([0.52911, 0.60004, 0.49229, 0.49805, 0.54502, 0.50680, 0.47777, 0.41028, 0.45304]						)

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






				def       __snake_case		( self						):
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint       , provider='''CPUExecutionProvider'''						)
											lowerCAmelCase   =		DPMSolverMultistepScheduler.from_config(pipe.scheduler.config						)
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

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

											assert image.shape == (1, 1_28, 1_28, 3)
											lowerCAmelCase   =		np.array([0.65331, 0.58277, 0.48204, 0.56059, 0.53665, 0.56235, 0.50969, 0.40009, 0.46552]						)

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




@nightly
@require_onnxruntime
@require_torch_gpu
class 							__UpperCamelCase			( unittest.TestCase  ):





				'''simple docstring'''



				@property
				def       __snake_case		( self						):
											return (
											    "CUDAExecutionProvider",
											    {
											        "gpu_mem_limit": "15000000000",  # 15GB
											        "arena_extend_strategy": "kSameAsRequested",
											    },
											)



				@property
				def       __snake_case		( self						):
											lowerCAmelCase   =		ort.SessionOptions()
											lowerCAmelCase   =		False
											return options



				def       __snake_case		( self						):
											lowerCAmelCase   =		load_image(
											    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
											    '''/img2img/sketch-mountains-input.jpg'''						)
											lowerCAmelCase   =		init_image.resize((7_68, 5_12)						)
											# using the PNDM scheduler by default
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(
											    '''CompVis/stable-diffusion-v1-4'''       , revision='''onnx'''       , safety_checker=UpperCAmelCase_       , feature_extractor=UpperCAmelCase_       , provider=self.gpu_provider       , sess_options=self.gpu_options       , )
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

											lowerCAmelCase   =		'''A fantasy landscape, trending on artstation'''

											lowerCAmelCase   =		np.random.RandomState(0						)
											lowerCAmelCase   =		pipe(
											    prompt=UpperCAmelCase_       , image=UpperCAmelCase_       , strength=0.75       , guidance_scale=7.5       , num_inference_steps=10       , generator=UpperCAmelCase_       , output_type='''np'''       , )
											lowerCAmelCase   =		output.images
											lowerCAmelCase   =		images[0, 2_55:2_58, 3_83:3_86, -1]

											assert images.shape == (1, 5_12, 7_68, 3)
											lowerCAmelCase   =		np.array([0.4909, 0.5059, 0.5372, 0.4623, 0.4876, 0.5049, 0.4820, 0.4956, 0.5019]						)
											# TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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






				def       __snake_case		( self						):
											lowerCAmelCase   =		load_image(
											    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
											    '''/img2img/sketch-mountains-input.jpg'''						)
											lowerCAmelCase   =		init_image.resize((7_68, 5_12)						)
											lowerCAmelCase   =		LMSDiscreteScheduler.from_pretrained(
											    '''runwayml/stable-diffusion-v1-5'''       , subfolder='''scheduler'''       , revision='''onnx'''						)
											lowerCAmelCase   =		OnnxStableDiffusionImgaImgPipeline.from_pretrained(
											    '''runwayml/stable-diffusion-v1-5'''       , revision='''onnx'''       , scheduler=UpperCAmelCase_       , safety_checker=UpperCAmelCase_       , feature_extractor=UpperCAmelCase_       , provider=self.gpu_provider       , sess_options=self.gpu_options       , )
											pipe.set_progress_bar_config(disable=UpperCAmelCase_						)

											lowerCAmelCase   =		'''A fantasy landscape, trending on artstation'''

											lowerCAmelCase   =		np.random.RandomState(0						)
											lowerCAmelCase   =		pipe(
											    prompt=UpperCAmelCase_       , image=UpperCAmelCase_       , strength=0.75       , guidance_scale=7.5       , num_inference_steps=20       , generator=UpperCAmelCase_       , output_type='''np'''       , )
											lowerCAmelCase   =		output.images
											lowerCAmelCase   =		images[0, 2_55:2_58, 3_83:3_86, -1]

											assert images.shape == (1, 5_12, 7_68, 3)
											lowerCAmelCase   =		np.array([0.8043, 0.926, 0.9581, 0.8119, 0.8954, 0.913, 0.7209, 0.7463, 0.7431]						)
											# TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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



