

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()


_lowerCamelCase							:  Union[str, Any]     =				"./tests/fixtures/elise_format0.mid"





@unittest.skip("""The note-seq package currently throws an error on import"""		)
class    __UpperCAmelCase	(			lowerCamelCase__     ,    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 			__magic_name__       (				self				:	int		):
     torch.manual_seed(0		)
     UpperCAmelCase			:				int										=					SpectrogramNotesEncoder(
         max_length=2_0_4_8,						vocab_size=1_5_3_6,						d_model=7_6_8,						dropout_rate=0.1,						num_layers=1,						num_heads=1,						d_kv=4,						d_ff=2_0_4_8,						feed_forward_proj='''gated-gelu''',						)

     UpperCAmelCase			:				str										=					SpectrogramContEncoder(
         input_dims=1_2_8,						targets_context_length=2_5_6,						d_model=7_6_8,						dropout_rate=0.1,						num_layers=1,						num_heads=1,						d_kv=4,						d_ff=2_0_4_8,						feed_forward_proj='''gated-gelu''',						)

     UpperCAmelCase			:				int										=					TaFilmDecoder(
         input_dims=1_2_8,						targets_length=2_5_6,						max_decoder_noise_time=2_0_0_0_0.0,						d_model=7_6_8,						num_layers=1,						num_heads=1,						d_kv=4,						d_ff=2_0_4_8,						dropout_rate=0.1,						)

     UpperCAmelCase			:				Optional[int]										=					DDPMScheduler()

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



   def 			__magic_name__       (				self				:	Tuple,						__A				:	Optional[Any],						__A				:	Optional[Any]=0		):
     if str(__A		).startswith('''mps'''		):
       UpperCAmelCase			:				int										=					torch.manual_seed(__A		)
     else:
       UpperCAmelCase			:				str										=					torch.Generator(device=__A		).manual_seed(__A		)
     UpperCAmelCase			:				Tuple										=					{
         '''input_tokens''': [
             [1_1_3_4, 9_0, 1_1_3_5, 1_1_3_3, 1_0_8_0, 1_1_2, 1_1_3_2, 1_0_8_0, 1_1_3_3, 1_0_7_9, 1_3_3, 1_1_3_2, 1_0_7_9, 1_1_3_3, 1] + [0] * 2_0_3_3
         ],
         '''generator''': generator,
         '''num_inference_steps''': 4,
         '''output_type''': '''mel''',
     }
     return inputs



   def 			__magic_name__       (				self				:	Optional[int]		):
     UpperCAmelCase			:				List[str]										=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
     UpperCAmelCase			:				Tuple										=					self.get_dummy_components()
     UpperCAmelCase			:				str										=					SpectrogramDiffusionPipeline(**__A		)
     UpperCAmelCase			:				Union[str, Any]										=					pipe.to(__A		)
     pipe.set_progress_bar_config(disable=__A		)

     UpperCAmelCase			:				Any										=					self.get_dummy_inputs(__A		)
     UpperCAmelCase			:				List[str]										=					pipe(**__A		)
     UpperCAmelCase			:				Union[str, Any]										=					output.audios

     UpperCAmelCase			:				int										=					mel[0, -3:, -3:]

     assert mel_slice.shape == (3, 3)
     UpperCAmelCase			:				Dict										=					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 			__magic_name__       (				self				:	str		):
     return super().test_save_load_local()



   @skip_mps
   def 			__magic_name__       (				self				:	int		):
     return super().test_dict_tuple_outputs_equivalent()



   @skip_mps
   def 			__magic_name__       (				self				:	int		):
     return super().test_save_load_optional_components()



   @skip_mps
   def 			__magic_name__       (				self				:	int		):
     return super().test_attention_slicing_forward_pass()



   def 			__magic_name__       (				self				:	Optional[int]		):
     pass



   def 			__magic_name__       (				self				:	Optional[Any]		):
     pass





   @skip_mps
   def 			__magic_name__       (				self				:	Tuple		):
     return super().test_progress_bar()






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



   def 			__magic_name__       (				self				:	List[Any]		):
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()



   def 			__magic_name__       (				self				:	Optional[Any]		):
     # TODO - test that pipeline can decode tokens in a callback
     # so that music can be played live
     UpperCAmelCase			:				Any										=					torch_device

     UpperCAmelCase			:				Tuple										=					SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''		)
     UpperCAmelCase			:				Optional[int]										=					pipe.melgan
     UpperCAmelCase			:				List[str]										=					None

     UpperCAmelCase			:				Optional[int]										=					pipe.to(__A		)
     pipe.set_progress_bar_config(disable=__A		)

     def callback(__A				:	Optional[Any],						__A				:	List[str]		):
       # decode mel to audio
       UpperCAmelCase			:				Optional[Any]										=					melgan(input_features=mel_output.astype(np.floataa		)		)[0]
       assert len(audio[0]		) == 8_1_9_2_0 * (step + 1)
       # simulate that audio is played
       return audio

     UpperCAmelCase			:				str										=					MidiProcessor()
     UpperCAmelCase			:				Optional[int]										=					processor(__A		)

     UpperCAmelCase			:				Optional[int]										=					input_tokens[:3]
     UpperCAmelCase			:				List[str]										=					torch.manual_seed(0		)
     pipe(__A,						num_inference_steps=5,						generator=__A,						callback=__A,						output_type='''mel'''		)



   def 			__magic_name__       (				self				:	Optional[Any]		):
     UpperCAmelCase			:				Dict										=					torch_device

     UpperCAmelCase			:				List[Any]										=					SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''		)
     UpperCAmelCase			:				List[Any]										=					pipe.to(__A		)
     pipe.set_progress_bar_config(disable=__A		)
     UpperCAmelCase			:				int										=					MidiProcessor()

     UpperCAmelCase			:				List[Any]										=					processor(__A		)
     # just run two denoising loops
     UpperCAmelCase			:				str										=					input_tokens[:2]

     UpperCAmelCase			:				Optional[int]										=					torch.manual_seed(0		)
     UpperCAmelCase			:				List[Any]										=					pipe(__A,						num_inference_steps=2,						generator=__A		)

     UpperCAmelCase			:				List[str]										=					output.audios[0]

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





   def 			__magic_name__       (				self				:	List[str]		):
     UpperCAmelCase			:				str										=					torch_device

     UpperCAmelCase			:				int										=					SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''		)
     UpperCAmelCase			:				Optional[Any]										=					pipe.to(__A		)
     pipe.set_progress_bar_config(disable=__A		)

     UpperCAmelCase			:				Dict										=					MidiProcessor()

     UpperCAmelCase			:				Union[str, Any]										=					processor(__A		)

     # just run 4 denoising loops
     UpperCAmelCase			:				Union[str, Any]										=					input_tokens[:4]

     UpperCAmelCase			:				Any										=					torch.manual_seed(0		)
     UpperCAmelCase			:				Optional[Any]										=					pipe(__A,						num_inference_steps=1_0_0,						generator=__A		)

     UpperCAmelCase			:				Tuple										=					output.audios[0]
     assert abs(np.abs(__A		).sum() - 9_3_8_9.1_1_1_1		) < 5E-2








def      a__	(						UpperCAmelCase				:     list     )    ->		list:
  UpperCAmelCase			:				Tuple										=					len(UpperCAmelCase     )

  # Set the gap value to be decreased by a factor of 1.3
  # after each iteration
  UpperCAmelCase			:				Optional[Any]										=					1.3

  # Continue sorting until the gap is 1
  while gap > 1:
    # Decrease the gap value
    UpperCAmelCase			:				int										=					int(gap / shrink     )

    # Sort the elements using insertion sort
    for i in range(UpperCAmelCase							,					len(UpperCAmelCase     )     ):
      UpperCAmelCase			:				Union[str, Any]										=					collection[i]
      UpperCAmelCase			:				List[Any]										=					i
      while j >= gap and collection[j - gap] > temp:
        UpperCAmelCase			:				int										=					collection[j - gap]
        j -= gap
      UpperCAmelCase			:				List[Any]										=					temp

  return collection


if __name__ == "__main__":
     import doctest

     doctest.testmod()





