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

import flax
import jax.numpy as jnp
from jax import random

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .scheduling_utils_flax import FlaxSchedulerMixin




@flax.struct.dataclass
class 					_lowerCamelCase			:





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	None
     SCREAMING_SNAKE_CASE_		 =	None
     SCREAMING_SNAKE_CASE_		 =	None  # sigma(t_i)






     @classmethod
     def  __SCREAMING_SNAKE_CASE      (				cls	)	-> Dict:




           """simple docstring"""







           return cls()




@dataclass
class 					_lowerCamelCase			(   UpperCamelCase_			):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	42
     SCREAMING_SNAKE_CASE_		 =	42
     SCREAMING_SNAKE_CASE_		 =	42




class 					_lowerCamelCase			(   UpperCamelCase_						,				UpperCamelCase_			):





     """simple docstring"""







     @property
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           return True






     @register_to_config
     def __init__(				self    ,    __SCREAMING_SNAKE_CASE = 0.02    ,    __SCREAMING_SNAKE_CASE = 1_0_0    ,    __SCREAMING_SNAKE_CASE = 1.007    ,    __SCREAMING_SNAKE_CASE = 8_0    ,    __SCREAMING_SNAKE_CASE = 0.05    ,    __SCREAMING_SNAKE_CASE = 5_0    ,    )	-> List[str]:




           """simple docstring"""







           pass






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[str]:




           """simple docstring"""







           return KarrasVeSchedulerState.create()






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = ()	)	-> KarrasVeSchedulerState:




           """simple docstring"""







           UpperCamelCase__     : Any											=					jnp.arange(0    ,    __SCREAMING_SNAKE_CASE	)[::-1].copy()
           UpperCamelCase__     : Optional[int]											=					[
               (
                   self.config.sigma_max**2
                   * (self.config.sigma_min**2 / self.config.sigma_max**2) ** (i / (num_inference_steps - 1))
               )
               for i in timesteps
           ]

           return state.replace(
               num_inference_steps=__SCREAMING_SNAKE_CASE    ,    schedule=jnp.array(__SCREAMING_SNAKE_CASE    ,    dtype=jnp.floataa	)    ,    timesteps=__SCREAMING_SNAKE_CASE    ,    )






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    )	-> Tuple[jnp.ndarray, float]:




           """simple docstring"""







           if self.config.s_min <= sigma <= self.config.s_max:
                 UpperCamelCase__     : Dict											=					min(self.config.s_churn / state.num_inference_steps    ,    2**0.5 - 1	)
           else:
                 UpperCamelCase__     : Union[str, Any]											=					0

           # sample eps ~ N(0, S_noise^2 * I)
           UpperCamelCase__     : Optional[Any]											=					random.split(__SCREAMING_SNAKE_CASE    ,    num=1	)
           UpperCamelCase__     : Tuple											=					self.config.s_noise * random.normal(key=__SCREAMING_SNAKE_CASE    ,    shape=sample.shape	)
           UpperCamelCase__     : Optional[int]											=					sigma + gamma * sigma
           UpperCamelCase__     : Any											=					sample + ((sigma_hat**2 - sigma**2) ** 0.5 * eps)

           return sample_hat, sigma_hat






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = True    ,    )	-> Union[FlaxKarrasVeOutput, Tuple]:




           """simple docstring"""







           UpperCamelCase__     : str											=					sample_hat + sigma_hat * model_output
           UpperCamelCase__     : Tuple											=					(sample_hat - pred_original_sample) / sigma_hat
           UpperCamelCase__     : List[Any]											=					sample_hat + (sigma_prev - sigma_hat) * derivative

           if not return_dict:
                 return (sample_prev, derivative, state)

           return FlaxKarrasVeOutput(prev_sample=__SCREAMING_SNAKE_CASE    ,    derivative=__SCREAMING_SNAKE_CASE    ,    state=__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE = True    ,    )	-> Union[FlaxKarrasVeOutput, Tuple]:




           """simple docstring"""







           UpperCamelCase__     : int											=					sample_prev + sigma_prev * model_output
           UpperCamelCase__     : str											=					(sample_prev - pred_original_sample) / sigma_prev
           UpperCamelCase__     : List[str]											=					sample_hat + (sigma_prev - sigma_hat) * (0.5 * derivative + 0.5 * derivative_corr)

           if not return_dict:
                 return (sample_prev, derivative, state)

           return FlaxKarrasVeOutput(prev_sample=__SCREAMING_SNAKE_CASE    ,    derivative=__SCREAMING_SNAKE_CASE    ,    state=__SCREAMING_SNAKE_CASE	)






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE	)	-> Tuple:




           """simple docstring"""







           raise NotImplementedError()




