
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 				UpperCamelCase__		(	lowerCamelCase__       ,    unittest.TestCase	):






       """simple docstring"""


       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     _UpperCamelCase	(							self	)				->				Tuple:
          torch.manual_seed(0	)
          SCREAMING_SNAKE_CASE_	=      SpectrogramNotesEncoder(
              max_length=2048 ,     vocab_size=1536 ,     d_model=768 ,     dropout_rate=0.1 ,     num_layers=1 ,     num_heads=1 ,     d_kv=4 ,     d_ff=2048 ,     feed_forward_proj='''gated-gelu''' ,     )

          SCREAMING_SNAKE_CASE_	=      SpectrogramContEncoder(
              input_dims=128 ,     targets_context_length=256 ,     d_model=768 ,     dropout_rate=0.1 ,     num_layers=1 ,     num_heads=1 ,     d_kv=4 ,     d_ff=2048 ,     feed_forward_proj='''gated-gelu''' ,     )

          SCREAMING_SNAKE_CASE_	=      TaFilmDecoder(
              input_dims=128 ,     targets_length=256 ,     max_decoder_noise_time=20000.0 ,     d_model=768 ,     num_layers=1 ,     num_heads=1 ,     d_kv=4 ,     d_ff=2048 ,     dropout_rate=0.1 ,     )

          SCREAMING_SNAKE_CASE_	=      DDPMScheduler()

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



       def     _UpperCamelCase	(							self ,     _A ,     _A=0	)				->				Union[str, Any]:
          if str(__A	).startswith('''mps'''	):
             SCREAMING_SNAKE_CASE_	=      torch.manual_seed(__A	)
          else:
             SCREAMING_SNAKE_CASE_	=      torch.Generator(device=__A	).manual_seed(__A	)
          SCREAMING_SNAKE_CASE_	=      {
              '''input_tokens''': [
                  [1134, 90, 1135, 1133, 1080, 112, 1132, 1080, 1133, 1079, 133, 1132, 1079, 1133, 1] + [0] * 2033
              ],
              '''generator''': generator,
              '''num_inference_steps''': 4,
              '''output_type''': '''mel''',
          }
          return inputs



       def     _UpperCamelCase	(							self	)				->				Tuple:
          SCREAMING_SNAKE_CASE_	=      '''cpu'''  # ensure determinism for the device-dependent torch.Generator
          SCREAMING_SNAKE_CASE_	=      self.get_dummy_components()
          SCREAMING_SNAKE_CASE_	=      SpectrogramDiffusionPipeline(**__A	)
          SCREAMING_SNAKE_CASE_	=      pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)

          SCREAMING_SNAKE_CASE_	=      self.get_dummy_inputs(__A	)
          SCREAMING_SNAKE_CASE_	=      pipe(**__A	)
          SCREAMING_SNAKE_CASE_	=      output.audios

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

          assert mel_slice.shape == (3, 3)
          SCREAMING_SNAKE_CASE_	=      np.array(
              [-11.512925, -4.78_8215, -0.4617_2905, -2.05_1715, -10.539147, -10.970963, -9.09_1634, 4.0, 4.0]	)
          assert np.abs(mel_slice.flatten() - expected_slice	).max() < 1E-2



       @skip_mps
       def     _UpperCamelCase	(							self	)				->				Optional[int]:
          return super().test_save_load_local()



       @skip_mps
       def     _UpperCamelCase	(							self	)				->				Any:
          return super().test_dict_tuple_outputs_equivalent()



       @skip_mps
       def     _UpperCamelCase	(							self	)				->				Any:
          return super().test_save_load_optional_components()



       @skip_mps
       def     _UpperCamelCase	(							self	)				->				int:
          return super().test_attention_slicing_forward_pass()



       def     _UpperCamelCase	(							self	)				->				Tuple:
          pass



       def     _UpperCamelCase	(							self	)				->				str:
          pass




       @skip_mps
       def     _UpperCamelCase	(							self	)				->				List[str]:
          return super().test_progress_bar()







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






       """simple docstring"""


       def     _UpperCamelCase	(							self	)				->				str:
          # clean up the VRAM after each test
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()



       def     _UpperCamelCase	(							self	)				->				Dict:
          # TODO - test that pipeline can decode tokens in a callback
          # so that music can be played live
          SCREAMING_SNAKE_CASE_	=      torch_device

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

          SCREAMING_SNAKE_CASE_	=      pipe.to(__A	)
          pipe.set_progress_bar_config(disable=__A	)

          def callback(_A ,     _A	):
             # decode mel to audio
             SCREAMING_SNAKE_CASE_	=      melgan(input_features=mel_output.astype(np.floataa	)	)[0]
             assert len(audio[0]	) == 81920 * (step + 1)
             # simulate that audio is played
             return audio

          SCREAMING_SNAKE_CASE_	=      MidiProcessor()
          SCREAMING_SNAKE_CASE_	=      processor(__A	)

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



       def     _UpperCamelCase	(							self	)				->				str:
          SCREAMING_SNAKE_CASE_	=      torch_device

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

          SCREAMING_SNAKE_CASE_	=      processor(__A	)
          # just run two denoising loops
          SCREAMING_SNAKE_CASE_	=      input_tokens[:2]

          SCREAMING_SNAKE_CASE_	=      torch.manual_seed(0	)
          SCREAMING_SNAKE_CASE_	=      pipe(__A ,     num_inference_steps=2 ,     generator=__A	)

          SCREAMING_SNAKE_CASE_	=      output.audios[0]

          assert abs(np.abs(__A	).sum() - 3612.841	) < 1E-1




       def     _UpperCamelCase	(							self	)				->				Union[str, Any]:
          SCREAMING_SNAKE_CASE_	=      torch_device

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

          SCREAMING_SNAKE_CASE_	=      MidiProcessor()

          SCREAMING_SNAKE_CASE_	=      processor(__A	)

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

          SCREAMING_SNAKE_CASE_	=      torch.manual_seed(0	)
          SCREAMING_SNAKE_CASE_	=      pipe(__A ,     num_inference_steps=100 ,     generator=__A	)

          SCREAMING_SNAKE_CASE_	=      output.audios[0]
          assert abs(np.abs(__A	).sum() - 9389.1111	) < 5E-2





