






'''simple docstring'''






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

import numpy as np
import torch

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


__A     						= logging.get_logger(__name__)  # pylint: disable=invalid-name




@dataclass
# Copied from diffusers.schedulers.scheduling_ddpm.DDPMSchedulerOutput with DDPM->EulerDiscrete
class a_     (     UpperCamelCase_				):
   _snake_case	   =				42
   _snake_case	   =				None




def 							_SCREAMING_SNAKE_CASE	(					A	:							Optional[Any]	,		A	:							Optional[int]=0.999	,		A	:							List[str]="cosine"	,		)     ->			Tuple:







   """simple docstring"""
   if alpha_transform_type == "cosine":

      def alpha_bar_fn(A	:							Optional[Any]	):
         return math.cos((t + 0.008) / 1.008 * math.pi / 2	) ** 2

   elif alpha_transform_type == "exp":

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

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

   __snake_case					:   Tuple  				=			[]
   for i in range(A	):
      __snake_case					:   str  				=			i / num_diffusion_timesteps
      __snake_case					:   Optional[Any]  				=			(i + 1) / num_diffusion_timesteps
      betas.append(min(1 - alpha_bar_fn(A	) / alpha_bar_fn(A	)	,		A	)	)
   return torch.tensor(A	,		dtype=torch.floataa	)





