

from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

import numpy as np
import PIL
from PIL import Image

from ...utils import BaseOutput, is_torch_available, is_transformers_available


@dataclass
class      lowerCamelCase_ (			_lowerCAmelCase):
 _lowerCAmelCase									=	{
     '''sld_warmup_steps''': 15,
     '''sld_guidance_scale''': 20,
     '''sld_threshold''': 0.0,
     '''sld_momentum_scale''': 0.0,
     '''sld_mom_beta''': 0.0,
 }
 _lowerCAmelCase									=	{
     '''sld_warmup_steps''': 10,
     '''sld_guidance_scale''': 10_00,
     '''sld_threshold''': 0.01,
     '''sld_momentum_scale''': 0.3,
     '''sld_mom_beta''': 0.4,
 }
 _lowerCAmelCase									=	{
     '''sld_warmup_steps''': 7,
     '''sld_guidance_scale''': 20_00,
     '''sld_threshold''': 0.0_25,
     '''sld_momentum_scale''': 0.5,
     '''sld_mom_beta''': 0.7,
 }
 _lowerCAmelCase									=	{
     '''sld_warmup_steps''': 0,
     '''sld_guidance_scale''': 50_00,
     '''sld_threshold''': 1.0,
     '''sld_momentum_scale''': 0.5,
     '''sld_mom_beta''': 0.7,
 }



@dataclass
class      lowerCamelCase_ (			_lowerCAmelCase):
 _lowerCAmelCase									=	42
 _lowerCAmelCase									=	42
 _lowerCAmelCase									=	42
 _lowerCAmelCase									=	42


if is_transformers_available() and is_torch_available():
      from .pipeline_stable_diffusion_safe import StableDiffusionPipelineSafe
      from .safety_checker import SafeStableDiffusionSafetyChecker










import gc
import unittest

import numpy as np
import torch

from diffusers import DDPMScheduler, MidiProcessor, SpectrogramDiffusionPipeline
from diffusers.pipelines.spectrogram_diffusion import SpectrogramContEncoder, SpectrogramNotesEncoder, TaFilmDecoder
from diffusers.utils import require_torch_gpu, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_note_seq, require_onnxruntime

from ..pipeline_params import TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS, TOKENS_TO_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


UpperCAmelCase__  =			"./tests/fixtures/elise_format0.mid"







