


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__     ( lowerCAmelCase     ,							unittest.TestCase):
     SCREAMING_SNAKE_CASE__       =    SpectrogramDiffusionPipeline
     SCREAMING_SNAKE_CASE__       =    PipelineTesterMixin.required_optional_params - {
         '''callback''',
         '''latents''',
         '''callback_steps''',
         '''output_type''',
         '''num_images_per_prompt''',
     }
     SCREAMING_SNAKE_CASE__       =    False
     SCREAMING_SNAKE_CASE__       =    False
     SCREAMING_SNAKE_CASE__       =    TOKENS_TO_AUDIO_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE__       =    TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS





     def   __A      (self )  ->     List[Any]:
      torch.manual_seed(0 )
      _lowercase							   =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'''	,  )

      _lowercase							   =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'''	,  )

      _lowercase							   =TaFilmDecoder(
          input_dims=1_2_8	,  targets_length=2_5_6	,  max_decoder_noise_time=2_0000.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	,  )

      _lowercase							   =DDPMScheduler()

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





     def   __A      (self	,  UpperCAmelCase	,  UpperCAmelCase=0 )  ->     List[Any]:
      if str(UpperCAmelCase ).startswith('''mps''' ):
       _lowercase							   =torch.manual_seed(UpperCAmelCase )
      else:
       _lowercase							   =torch.Generator(device=UpperCAmelCase ).manual_seed(UpperCAmelCase )
      _lowercase							   ={
          '''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   __A      (self )  ->     List[Any]:
      _lowercase							   ='''cpu'''  # ensure determinism for the device-dependent torch.Generator
      _lowercase							   =self.get_dummy_components()
      _lowercase							   =SpectrogramDiffusionPipeline(**UpperCAmelCase )
      _lowercase							   =pipe.to(UpperCAmelCase )
      pipe.set_progress_bar_config(disable=UpperCAmelCase )

      _lowercase							   =self.get_dummy_inputs(UpperCAmelCase )
      _lowercase							   =pipe(**UpperCAmelCase )
      _lowercase							   =output.audios

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

      assert mel_slice.shape == (3, 3)
      _lowercase							   =np.array(
          [-11.51_2925, -4.78_8215, -0.4617_2905, -2.05_1715, -10.53_9147, -10.97_0963, -9.09_1634, 4.0, 4.0] )
      assert np.abs(mel_slice.flatten() - expected_slice ).max() < 1e-2





     @skip_mps
     def   __A      (self )  ->     Dict:
      return super().test_save_load_local()





     @skip_mps
     def   __A      (self )  ->     Dict:
      return super().test_dict_tuple_outputs_equivalent()





     @skip_mps
     def   __A      (self )  ->     Tuple:
      return super().test_save_load_optional_components()





     @skip_mps
     def   __A      (self )  ->     Optional[Any]:
      return super().test_attention_slicing_forward_pass()





     def   __A      (self )  ->     List[Any]:
      pass





     def   __A      (self )  ->     Tuple:
      pass





     @skip_mps
     def   __A      (self )  ->     str:
      return super().test_progress_bar()







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





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





     def   __A      (self )  ->     List[Any]:
      # TODO - test that pipeline can decode tokens in a callback
      # so that music can be played live
      _lowercase							   =torch_device

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

      _lowercase							   =pipe.to(UpperCAmelCase )
      pipe.set_progress_bar_config(disable=UpperCAmelCase )

      def callback(UpperCAmelCase	,  UpperCAmelCase ):
       # decode mel to audio
       _lowercase							   =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

      _lowercase							   =MidiProcessor()
      _lowercase							   =processor(UpperCAmelCase )

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





     def   __A      (self )  ->     Union[str, Any]:
      _lowercase							   =torch_device

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

      _lowercase							   =processor(UpperCAmelCase )
      # just run two denoising loops
      _lowercase							   =input_tokens[:2]

      _lowercase							   =torch.manual_seed(0 )
      _lowercase							   =pipe(UpperCAmelCase	,  num_inference_steps=2	,  generator=UpperCAmelCase )

      _lowercase							   =output.audios[0]

      assert abs(np.abs(UpperCAmelCase ).sum() - 3612.841 ) < 1e-1





     def   __A      (self )  ->     Tuple:
      _lowercase							   =torch_device

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

      _lowercase							   =MidiProcessor()

      _lowercase							   =processor(UpperCAmelCase )

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

      _lowercase							   =torch.manual_seed(0 )
      _lowercase							   =pipe(UpperCAmelCase	,  num_inference_steps=1_0_0	,  generator=UpperCAmelCase )

      _lowercase							   =output.audios[0]
      assert abs(np.abs(UpperCAmelCase ).sum() - 9389.1111 ) < 5e-2





import unittest

import numpy as np

from transformers import is_tf_available
from transformers.testing_utils import require_tf


if is_tf_available():
  import tensorflow as tf

  from transformers.activations_tf import get_tf_activation







@require_tf
class    lowerCamelCase__     ( unittest.TestCase):





     def   __A      (self )  ->     Optional[Any]:
      _lowercase							   =tf.constant([-1_0_0, -1.0, -0.1, 0, 0.1, 1.0, 100.0] )
      _lowercase							   =get_tf_activation('''gelu''' )
      _lowercase							   =get_tf_activation('''gelu_10''' )

      _lowercase							   =gelu(UpperCAmelCase )
      _lowercase							   =geluaa(UpperCAmelCase )

      _lowercase							   =tf.where(y_gelu_aa < 10.0	,  1.0	,  0.0 )

      self.assertEqual(tf.math.reduce_max(UpperCAmelCase ).numpy().item()	,  10.0 )
      self.assertTrue(np.allclose(y_gelu * clipped_mask	,  y_gelu_aa * clipped_mask ) )





     def   __A      (self )  ->     Optional[Any]:
      get_tf_activation('''gelu''' )
      get_tf_activation('''gelu_10''' )
      get_tf_activation('''gelu_fast''' )
      get_tf_activation('''gelu_new''' )
      get_tf_activation('''glu''' )
      get_tf_activation('''mish''' )
      get_tf_activation('''quick_gelu''' )
      get_tf_activation('''relu''' )
      get_tf_activation('''sigmoid''' )
      get_tf_activation('''silu''' )
      get_tf_activation('''swish''' )
      get_tf_activation('''tanh''' )
      with self.assertRaises(UpperCAmelCase ):
       get_tf_activation('''bogus''' )
      with self.assertRaises(UpperCAmelCase ):
       get_tf_activation(UpperCAmelCase )