class a_     (     UpperCamelCase_		,   UpperCamelCase_				):
   _snake_case	   =				[e.name for e in KarrasDiffusionSchedulers]
   _snake_case	   =				1







   @register_to_config
   def __init__(self       , __a = 1_0_0_0       , __a = 0.0_001       , __a = 0.02       , __a = "linear"       , __a = None       , __a = "epsilon"       , __a = "linear"       , __a = False       , __a = "linspace"       , __a = 0       , )      ->   Optional[int]:
      """simple docstring"""






      if trained_betas is not None:
         __snake_case					:   str  				=			torch.tensor(__a       , dtype=torch.floataa)
      elif beta_schedule == "linear":
         __snake_case					:   int  				=			torch.linspace(__a       , __a       , __a       , dtype=torch.floataa)
      elif beta_schedule == "scaled_linear":
         # this schedule is very specific to the latent diffusion model.
         __snake_case					:   str  				=			(
             torch.linspace(beta_start**0.5       , beta_end**0.5       , __a       , dtype=torch.floataa) ** 2
         )
      elif beta_schedule == "squaredcos_cap_v2":
         # Glide cosine schedule
         __snake_case					:   Dict  				=			betas_for_alpha_bar(__a)
      else:
         raise NotImplementedError(F"""{beta_schedule} does is not implemented for {self.__class__}""")

      __snake_case					:   Optional[Any]  				=			1.0 - self.betas
      __snake_case					:   Union[str, Any]  				=			torch.cumprod(self.alphas       , dim=0)

      __snake_case					:   str  				=			np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5)
      __snake_case					:   Dict  				=			np.concatenate([sigmas[::-1], [0.0]]).astype(np.floataa)
      __snake_case					:   Tuple  				=			torch.from_numpy(__a)

      # setable values
      __snake_case					:   List[Any]  				=			None
      __snake_case					:   Any  				=			np.linspace(0       , num_train_timesteps - 1       , __a       , dtype=__a)[::-1].copy()
      __snake_case					:   List[str]  				=			torch.from_numpy(__a)
      __snake_case					:   int  				=			False
      __snake_case					:   Tuple  				=			use_karras_sigmas







   @property
   def      SCREAMING_SNAKE_CASE__       (self)      ->   Any:
      """simple docstring"""






      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       , __a       , __a)      ->   torch.FloatTensor:
      """simple docstring"""






      if isinstance(__a       , torch.Tensor):
         __snake_case					:   Dict  				=			timestep.to(self.timesteps.device)
      __snake_case					:   Tuple  				=			(self.timesteps == timestep).nonzero().item()
      __snake_case					:   List[str]  				=			self.sigmas[step_index]

      __snake_case					:   int  				=			sample / ((sigma**2 + 1) ** 0.5)

      __snake_case					:   Union[str, Any]  				=			True
      return sample







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a = None)      ->   Union[str, Any]:
      """simple docstring"""






      __snake_case					:   Optional[Any]  				=			num_inference_steps

      # "linspace", "leading", "trailing" corresponds to annotation of Table 2. of https://arxiv.org/abs/2305.08891
      if self.config.timestep_spacing == "linspace":
         __snake_case					:   Union[str, Any]  				=			np.linspace(0       , self.config.num_train_timesteps - 1       , __a       , dtype=__a)[
             ::-1
         ].copy()
      elif self.config.timestep_spacing == "leading":
         __snake_case					:   str  				=			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
         __snake_case					:   Any  				=			(np.arange(0       , __a) * step_ratio).round()[::-1].copy().astype(__a)
         timesteps += self.config.steps_offset
      elif self.config.timestep_spacing == "trailing":
         __snake_case					:   Dict  				=			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
         __snake_case					:   Any  				=			(np.arange(self.config.num_train_timesteps       , 0       , -step_ratio)).round().copy().astype(__a)
         timesteps -= 1
      else:
         raise ValueError(
             F"""{self.config.timestep_spacing} is not supported. Please make sure to choose one of 'linspace', 'leading' or 'trailing'.""")

      __snake_case					:   Union[str, Any]  				=			np.array(((1 - self.alphas_cumprod) / self.alphas_cumprod) ** 0.5)
      __snake_case					:   Any  				=			np.log(__a)

      if self.config.interpolation_type == "linear":
         __snake_case					:   Dict  				=			np.interp(__a       , np.arange(0       , len(__a))       , __a)
      elif self.config.interpolation_type == "log_linear":
         __snake_case					:   Dict  				=			torch.linspace(np.log(sigmas[-1])       , np.log(sigmas[0])       , num_inference_steps + 1).exp()
      else:
         raise ValueError(
             F"""{self.config.interpolation_type} is not implemented. Please specify interpolation_type to either"""
             ' \'linear\' or \'log_linear\'')

      if self.use_karras_sigmas:
         __snake_case					:   Optional[Any]  				=			self._convert_to_karras(in_sigmas=__a       , num_inference_steps=self.num_inference_steps)
         __snake_case					:   Any  				=			np.array([self._sigma_to_t(__a       , __a) for sigma in sigmas])

      __snake_case					:   Optional[Any]  				=			np.concatenate([sigmas, [0.0]]).astype(np.floataa)
      __snake_case					:   List[str]  				=			torch.from_numpy(__a).to(device=__a)
      if str(__a).startswith('mps'):
         # mps does not support float64
         __snake_case					:   Union[str, Any]  				=			torch.from_numpy(__a).to(__a       , dtype=torch.floataa)
      else:
         __snake_case					:   Optional[Any]  				=			torch.from_numpy(__a).to(device=__a)







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a)      ->   int:
      """simple docstring"""






      __snake_case					:   Any  				=			np.log(__a)

      # get distribution
      __snake_case					:   str  				=			log_sigma - log_sigmas[:, np.newaxis]

      # get sigmas range
      __snake_case					:   List[Any]  				=			np.cumsum((dists >= 0)       , axis=0).argmax(axis=0).clip(max=log_sigmas.shape[0] - 2)
      __snake_case					:   int  				=			low_idx + 1

      __snake_case					:   Union[str, Any]  				=			log_sigmas[low_idx]
      __snake_case					:   Optional[int]  				=			log_sigmas[high_idx]

      # interpolate sigmas
      __snake_case					:   str  				=			(low - log_sigma) / (low - high)
      __snake_case					:   Union[str, Any]  				=			np.clip(__a       , 0       , 1)

      # transform interpolation to time range
      __snake_case					:   Optional[Any]  				=			(1 - w) * low_idx + w * high_idx
      __snake_case					:   List[Any]  				=			t.reshape(sigma.shape)
      return t







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a)      ->   torch.FloatTensor:
      """simple docstring"""






      __snake_case					:   float  				=			in_sigmas[-1].item()
      __snake_case					:   float  				=			in_sigmas[0].item()

      __snake_case					:   Tuple  				=			7.0  # 7.0 is the value used in the paper
      __snake_case					:   Any  				=			np.linspace(0       , 1       , __a)
      __snake_case					:   Optional[int]  				=			sigma_min ** (1 / rho)
      __snake_case					:   int  				=			sigma_max ** (1 / rho)
      __snake_case					:   List[Any]  				=			(max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
      return sigmas







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a       , __a = 0.0       , __a = 0.0       , __a = float('inf')       , __a = 1.0       , __a = None       , __a = True       , )      ->   Union[EulerDiscreteSchedulerOutput, Tuple]:
      """simple docstring"""






      if (
          isinstance(__a       , __a)
          or isinstance(__a       , torch.IntTensor)
          or isinstance(__a       , torch.LongTensor)
      ):
         raise ValueError(
             (
                 'Passing integer indices (e.g. from `enumerate(timesteps)`) as timesteps to'
                 ' `EulerDiscreteScheduler.step()` is not supported. Make sure to pass'
                 ' one of the `scheduler.timesteps` as a timestep.'
             )       , )

      if not self.is_scale_input_called:
         logger.warning(
             'The `scale_model_input` function should be called before `step` to ensure correct denoising. '
             'See `StableDiffusionPipeline` for a usage example.')

      if isinstance(__a       , torch.Tensor):
         __snake_case					:   Union[str, Any]  				=			timestep.to(self.timesteps.device)

      __snake_case					:   Tuple  				=			(self.timesteps == timestep).nonzero().item()
      __snake_case					:   List[str]  				=			self.sigmas[step_index]

      __snake_case					:   Any  				=			min(s_churn / (len(self.sigmas) - 1)       , 2**0.5 - 1) if s_tmin <= sigma <= s_tmax else 0.0

      __snake_case					:   List[Any]  				=			randn_tensor(
          model_output.shape       , dtype=model_output.dtype       , device=model_output.device       , generator=__a)

      __snake_case					:   Tuple  				=			noise * s_noise
      __snake_case					:   List[Any]  				=			sigma * (gamma + 1)

      if gamma > 0:
         __snake_case					:   List[str]  				=			sample + eps * (sigma_hat**2 - sigma**2) ** 0.5

      # 1. compute predicted original sample (x_0) from sigma-scaled predicted noise
      # NOTE: "original_sample" should not be an expected prediction_type but is left in for
      # backwards compatibility
      if self.config.prediction_type == "original_sample" or self.config.prediction_type == "sample":
         __snake_case					:   int  				=			model_output
      elif self.config.prediction_type == "epsilon":
         __snake_case					:   Optional[Any]  				=			sample - sigma_hat * model_output
      elif self.config.prediction_type == "v_prediction":
         # * c_out + input * c_skip
         __snake_case					:   int  				=			model_output * (-sigma / (sigma**2 + 1) ** 0.5) + (sample / (sigma**2 + 1))
      else:
         raise ValueError(
             F"""prediction_type given as {self.config.prediction_type} must be one of `epsilon`, or `v_prediction`""")

      # 2. Convert to an ODE derivative
      __snake_case					:   Optional[Any]  				=			(sample - pred_original_sample) / sigma_hat

      __snake_case					:   Union[str, Any]  				=			self.sigmas[step_index + 1] - sigma_hat

      __snake_case					:   Optional[Any]  				=			sample + derivative * dt

      if not return_dict:
         return (prev_sample,)

      return EulerDiscreteSchedulerOutput(prev_sample=__a       , pred_original_sample=__a)







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a       , )      ->   torch.FloatTensor:
      """simple docstring"""






      __snake_case					:   Optional[Any]  				=			self.sigmas.to(device=original_samples.device       , dtype=original_samples.dtype)
      if original_samples.device.type == "mps" and torch.is_floating_point(__a):
         # mps does not support float64
         __snake_case					:   List[Any]  				=			self.timesteps.to(original_samples.device       , dtype=torch.floataa)
         __snake_case					:   Tuple  				=			timesteps.to(original_samples.device       , dtype=torch.floataa)
      else:
         __snake_case					:   Tuple  				=			self.timesteps.to(original_samples.device)
         __snake_case					:   int  				=			timesteps.to(original_samples.device)

      __snake_case					:   List[str]  				=			[(schedule_timesteps == t).nonzero().item() for t in timesteps]

      __snake_case					:   str  				=			sigmas[step_indices].flatten()
      while len(sigma.shape) < len(original_samples.shape):
         __snake_case					:   Optional[Any]  				=			sigma.unsqueeze(-1)

      __snake_case					:   str  				=			original_samples + noise * sigma
      return noisy_samples







   def __len__(self)      ->   Optional[int]:
      """simple docstring"""






      return self.config.num_train_timesteps







