
'''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__						:				Optional[int]           =    "./tests/fixtures/elise_format0.mid"


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






 def 		lowercase_  (     self					:List[Any]     )			->     List[str]:







       '''simple docstring'''



       torch.manual_seed(0     )
       __A    		=     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    		=     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    		=     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		,			)

       __A    		=     DDPMScheduler()

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






 def 		lowercase_  (     self					:Dict		,			_A					:int		,			_A					:int=0     )			->     List[Any]:







       '''simple docstring'''



       if str(_A     ).startswith('mps'     ):
             __A    		=     torch.manual_seed(_A     )
       else:
             __A    		=     torch.Generator(device=_A     ).manual_seed(_A     )
       __A    		=     {
           '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 		lowercase_  (     self					:Optional[int]     )			->     Optional[int]:







       '''simple docstring'''



       __A    		=     'cpu'  # ensure determinism for the device-dependent torch.Generator
       __A    		=     self.get_dummy_components()
       __A    		=     SpectrogramDiffusionPipeline(**_A     )
       __A    		=     pipe.to(_A     )
       pipe.set_progress_bar_config(disable=_A     )

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

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

       assert mel_slice.shape == (3, 3)
       __A    		=     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 		lowercase_  (     self					:Optional[int]     )			->     str:







       '''simple docstring'''



       return super().test_save_load_local()






 @skip_mps
 def 		lowercase_  (     self					:Optional[Any]     )			->     Union[str, Any]:







       '''simple docstring'''



       return super().test_dict_tuple_outputs_equivalent()






 @skip_mps
 def 		lowercase_  (     self					:Tuple     )			->     Any:







       '''simple docstring'''



       return super().test_save_load_optional_components()






 @skip_mps
 def 		lowercase_  (     self					:Tuple     )			->     List[str]:







       '''simple docstring'''



       return super().test_attention_slicing_forward_pass()






 def 		lowercase_  (     self					:str     )			->     List[str]:







       '''simple docstring'''



       pass






 def 		lowercase_  (     self					:Optional[Any]     )			->     List[str]:







       '''simple docstring'''



       pass






 @skip_mps
 def 		lowercase_  (     self					:Optional[Any]     )			->     Tuple:







       '''simple docstring'''



       return super().test_progress_bar()


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






 def 		lowercase_  (     self					:Any     )			->     Tuple:







       '''simple docstring'''



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






 def 		lowercase_  (     self					:List[Any]     )			->     Union[str, Any]:







       '''simple docstring'''



       __A    		=     torch_device

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

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

       def callback(_A					:List[Any]		,			_A					:Tuple     ):
             # decode mel to audio
             __A    		=     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    		=     MidiProcessor()
       __A    		=     processor(_A     )

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






 def 		lowercase_  (     self					:Optional[Any]     )			->     int:







       '''simple docstring'''



       __A    		=     torch_device

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

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

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

       __A    		=     output.audios[0]

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






 def 		lowercase_  (     self					:str     )			->     str:







       '''simple docstring'''



       __A    		=     torch_device

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

       __A    		=     MidiProcessor()

       __A    		=     processor(_A     )

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

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

       __A    		=     output.audios[0]
       assert abs(np.abs(_A     ).sum() - 9_389.1_111     ) < 5E-2








'''simple docstring'''





import concurrent.futures
import json
import os
import shutil
import tempfile
import unittest

from transformers import AutoTokenizer, PreTrainedTokenizerFast
from transformers.testing_utils import require_tokenizers

from ..test_tokenization_common import TokenizerTesterMixin


@require_tokenizers
class 	UpperCamelCase__					(      SCREAMING_SNAKE_CASE ,       unittest.TestCase):
 UpperCAmelCase__      :      Tuple  			=   PreTrainedTokenizerFast
 UpperCAmelCase__      :      Tuple  			=   False
 UpperCAmelCase__      :      str  			=   True
 UpperCAmelCase__      :      str  			=   'tokenizer_file'






 def 		lowercase_  (     self					:Union[str, Any]     )			->     List[Any]:







       '''simple docstring'''



       __A    		=     False  # because we don't have pretrained_vocab_files_map
       super().setUp()
       __A    		=     True

       __A    		=     ['robot-test/dummy-tokenizer-fast', 'robot-test/dummy-tokenizer-wordlevel']
       __A    		=     'SaulLu/dummy-tokenizer-bytelevel-bpe'

       # Inclusion of 2 tokenizers to test different types of models (Unigram and WordLevel for the moment)
       __A    		=     [(PreTrainedTokenizerFast, model_path, {}) for model_path in model_paths]

       __A    		=     PreTrainedTokenizerFast.from_pretrained(model_paths[0]     )
       tokenizer.save_pretrained(self.tmpdirname     )






 def 		lowercase_  (     self					:Any     )			->     List[str]:







       '''simple docstring'''



       pass






 def 		lowercase_  (     self					:Optional[Any]     )			->     Optional[Any]:







       '''simple docstring'''



       pass






 def 		lowercase_  (     self					:Tuple     )			->     str:







       '''simple docstring'''



       pass






 def 		lowercase_  (     self					:List[Any]     )			->     Tuple:







       '''simple docstring'''



       pass






 def 		lowercase_  (     self					:List[Any]     )			->     str:







       '''simple docstring'''



       __A    		=     self.tmpdirname
       # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
       for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
             with self.subTest(F'{tokenizer.__class__.__name__} ({pretrained_name})'     ):
                   try:
                         __A    		=     tempfile.mkdtemp()
                         __A    		=     self.rust_tokenizer_class.from_pretrained(_A		,			**_A     )

                         tokenizer.save_pretrained(self.tmpdirname     )
                         super().test_training_new_tokenizer()
                   finally:
                         # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                         # is restored
                         shutil.rmtree(self.tmpdirname     )
                         __A    		=     tmpdirname_orig






 def 		lowercase_  (     self					:int     )			->     int:







       '''simple docstring'''



       __A    		=     self.tmpdirname
       # Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
       for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
             with self.subTest(F'{tokenizer.__class__.__name__} ({pretrained_name})'     ):
                   try:
                         __A    		=     tempfile.mkdtemp()
                         __A    		=     self.rust_tokenizer_class.from_pretrained(_A		,			**_A     )

                         tokenizer.save_pretrained(self.tmpdirname     )
                         super().test_training_new_tokenizer_with_special_tokens_change()
                   finally:
                         # Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
                         # is restored
                         shutil.rmtree(self.tmpdirname     )
                         __A    		=     tmpdirname_orig






 def 		lowercase_  (     self					:Optional[Any]     )			->     Any:







       '''simple docstring'''



       __A    		=     self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name     )

       __A    		=     ('a' for _ in range(1_000     ))
       __A    		=     tokenizer.train_new_from_iterator(text_iterator=_A		,			length=1_000		,			vocab_size=50     )

       __A    		=     new_tokenizer.encode('a🤗'     )
       self.assertEqual(_A		,			[64, 172, 253, 97, 245]     )


@require_tokenizers
class 	UpperCamelCase__					(      unittest.TestCase):






 def 		lowercase_  (     self					:Tuple     )			->     List[Any]:







       '''simple docstring'''



       __A    		=     AutoTokenizer.from_pretrained('bert-base-cased'     )
       __A    		=     json.loads(tokenizer._tokenizer.to_str()     )
       __A    		=     len(_A     )

       with tempfile.TemporaryDirectory() as tmp_dir:
             # Hack to save this in the tokenizer_config.json
             __A    		=     ['tokenizer.4.0.0.json']
             tokenizer.save_pretrained(_A     )
             json.dump(_A		,			open(os.path.join(_A		,			'tokenizer.4.0.0.json'     )		,			'w'     )     )

             # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
             __A    		=     AutoTokenizer.from_pretrained(_A     )
             self.assertEqual(len(_A     )		,			len(_A     ) + 1     )
             __A    		=     json.loads(new_tokenizer._tokenizer.to_str()     )
             self.assertIn('huggingface'		,			json_tokenizer['model']['vocab']     )

             # Will need to be adjusted if we reach v42 and this test is still here.
             # Should pick the old tokenizer file as the version of Transformers is < 4.0.0
             shutil.move(os.path.join(_A		,			'tokenizer.4.0.0.json'     )		,			os.path.join(_A		,			'tokenizer.42.0.0.json'     )     )
             __A    		=     ['tokenizer.42.0.0.json']
             tokenizer.save_pretrained(_A     )
             __A    		=     AutoTokenizer.from_pretrained(_A     )
             self.assertEqual(len(_A     )		,			len(_A     )     )
             __A    		=     json.loads(new_tokenizer._tokenizer.to_str()     )
             self.assertNotIn('huggingface'		,			json_tokenizer['model']['vocab']     )






 def 		lowercase_  (     self					:List[Any]     )			->     Optional[int]:







       '''simple docstring'''



       __A    		=     'hf-internal-testing/test-two-tokenizers'

       # This should pick the new tokenizer file as the version of Transformers is > 4.0.0
       __A    		=     AutoTokenizer.from_pretrained(_A     )
       self.assertEqual(len(_A     )		,			28_997     )
       __A    		=     json.loads(tokenizer._tokenizer.to_str()     )
       self.assertIn('huggingface'		,			json_tokenizer['model']['vocab']     )

       # Testing an older version by monkey-patching the version in the module it's used.
       import transformers as old_transformers

       __A    		=     '3.0.0'
       __A    		=     old_transformers.models.auto.AutoTokenizer.from_pretrained(_A     )
       self.assertEqual(len(_A     )		,			28_996     )
       __A    		=     json.loads(old_tokenizer._tokenizer.to_str()     )
       self.assertNotIn('huggingface'		,			json_tokenizer['model']['vocab']     )


@require_tokenizers
class 	UpperCamelCase__					(      unittest.TestCase):






 def 		lowercase_  (     self					:List[str]     )			->     List[Any]:







       '''simple docstring'''



       __A    		=     PreTrainedTokenizerFast.from_pretrained('robot-test/dummy-tokenizer-wordlevel'     )
       __A    		=     'The Matrix is a 1999 science fiction action film.'

       with concurrent.futures.ThreadPoolExecutor() as executor:
             __A    		=     [executor.submit(self.fetch		,			_A		,			_A     ) for i in range(10     )]
             __A    		=     [future.result() for future in futures]
             self.assertEqual(_A		,			[[1, 10, 0, 8, 0, 18, 0, 0, 0, 2] for i in range(10     )]     )






 def 		lowercase_  (     self					:Optional[int]		,			_A					:str		,			_A					:Any     )			->     Optional[int]:







       '''simple docstring'''



       return tokenizer.encode(_A		,			truncation='longest_first'		,			padding='longest'     )