from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


_lowerCamelCase							:  Union[str, Any]     =				logging.get_logger(__name__)

_lowerCamelCase							:  Optional[int]     =				{
    "facebook/m2m100_418M": "https://huggingface.co/facebook/m2m100_418M/resolve/main/config.json",
    # See all M2M100 models at https://huggingface.co/models?filter=m2m_100
}





class    __UpperCAmelCase	(			lowerCamelCase__		):
   UpperCamelCase  	=	"""m2m_100"""
   UpperCamelCase  	=	["""past_key_values"""]
   UpperCamelCase  	=	{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}



   def __init__(				self				:	int,						__A				:	Optional[Any]=1_2_8_1_1_2,						__A				:	str=1_0_2_4,						__A				:	Tuple=1_2,						__A				:	Any=4_0_9_6,						__A				:	str=1_6,						__A				:	List[str]=1_2,						__A				:	int=4_0_9_6,						__A				:	List[Any]=1_6,						__A				:	Tuple=0.0_5,						__A				:	Tuple=0.0_5,						__A				:	List[Any]=True,						__A				:	int=True,						__A				:	str="relu",						__A				:	str=1_0_2_4,						__A				:	str=0.1,						__A				:	str=0.1,						__A				:	List[str]=0.0,						__A				:	List[str]=0.0_2,						__A				:	Any=2,						__A				:	List[str]=True,						__A				:	Tuple=1,						__A				:	Optional[Any]=0,						__A				:	Optional[Any]=2,						**__A				:	Optional[Any],						):
     UpperCAmelCase			:				str										=					vocab_size
     UpperCAmelCase			:				Optional[Any]										=					max_position_embeddings
     UpperCAmelCase			:				List[str]										=					d_model
     UpperCAmelCase			:				Any										=					encoder_ffn_dim
     UpperCAmelCase			:				Optional[int]										=					encoder_layers
     UpperCAmelCase			:				List[Any]										=					encoder_attention_heads
     UpperCAmelCase			:				str										=					decoder_ffn_dim
     UpperCAmelCase			:				Any										=					decoder_layers
     UpperCAmelCase			:				List[str]										=					decoder_attention_heads
     UpperCAmelCase			:				str										=					dropout
     UpperCAmelCase			:				int										=					attention_dropout
     UpperCAmelCase			:				Dict										=					activation_dropout
     UpperCAmelCase			:				Tuple										=					activation_function
     UpperCAmelCase			:				Any										=					init_std
     UpperCAmelCase			:				Union[str, Any]										=					encoder_layerdrop
     UpperCAmelCase			:				List[Any]										=					decoder_layerdrop
     UpperCAmelCase			:				Optional[Any]										=					use_cache
     UpperCAmelCase			:				str										=					encoder_layers
     UpperCAmelCase			:				List[Any]										=					scale_embedding  # scale factor will be sqrt(d_model) if True

     super().__init__(
         pad_token_id=__A,						bos_token_id=__A,						eos_token_id=__A,						is_encoder_decoder=__A,						decoder_start_token_id=__A,						**__A,						)