'''simple docstring'''






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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block




@dataclass
class a_     (     UpperCamelCase_				):
   _snake_case	   =				42




class a_     (     UpperCamelCase_		,   UpperCamelCase_				):







   @register_to_config
   def __init__(self       , __a = None       , __a = 3       , __a = 3       , __a = False       , __a = "positional"       , __a = 0       , __a = True       , __a = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")       , __a = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")       , __a = (2_2_4, 4_4_8, 6_7_2, 8_9_6)       , __a = 2       , __a = 1       , __a = 1       , __a = "conv"       , __a = "conv"       , __a = "silu"       , __a = 8       , __a = 3_2       , __a = 1E-5       , __a = "default"       , __a = True       , __a = None       , __a = None       , )      ->   List[str]:
      """simple docstring"""






      super().__init__()

      __snake_case					:   Optional[int]  				=			sample_size
      __snake_case					:   Union[str, Any]  				=			block_out_channels[0] * 4

      # Check inputs
      if len(__a) != len(__a):
         raise ValueError(
             F"""Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}.""")

      if len(__a) != len(__a):
         raise ValueError(
             F"""Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}.""")

      # input
      __snake_case					:   Union[str, Any]  				=			nn.Convad(__a       , block_out_channels[0]       , kernel_size=3       , padding=(1, 1))

      # time
      if time_embedding_type == "fourier":
         __snake_case					:   str  				=			GaussianFourierProjection(embedding_size=block_out_channels[0]       , scale=1_6)
         __snake_case					:   Tuple  				=			2 * block_out_channels[0]
      elif time_embedding_type == "positional":
         __snake_case					:   List[str]  				=			Timesteps(block_out_channels[0]       , __a       , __a)
         __snake_case					:   Optional[Any]  				=			block_out_channels[0]

      __snake_case					:   List[Any]  				=			TimestepEmbedding(__a       , __a)

      # class embedding
      if class_embed_type is None and num_class_embeds is not None:
         __snake_case					:   Optional[Any]  				=			nn.Embedding(__a       , __a)
      elif class_embed_type == "timestep":
         __snake_case					:   int  				=			TimestepEmbedding(__a       , __a)
      elif class_embed_type == "identity":
         __snake_case					:   Any  				=			nn.Identity(__a       , __a)
      else:
         __snake_case					:   Dict  				=			None

      __snake_case					:   Union[str, Any]  				=			nn.ModuleList([])
      __snake_case					:   Dict  				=			None
      __snake_case					:   Union[str, Any]  				=			nn.ModuleList([])

      # down
      __snake_case					:   List[str]  				=			block_out_channels[0]
      for i, down_block_type in enumerate(__a):
         __snake_case					:   List[str]  				=			output_channel
         __snake_case					:   Tuple  				=			block_out_channels[i]
         __snake_case					:   Union[str, Any]  				=			i == len(__a) - 1

         __snake_case					:   Dict  				=			get_down_block(
             __a       , num_layers=__a       , in_channels=__a       , out_channels=__a       , temb_channels=__a       , add_downsample=not is_final_block       , resnet_eps=__a       , resnet_act_fn=__a       , resnet_groups=__a       , attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel       , downsample_padding=__a       , resnet_time_scale_shift=__a       , downsample_type=__a       , )
         self.down_blocks.append(__a)

      # mid
      __snake_case					:   List[str]  				=			UNetMidBlockaD(
          in_channels=block_out_channels[-1]       , temb_channels=__a       , resnet_eps=__a       , resnet_act_fn=__a       , output_scale_factor=__a       , resnet_time_scale_shift=__a       , attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]       , resnet_groups=__a       , add_attention=__a       , )

      # up
      __snake_case					:   List[Any]  				=			list(reversed(__a))
      __snake_case					:   Any  				=			reversed_block_out_channels[0]
      for i, up_block_type in enumerate(__a):
         __snake_case					:   str  				=			output_channel
         __snake_case					:   List[Any]  				=			reversed_block_out_channels[i]
         __snake_case					:   Optional[Any]  				=			reversed_block_out_channels[min(i + 1       , len(__a) - 1)]

         __snake_case					:   Optional[int]  				=			i == len(__a) - 1

         __snake_case					:   List[Any]  				=			get_up_block(
             __a       , num_layers=layers_per_block + 1       , in_channels=__a       , out_channels=__a       , prev_output_channel=__a       , temb_channels=__a       , add_upsample=not is_final_block       , resnet_eps=__a       , resnet_act_fn=__a       , resnet_groups=__a       , attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel       , resnet_time_scale_shift=__a       , upsample_type=__a       , )
         self.up_blocks.append(__a)
         __snake_case					:   Dict  				=			output_channel

      # out
      __snake_case					:   Tuple  				=			norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4       , 3_2)
      __snake_case					:   Tuple  				=			nn.GroupNorm(num_channels=block_out_channels[0]       , num_groups=__a       , eps=__a)
      __snake_case					:   List[str]  				=			nn.SiLU()
      __snake_case					:   List[Any]  				=			nn.Convad(block_out_channels[0]       , __a       , kernel_size=3       , padding=1)







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a = None       , __a = True       , )      ->   Union[UNetaDOutput, Tuple]:
      """simple docstring"""






      if self.config.center_input_sample:
         __snake_case					:   Optional[Any]  				=			2 * sample - 1.0

      # 1. time
      __snake_case					:   Any  				=			timestep
      if not torch.is_tensor(__a):
         __snake_case					:   Dict  				=			torch.tensor([timesteps]       , dtype=torch.long       , device=sample.device)
      elif torch.is_tensor(__a) and len(timesteps.shape) == 0:
         __snake_case					:   Dict  				=			timesteps[None].to(sample.device)

      # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
      __snake_case					:   Optional[int]  				=			timesteps * torch.ones(sample.shape[0]       , dtype=timesteps.dtype       , device=timesteps.device)

      __snake_case					:   Dict  				=			self.time_proj(__a)

      # timesteps does not contain any weights and will always return f32 tensors
      # but time_embedding might actually be running in fp16. so we need to cast here.
      # there might be better ways to encapsulate this.
      __snake_case					:   Optional[Any]  				=			t_emb.to(dtype=self.dtype)
      __snake_case					:   List[Any]  				=			self.time_embedding(__a)

      if self.class_embedding is not None:
         if class_labels is None:
            raise ValueError('class_labels should be provided when doing class conditioning')

         if self.config.class_embed_type == "timestep":
            __snake_case					:   str  				=			self.time_proj(__a)

         __snake_case					:   List[str]  				=			self.class_embedding(__a).to(dtype=self.dtype)
         __snake_case					:   Optional[Any]  				=			emb + class_emb

      # 2. pre-process
      __snake_case					:   List[str]  				=			sample
      __snake_case					:   Optional[Any]  				=			self.conv_in(__a)

      # 3. down
      __snake_case					:   List[str]  				=			(sample,)
      for downsample_block in self.down_blocks:
         if hasattr(__a       , 'skip_conv'):
            __snake_case				,__snake_case				,__snake_case					:   List[str]  				=			downsample_block(
                hidden_states=__a       , temb=__a       , skip_sample=__a)
         else:
            __snake_case				,__snake_case					:   Dict  				=			downsample_block(hidden_states=__a       , temb=__a)

         down_block_res_samples += res_samples

      # 4. mid
      __snake_case					:   Any  				=			self.mid_block(__a       , __a)

      # 5. up
      __snake_case					:   Dict  				=			None
      for upsample_block in self.up_blocks:
         __snake_case					:   int  				=			down_block_res_samples[-len(upsample_block.resnets) :]
         __snake_case					:   Union[str, Any]  				=			down_block_res_samples[: -len(upsample_block.resnets)]

         if hasattr(__a       , 'skip_conv'):
            __snake_case				,__snake_case					:   List[str]  				=			upsample_block(__a       , __a       , __a       , __a)
         else:
            __snake_case					:   int  				=			upsample_block(__a       , __a       , __a)

        # 6. post-process
      __snake_case					:   List[str]  				=			self.conv_norm_out(__a)
      __snake_case					:   Optional[int]  				=			self.conv_act(__a)
      __snake_case					:   Optional[int]  				=			self.conv_out(__a)

      if skip_sample is not None:
         sample += skip_sample

      if self.config.time_embedding_type == "fourier":
         __snake_case					:   str  				=			timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]))))
         __snake_case					:   Union[str, Any]  				=			sample / timesteps

      if not return_dict:
         return (sample,)

      return UNetaDOutput(sample=__a)