




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__    :							Any       		=     './tests/fixtures/elise_format0.mid'

@unittest.skip("The note-seq package currently throws an error on import" )
class   UpperCAmelCase    (      A ,       unittest.TestCase ):
   __SCREAMING_SNAKE_CASE		:				Dict      					=       SpectrogramDiffusionPipeline
   __SCREAMING_SNAKE_CASE		:				Any      					=       PipelineTesterMixin.required_optional_params - {
       "callback",
       "latents",
       "callback_steps",
       "output_type",
       "num_images_per_prompt",
   }
   __SCREAMING_SNAKE_CASE		:				int      					=       False
   __SCREAMING_SNAKE_CASE		:				Optional[int]      					=       False
   __SCREAMING_SNAKE_CASE		:				Optional[int]      					=       TOKENS_TO_AUDIO_GENERATION_PARAMS
   __SCREAMING_SNAKE_CASE		:				Dict      					=       TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS






   def 			_a    (						self :							Tuple  )->      Tuple:
     torch.manual_seed(0  )
     _lowercase:			int      =     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'    ,       )

     _lowercase:			Optional[Any]      =     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'    ,       )

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

     _lowercase:			List[str]      =     DDPMScheduler()

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






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

     _lowercase:			Optional[Any]      =     self.get_dummy_inputs(UpperCamelCase_  )
     _lowercase:			List[Any]      =     pipe(**UpperCamelCase_  )
     _lowercase:			List[str]      =     output.audios

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

     assert mel_slice.shape == (3, 3)
     _lowercase:			int      =     np.array(
         [-11.512_925, -4.788_215, -0.46_172_905, -2.051_715, -10.539_147, -10.970_963, -9.091_634, 4.0, 4.0]  )
     assert np.abs(mel_slice.flatten() - expected_slice  ).max() < 1E-2






   @skip_mps
   def 			_a    (						self :							Any  )->      int:
     return super().test_save_load_local()






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






   @skip_mps
   def 			_a    (						self :							List[Any]  )->      List[str]:
     return super().test_save_load_optional_components()






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






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






   def 			_a    (						self :							Tuple  )->      Union[str, Any]:
     pass






   @skip_mps
   def 			_a    (						self :							Dict  )->      Optional[Any]:
     return super().test_progress_bar()

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






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






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

     _lowercase:			Any      =     SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'  )
     _lowercase:			List[Any]      =     pipe.melgan
     _lowercase:			str      =     None

     _lowercase:			int      =     pipe.to(UpperCamelCase_  )
     pipe.set_progress_bar_config(disable=UpperCamelCase_  )

     def callback(UpperCamelCase_ :							List[Any]    ,       UpperCamelCase_ :							Union[str, Any]  ):
       # decode mel to audio
       _lowercase:			Dict      =     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

     _lowercase:			Any      =     MidiProcessor()
     _lowercase:			Tuple      =     processor(UpperCamelCase_  )

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






   def 			_a    (						self :							Union[str, Any]  )->      Optional[int]:
     _lowercase:			List[Any]      =     torch_device

     _lowercase:			List[Any]      =     SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'  )
     _lowercase:			Any      =     pipe.to(UpperCamelCase_  )
     pipe.set_progress_bar_config(disable=UpperCamelCase_  )
     _lowercase:			Dict      =     MidiProcessor()

     _lowercase:			Union[str, Any]      =     processor(UpperCamelCase_  )
     # just run two denoising loops
     _lowercase:			List[str]      =     input_tokens[:2]

     _lowercase:			Optional[Any]      =     torch.manual_seed(0  )
     _lowercase:			Optional[Any]      =     pipe(UpperCamelCase_    ,       num_inference_steps=2    ,       generator=UpperCamelCase_  )

     _lowercase:			Union[str, Any]      =     output.audios[0]

     assert abs(np.abs(UpperCamelCase_  ).sum() - 3_612.841  ) < 1E-1






   def 			_a    (						self :							Any  )->      Tuple:
     _lowercase:			int      =     torch_device

     _lowercase:			Union[str, Any]      =     SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'  )
     _lowercase:			Union[str, Any]      =     pipe.to(UpperCamelCase_  )
     pipe.set_progress_bar_config(disable=UpperCamelCase_  )

     _lowercase:			Union[str, Any]      =     MidiProcessor()

     _lowercase:			Dict      =     processor(UpperCamelCase_  )

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

     _lowercase:			int      =     torch.manual_seed(0  )
     _lowercase:			Any      =     pipe(UpperCamelCase_    ,       num_inference_steps=1_00    ,       generator=UpperCamelCase_  )

     _lowercase:			str      =     output.audios[0]
     assert abs(np.abs(UpperCamelCase_  ).sum() - 9_389.1_111  ) < 5E-2






def 		_lowerCamelCase					(       UpperCamelCase       ,  UpperCamelCase				)      ->		int:





  """simple docstring"""





  return (input_a, input_a).count(0				) % 2


def 		_lowerCamelCase					(       )      ->		None:





  """simple docstring"""





  assert xor_gate(0       ,  0				) == 0
  assert xor_gate(0       ,  1				) == 1
  assert xor_gate(1       ,  0				) == 1
  assert xor_gate(1       ,  1				) == 0


if __name__ == "__main__":
   print(xor_gate(0, 0))
   print(xor_gate(0, 1))
