




# 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  _lowerCAmelCase     (  _lowercase	):
      A__		     =		42
      A__		     =		None



def      __lowerCAmelCase			(					UpperCamelCase	,      UpperCamelCase=0.999	,      UpperCamelCase="cosine"	,      )     ->	Dict:
  if alpha_transform_type == "cosine":

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

  elif alpha_transform_type == "exp":

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

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

  lowerCAmelCase__				:  str		       =     []
  for i in range(UpperCamelCase      ):
    lowerCAmelCase__				:  int		       =     i / num_diffusion_timesteps
    lowerCAmelCase__				:  Optional[int]		       =     (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  _lowerCAmelCase     (  _lowercase    ,			_lowercase	):
      A__		     =		1

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

        lowerCAmelCase__				:  List[str]		       =     1.0 - self.betas
        lowerCAmelCase__				:  List[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.
        lowerCAmelCase__				:  Dict		       =     torch.tensor(0.0		) if set_alpha_to_zero else self.alphas_cumprod[-1]

        # standard deviation of the initial noise distribution
        lowerCAmelCase__				:  int		       =     1.0

        # setable values
        lowerCAmelCase__				:  Tuple		       =     None
        lowerCAmelCase__				:  List[Any]		       =     torch.from_numpy(np.arange(0      ,				__UpperCAmelCase		).copy().astype(np.intaa		)		)

      def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase = None		):
        return sample

      def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase = None		):

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

        lowerCAmelCase__				:  List[Any]		       =     num_inference_steps
        lowerCAmelCase__				:  Union[str, 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
        lowerCAmelCase__				:  Union[str, Any]		       =     (np.arange(0      ,				__UpperCAmelCase		) * step_ratio).round().copy().astype(np.intaa		)
        lowerCAmelCase__				:  int		       =     torch.from_numpy(__UpperCAmelCase		).to(__UpperCAmelCase		)
        self.timesteps += self.config.steps_offset

      def 	__magic_name__(   self      ,				__UpperCAmelCase      ,				__UpperCAmelCase      ,				__UpperCAmelCase      ,				__UpperCAmelCase = 0.0      ,				__UpperCAmelCase = False      ,				__UpperCAmelCase = None      ,				__UpperCAmelCase = True      ,				):
        # 1. get previous step value (=t+1)
        lowerCAmelCase__				:  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
        lowerCAmelCase__				:  List[Any]		       =     self.alphas_cumprod[timestep]
        lowerCAmelCase__				:  int		       =     (
            self.alphas_cumprod[prev_timestep]
            if prev_timestep < self.config.num_train_timesteps
            else self.final_alpha_cumprod
        )

        lowerCAmelCase__				:  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":
          lowerCAmelCase__				:  Union[str, Any]		       =     (sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
          lowerCAmelCase__				:  Optional[Any]		       =     model_output
        elif self.config.prediction_type == "sample":
          lowerCAmelCase__				:  Optional[Any]		       =     model_output
          lowerCAmelCase__				:  str		       =     (sample - alpha_prod_t ** 0.5 * pred_original_sample) / beta_prod_t ** 0.5
        elif self.config.prediction_type == "v_prediction":
          lowerCAmelCase__				:  List[Any]		       =     (alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
          lowerCAmelCase__				:  Any		       =     (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:
          lowerCAmelCase__				:  Optional[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
        lowerCAmelCase__				:  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
        lowerCAmelCase__				:  Union[str, Any]		       =     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=__UpperCAmelCase      ,				pred_original_sample=__UpperCAmelCase		)







      def __len__(   self		):
        return self.config.num_train_timesteps







def      __lowerCAmelCase			(					UpperCamelCase      )     ->	None:
  lowerCAmelCase__				:  Dict		       =     generate_pascal_triangle(UpperCamelCase      )
  for row_idx in range(UpperCamelCase      ):
    # Print left spaces
    for _ in range(num_rows - row_idx - 1      ):
      print(end=''' '''      )
    # Print row values
    for col_idx in range(row_idx + 1      ):
      if col_idx != row_idx:
        print(triangle[row_idx][col_idx]	,      end=''' '''      )
      else:
        print(triangle[row_idx][col_idx]	,      end=''''''      )
    print()



def      __lowerCAmelCase			(					UpperCamelCase      )     ->	list[list[int]]:

  if not isinstance(UpperCamelCase	,      UpperCamelCase      ):
    raise TypeError('''The input value of \'num_rows\' should be \'int\''''      )

  if num_rows == 0:
    return []
  elif num_rows < 0:
    raise ValueError(
        '''The input value of \'num_rows\' should be greater than or equal to 0'''      )

  lowerCAmelCase__				:  list[list[int]]		       =     []
  for current_row_idx in range(UpperCamelCase      ):
    lowerCAmelCase__				:  List[str]		       =     populate_current_row(UpperCamelCase	,      UpperCamelCase      )
    triangle.append(UpperCamelCase      )
  return triangle



def      __lowerCAmelCase			(					UpperCamelCase	,      UpperCamelCase      )     ->	list[int]:
  lowerCAmelCase__				:  Optional[Any]		       =     [-1] * (current_row_idx + 1)
  # first and last elements of current row are equal to 1
  lowerCAmelCase__       ,			lowerCAmelCase__				:  int		       =     1, 1
  for current_col_idx in range(1	,      UpperCamelCase      ):
    calculate_current_element(
        UpperCamelCase	,      UpperCamelCase	,      UpperCamelCase	,      UpperCamelCase      )
  return current_row



def      __lowerCAmelCase			(					UpperCamelCase	,      UpperCamelCase	,      UpperCamelCase	,      UpperCamelCase	,      )     ->	None:
  lowerCAmelCase__				:  Optional[int]		       =     triangle[current_row_idx - 1][current_col_idx - 1]
  lowerCAmelCase__				:  List[str]		       =     triangle[current_row_idx - 1][current_col_idx]
  lowerCAmelCase__				:  List[Any]		       =     above_to_left_elt + above_to_right_elt



def      __lowerCAmelCase			(					UpperCamelCase      )     ->	list[list[int]]:

  if not isinstance(UpperCamelCase	,      UpperCamelCase      ):
    raise TypeError('''The input value of \'num_rows\' should be \'int\''''      )

  if num_rows == 0:
    return []
  elif num_rows < 0:
    raise ValueError(
        '''The input value of \'num_rows\' should be greater than or equal to 0'''      )

  lowerCAmelCase__				:  list[list[int]]		       =     [[1]]

  for row_index in range(1	,      UpperCamelCase      ):
    lowerCAmelCase__				:  Optional[int]		       =     [0] + result[-1] + [0]
    lowerCAmelCase__				:  List[str]		       =     row_index + 1
    # Calculate the number of distinct elements in a row
    lowerCAmelCase__				:  Tuple		       =     sum(divmod(UpperCamelCase	,      2      )      )
    lowerCAmelCase__				:  Tuple		       =     [
        temp_row[i - 1] + temp_row[i] for i in range(1	,      distinct_elements + 1      )
    ]
    lowerCAmelCase__				:  int		       =     row_first_half[: (row_index + 1) // 2]
    row_second_half.reverse()
    lowerCAmelCase__				:  Dict		       =     row_first_half + row_second_half
    result.append(UpperCamelCase      )

  return result



def      __lowerCAmelCase			(					)     ->	None:
  from collections.abc import Callable
  from timeit import timeit

  def benchmark_a_function(UpperCamelCase	,      UpperCamelCase      ) -> None:
    lowerCAmelCase__				:  Tuple		       =     F"""{func.__name__}({value})"""
    lowerCAmelCase__				:  Dict		       =     timeit(F"""__main__.{call}"""	,      setup='''import __main__'''      )
    # print(f"{call:38} = {func(value)} -- {timing:.4f} seconds")
    print(F"""{call:38} -- {timing:.4f} seconds"""      )

  for value in range(15      ):  # (1, 7, 14):
    for func in (generate_pascal_triangle, generate_pascal_triangle_optimized):
      benchmark_a_function(UpperCamelCase	,      UpperCamelCase      )
    print()


if __name__ == "__main__":
     import doctest

     doctest.testmod()
     benchmark()

