


import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest





class   __a			(    __UpperCamelCase      ):
 __lowercase :							Tuple        = (UniPCMultistepScheduler,)
 __lowercase :							Dict        = (('num_inference_steps', 25),)
 def 		SCREAMING_SNAKE_CASE__   (   self				, **lowerCAmelCase__ )					->				int:




        '''simple docstring'''
        lowercase__:   List[str]  			=   {
            'num_train_timesteps': 1_000,
            'beta_start': 0.0_0_0_1,
            'beta_end': 0.0_2,
            'beta_schedule': 'linear',
            'solver_order': 2,
            'solver_type': 'bh2',
        }

        config.update(**lowerCAmelCase__ )
        return config
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__=0				, **lowerCAmelCase__ )					->				Optional[Any]:




        '''simple docstring'''
        lowercase__:   Dict  			=   dict(self.forward_default_kwargs )
        lowercase__:   List[str]  			=   kwargs.pop('num_inference_steps'				, lowerCAmelCase__ )
        lowercase__:   Dict  			=   self.dummy_sample
        lowercase__:   Tuple  			=   0.1 * sample
        lowercase__:   Optional[Any]  			=   [residual + 0.2, residual + 0.1_5, residual + 0.1_0]

        for scheduler_class in self.scheduler_classes:
               lowercase__:   Tuple  			=   self.get_scheduler_config(**lowerCAmelCase__ )
               lowercase__:   List[Any]  			=   scheduler_class(**lowerCAmelCase__ )
               scheduler.set_timesteps(lowerCAmelCase__ )
               # copy over dummy past residuals
               lowercase__:   Tuple  			=   dummy_past_residuals[: scheduler.config.solver_order]

               with tempfile.TemporaryDirectory() as tmpdirname:
                      scheduler.save_config(lowerCAmelCase__ )
                      lowercase__:   List[str]  			=   scheduler_class.from_pretrained(lowerCAmelCase__ )
                      new_scheduler.set_timesteps(lowerCAmelCase__ )
                      # copy over dummy past residuals
                      lowercase__:   int  			=   dummy_past_residuals[: new_scheduler.config.solver_order]

               lowercase__			,						lowercase__:   Optional[int]  			=   sample, sample
               for t in range(lowerCAmelCase__				, time_step + scheduler.config.solver_order + 1 ):
                      lowercase__:   Union[str, Any]  			=   scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, **lowerCAmelCase__ ).prev_sample
                      lowercase__:   List[Any]  			=   new_scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, **lowerCAmelCase__ ).prev_sample

                      assert torch.sum(torch.abs(output - new_output ) ) < 1E-5, "Scheduler outputs are not identical"
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__=0				, **lowerCAmelCase__ )					->				Any:




        '''simple docstring'''
        lowercase__:   str  			=   dict(self.forward_default_kwargs )
        lowercase__:   Any  			=   kwargs.pop('num_inference_steps'				, lowerCAmelCase__ )
        lowercase__:   Optional[int]  			=   self.dummy_sample
        lowercase__:   Optional[int]  			=   0.1 * sample
        lowercase__:   Union[str, Any]  			=   [residual + 0.2, residual + 0.1_5, residual + 0.1_0]

        for scheduler_class in self.scheduler_classes:
               lowercase__:   List[Any]  			=   self.get_scheduler_config()
               lowercase__:   List[Any]  			=   scheduler_class(**lowerCAmelCase__ )
               scheduler.set_timesteps(lowerCAmelCase__ )

               # copy over dummy past residuals (must be after setting timesteps)
               lowercase__:   int  			=   dummy_past_residuals[: scheduler.config.solver_order]

               with tempfile.TemporaryDirectory() as tmpdirname:
                      scheduler.save_config(lowerCAmelCase__ )
                      lowercase__:   List[Any]  			=   scheduler_class.from_pretrained(lowerCAmelCase__ )
                      # copy over dummy past residuals
                      new_scheduler.set_timesteps(lowerCAmelCase__ )

                      # copy over dummy past residual (must be after setting timesteps)
                      lowercase__:   Any  			=   dummy_past_residuals[: new_scheduler.config.solver_order]

               lowercase__:   Tuple  			=   scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, **lowerCAmelCase__ ).prev_sample
               lowercase__:   Optional[int]  			=   new_scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, **lowerCAmelCase__ ).prev_sample

               assert torch.sum(torch.abs(output - new_output ) ) < 1E-5, "Scheduler outputs are not identical"
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__=None				, **lowerCAmelCase__ )					->				int:




        '''simple docstring'''
        if scheduler is None:
               lowercase__:   int  			=   self.scheduler_classes[0]
               lowercase__:   List[str]  			=   self.get_scheduler_config(**lowerCAmelCase__ )
               lowercase__:   Any  			=   scheduler_class(**lowerCAmelCase__ )

        lowercase__:   Optional[int]  			=   self.scheduler_classes[0]
        lowercase__:   Optional[int]  			=   self.get_scheduler_config(**lowerCAmelCase__ )
        lowercase__:   Union[str, Any]  			=   scheduler_class(**lowerCAmelCase__ )

        lowercase__:   int  			=   10
        lowercase__:   int  			=   self.dummy_model()
        lowercase__:   Tuple  			=   self.dummy_sample_deter
        scheduler.set_timesteps(lowerCAmelCase__ )

        for i, t in enumerate(scheduler.timesteps ):
               lowercase__:   Optional[int]  			=   model(lowerCAmelCase__				, lowerCAmelCase__ )
               lowercase__:   List[Any]  			=   scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__ ).prev_sample

        return sample
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Optional[int]:




        '''simple docstring'''
        lowercase__:   Union[str, Any]  			=   dict(self.forward_default_kwargs )

        lowercase__:   Optional[Any]  			=   kwargs.pop('num_inference_steps'				, lowerCAmelCase__ )

        for scheduler_class in self.scheduler_classes:
               lowercase__:   Tuple  			=   self.get_scheduler_config()
               lowercase__:   Dict  			=   scheduler_class(**lowerCAmelCase__ )

               lowercase__:   int  			=   self.dummy_sample
               lowercase__:   Any  			=   0.1 * sample

               if num_inference_steps is not None and hasattr(lowerCAmelCase__				, 'set_timesteps' ):
                      scheduler.set_timesteps(lowerCAmelCase__ )
               elif num_inference_steps is not None and not hasattr(lowerCAmelCase__				, 'set_timesteps' ):
                      lowercase__:   Optional[Any]  			=   num_inference_steps

               # copy over dummy past residuals (must be done after set_timesteps)
               lowercase__:   List[str]  			=   [residual + 0.2, residual + 0.1_5, residual + 0.1_0]
               lowercase__:   List[Any]  			=   dummy_past_residuals[: scheduler.config.solver_order]

               lowercase__:   Any  			=   scheduler.timesteps[5]
               lowercase__:   List[Any]  			=   scheduler.timesteps[6]

               lowercase__:   int  			=   scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, **lowerCAmelCase__ ).prev_sample
               lowercase__:   List[Any]  			=   scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, **lowerCAmelCase__ ).prev_sample

               self.assertEqual(output_a.shape				, sample.shape )
               self.assertEqual(output_a.shape				, output_a.shape )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Dict:




        '''simple docstring'''
        # make sure that iterating over schedulers with same config names gives same results
        # for defaults
        lowercase__:   Tuple  			=   UniPCMultistepScheduler(**self.get_scheduler_config() )
        lowercase__:   Optional[int]  			=   self.full_loop(scheduler=lowerCAmelCase__ )
        lowercase__:   List[str]  			=   torch.mean(torch.abs(lowerCAmelCase__ ) )

        assert abs(result_mean.item() - 0.2_4_6_4 ) < 1E-3

        lowercase__:   Tuple  			=   DPMSolverSinglestepScheduler.from_config(scheduler.config )
        lowercase__:   List[str]  			=   DEISMultistepScheduler.from_config(scheduler.config )
        lowercase__:   Any  			=   DPMSolverMultistepScheduler.from_config(scheduler.config )
        lowercase__:   Union[str, Any]  			=   UniPCMultistepScheduler.from_config(scheduler.config )

        lowercase__:   Optional[int]  			=   self.full_loop(scheduler=lowerCAmelCase__ )
        lowercase__:   Any  			=   torch.mean(torch.abs(lowerCAmelCase__ ) )

        assert abs(result_mean.item() - 0.2_4_6_4 ) < 1E-3
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				List[Any]:




        '''simple docstring'''
        for timesteps in [25, 50, 100, 999, 1_000]:
               self.check_over_configs(num_train_timesteps=lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				List[Any]:




        '''simple docstring'''
        self.check_over_configs(thresholding=lowerCAmelCase__ )
        for order in [1, 2, 3]:
               for solver_type in ["bh1", "bh2"]:
                      for threshold in [0.5, 1.0, 2.0]:
                             for prediction_type in ["epsilon", "sample"]:
                                    self.check_over_configs(
                                        thresholding=lowerCAmelCase__				, prediction_type=lowerCAmelCase__				, sample_max_value=lowerCAmelCase__				, solver_order=lowerCAmelCase__				, solver_type=lowerCAmelCase__				, )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Any:




        '''simple docstring'''
        for prediction_type in ["epsilon", "v_prediction"]:
               self.check_over_configs(prediction_type=lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Tuple:




        '''simple docstring'''
        for solver_type in ["bh1", "bh2"]:
               for order in [1, 2, 3]:
                      for prediction_type in ["epsilon", "sample"]:
                             self.check_over_configs(
                                 solver_order=lowerCAmelCase__				, solver_type=lowerCAmelCase__				, prediction_type=lowerCAmelCase__				, )
                             lowercase__:   Tuple  			=   self.full_loop(
                                 solver_order=lowerCAmelCase__				, solver_type=lowerCAmelCase__				, prediction_type=lowerCAmelCase__				, )
                             assert not torch.isnan(lowerCAmelCase__ ).any(), "Samples have nan numbers"
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Optional[Any]:




        '''simple docstring'''
        self.check_over_configs(lower_order_final=lowerCAmelCase__ )
        self.check_over_configs(lower_order_final=lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Tuple:




        '''simple docstring'''
        for num_inference_steps in [1, 2, 3, 5, 10, 50, 100, 999, 1_000]:
               self.check_over_forward(num_inference_steps=lowerCAmelCase__				, time_step=0 )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				Optional[Any]:




        '''simple docstring'''
        lowercase__:   Union[str, Any]  			=   self.full_loop()
        lowercase__:   Optional[int]  			=   torch.mean(torch.abs(lowerCAmelCase__ ) )

        assert abs(result_mean.item() - 0.2_4_6_4 ) < 1E-3
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				int:




        '''simple docstring'''
        lowercase__:   Optional[Any]  			=   self.full_loop(prediction_type='v_prediction' )
        lowercase__:   Dict  			=   torch.mean(torch.abs(lowerCAmelCase__ ) )

        assert abs(result_mean.item() - 0.1_0_1_4 ) < 1E-3
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				str:




        '''simple docstring'''
        lowercase__:   str  			=   self.scheduler_classes[0]
        lowercase__:   List[str]  			=   self.get_scheduler_config(thresholding=lowerCAmelCase__				, dynamic_thresholding_ratio=0 )
        lowercase__:   Optional[Any]  			=   scheduler_class(**lowerCAmelCase__ )

        lowercase__:   List[str]  			=   10
        lowercase__:   Optional[int]  			=   self.dummy_model()
        lowercase__:   Optional[int]  			=   self.dummy_sample_deter.half()
        scheduler.set_timesteps(lowerCAmelCase__ )

        for i, t in enumerate(scheduler.timesteps ):
               lowercase__:   Dict  			=   model(lowerCAmelCase__				, lowerCAmelCase__ )
               lowercase__:   Optional[int]  			=   scheduler.step(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__ ).prev_sample

        assert sample.dtype == torch.floataa
 def 		SCREAMING_SNAKE_CASE__   (   self				, **lowerCAmelCase__ )					->				List[str]:




        '''simple docstring'''
        for scheduler_class in self.scheduler_classes:
               lowercase__:   int  			=   self.get_scheduler_config(**lowerCAmelCase__ )
               lowercase__:   Optional[Any]  			=   scheduler_class(**lowerCAmelCase__ )

               scheduler.set_timesteps(scheduler.config.num_train_timesteps )
               assert len(scheduler.timesteps.unique() ) == scheduler.num_inference_steps






from math import acos, sin
from typing import List, Tuple, Union

import numpy as np
import torch
from PIL import Image

from ...models import AutoencoderKL, UNetaDConditionModel
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import AudioPipelineOutput, BaseOutput, DiffusionPipeline, ImagePipelineOutput
from .mel import Mel





class   __a			(    __UpperCamelCase      ):
 __lowercase :							Any        = ['vqvae']
 def __init__(   self				, lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__				, )					->				Union[str, Any]:




        '''simple docstring'''
        super().__init__()
        self.register_modules(unet=lowerCAmelCase__				, scheduler=lowerCAmelCase__				, mel=lowerCAmelCase__				, vqvae=lowerCAmelCase__ )
 def 		SCREAMING_SNAKE_CASE__   (   self )					->				int:




        '''simple docstring'''
        return 50 if isinstance(self.scheduler				, lowerCAmelCase__ ) else 1_000
 @torch.no_grad()
 def __call__(   self				, lowerCAmelCase__ = 1				, lowerCAmelCase__ = None				, lowerCAmelCase__ = None				, lowerCAmelCase__ = 0				, lowerCAmelCase__ = 0				, lowerCAmelCase__ = None				, lowerCAmelCase__ = None				, lowerCAmelCase__ = 0				, lowerCAmelCase__ = 0				, lowerCAmelCase__ = None				, lowerCAmelCase__ = 0				, lowerCAmelCase__ = None				, lowerCAmelCase__ = None				, lowerCAmelCase__=True				, )					->				Union[
     Union[AudioPipelineOutput, ImagePipelineOutput],
     Tuple[List[Image.Image], Tuple[int, List[np.ndarray]]],
 ]:




        '''simple docstring'''
        lowercase__:   Union[str, Any]  			=   steps or self.get_default_steps()
        self.scheduler.set_timesteps(lowerCAmelCase__ )
        lowercase__:   Optional[int]  			=   step_generator or generator
        # For backwards compatibility
        if type(self.unet.config.sample_size ) == int:
               lowercase__:   Optional[int]  			=   (self.unet.config.sample_size, self.unet.config.sample_size)
        if noise is None:
               lowercase__:   List[str]  			=   randn_tensor(
                   (
                       batch_size,
                       self.unet.config.in_channels,
                       self.unet.config.sample_size[0],
                       self.unet.config.sample_size[1],
                   )				, generator=lowerCAmelCase__				, device=self.device				, )
        lowercase__:   List[Any]  			=   noise
        lowercase__:   int  			=   None

        if audio_file is not None or raw_audio is not None:
               self.mel.load_audio(lowerCAmelCase__				, lowerCAmelCase__ )
               lowercase__:   int  			=   self.mel.audio_slice_to_image(lowerCAmelCase__ )
               lowercase__:   int  			=   np.frombuffer(input_image.tobytes()				, dtype='uint8' ).reshape(
                   (input_image.height, input_image.width) )
               lowercase__:   str  			=   (input_image / 255) * 2 - 1
               lowercase__:   Union[str, Any]  			=   torch.tensor(input_image[np.newaxis, :, :]				, dtype=torch.float ).to(self.device )

               if self.vqvae is not None:
                      lowercase__:   Optional[int]  			=   self.vqvae.encode(torch.unsqueeze(lowerCAmelCase__				, 0 ) ).latent_dist.sample(
                          generator=lowerCAmelCase__ )[0]
                      lowercase__:   Dict  			=   self.vqvae.config.scaling_factor * input_images

               if start_step > 0:
                      lowercase__:   List[Any]  			=   self.scheduler.add_noise(lowerCAmelCase__				, lowerCAmelCase__				, self.scheduler.timesteps[start_step - 1] )

               lowercase__:   str  			=   (
                   self.unet.config.sample_size[1] * self.mel.get_sample_rate() / self.mel.x_res / self.mel.hop_length
               )
               lowercase__:   Dict  			=   int(mask_start_secs * pixels_per_second )
               lowercase__:   Tuple  			=   int(mask_end_secs * pixels_per_second )
               lowercase__:   List[Any]  			=   self.scheduler.add_noise(lowerCAmelCase__				, lowerCAmelCase__				, torch.tensor(self.scheduler.timesteps[start_step:] ) )

        for step, t in enumerate(self.progress_bar(self.scheduler.timesteps[start_step:] ) ):
               if isinstance(self.unet				, lowerCAmelCase__ ):
                      lowercase__:   Union[str, Any]  			=   self.unet(lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__ )['sample']
               else:
                      lowercase__:   Optional[Any]  			=   self.unet(lowerCAmelCase__				, lowerCAmelCase__ )['sample']

               if isinstance(self.scheduler				, lowerCAmelCase__ ):
                      lowercase__:   List[str]  			=   self.scheduler.step(
                          model_output=lowerCAmelCase__				, timestep=lowerCAmelCase__				, sample=lowerCAmelCase__				, eta=lowerCAmelCase__				, generator=lowerCAmelCase__				, )['prev_sample']
               else:
                      lowercase__:   int  			=   self.scheduler.step(
                          model_output=lowerCAmelCase__				, timestep=lowerCAmelCase__				, sample=lowerCAmelCase__				, generator=lowerCAmelCase__				, )['prev_sample']

               if mask is not None:
                      if mask_start > 0:
                             lowercase__:   List[Any]  			=   mask[:, step, :, :mask_start]
                      if mask_end > 0:
                             lowercase__:   Optional[int]  			=   mask[:, step, :, -mask_end:]

        if self.vqvae is not None:
               # 0.18215 was scaling factor used in training to ensure unit variance
               lowercase__:   Union[str, Any]  			=   1 / self.vqvae.config.scaling_factor * images
               lowercase__:   Optional[Any]  			=   self.vqvae.decode(lowerCAmelCase__ )['sample']

        lowercase__:   Dict  			=   (images / 2 + 0.5).clamp(0				, 1 )
        lowercase__:   List[Any]  			=   images.cpu().permute(0				, 2				, 3				, 1 ).numpy()
        lowercase__:   int  			=   (images * 255).round().astype('uint8' )
        lowercase__:   Optional[Any]  			=   list(
            (Image.fromarray(_[:, :, 0] ) for _ in images)
            if images.shape[3] == 1
            else (Image.fromarray(lowerCAmelCase__				, mode='RGB' ).convert('L' ) for _ in images) )

        lowercase__:   Dict  			=   [self.mel.image_to_audio(lowerCAmelCase__ ) for _ in images]
        if not return_dict:
               return images, (self.mel.get_sample_rate(), audios)

        return BaseOutput(**AudioPipelineOutput(np.array(lowerCAmelCase__ )[:, np.newaxis, :] )				, **ImagePipelineOutput(lowerCAmelCase__ ) )
 @torch.no_grad()
 def 		SCREAMING_SNAKE_CASE__   (   self				, lowerCAmelCase__				, lowerCAmelCase__ = 50 )					->				np.ndarray:




        '''simple docstring'''
        assert isinstance(self.scheduler				, lowerCAmelCase__ )
        self.scheduler.set_timesteps(lowerCAmelCase__ )
        lowercase__:   List[str]  			=   np.array(
            [np.frombuffer(image.tobytes()				, dtype='uint8' ).reshape((1, image.height, image.width) ) for image in images] )
        lowercase__:   str  			=   (sample / 255) * 2 - 1
        lowercase__:   str  			=   torch.Tensor(lowerCAmelCase__ ).to(self.device )

        for t in self.progress_bar(torch.flip(self.scheduler.timesteps				, (0,) ) ):
               lowercase__:   Union[str, Any]  			=   t - self.scheduler.config.num_train_timesteps // self.scheduler.num_inference_steps
               lowercase__:   Optional[Any]  			=   self.scheduler.alphas_cumprod[t]
               lowercase__:   str  			=   (
                   self.scheduler.alphas_cumprod[prev_timestep]
                   if prev_timestep >= 0
                   else self.scheduler.final_alpha_cumprod
               )
               lowercase__:   str  			=   1 - alpha_prod_t
               lowercase__:   int  			=   self.unet(lowerCAmelCase__				, lowerCAmelCase__ )['sample']
               lowercase__:   int  			=   (1 - alpha_prod_t_prev) ** 0.5 * model_output
               lowercase__:   Optional[int]  			=   (sample - pred_sample_direction) * alpha_prod_t_prev ** (-0.5)
               lowercase__:   Any  			=   sample * alpha_prod_t ** 0.5 + beta_prod_t ** 0.5 * model_output

        return sample
 @staticmethod
 def 		SCREAMING_SNAKE_CASE__   (   lowerCAmelCase__				, lowerCAmelCase__				, lowerCAmelCase__ )					->				torch.Tensor:




        '''simple docstring'''
        lowercase__:   Any  			=   acos(torch.dot(torch.flatten(lowerCAmelCase__ )				, torch.flatten(lowerCAmelCase__ ) ) / torch.norm(lowerCAmelCase__ ) / torch.norm(lowerCAmelCase__ ) )
        return sin((1 - alpha) * theta ) * xa / sin(lowerCAmelCase__ ) + sin(alpha * theta ) * xa / sin(lowerCAmelCase__ )


