





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


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




@unittest.skip('''The note-seq package currently throws an error on import'''							)
class 					_a		( _lowerCAmelCase      ,							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						:       int				)      -> str:




										'''simple docstring'''



										torch.manual_seed(0				)
										_UpperCamelCase     :   int      =       SpectrogramNotesEncoder(
										    max_length=2_0_4_8,					vocab_size=1_5_3_6,					d_model=7_6_8,					dropout_rate=0.1,					num_layers=1,					num_heads=1,					d_kv=4,					d_ff=2_0_4_8,					feed_forward_proj='''gated-gelu''',					)

										_UpperCamelCase     :   Optional[Any]      =       SpectrogramContEncoder(
										    input_dims=1_2_8,					targets_context_length=2_5_6,					d_model=7_6_8,					dropout_rate=0.1,					num_layers=1,					num_heads=1,					d_kv=4,					d_ff=2_0_4_8,					feed_forward_proj='''gated-gelu''',					)

										_UpperCamelCase     :   List[str]      =       TaFilmDecoder(
										    input_dims=1_2_8,					targets_length=2_5_6,					max_decoder_noise_time=20_000.0,					d_model=7_6_8,					num_layers=1,					num_heads=1,					d_kv=4,					d_ff=2_0_4_8,					dropout_rate=0.1,					)

										_UpperCamelCase     :   Optional[Any]      =       DDPMScheduler()

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






				def 							snake_case      (					self						:       str,					lowerCAmelCase__						:       List[str],					lowerCAmelCase__						:       str=0				)      -> Optional[int]:




										'''simple docstring'''



										if str(lowerCAmelCase__				).startswith('''mps'''				):
																_UpperCamelCase     :   Union[str, Any]      =       torch.manual_seed(lowerCAmelCase__				)
										else:
																_UpperCamelCase     :   List[str]      =       torch.Generator(device=lowerCAmelCase__				).manual_seed(lowerCAmelCase__				)
										_UpperCamelCase     :   List[str]      =       {
										    '''input_tokens''': [
										        [1_1_3_4, 9_0, 1_1_3_5, 1_1_3_3, 1_0_8_0, 1_1_2, 1_1_3_2, 1_0_8_0, 1_1_3_3, 1_0_7_9, 1_3_3, 1_1_3_2, 1_0_7_9, 1_1_3_3, 1] + [0] * 2_0_3_3
										    ],
										    '''generator''': generator,
										    '''num_inference_steps''': 4,
										    '''output_type''': '''mel''',
										}
										return inputs






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




										'''simple docstring'''



										_UpperCamelCase     :   Dict      =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
										_UpperCamelCase     :   Optional[int]      =       self.get_dummy_components()
										_UpperCamelCase     :   Tuple      =       SpectrogramDiffusionPipeline(**lowerCAmelCase__				)
										_UpperCamelCase     :   Dict      =       pipe.to(lowerCAmelCase__				)
										pipe.set_progress_bar_config(disable=lowerCAmelCase__				)

										_UpperCamelCase     :   Tuple      =       self.get_dummy_inputs(lowerCAmelCase__				)
										_UpperCamelCase     :   Dict      =       pipe(**lowerCAmelCase__				)
										_UpperCamelCase     :   Dict      =       output.audios

										_UpperCamelCase     :   Tuple      =       mel[0, -3:, -3:]

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




										'''simple docstring'''



										return super().test_save_load_local()






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




										'''simple docstring'''



										return super().test_dict_tuple_outputs_equivalent()






				@skip_mps
				def 							snake_case      (					self						:       List[Any]				)      -> int:




										'''simple docstring'''



										return super().test_save_load_optional_components()






				@skip_mps
				def 							snake_case      (					self						:       int				)      -> Optional[int]:




										'''simple docstring'''



										return super().test_attention_slicing_forward_pass()






				def 							snake_case      (					self						:       Any				)      -> Dict:




										'''simple docstring'''



										pass






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




										'''simple docstring'''



										pass






				@skip_mps
				def 							snake_case      (					self						:       int				)      -> List[Any]:




										'''simple docstring'''



										return super().test_progress_bar()




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






				def 							snake_case      (					self						:       Optional[int]				)      -> List[str]:




										'''simple docstring'''



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






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




										'''simple docstring'''



										_UpperCamelCase     :   List[Any]      =       torch_device

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

										_UpperCamelCase     :   Tuple      =       pipe.to(lowerCAmelCase__				)
										pipe.set_progress_bar_config(disable=lowerCAmelCase__				)

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

										_UpperCamelCase     :   Dict      =       MidiProcessor()
										_UpperCamelCase     :   Tuple      =       processor(lowerCAmelCase__				)

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






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




										'''simple docstring'''



										_UpperCamelCase     :   Dict      =       torch_device

										_UpperCamelCase     :   str      =       SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''				)
										_UpperCamelCase     :   Any      =       pipe.to(lowerCAmelCase__				)
										pipe.set_progress_bar_config(disable=lowerCAmelCase__				)
										_UpperCamelCase     :   List[str]      =       MidiProcessor()

										_UpperCamelCase     :   Tuple      =       processor(lowerCAmelCase__				)
										# just run two denoising loops
										_UpperCamelCase     :   Tuple      =       input_tokens[:2]

										_UpperCamelCase     :   List[str]      =       torch.manual_seed(0				)
										_UpperCamelCase     :   Any      =       pipe(lowerCAmelCase__,					num_inference_steps=2,					generator=lowerCAmelCase__				)

										_UpperCamelCase     :   Dict      =       output.audios[0]

										assert abs(np.abs(lowerCAmelCase__				).sum() - 3_612.841				) < 1e-1






				def 							snake_case      (					self						:       Tuple				)      -> List[str]:




										'''simple docstring'''



										_UpperCamelCase     :   int      =       torch_device

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

										_UpperCamelCase     :   List[str]      =       MidiProcessor()

										_UpperCamelCase     :   Tuple      =       processor(lowerCAmelCase__				)

										# just run 4 denoising loops
										_UpperCamelCase     :   Tuple      =       input_tokens[:4]

										_UpperCamelCase     :   List[Any]      =       torch.manual_seed(0				)
										_UpperCamelCase     :   Dict      =       pipe(lowerCAmelCase__,					num_inference_steps=1_0_0,					generator=lowerCAmelCase__				)

										_UpperCamelCase     :   Optional[int]      =       output.audios[0]
										assert abs(np.abs(lowerCAmelCase__				).sum() - 9_389.1_111				) < 5e-2