import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 					_lowerCamelCase			(   UpperCamelCase_						,				unittest.TestCase			):





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	DanceDiffusionPipeline
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE_		 =	PipelineTesterMixin.required_optional_params - {
         '''callback''',
         '''latents''',
         '''callback_steps''',
         '''output_type''',
         '''num_images_per_prompt''',
     }
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
     SCREAMING_SNAKE_CASE_		 =	False
     SCREAMING_SNAKE_CASE_		 =	False






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           torch.manual_seed(0	)
           UpperCamelCase__     : List[str]											=					UNetaDModel(
               block_out_channels=(3_2, 3_2, 6_4)    ,    extra_in_channels=1_6    ,    sample_size=5_1_2    ,    sample_rate=1_6_0_0_0    ,    in_channels=2    ,    out_channels=2    ,    flip_sin_to_cos=__SCREAMING_SNAKE_CASE    ,    use_timestep_embedding=__SCREAMING_SNAKE_CASE    ,    time_embedding_type='''fourier'''    ,    mid_block_type='''UNetMidBlock1D'''    ,    down_block_types=('''DownBlock1DNoSkip''', '''DownBlock1D''', '''AttnDownBlock1D''')    ,    up_block_types=('''AttnUpBlock1D''', '''UpBlock1D''', '''UpBlock1DNoSkip''')    ,    )
           UpperCamelCase__     : Union[str, Any]											=					IPNDMScheduler()

           UpperCamelCase__     : List[str]											=					{
               '''unet''': unet,
               '''scheduler''': scheduler,
           }
           return components






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=0	)	-> Any:




           """simple docstring"""







           if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 UpperCamelCase__     : Optional[Any]											=					torch.manual_seed(__SCREAMING_SNAKE_CASE	)
           else:
                 UpperCamelCase__     : Optional[Any]											=					torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : int											=					{
               '''batch_size''': 1,
               '''generator''': generator,
               '''num_inference_steps''': 4,
           }
           return inputs






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           UpperCamelCase__     : str											=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCamelCase__     : Dict											=					self.get_dummy_components()
           UpperCamelCase__     : str											=					DanceDiffusionPipeline(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[Any]											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Dict											=					pipe(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[str]											=					output.audios

           UpperCamelCase__     : Dict											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, components["unet"].sample_size)
           UpperCamelCase__     : List[Any]											=					np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]	)
           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> List[Any]:




           """simple docstring"""







           return super().test_save_load_local()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3	)






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           return super().test_save_load_optional_components()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Optional[int]:




           """simple docstring"""







           return super().test_attention_slicing_forward_pass()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







           super().test_inference_batch_single_identical(expected_max_diff=3e-3	)




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





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







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






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Any											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''	)
           UpperCamelCase__     : int											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Optional[Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : str											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Tuple											=					np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]	)

           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Union[str, Any]:




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Union[str, Any]											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''    ,    torch_dtype=torch.floataa	)
           UpperCamelCase__     : Tuple											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Union[str, Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : List[Any]											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Optional[Any]											=					np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]	)

           assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2


