code: import math
  from collections import defaultdict
  from typing import List, Optional, Tuple, Union
  
  import numpy as np
  import torch
  
  from ..configuration_utils import ConfigMixin, register_to_config
  from ..utils import randn_tensor
  from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin, SchedulerOutput
  
  def      SCREAMING_SNAKE_CASE_  (		snake_case__     ,						snake_case__=0.9_99     ,						snake_case__="cosine"     ,						)  ->			List[Any]:
  							if alpha_transform_type == "cosine":
  
  														def alpha_bar_fn(snake_case__							):
  																					return math.cos((t + 0.0_08) / 1.0_08 * 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(snake_case__							):
  														lowerCAmelCase				=	i / num_diffusion_timesteps
  														lowerCAmelCase				=	(i + 1) / num_diffusion_timesteps
  														betas.append(min(1 - alpha_bar_fn(snake_case__							) / alpha_bar_fn(snake_case__							)     ,						snake_case__							)							)
  							return torch.tensor(snake_case__     ,						dtype=torch.floataa							)
  
  
  
  
  
  
  class      lowercase_						(						a_   ,  a_		):
  	UpperCAmelCase_						:      Any     =       [e.name for e in KarrasDiffusionSchedulers]
  	UpperCAmelCase_						:      Optional[int]     =       2
  
  
  	@register_to_config
  	def __init__(							self	,    __SCREAMING_SNAKE_CASE = 1000	,    __SCREAMING_SNAKE_CASE = 0.0_0_0_8_5	,    __SCREAMING_SNAKE_CASE = 0.0_1_2	,    __SCREAMING_SNAKE_CASE = "linear"	,    __SCREAMING_SNAKE_CASE = None	,    __SCREAMING_SNAKE_CASE = "epsilon"	,    __SCREAMING_SNAKE_CASE = "linspace"	,    __SCREAMING_SNAKE_CASE = 0	,    )		->List[Any]:
  								if trained_betas is not None:
  															lowerCAmelCase				=	torch.tensor(_snake_case	,    dtype=torch.floataa       )
  								elif beta_schedule == "linear":
  															lowerCAmelCase				=	torch.linspace(_snake_case	,    _snake_case	,    _snake_case	,    dtype=torch.floataa       )
  								elif beta_schedule == "scaled_linear":
  															# this schedule is very specific to the latent diffusion model.
  															lowerCAmelCase				=	(
  															    torch.linspace(beta_start**0.5	,    beta_end**0.5	,    _snake_case	,    dtype=torch.floataa       ) ** 2
  															)
  								elif beta_schedule == "squaredcos_cap_v2":
  															# Glide cosine schedule
  															lowerCAmelCase				=	betas_for_alpha_bar(_snake_case       )
  								else:
  															raise NotImplementedError(F"{beta_schedule} does is not implemented for {self.__class__}"       )
  
  								lowerCAmelCase				=	1.0 - self.betas
  								lowerCAmelCase				=	torch.cumprod(self.alphas	,    dim=0       )
  
  								#  set all values
  								self.set_timesteps(_snake_case	,    _snake_case	,    _snake_case       )
  
  
  	def     SCREAMING_SNAKE_CASE_       (							self	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE=None       )		->Any:
  								if schedule_timesteps is None:
  															lowerCAmelCase				=	self.timesteps
  
  								lowerCAmelCase				=	(schedule_timesteps == timestep).nonzero()
  
  								# The sigma index that is taken for the **very** first `step`
  								# is always the second index (or the last index if there is only 1)
  								# This way we can ensure we don't accidentally skip a sigma in
  								# case we start in the middle of the denoising schedule (e.g. for image-to-image)
  								if len(self._index_counter       ) == 0:
  															lowerCAmelCase				=	1 if len(_snake_case       ) > 1 else 0
  								else:
  															lowerCAmelCase				=	timestep.cpu().item() if torch.is_tensor(_snake_case       ) else timestep
  															lowerCAmelCase				=	self._index_counter[timestep_int]
  
  								return indices[pos].item()
  
  
  	@property
  	def     SCREAMING_SNAKE_CASE_       (							self       )		->Union[str, Any]:
  								# standard deviation of the initial noise distribution
  								if self.config.timestep_spacing in ["linspace", "trailing"]:
  															return self.sigmas.max()
  
  								return (self.sigmas.max() ** 2 + 1) ** 0.5
  
  
  	def     SCREAMING_SNAKE_CASE_       (							self	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    )		->torch.FloatTensor:
  								lowerCAmelCase				=	self.index_for_timestep(_snake_case       )
  
  								if self.state_in_first_order:
  															lowerCAmelCase				=	self.sigmas[step_index]
  								else:
  															lowerCAmelCase				=	self.sigmas_interpol[step_index - 1]
  
  								lowerCAmelCase				=	sample / ((sigma**2 + 1) ** 0.5)
  								return sample
  
  
  	def     SCREAMING_SNAKE_CASE_       (							self	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE = None	,    __SCREAMING_SNAKE_CASE = None	,    )		->Dict:
  								lowerCAmelCase				=	num_inference_steps
  
  								lowerCAmelCase				=	num_train_timesteps or self.config.num_train_timesteps
  
  								# "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
  								if self.config.timestep_spacing == "linspace":
  															lowerCAmelCase				=	np.linspace(0	,    num_train_timesteps - 1	,    _snake_case	,    dtype=_snake_case       )[::-1].copy()
  								elif self.config.timestep_spacing == "leading":
  															lowerCAmelCase				=	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
  															lowerCAmelCase				=	(np.arange(0	,    _snake_case       ) * step_ratio).round()[::-1].copy().astype(_snake_case       )
  															timesteps += self.config.steps_offset
  								elif self.config.timestep_spacing == "trailing":
  															lowerCAmelCase				=	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
  															lowerCAmelCase				=	(np.arange(_snake_case	,    0	,    -step_ratio       )).round().copy().astype(_snake_case       )
  															timesteps -= 1
  								else:
  															raise ValueError(
  															    F"{self.config.timestep_spacing} is not supported. Please make sure to choose one of \'linspace\', \'leading\' or \'trailing\'."       )
  
  								lowerCAmelCase				=	np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5       )
  								lowerCAmelCase				=	torch.from_numpy(np.log(_snake_case       )       ).to(_snake_case       )
  
  								lowerCAmelCase				=	np.interp(_snake_case	,    np.arange(0	,    len(_snake_case       )       )	,    _snake_case       )
  								lowerCAmelCase				=	np.concatenate([sigmas, [0.0]]       ).astype(np.floataa       )
  								lowerCAmelCase				=	torch.from_numpy(_snake_case       ).to(device=_snake_case       )
  
  								# compute up and down sigmas
  								lowerCAmelCase				=	sigmas.roll(-1       )
  								lowerCAmelCase				=	0.0
  								lowerCAmelCase				=	(sigmas_next**2 * (sigmas**2 - sigmas_next**2) / sigmas**2) ** 0.5
  								lowerCAmelCase				=	(sigmas_next**2 - sigmas_up**2) ** 0.5
  								lowerCAmelCase				=	0.0
  
  								# compute interpolated sigmas
  								lowerCAmelCase				=	sigmas.log().lerp(sigmas_down.log()	,    0.5       ).exp()
  								lowerCAmelCase				=	0.0
  
  								# set sigmas
  								lowerCAmelCase				=	torch.cat([sigmas[:1], sigmas[1:].repeat_interleave(2       ), sigmas[-1:]]       )
  								lowerCAmelCase				=	torch.cat(
  								    [sigmas_interpol[:1], sigmas_interpol[1:].repeat_interleave(2       ), sigmas_interpol[-1:]]       )
  								lowerCAmelCase				=	torch.cat([sigmas_up[:1], sigmas_up[1:].repeat_interleave(2       ), sigmas_up[-1:]]       )
  								lowerCAmelCase				=	torch.cat([sigmas_down[:1], sigmas_down[1:].repeat_interleave(2       ), sigmas_down[-1:]]       )
  
  								if str(_snake_case       ).startswith('''mps'''       ):
  															# mps does not support float64
  															lowerCAmelCase				=	torch.from_numpy(_snake_case       ).to(_snake_case	,    dtype=torch.floataa       )
  								else:
  															lowerCAmelCase				=	torch.from_numpy(_snake_case       ).to(_snake_case       )
  
  								lowerCAmelCase				=	self.sigma_to_t(_snake_case       ).to(_snake_case	,    dtype=timesteps.dtype       )
  								lowerCAmelCase				=	torch.stack((timesteps_interpol[:-2, None], timesteps[1:, None])	,    dim=-1       ).flatten()
  
  								lowerCAmelCase				=	torch.cat([timesteps[:1], interleaved_timesteps]       )
  
  								lowerCAmelCase				=	None
  
  								# for exp beta schedules, such as the one for `pipeline_shap_e.py`
  								# we need an index counter
  								lowerCAmelCase				=	defaultdict(_snake_case       )
  
  
  	def     SCREAMING_SNAKE_CASE_       (							self	,    __SCREAMING_SNAKE_CASE       )		->List[Any]:
  								# get log sigma
  								lowerCAmelCase				=	sigma.log()
  
  								# get distribution
  								lowerCAmelCase				=	log_sigma - self.log_sigmas[:, None]
  
  								# get sigmas range
  								lowerCAmelCase				=	dists.ge(0       ).cumsum(dim=0       ).argmax(dim=0       ).clamp(max=self.log_sigmas.shape[0] - 2       )
  								lowerCAmelCase				=	low_idx + 1
  
  								lowerCAmelCase				=	self.log_sigmas[low_idx]
  								lowerCAmelCase				=	self.log_sigmas[high_idx]
  
  								# interpolate sigmas
  								lowerCAmelCase				=	(low - log_sigma) / (low - high)
  								lowerCAmelCase				=	w.clamp(0	,    1       )
  
  								# transform interpolation to time range
  								lowerCAmelCase				=	(1 - w) * low_idx + w * high_idx
  								lowerCAmelCase				=	t.view(sigma.shape       )
  								return t
  
  
  	@property
  	def     SCREAMING_SNAKE_CASE_       (							self       )		->Any:
  								return self.sample is None
  
  
  	def     SCREAMING_SNAKE_CASE_       (							self	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE = None	,    __SCREAMING_SNAKE_CASE = True	,    )		->Union[SchedulerOutput, Tuple]:
  								lowerCAmelCase				=	self.index_for_timestep(_snake_case       )
  
  								# advance index counter by 1
  								lowerCAmelCase				=	timestep.cpu().item() if torch.is_tensor(_snake_case       ) else timestep
  								self._index_counter[timestep_int] += 1
  
  								if self.state_in_first_order:
  															lowerCAmelCase				=	self.sigmas[step_index]
  															lowerCAmelCase				=	self.sigmas_interpol[step_index]
  															lowerCAmelCase				=	self.sigmas_up[step_index]
  															lowerCAmelCase				=	self.sigmas_down[step_index - 1]
  								else:
  															# 2nd order / KPDM2's method
  															lowerCAmelCase				=	self.sigmas[step_index - 1]
  															lowerCAmelCase				=	self.sigmas_interpol[step_index - 1]
  															lowerCAmelCase				=	self.sigmas_up[step_index - 1]
  															lowerCAmelCase				=	self.sigmas_down[step_index - 1]
  
  								# currently only gamma=0 is supported. This usually works best anyways.
  								# We can support gamma in the future but then need to scale the timestep before
  								# passing it to the model which requires a change in API
  								lowerCAmelCase				=	0
  								lowerCAmelCase				=	sigma * (gamma + 1)  # Note: sigma_hat == sigma for now
  
  								lowerCAmelCase				=	model_output.device
  								lowerCAmelCase				=	randn_tensor(model_output.shape	,    dtype=model_output.dtype	,    device=_snake_case	,    generator=_snake_case       )
  
  								# 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
  								if self.config.prediction_type == "epsilon":
  															lowerCAmelCase				=	sigma_hat if self.state_in_first_order else sigma_interpol
  															lowerCAmelCase				=	sample - sigma_input * model_output
  								elif self.config.prediction_type == "v_prediction":
  															lowerCAmelCase				=	sigma_hat if self.state_in_first_order else sigma_interpol
  															lowerCAmelCase				=	model_output * (-sigma_input / (sigma_input**2 + 1) ** 0.5) + (
  															    sample / (sigma_input**2 + 1)
  															)
  								elif self.config.prediction_type == "sample":
  															raise NotImplementedError('''prediction_type not implemented yet: sample'''       )
  								else:
  															raise ValueError(
  															    F"prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`"       )
  
  								if self.state_in_first_order:
  															# 2. Convert to an ODE derivative for 1st order
  															lowerCAmelCase				=	(sample - pred_original_sample) / sigma_hat
  															# 3. delta timestep
  															lowerCAmelCase				=	sigma_interpol - sigma_hat
  
  															# store for 2nd order step
  															lowerCAmelCase				=	sample
  															lowerCAmelCase				=	dt
  															lowerCAmelCase				=	sample + derivative * dt
  								else:
  															# DPM-Solver-2
  															# 2. Convert to an ODE derivative for 2nd order
  															lowerCAmelCase				=	(sample - pred_original_sample) / sigma_interpol
  															# 3. delta timestep
  															lowerCAmelCase				=	sigma_down - sigma_hat
  
  															lowerCAmelCase				=	self.sample
  															lowerCAmelCase				=	None
  
  															lowerCAmelCase				=	sample + derivative * dt
  															lowerCAmelCase				=	prev_sample + noise * sigma_up
  
  								if not return_dict:
  															return (prev_sample,)
  
  								return SchedulerOutput(prev_sample=_snake_case       )
  
  
  	def     SCREAMING_SNAKE_CASE_       (							self	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    __SCREAMING_SNAKE_CASE	,    )		->torch.FloatTensor:
  								# Make sure sigmas and timesteps have the same device and dtype as original_samples
  								lowerCAmelCase				=	self.sigmas.to(device=original_samples.device	,    dtype=original_samples.dtype       )
  								if original_samples.device.type == "mps" and torch.is_floating_point(_snake_case       ):
  															# mps does not support float64
  															lowerCAmelCase				=	self.timesteps.to(original_samples.device	,    dtype=torch.floataa       )
  															lowerCAmelCase				=	timesteps.to(original_samples.device	,    dtype=torch.floataa       )
  								else:
  															lowerCAmelCase				=	self.timesteps.to(original_samples.device       )
  															lowerCAmelCase				=	timesteps.to(original_samples.device       )
  
  								lowerCAmelCase				=	[self.index_for_timestep(_snake_case	,    _snake_case       ) for t in timesteps]
  
  								lowerCAmelCase				=	sigmas[step_indices].flatten()
  								while len(sigma.shape       ) < len(original_samples.shape       ):
  															lowerCAmelCase				=	sigma.unsqueeze(-1       )
  
  								lowerCAmelCase				=	original_samples + noise * sigma
  								return noisy_samples
  
  
  
  
  	def __len__(							self       )		->str:
  								return self.config.num_train_timesteps
  
  
  
  
style_context: from dataclasses import dataclass
  from typing import List, Optional, Union
  
  import numpy as np
  import PIL
  
  from ...utils import BaseOutput, is_invisible_watermark_available, is_torch_available, is_transformers_available
  
  
  
  
  @dataclass
  class      lowercase_						(						UpperCamelCase_		):
  
  
  
  	"""simple docstring"""
  
  
  
  
  
  	UpperCAmelCase_						:      Union[List[PIL.Image.Image], np.ndarray]
  
  
  if is_transformers_available() and is_torch_available() and is_invisible_watermark_available():
  	from .pipeline_stable_diffusion_xl import StableDiffusionXLPipeline
  	from .pipeline_stable_diffusion_xl_imgaimg import StableDiffusionXLImgaImgPipeline
  
  
  
  