

# 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  _A		(  __magic_name__):
  SCREAMING_SNAKE_CASE    :      torch.FloatTensor
  SCREAMING_SNAKE_CASE    :      Optional[torch.FloatTensor]       =  None

def    A_  (							a					,   a=0.9_99					,   a="cosine"					,   ):

      """simple docstring"""



      if alpha_transform_type == "cosine":

            def alpha_bar_fn(a  ):
                  return math.cos((t + 0.0_08) / 1.0_08 * math.pi / 2  ) ** 2

      elif alpha_transform_type == "exp":

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

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

      SCREAMING_SNAKE_CASE_				:  Any           =     []
      for i in range(a  ):
            SCREAMING_SNAKE_CASE_				:  List[Any]           =     i / num_diffusion_timesteps
            SCREAMING_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		(  __magic_name__   ,     __magic_name__):
  SCREAMING_SNAKE_CASE    :      int       =  1

  @register_to_config
  def __init__(				self ,					_SCREAMING_SNAKE_CASE = 1000 ,					_SCREAMING_SNAKE_CASE = 0.0001 ,					_SCREAMING_SNAKE_CASE = 0.02 ,					_SCREAMING_SNAKE_CASE = "linear" ,					_SCREAMING_SNAKE_CASE = None ,					_SCREAMING_SNAKE_CASE = True ,					_SCREAMING_SNAKE_CASE = True ,					_SCREAMING_SNAKE_CASE = 0 ,					_SCREAMING_SNAKE_CASE = "epsilon" ,					_SCREAMING_SNAKE_CASE = 1.0 ,					**_SCREAMING_SNAKE_CASE ,					):





        """simple docstring"""







        if kwargs.get('set_alpha_to_one' ,					_SCREAMING_SNAKE_CASE					) is not None:
              SCREAMING_SNAKE_CASE_				:  Optional[int]           =     (
                  'The `set_alpha_to_one` argument is deprecated. Please use `set_alpha_to_zero` instead.'
              )
              deprecate('set_alpha_to_one' ,					'1.0.0' ,					_SCREAMING_SNAKE_CASE ,					standard_warn=_SCREAMING_SNAKE_CASE					)
              SCREAMING_SNAKE_CASE_				:  str           =     kwargs['set_alpha_to_one']
        if trained_betas is not None:
              SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     torch.tensor(_SCREAMING_SNAKE_CASE ,					dtype=torch.floataa					)
        elif beta_schedule == "linear":
              SCREAMING_SNAKE_CASE_				:  List[Any]           =     torch.linspace(_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE ,					dtype=torch.floataa					)
        elif beta_schedule == "scaled_linear":
              # this schedule is very specific to the latent diffusion model.
              SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     (
                  torch.linspace(beta_start**0.5 ,					beta_end**0.5 ,					_SCREAMING_SNAKE_CASE ,					dtype=torch.floataa					) ** 2
              )
        elif beta_schedule == "squaredcos_cap_v2":
              # Glide cosine schedule
              SCREAMING_SNAKE_CASE_				:  Any           =     betas_for_alpha_bar(_SCREAMING_SNAKE_CASE					)
        else:
              raise NotImplementedError(f"{beta_schedule} does is not implemented for {self.__class__}"					)

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

        # standard deviation of the initial noise distribution
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     1.0

        # setable values
        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     None
        SCREAMING_SNAKE_CASE_				:  Tuple           =     torch.from_numpy(np.arange(0 ,					_SCREAMING_SNAKE_CASE					).copy().astype(np.intaa					)					)

  def  UpperCAmelCase    (				self ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE = None					):





        """simple docstring"""







        return sample

  def  UpperCAmelCase    (				self ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_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."					)

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

  def  UpperCAmelCase    (				self ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE ,					_SCREAMING_SNAKE_CASE = 0.0 ,					_SCREAMING_SNAKE_CASE = False ,					_SCREAMING_SNAKE_CASE = None ,					_SCREAMING_SNAKE_CASE = True ,					):





        """simple docstring"""







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

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



  def __len__(				self					):





        """simple docstring"""







        return self.config.num_train_timesteps









import gc
import unittest

import numpy as np
import torch

from diffusers import (
    AudioDiffusionPipeline,
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    DiffusionPipeline,
    Mel,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()
class  _A		(  unittest.TestCase):

  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()

  @property
  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        torch.manual_seed(0					)
        SCREAMING_SNAKE_CASE_				:  Optional[int]           =     UNetaDModel(
            sample_size=(32, 64) ,					in_channels=1 ,					out_channels=1 ,					layers_per_block=2 ,					block_out_channels=(128, 128) ,					down_block_types=('AttnDownBlock2D', 'DownBlock2D') ,					up_block_types=('UpBlock2D', 'AttnUpBlock2D') ,					)
        return model

  @property
  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        torch.manual_seed(0					)
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     UNetaDConditionModel(
            sample_size=(64, 32) ,					in_channels=1 ,					out_channels=1 ,					layers_per_block=2 ,					block_out_channels=(128, 128) ,					down_block_types=('CrossAttnDownBlock2D', 'DownBlock2D') ,					up_block_types=('UpBlock2D', 'CrossAttnUpBlock2D') ,					cross_attention_dim=10 ,					)
        return model

  @property
  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        torch.manual_seed(0					)
        SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     AutoencoderKL(
            sample_size=(128, 64) ,					in_channels=1 ,					out_channels=1 ,					latent_channels=1 ,					layers_per_block=2 ,					block_out_channels=(128, 128) ,					down_block_types=('DownEncoderBlock2D', 'DownEncoderBlock2D') ,					up_block_types=('UpDecoderBlock2D', 'UpDecoderBlock2D') ,					)
        SCREAMING_SNAKE_CASE_				:  Dict           =     UNetaDModel(
            sample_size=(64, 32) ,					in_channels=1 ,					out_channels=1 ,					layers_per_block=2 ,					block_out_channels=(128, 128) ,					down_block_types=('AttnDownBlock2D', 'DownBlock2D') ,					up_block_types=('UpBlock2D', 'AttnUpBlock2D') ,					)
        return vqvae, unet



  @slow
  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  Optional[int]           =     'cpu'  # ensure determinism for the device-dependent torch.Generator
        SCREAMING_SNAKE_CASE_				:  Any           =     Mel(
            x_res=self.dummy_unet.config.sample_size[1] ,					y_res=self.dummy_unet.config.sample_size[0] ,					)

        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     DDPMScheduler()
        SCREAMING_SNAKE_CASE_				:  str           =     AudioDiffusionPipeline(vqvae=_SCREAMING_SNAKE_CASE ,					unet=self.dummy_unet ,					mel=_SCREAMING_SNAKE_CASE ,					scheduler=_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  Dict           =     pipe.to(_SCREAMING_SNAKE_CASE					)
        pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

        SCREAMING_SNAKE_CASE_				:  Dict           =     torch.Generator(device=_SCREAMING_SNAKE_CASE					).manual_seed(42					)
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     pipe(generator=_SCREAMING_SNAKE_CASE ,					steps=4					)
        SCREAMING_SNAKE_CASE_				:  Any           =     output.audios[0]
        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     output.images[0]

        SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     torch.Generator(device=_SCREAMING_SNAKE_CASE					).manual_seed(42					)
        SCREAMING_SNAKE_CASE_				:  Optional[int]           =     pipe(generator=_SCREAMING_SNAKE_CASE ,					steps=4 ,					return_dict=_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     output[0][0]

        assert audio.shape == (1, (self.dummy_unet.config.sample_size[1] - 1) * mel.hop_length)
        assert (
            image.height == self.dummy_unet.config.sample_size[0]
            and image.width == self.dummy_unet.config.sample_size[1]
        )
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     np.frombuffer(image.tobytes() ,					dtype='uint8'					)[:10]
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     np.frombuffer(image_from_tuple.tobytes() ,					dtype='uint8'					)[:10]
        SCREAMING_SNAKE_CASE_				:  Any           =     np.array([69, 255, 255, 255, 0, 0, 77, 181, 12, 127]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() == 0
        assert np.abs(image_from_tuple_slice.flatten() - expected_slice					).max() == 0

        SCREAMING_SNAKE_CASE_				:  Optional[int]           =     Mel(
            x_res=self.dummy_vqvae_and_unet[0].config.sample_size[1] ,					y_res=self.dummy_vqvae_and_unet[0].config.sample_size[0] ,					)

        SCREAMING_SNAKE_CASE_				:  str           =     DDIMScheduler()
        SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     self.dummy_vqvae_and_unet
        SCREAMING_SNAKE_CASE_				:  int           =     AudioDiffusionPipeline(
            vqvae=self.dummy_vqvae_and_unet[0] ,					unet=dummy_vqvae_and_unet[1] ,					mel=_SCREAMING_SNAKE_CASE ,					scheduler=_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  str           =     pipe.to(_SCREAMING_SNAKE_CASE					)
        pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

        np.random.seed(0					)
        SCREAMING_SNAKE_CASE_				:  Dict           =     np.random.uniform(-1 ,					1 ,					((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)					)
        SCREAMING_SNAKE_CASE_				:  str           =     torch.Generator(device=_SCREAMING_SNAKE_CASE					).manual_seed(42					)
        SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     pipe(raw_audio=_SCREAMING_SNAKE_CASE ,					generator=_SCREAMING_SNAKE_CASE ,					start_step=5 ,					steps=10					)
        SCREAMING_SNAKE_CASE_				:  int           =     output.images[0]

        assert (
            image.height == self.dummy_vqvae_and_unet[0].config.sample_size[0]
            and image.width == self.dummy_vqvae_and_unet[0].config.sample_size[1]
        )
        SCREAMING_SNAKE_CASE_				:  Optional[int]           =     np.frombuffer(image.tobytes() ,					dtype='uint8'					)[:10]
        SCREAMING_SNAKE_CASE_				:  int           =     np.array([120, 117, 110, 109, 138, 167, 138, 148, 132, 121]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() == 0

        SCREAMING_SNAKE_CASE_				:  Dict           =     self.dummy_unet_condition
        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     AudioDiffusionPipeline(
            vqvae=self.dummy_vqvae_and_unet[0] ,					unet=_SCREAMING_SNAKE_CASE ,					mel=_SCREAMING_SNAKE_CASE ,					scheduler=_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  Dict           =     pipe.to(_SCREAMING_SNAKE_CASE					)
        pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

        np.random.seed(0					)
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     torch.rand((1, 1, 10)					)
        SCREAMING_SNAKE_CASE_				:  Any           =     pipe(generator=_SCREAMING_SNAKE_CASE ,					encoding=_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     output.images[0]
        SCREAMING_SNAKE_CASE_				:  Dict           =     np.frombuffer(image.tobytes() ,					dtype='uint8'					)[:10]
        SCREAMING_SNAKE_CASE_				:  Union[str, Any]           =     np.array([107, 103, 120, 127, 142, 122, 113, 122, 97, 111]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() == 0

@slow
@require_torch_gpu
class  _A		(  unittest.TestCase):

  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()



  def  UpperCAmelCase    (				self					):





        """simple docstring"""







        SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     torch_device

        SCREAMING_SNAKE_CASE_				:  str           =     DiffusionPipeline.from_pretrained('teticio/audio-diffusion-ddim-256'					)
        SCREAMING_SNAKE_CASE_				:  Dict           =     pipe.to(_SCREAMING_SNAKE_CASE					)
        pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

        SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     torch.Generator(device=_SCREAMING_SNAKE_CASE					).manual_seed(42					)
        SCREAMING_SNAKE_CASE_				:  List[Any]           =     pipe(generator=_SCREAMING_SNAKE_CASE					)
        SCREAMING_SNAKE_CASE_				:  Any           =     output.audios[0]
        SCREAMING_SNAKE_CASE_				:  Optional[Any]           =     output.images[0]

        assert audio.shape == (1, (pipe.unet.config.sample_size[1] - 1) * pipe.mel.hop_length)
        assert image.height == pipe.unet.config.sample_size[0] and image.width == pipe.unet.config.sample_size[1]
        SCREAMING_SNAKE_CASE_				:  Optional[int]           =     np.frombuffer(image.tobytes() ,					dtype='uint8'					)[:10]
        SCREAMING_SNAKE_CASE_				:  Any           =     np.array([151, 167, 154, 144, 122, 134, 121, 105, 70, 26]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() == 0