import itertools
import random
import unittest

import numpy as np

from transformers import EncodecFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
     import torch


lowercase_									=    random.Random()
def 		_snake_case(       SCREAMING_SNAKE_CASE__     :      Optional[int]       ,     SCREAMING_SNAKE_CASE__     :      List[str]=1.0       ,     SCREAMING_SNAKE_CASE__     :      List[Any]=None       ,     SCREAMING_SNAKE_CASE__     :      List[Any]=None )				-> List[str]:





    '''simple docstring'''






    if rng is None:
        A__        =  global_rng

    A__        =  []
    for batch_idx in range(shape[0] ):
        values.append([] )
        for _ in range(shape[1] ):
            values[-1].append(rng.random() * scale )

    return values


@require_torch
class 					A		(				unittest.TestCase					):







    """simple docstring"""





    def __init__( self					:			Any,lowercase_					:			Any,lowercase_					:			Tuple=7,lowercase_					:			Union[str, Any]=4_0_0,lowercase_					:			Union[str, Any]=2_0_0_0,lowercase_					:			int=1,lowercase_					:			Optional[Any]=0.0,lowercase_					:			Optional[int]=2_4_0_0_0,lowercase_					:			List[str]=True,)->       Any:





        '''simple docstring'''







        A__        =  parent
        A__        =  batch_size
        A__        =  min_seq_length
        A__        =  max_seq_length
        A__        =  (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        A__        =  feature_size
        A__        =  padding_value
        A__        =  sampling_rate
        A__        =  return_attention_mask


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





        '''simple docstring'''







        return {
            "feature_size": self.feature_size,
            "padding_value": self.padding_value,
            "sampling_rate": self.sampling_rate,
            "return_attention_mask": self.return_attention_mask,
        }


    def 		snake_case__	( self					:			Optional[Any],lowercase_					:			str=False,lowercase_					:			Tuple=False						)->       List[str]:





        '''simple docstring'''







        def _flatten(lowercase_					:			Any						):
            return list(itertools.chain(*lowercase_						)						)

        if equal_length:
            A__        =  floats_list((self.batch_size, self.max_seq_length)						)
        else:
            # make sure that inputs increase in size
            A__        =  [
                _flatten(floats_list((x, self.feature_size)						)						)
                for x in range(self.min_seq_length,self.max_seq_length,self.seq_length_diff						)
            ]

        if numpify:
            A__        =  [np.asarray(lowercase_						) for x in audio_inputs]

        return audio_inputs




@require_torch
class 					A		(				_UpperCAmelCase       ,				unittest.TestCase					):







    """simple docstring"""





    lowerCamelCase						      =				EncodecFeatureExtractor


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





        '''simple docstring'''







        A__        =  EnCodecFeatureExtractionTester(self						)


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





        '''simple docstring'''







        A__        =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
        # create three inputs of length 800, 1000, and 1200
        A__        =  [floats_list((1, x)						)[0] for x in range(8_0_0,1_4_0_0,2_0_0						)]
        A__        =  [np.asarray(lowercase_						) for audio_input in audio_inputs]

        # Test not batched input
        A__        =  feat_extract(audio_inputs[0],return_tensors='np'						).input_values
        A__        =  feat_extract(np_audio_inputs[0],return_tensors='np'						).input_values
        self.assertTrue(np.allclose(lowercase_,lowercase_,atol=1E-3						)						)

        # Test batched
        A__        =  feat_extract(lowercase_,padding=lowercase_,return_tensors='np'						).input_values
        A__        =  feat_extract(lowercase_,padding=lowercase_,return_tensors='np'						).input_values
        for enc_seq_a, enc_seq_a in zip(lowercase_,lowercase_						):
            self.assertTrue(np.allclose(lowercase_,lowercase_,atol=1E-3						)						)


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





        '''simple docstring'''







        A__        =  self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
        A__        =  np.random.rand(1_0_0						).astype(np.floataa						)
        A__        =  np_audio_inputs.tolist()

        for inputs in [py_audio_inputs, np_audio_inputs]:
            A__        =  feature_extractor.pad([{'input_values': inputs}],return_tensors='np'						)
            self.assertTrue(np_processed.input_values.dtype == np.floataa						)
            A__        =  feature_extractor.pad([{'input_values': inputs}],return_tensors='pt'						)
            self.assertTrue(pt_processed.input_values.dtype == torch.floataa						)


    def 		snake_case__	( self					:			int,lowercase_					:			Tuple						)->       Dict:





        '''simple docstring'''







        from datasets import load_dataset

        A__        =  load_dataset('hf-internal-testing/librispeech_asr_dummy','clean',split='validation'						)
        # automatic decoding with librispeech
        A__        =  ds.sort('id'						).select(range(lowercase_						)						)[:num_samples]['audio']

        return [x["array"] for x in audio_samples]


    def 		snake_case__	( self					:			int						)->       List[str]:





        '''simple docstring'''







        A__        =  torch.tensor(
            [2.3_804E-03, 2.0_752E-03, 1.9_836E-03, 2.1_057E-03, 1.6_174E-03,
             3.0_518E-04, 9.1_553E-05, 3.3_569E-04, 9.7_656E-04, 1.8_311E-03,
             2.0_142E-03, 2.1_057E-03, 1.7_395E-03, 4.5_776E-04, -3.9_673E-04,
             4.5_776E-04, 1.0_071E-03, 9.1_553E-05, 4.8_828E-04, 1.1_597E-03,
             7.3_242E-04, 9.4_604E-04, 1.8_005E-03, 1.8_311E-03, 8.8_501E-04,
             4.2_725E-04, 4.8_828E-04, 7.3_242E-04, 1.0_986E-03, 2.1_057E-03]						)
        # fmt: on
        A__        =  self._load_datasamples(1						)
        A__        =  EncodecFeatureExtractor()
        A__        =  feature_extractor(lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(input_values.shape,(1, 1, 9_3_6_8_0)						)
        self.assertTrue(torch.allclose(input_values[0, 0, :3_0],lowercase_,atol=1E-6						)						)


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





        '''simple docstring'''







        A__        =  torch.tensor(
            [2.3_804E-03, 2.0_752E-03, 1.9_836E-03, 2.1_057E-03, 1.6_174E-03,
             3.0_518E-04, 9.1_553E-05, 3.3_569E-04, 9.7_656E-04, 1.8_311E-03,
             2.0_142E-03, 2.1_057E-03, 1.7_395E-03, 4.5_776E-04, -3.9_673E-04,
             4.5_776E-04, 1.0_071E-03, 9.1_553E-05, 4.8_828E-04, 1.1_597E-03,
             7.3_242E-04, 9.4_604E-04, 1.8_005E-03, 1.8_311E-03, 8.8_501E-04,
             4.2_725E-04, 4.8_828E-04, 7.3_242E-04, 1.0_986E-03, 2.1_057E-03]						)
        # fmt: on
        A__        =  self._load_datasamples(1						)
        A__        =  [np.tile(input_audio[0][None],reps=(2, 1)						)]
        input_audio[0][1] *= 0.5
        A__        =  EncodecFeatureExtractor(feature_size=2						)
        A__        =  feature_extractor(lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(input_values.shape,(1, 2, 9_3_6_8_0)						)
        self.assertTrue(torch.allclose(input_values[0, 0, :3_0],lowercase_,atol=1E-6						)						)
        self.assertTrue(torch.allclose(input_values[0, 1, :3_0],EXPECTED_INPUT_VALUES * 0.5,atol=1E-6						)						)


    def 		snake_case__	( self					:			Any						)->       int:





        '''simple docstring'''







        A__        =  self._load_datasamples(2						)
        # would be easier if the stride was like
        A__        =  EncodecFeatureExtractor(feature_size=1,chunk_length_s=1,overlap=0.01						)

        # pad and trunc raise an error ?
        with self.assertRaisesRegex(
            lowercase_,'^Both padding and truncation were set. Make sure you only set one.$',):
            A__        =  feature_extractor(
                lowercase_,padding='max_length',truncation=lowercase_,return_tensors='pt'						).input_values

        # truncate to chunk
        A__        =  feature_extractor(lowercase_,truncation=lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(truncated_outputs.shape,(2, 1, 7_1_5_2_0)						)  # 2 chunks

        # force truncate to max_length
        A__        =  feature_extractor(
            lowercase_,truncation=lowercase_,max_length=4_8_0_0_0,return_tensors='pt'						).input_values
        self.assertEquals(truncated_outputs.shape,(2, 1, 4_8_0_0_0)						)

        # pad to chunk
        A__        =  feature_extractor(lowercase_,padding=lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(padded_outputs.shape,(2, 1, 9_5_2_8_0)						)

        # pad to chunk
        A__        =  feature_extractor(lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(truncated_outputs.shape,(2, 1, 9_5_2_8_0)						)

        # force pad to max length
        A__        =  feature_extractor(
            lowercase_,padding='max_length',max_length=1_0_0_0_0_0,return_tensors='pt'						).input_values
        self.assertEquals(truncated_outputs.shape,(2, 1, 1_0_0_0_0_0)						)

        # force no pad
        with self.assertRaisesRegex(
            lowercase_,'^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$',):
            A__        =  feature_extractor(lowercase_,padding=lowercase_,return_tensors='pt'						).input_values

        A__        =  feature_extractor(input_audio[0],padding=lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(truncated_outputs.shape,(1, 1, 9_3_6_8_0)						)

        # no pad if no chunk_length_s
        A__        =  None
        with self.assertRaisesRegex(
            lowercase_,'^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$',):
            A__        =  feature_extractor(lowercase_,padding=lowercase_,return_tensors='pt'						).input_values

        A__        =  feature_extractor(input_audio[0],padding=lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(truncated_outputs.shape,(1, 1, 9_3_6_8_0)						)

        # no pad if no overlap
        A__        =  2
        A__        =  None
        with self.assertRaisesRegex(
            lowercase_,'^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$',):
            A__        =  feature_extractor(lowercase_,padding=lowercase_,return_tensors='pt'						).input_values

        A__        =  feature_extractor(input_audio[0],padding=lowercase_,return_tensors='pt'						).input_values
        self.assertEquals(truncated_outputs.shape,(1, 1, 9_3_6_8_0)						)

