
# 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 __snake_case		(      lowerCAmelCase     ):
    _a		:						torch.FloatTensor
    _a		:						Optional[torch.FloatTensor]=				None







def 						_snake_case(  SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__=0.999					,	SCREAMING_SNAKE_CASE__="cosine"					,	) ->    Optional[Any]:
       if alpha_transform_type == "cosine":

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

       elif alpha_transform_type == "exp":

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

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

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




class __snake_case		(      lowerCAmelCase					,	lowerCAmelCase     ):
    _a		:						Union[str, Any]=				1
    @register_to_config
    def __init__(  self		,snake_case = 1000		,snake_case = 0.0_001		,snake_case = 0.02		,snake_case = "linear"		,snake_case = None		,snake_case = True		,snake_case = True		,snake_case = 0		,snake_case = "epsilon"		,snake_case = 1.0		,**snake_case		,):



           '''simple docstring'''







           if kwargs.get("""set_alpha_to_one"""		,snake_case      ) is not None:
                  lowercase							:     Any       	=      (
                      """The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead."""
                  )
                  deprecate("""set_alpha_to_one"""		,"""1.0.0"""		,snake_case		,standard_warn=snake_case      )
                  lowercase							:     List[str]       	=      kwargs["""set_alpha_to_one"""]
           if trained_betas is not None:
                  lowercase							:     str       	=      torch.tensor(snake_case		,dtype=torch.floataa      )
           elif beta_schedule == "linear":
                  lowercase							:     Union[str, Any]       	=      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.
                  lowercase							:     int       	=      (
                      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
                  lowercase							:     List[Any]       	=      betas_for_alpha_bar(snake_case      )
           else:
                  raise NotImplementedError(f"{beta_schedule} does is not implemented for {self.__class__}"      )

           lowercase							:     Optional[Any]       	=      1.0 - self.betas
           lowercase							:     str       	=      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.
           lowercase							:     Optional[int]       	=      torch.tensor(0.0      ) if set_alpha_to_zero else self.alphas_cumprod[-1]

           # standard deviation of the initial noise distribution
           lowercase							:     Dict       	=      1.0

           # setable values
           lowercase							:     List[Any]       	=      None
           lowercase							:     Dict       	=      torch.from_numpy(np.arange(0		,snake_case      ).copy().astype(np.intaa      )      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = None      ):



           '''simple docstring'''







           return sample
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = None      ):



           '''simple docstring'''







           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."      )

           lowercase							:     Any       	=      num_inference_steps
           lowercase							:     Any       	=      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
           lowercase							:     Union[str, Any]       	=      (np.arange(0		,snake_case      ) * step_ratio).round().copy().astype(np.intaa      )
           lowercase							:     Dict       	=      torch.from_numpy(snake_case      ).to(snake_case      )
           self.timesteps += self.config.steps_offset
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,snake_case = 0.0		,snake_case = False		,snake_case = None		,snake_case = True		,):



           '''simple docstring'''







           lowercase							:     List[str]       	=      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
           lowercase							:     Optional[Any]       	=      self.alphas_cumprod[timestep]
           lowercase							:     str       	=      (
               self.alphas_cumprod[prev_timestep]
               if prev_timestep < self.config.num_train_timesteps
               else self.final_alpha_cumprod
           )

           lowercase							:     Any       	=      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":
                  lowercase							:     Union[str, Any]       	=      (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
                  lowercase							:     str       	=      model_output
           elif self.config.prediction_type == "sample":
                  lowercase							:     str       	=      model_output
                  lowercase							:     str       	=      (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
           elif self.config.prediction_type == "v_prediction":
                  lowercase							:     str       	=      (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
                  lowercase							:     int       	=      (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:
                  lowercase							:     int       	=      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
           lowercase							:     Dict       	=      (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
           lowercase							:     int       	=      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=snake_case		,pred_original_sample=snake_case      )
    def __len__(  self      ):



           '''simple docstring'''







           return self.config.num_train_timesteps





"""simple docstring"""
from ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCAmelCase    :List[str]     =					logging.get_logger(__name__)

_lowerCAmelCase    :Any     =					{
    'tiiuae/falcon-40b': 'https://huggingface.co/tiiuae/falcon-40b/resolve/main/config.json',
    'tiiuae/falcon-7b': 'https://huggingface.co/tiiuae/falcon-7b/resolve/main/config.json',
}
class 							_UpperCAmelCase							( a      ):





     '''simple docstring'''



     a__									='''falcon'''
     a__									=['''past_key_values''']





     def __init__(					self       ,	A=6_5_0_2_4       ,	A=4_5_4_4       ,	A=3_2       ,	A=7_1       ,	A=1E-5       ,	A=0.02       ,	A=True       ,	A=0.0       ,	A=0.0       ,	A=None       ,	A=False       ,	A=False       ,	A=True       ,	A=True       ,	A=False       ,	A=1_1       ,	A=1_1       ,	**A       ,	)							->					Any:
         _UpperCAmelCase							:  int											=    vocab_size
         # Backward compatibility with n_embed kwarg
         _UpperCAmelCase							:  Optional[Any]											=    kwargs.pop('''n_embed'''       ,	A		)
         _UpperCAmelCase							:  int											=    hidden_size if n_embed is None else n_embed
         _UpperCAmelCase							:  List[str]											=    num_hidden_layers
         _UpperCAmelCase							:  Tuple											=    num_attention_heads
         _UpperCAmelCase							:  Optional[int]											=    layer_norm_epsilon
         _UpperCAmelCase							:  Tuple											=    initializer_range
         _UpperCAmelCase							:  Optional[int]											=    use_cache
         _UpperCAmelCase							:  Any											=    hidden_dropout
         _UpperCAmelCase							:  Dict											=    attention_dropout

         _UpperCAmelCase							:  Any											=    bos_token_id
         _UpperCAmelCase							:  List[Any]											=    eos_token_id
         _UpperCAmelCase							:  Tuple											=    num_attention_heads if num_kv_heads is None else num_kv_heads
         _UpperCAmelCase							:  Dict											=    alibi
         _UpperCAmelCase							:  Optional[int]											=    new_decoder_architecture
         _UpperCAmelCase							:  str											=    multi_query  # Ignored when new_decoder_architecture is True
         _UpperCAmelCase							:  Optional[int]											=    parallel_attn
         _UpperCAmelCase							:  Optional[int]											=    bias

         super().__init__(bos_token_id=A       ,	eos_token_id=A       ,	**A		)





     @property
     def  __lowerCAmelCase	(					self		)							->					List[str]:
         return self.hidden_size // self.num_attention_heads





     @property
     def  __lowerCAmelCase	(					self		)							->					List[Any]:
         return not self.alibi


