





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


snake_case_     			=						"""./tests/fixtures/elise_format0.mid"""


@unittest.skip("""The note-seq package currently throws an error on import"""					)
class 	_lowercase    (				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       ):
       torch.manual_seed(0       )
       A							:							str				=    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'''     , )

       A							:							Any				=    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'''     , )

       A							:							Optional[int]				=    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=2_048     , dropout_rate=0.1     , )

       A							:							Optional[Any]				=    DDPMScheduler()

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


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

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

       A							:							str				=    mel[0, -3:, -3:]

       assert mel_slice.shape == (3, 3)
       A							:							Optional[Any]				=    np.array(
           [-11.512925, -4.788215, -0.46172905, -2.051715, -10.539147, -10.970963, -9.091634, 4.0, 4.0]       )
       assert np.abs(mel_slice.flatten() - expected_slice       ).max() < 1E-2


   @skip_mps
   def        snake_case					(			self       ):
       return super().test_save_load_local()


   @skip_mps
   def        snake_case					(			self       ):
       return super().test_dict_tuple_outputs_equivalent()


   @skip_mps
   def        snake_case					(			self       ):
       return super().test_save_load_optional_components()


   @skip_mps
   def        snake_case					(			self       ):
       return super().test_attention_slicing_forward_pass()


   def        snake_case					(			self       ):
       pass


   def        snake_case					(			self       ):
       pass


   @skip_mps
   def        snake_case					(			self       ):
       return super().test_progress_bar()







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


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


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

       A							:							Union[str, Any]				=    SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''       )
       A							:							Tuple				=    pipe.melgan
       A							:							Dict				=    None

       A							:							Tuple				=    pipe.to(_UpperCAmelCase       )
       pipe.set_progress_bar_config(disable=_UpperCAmelCase       )

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

       A							:							Optional[Any]				=    MidiProcessor()
       A							:							Union[str, Any]				=    processor(_UpperCAmelCase       )

       A							:							Optional[Any]				=    input_tokens[:3]
       A							:							str				=    torch.manual_seed(0       )
       pipe(_UpperCAmelCase     , num_inference_steps=5     , generator=_UpperCAmelCase     , callback=_UpperCAmelCase     , output_type='''mel'''       )


   def        snake_case					(			self       ):
       A							:							Optional[Any]				=    torch_device

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

       A							:							List[Any]				=    processor(_UpperCAmelCase       )
       # just run two denoising loops
       A							:							List[Any]				=    input_tokens[:2]

       A							:							str				=    torch.manual_seed(0       )
       A							:							List[str]				=    pipe(_UpperCAmelCase     , num_inference_steps=2     , generator=_UpperCAmelCase       )

       A							:							Optional[Any]				=    output.audios[0]

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


   def        snake_case					(			self       ):
       A							:							List[str]				=    torch_device

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

       A							:							str				=    MidiProcessor()

       A							:							Any				=    processor(_UpperCAmelCase       )

       # just run 4 denoising loops
       A							:							Dict				=    input_tokens[:4]

       A							:							List[Any]				=    torch.manual_seed(0       )
       A							:							Optional[Any]				=    pipe(_UpperCAmelCase     , num_inference_steps=100     , generator=_UpperCAmelCase       )

       A							:							List[Any]				=    output.audios[0]
       assert abs(np.abs(_UpperCAmelCase       ).sum() - 9389.1111       ) < 5E-2







'''simple docstring'''






import timeit


class 	_lowercase    :


   def __init__(			self     , _UpperCAmelCase       ):
       if isinstance(_UpperCAmelCase     , _UpperCAmelCase       ):  # Initializes a matrix identical to the one provided.
           A							:							List[Any]				=    arg
           A							:							Dict				=    len(_UpperCAmelCase       )
       else:  # Initializes a square matrix of the given size and set values to zero.
           A							:							str				=    arg
           A							:							Union[str, Any]				=    [[0 for _ in range(self.n       )] for _ in range(self.n       )]


   def __mul__(			self     , _UpperCAmelCase       ):
       A							:							List[str]				=    Matrix(self.n       )
       for i in range(self.n       ):
           for j in range(self.n       ):
               for k in range(self.n       ):
                   matrix.t[i][j] += self.t[i][k] * b.t[k][j]
       return matrix







def 				_lowerCamelCase(	UpperCamelCase__     :			Union[str, Any]     ,							UpperCamelCase__     :			int		)		->	Union[str, Any]:
    A							:							Tuple				=    Matrix([[1, 0], [0, 1]]		)
    while b > 0:
        if b & 1:
            matrix *= a
        a *= a
        b >>= 1
    return matrix
def 				_lowerCamelCase(	UpperCamelCase__     :			int     ,							UpperCamelCase__     :			Tuple     ,							UpperCamelCase__     :			Any		)		->	Any:
    # Trivial Cases
    if n == 1:
        return fa
    elif n == 2:
        return fa
    A							:							Optional[Any]				=    Matrix([[1, 1], [1, 0]]		)
    A							:							int				=    modular_exponentiation(UpperCamelCase__     ,							n - 2		)
    return fa * matrix.t[0][0] + fa * matrix.t[0][1]
def 				_lowerCamelCase(	UpperCamelCase__     :			str     ,							UpperCamelCase__     :			Optional[Any]     ,							UpperCamelCase__     :			Any		)		->	List[Any]:
    # Trivial Cases
    if n == 1:
        return fa
    elif n == 2:
        return fa

    A							:							str				=    fa
    A							:							List[Any]				=    fa
    n -= 2

    while n > 0:
        A,						A							:							List[str]				=    fn_a + fn_a, fn_a
        n -= 1

    return fn_a
def 				_lowerCamelCase(	)		->	Union[str, Any]:
    A							:							Any				=    '''
from random import randint
from __main__ import fibonacci_with_matrix_exponentiation
'''
    A							:							Optional[int]				=    '''fibonacci_with_matrix_exponentiation(randint(1,70000), 1, 1)'''
    A							:							List[Any]				=    timeit.timeit(setup=UpperCamelCase__     ,							stmt=UpperCamelCase__     ,							number=100		)
    print('''With matrix exponentiation the average execution time is '''     ,							exec_time / 100		)
    return exec_time
def 				_lowerCamelCase(	)		->	List[Any]:
    A							:							str				=    '''
from random import randint
from __main__ import simple_fibonacci
'''
    A							:							Any				=    '''simple_fibonacci(randint(1,70000), 1, 1)'''
    A							:							str				=    timeit.timeit(setup=UpperCamelCase__     ,							stmt=UpperCamelCase__     ,							number=100		)
    print(
        '''Without matrix exponentiation the average execution time is '''     ,							exec_time / 100		)
    return exec_time
def 				_lowerCamelCase(	)		->	Optional[Any]:
    matrix_exponentiation_time()
    simple_fibonacci_time()


if __name__ == "__main__":
     main()