@unittest.skip('''The note-seq package currently throws an error on import'''		)
class      __lowerCAmelCase       (     A							,		unittest.TestCase		):
			UpperCamelCase     =	SpectrogramDiffusionPipeline
			UpperCamelCase     =	PipelineTesterMixin.required_optional_params - {
			    '''callback''',
			    '''latents''',
			    '''callback_steps''',
			    '''output_type''',
			    '''num_images_per_prompt''',
			}
			UpperCamelCase     =	False
			UpperCamelCase     =	False
			UpperCamelCase     =	TOKENS_TO_AUDIO_GENERATION_PARAMS
			UpperCamelCase     =	TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS



			def        _lowerCamelCase			(		self  :    str)  ->       Any:



						"""simple docstring"""






						torch.manual_seed(0)
						_UpperCAmelCase									=  SpectrogramNotesEncoder(
						    max_length=20_48					,						vocab_size=15_36					,						d_model=7_68					,						dropout_rate=0.1					,						num_layers=1					,						num_heads=1					,						d_kv=4					,						d_ff=20_48					,						feed_forward_proj='gated-gelu'					,						)

						_UpperCAmelCase									=  SpectrogramContEncoder(
						    input_dims=1_28					,						targets_context_length=2_56					,						d_model=7_68					,						dropout_rate=0.1					,						num_layers=1					,						num_heads=1					,						d_kv=4					,						d_ff=20_48					,						feed_forward_proj='gated-gelu'					,						)

						_UpperCAmelCase									=  TaFilmDecoder(
						    input_dims=1_28					,						targets_length=2_56					,						max_decoder_noise_time=2_0_0_0_0.0					,						d_model=7_68					,						num_layers=1					,						num_heads=1					,						d_kv=4					,						d_ff=20_48					,						dropout_rate=0.1					,						)

						_UpperCAmelCase									=  DDPMScheduler()

						_UpperCAmelCase									=  {
						    'notes_encoder': notes_encoder.eval(),
						    'continuous_encoder': continuous_encoder.eval(),
						    'decoder': decoder.eval(),
						    'scheduler': scheduler,
						    'melgan': None,
						}
						return components



			def        _lowerCamelCase			(		self  :    Optional[Any]					,						A  :    str					,						A  :    Union[str, Any]=0)  ->       Optional[Any]:



						"""simple docstring"""






						if str(A).startswith('mps'):
									_UpperCAmelCase									=  torch.manual_seed(A)
						else:
									_UpperCAmelCase									=  torch.Generator(device=A).manual_seed(A)
						_UpperCAmelCase									=  {
						    'input_tokens': [
						        [11_34, 90, 11_35, 11_33, 10_80, 1_12, 11_32, 10_80, 11_33, 10_79, 1_33, 11_32, 10_79, 11_33, 1] + [0] * 20_33
						    ],
						    'generator': generator,
						    'num_inference_steps': 4,
						    'output_type': 'mel',
						}
						return inputs



			def        _lowerCamelCase			(		self  :    str)  ->       Optional[int]:



						"""simple docstring"""






						_UpperCAmelCase									=  'cpu'  # ensure determinism for the device-dependent torch.Generator
						_UpperCAmelCase									=  self.get_dummy_components()
						_UpperCAmelCase									=  SpectrogramDiffusionPipeline(**A)
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase									=  self.get_dummy_inputs(A)
						_UpperCAmelCase									=  pipe(**A)
						_UpperCAmelCase									=  output.audios

						_UpperCAmelCase									=  mel[0, -3:, -3:]

						assert mel_slice.shape == (3, 3)
						_UpperCAmelCase									=  np.array(
						    [-1_1.5_1_2_9_2_5, -4.7_8_8_2_1_5, -0.4_6_1_7_2_9_0_5, -2.0_5_1_7_1_5, -1_0.5_3_9_1_4_7, -1_0.9_7_0_9_6_3, -9.0_9_1_6_3_4, 4.0, 4.0])
						assert np.abs(mel_slice.flatten() - expected_slice).max() < 1E-2



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[Any])  ->       Optional[Any]:



						"""simple docstring"""






						return super().test_save_load_local()



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[Any])  ->       List[str]:



						"""simple docstring"""






						return super().test_dict_tuple_outputs_equivalent()



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[Any])  ->       Dict:



						"""simple docstring"""






						return super().test_save_load_optional_components()



			@skip_mps
			def        _lowerCamelCase			(		self  :    Union[str, Any])  ->       List[str]:



						"""simple docstring"""






						return super().test_attention_slicing_forward_pass()



			def        _lowerCamelCase			(		self  :    List[Any])  ->       Dict:



						"""simple docstring"""






						pass



			def        _lowerCamelCase			(		self  :    str)  ->       Any:



						"""simple docstring"""






						pass



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



						"""simple docstring"""






						return super().test_progress_bar()







@slow
@require_torch_gpu
@require_onnxruntime
@require_note_seq
class      __lowerCAmelCase       (     unittest.TestCase		):



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



						"""simple docstring"""






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



			def        _lowerCamelCase			(		self  :    int)  ->       Tuple:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.melgan
						_UpperCAmelCase									=  None

						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						def callback(A  :    Dict					,						A  :    Optional[int]):
									# decode mel to audio
									_UpperCAmelCase									=  melgan(input_features=mel_output.astype(np.floataa))[0]
									assert len(audio[0]) == 8_19_20 * (step + 1)
									# simulate that audio is played
									return audio

						_UpperCAmelCase									=  MidiProcessor()
						_UpperCAmelCase									=  processor(A)

						_UpperCAmelCase									=  input_tokens[:3]
						_UpperCAmelCase									=  torch.manual_seed(0)
						pipe(A					,						num_inference_steps=5					,						generator=A					,						callback=A					,						output_type='mel')



			def        _lowerCamelCase			(		self  :    str)  ->       Optional[int]:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)
						_UpperCAmelCase									=  MidiProcessor()

						_UpperCAmelCase									=  processor(A)
						# just run two denoising loops
						_UpperCAmelCase									=  input_tokens[:2]

						_UpperCAmelCase									=  torch.manual_seed(0)
						_UpperCAmelCase									=  pipe(A					,						num_inference_steps=2					,						generator=A)

						_UpperCAmelCase									=  output.audios[0]

						assert abs(np.abs(A).sum() - 3_6_1_2.8_4_1) < 1E-1



			def        _lowerCamelCase			(		self  :    List[str])  ->       Tuple:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase									=  MidiProcessor()

						_UpperCAmelCase									=  processor(A)

						# just run 4 denoising loops
						_UpperCAmelCase									=  input_tokens[:4]

						_UpperCAmelCase									=  torch.manual_seed(0)
						_UpperCAmelCase									=  pipe(A					,						num_inference_steps=1_00					,						generator=A)

						_UpperCAmelCase									=  output.audios[0]
						assert abs(np.abs(A).sum() - 9_3_8_9.1_1_1_1) < 5E-2