class    __UpperCAmelCase	(			lowerCamelCase__		):



   @property
   def 			__magic_name__       (				self				:	List[str]		):
     UpperCAmelCase			:				List[Any]										=					OrderedDict(
         [
             ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
             ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
         ]		)

     if self.use_past:
       UpperCAmelCase			:				List[str]										=					{0: '''batch'''}
       UpperCAmelCase			:				Any										=					{0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
     else:
       UpperCAmelCase			:				Any										=					{0: '''batch''', 1: '''decoder_sequence'''}
       UpperCAmelCase			:				List[Any]										=					{0: '''batch''', 1: '''decoder_sequence'''}

     if self.use_past:
       self.fill_with_past_key_values_(__A,						direction='''inputs'''		)
     return common_inputs



   def 			__magic_name__       (				self				:	Dict,						__A				:	PreTrainedTokenizer,						__A				:	int = -1,						__A				:	int = -1,						__A				:	bool = False,						__A				:	Optional[TensorType] = None,						):
     # Copied from OnnxConfig.generate_dummy_inputs
     # Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
     # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
     UpperCAmelCase			:				List[str]										=					compute_effective_axis_dimension(
         __A,						fixed_dimension=OnnxConfig.default_fixed_batch,						num_token_to_add=0		)

     # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
     UpperCAmelCase			:				Optional[int]										=					tokenizer.num_special_tokens_to_add(__A		)
     UpperCAmelCase			:				int										=					compute_effective_axis_dimension(
         __A,						fixed_dimension=OnnxConfig.default_fixed_sequence,						num_token_to_add=__A		)

     # Generate dummy inputs according to compute batch and sequence
     UpperCAmelCase			:				List[str]										=					[''' '''.join([tokenizer.unk_token]		) * seq_length] * batch_size
     UpperCAmelCase			:				Tuple										=					dict(tokenizer(__A,						return_tensors=__A		)		)
     return common_inputs



   def 			__magic_name__       (				self				:	Any,						__A				:	PreTrainedTokenizer,						__A				:	int = -1,						__A				:	int = -1,						__A				:	bool = False,						__A				:	Optional[TensorType] = None,						):
     UpperCAmelCase			:				int										=					self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
         __A,						__A,						__A,						__A,						__A		)

     # Generate decoder inputs
     UpperCAmelCase			:				Union[str, Any]										=					seq_length if not self.use_past else 1
     UpperCAmelCase			:				Optional[int]										=					self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
         __A,						__A,						__A,						__A,						__A		)
     UpperCAmelCase			:				Dict										=					{F'''decoder_{name}''': tensor for name, tensor in decoder_inputs.items()}
     UpperCAmelCase			:				Optional[int]										=					dict(**__A,						**__A		)

     if self.use_past:
       if not is_torch_available():
         raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''		)
       else:
         import torch
       UpperCAmelCase					,		UpperCAmelCase			:				List[str]										=					common_inputs['''input_ids'''].shape
       UpperCAmelCase			:				Dict										=					common_inputs['''decoder_input_ids'''].shape[1]
       UpperCAmelCase					,		UpperCAmelCase			:				Dict										=					self.num_attention_heads
       UpperCAmelCase			:				Union[str, Any]										=					(
           batch,
           num_encoder_attention_heads,
           encoder_seq_length,
           self._config.hidden_size // num_encoder_attention_heads,
       )
       UpperCAmelCase			:				str										=					decoder_seq_length + 3
       UpperCAmelCase			:				Union[str, Any]										=					(
           batch,
           num_decoder_attention_heads,
           decoder_past_length,
           self._config.hidden_size // num_decoder_attention_heads,
       )

       UpperCAmelCase			:				Union[str, Any]										=					torch.cat(
           [common_inputs['''decoder_attention_mask'''], torch.ones(__A,						__A		)],						dim=1		)

       UpperCAmelCase			:				Any										=					[]
       # If the number of encoder and decoder layers are present in the model configuration, both are considered
       UpperCAmelCase					,		UpperCAmelCase			:				str										=					self.num_layers
       UpperCAmelCase			:				Any										=					min(__A,						__A		)
       UpperCAmelCase			:				Dict										=					max(__A,						__A		) - min_num_layers
       UpperCAmelCase			:				Tuple										=					'''encoder''' if num_encoder_layers > num_decoder_layers else '''decoder'''

       for _ in range(__A		):
         common_inputs["past_key_values"].append(
             (
                 torch.zeros(__A		),
                 torch.zeros(__A		),
                 torch.zeros(__A		),
                 torch.zeros(__A		),
             )		)
       # TODO: test this.
       UpperCAmelCase			:				List[str]										=					encoder_shape if remaining_side_name == '''encoder''' else decoder_shape
       for _ in range(__A,						__A		):
         common_inputs["past_key_values"].append((torch.zeros(__A		), torch.zeros(__A		))		)
     return common_inputs





   UpperCamelCase  	=	_generate_dummy_inputs_for_default_and_seqaseq_lm





