



'''simple docstring'''





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


A			:   Union[str, Any]			=      """./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 					__snake_case       (    self    :    Union[str, Any]     ):



       '''simple docstring'''







       torch.manual_seed(0     )
       snake_case   :							Tuple						  =SpectrogramNotesEncoder(
           max_length=2_048,    vocab_size=1_536,    d_model=768,    dropout_rate=0.1,    num_layers=1,    num_heads=1,    d_kv=4,    d_ff=2_048,    feed_forward_proj='''gated-gelu''',    )

       snake_case   :							Tuple						  =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=2_048,    feed_forward_proj='''gated-gelu''',    )

       snake_case   :							List[Any]						  =TaFilmDecoder(
           input_dims=128,    targets_length=256,    max_decoder_noise_time=20_000.0,    d_model=768,    num_layers=1,    num_heads=1,    d_kv=4,    d_ff=2_048,    dropout_rate=0.1,    )

       snake_case   :							Dict						  =DDPMScheduler()

       snake_case   :							Union[str, Any]						  ={
           '''notes_encoder''': notes_encoder.eval(),
           '''continuous_encoder''': continuous_encoder.eval(),
           '''decoder''': decoder.eval(),
           '''scheduler''': scheduler,
           '''melgan''': None,
       }
       return components





    def 					__snake_case       (    self    :    Optional[int],    _snake_case    :    Tuple,    _snake_case    :    List[Any]=0     ):



       '''simple docstring'''







       if str(_snake_case     ).startswith('''mps'''     ):
          snake_case   :							List[Any]						  =torch.manual_seed(_snake_case     )
       else:
          snake_case   :							Any						  =torch.Generator(device=_snake_case     ).manual_seed(_snake_case     )
       snake_case   :							Any						  ={
           '''input_tokens''': [
               [1_134, 90, 1_135, 1_133, 1_080, 112, 1_132, 1_080, 1_133, 1_079, 133, 1_132, 1_079, 1_133, 1] + [0] * 2_033
           ],
           '''generator''': generator,
           '''num_inference_steps''': 4,
           '''output_type''': '''mel''',
       }
       return inputs





    def 					__snake_case       (    self    :    Dict     ):



       '''simple docstring'''







       snake_case   :							List[Any]						  ='''cpu'''  # ensure determinism for the device-dependent torch.Generator
       snake_case   :							str						  =self.get_dummy_components()
       snake_case   :							Any						  =SpectrogramDiffusionPipeline(**_snake_case     )
       snake_case   :							List[Any]						  =pipe.to(_snake_case     )
       pipe.set_progress_bar_config(disable=_snake_case     )

       snake_case   :							Optional[Any]						  =self.get_dummy_inputs(_snake_case     )
       snake_case   :							Union[str, Any]						  =pipe(**_snake_case     )
       snake_case   :							Union[str, Any]						  =output.audios

       snake_case   :							Optional[int]						  =mel[0, -3:, -3:]

       assert mel_slice.shape == (3, 3)
       snake_case   :							Union[str, Any]						  =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 					__snake_case       (    self    :    Union[str, Any]     ):



       '''simple docstring'''







       return super().test_save_load_local()





    @skip_mps
    def 					__snake_case       (    self    :    str     ):



       '''simple docstring'''







       return super().test_dict_tuple_outputs_equivalent()





    @skip_mps
    def 					__snake_case       (    self    :    Optional[Any]     ):



       '''simple docstring'''







       return super().test_save_load_optional_components()





    @skip_mps
    def 					__snake_case       (    self    :    Optional[int]     ):



       '''simple docstring'''







       return super().test_attention_slicing_forward_pass()





    def 					__snake_case       (    self    :    Dict     ):



       '''simple docstring'''







       pass





    def 					__snake_case       (    self    :    Optional[int]     ):



       '''simple docstring'''







       pass






    @skip_mps
    def 					__snake_case       (    self    :    Union[str, Any]     ):



       '''simple docstring'''







       return super().test_progress_bar()





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





    def 					__snake_case       (    self    :    Dict     ):



       '''simple docstring'''







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





    def 					__snake_case       (    self    :    List[str]     ):



       '''simple docstring'''







       snake_case   :							int						  =torch_device

       snake_case   :							Any						  =SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''     )
       snake_case   :							int						  =pipe.melgan
       snake_case   :							Any						  =None

       snake_case   :							str						  =pipe.to(_snake_case     )
       pipe.set_progress_bar_config(disable=_snake_case     )

       def callback(_snake_case    :    int,    _snake_case    :    List[Any]     ):
          # decode mel to audio
          snake_case   :							Optional[Any]						  =melgan(input_features=mel_output.astype(np.floataa     )     )[0]
          assert len(audio[0]     ) == 81_920 * (step + 1)
          # simulate that audio is played
          return audio

       snake_case   :							Tuple						  =MidiProcessor()
       snake_case   :							int						  =processor(_snake_case     )

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





    def 					__snake_case       (    self    :    Optional[Any]     ):



       '''simple docstring'''







       snake_case   :							Any						  =torch_device

       snake_case   :							Optional[Any]						  =SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''     )
       snake_case   :							Union[str, Any]						  =pipe.to(_snake_case     )
       pipe.set_progress_bar_config(disable=_snake_case     )
       snake_case   :							Optional[int]						  =MidiProcessor()

       snake_case   :							int						  =processor(_snake_case     )
       # just run two denoising loops
       snake_case   :							List[str]						  =input_tokens[:2]

       snake_case   :							Union[str, Any]						  =torch.manual_seed(0     )
       snake_case   :							str						  =pipe(_snake_case,    num_inference_steps=2,    generator=_snake_case     )

       snake_case   :							Optional[int]						  =output.audios[0]

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






    def 					__snake_case       (    self    :    str     ):



       '''simple docstring'''







       snake_case   :							Optional[int]						  =torch_device

       snake_case   :							str						  =SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''     )
       snake_case   :							Dict						  =pipe.to(_snake_case     )
       pipe.set_progress_bar_config(disable=_snake_case     )

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

       snake_case   :							Optional[Any]						  =processor(_snake_case     )

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

       snake_case   :							int						  =torch.manual_seed(0     )
       snake_case   :							List[Any]						  =pipe(_snake_case,    num_inference_steps=100,    generator=_snake_case     )

       snake_case   :							Tuple						  =output.audios[0]
       assert abs(np.abs(_snake_case     ).sum() - 9389.1111     ) < 5E-2






'''simple docstring'''





import doctest

import projectq
from projectq.ops import H, Measure
def 		_a   (  lowerCamelCase_							):
   snake_case   :							str						  =quantum_engine.allocate_qubit()
   H | qubit
   Measure | qubit
   return int(lowerCamelCase_							)


if __name__ == "__main__":
      doctest.testmod()

      # initialises a new quantum backend
      A			:   Dict			=      projectq.MainEngine()

      # Generate a list of 10 random numbers
      A			:   Dict			=      [get_random_number(quantum_engine) for _ in range(10)]

      # Flushes the quantum engine from memory
      quantum_engine.flush()

      print("""Random numbers""", random_numbers_list)





