



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


_SCREAMING_SNAKE_CASE =	"""./tests/fixtures/elise_format0.mid"""
@unittest.skip("""The note-seq package currently throws an error on import"""							)
class      __UpperCAmelCase							(    lowerCAmelCase_ ,unittest.TestCase							):



					'''simple docstring'''





					_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    :	List[str])						->       Tuple:
								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 						__snake_case	(  self    :	List[str]			,					_lowercase    :	Dict			,					_lowercase    :	Optional[Any]=0)						->       Optional[Any]:

								if str(__SCREAMING_SNAKE_CASE).startswith('mps'):
											A_       		=   torch.manual_seed(__SCREAMING_SNAKE_CASE)
								else:
											A_       		=   torch.Generator(device=__SCREAMING_SNAKE_CASE).manual_seed(__SCREAMING_SNAKE_CASE)
								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 						__snake_case	(  self    :	str)						->       int:
								A_       		=   'cpu'  # ensure determinism for the device-dependent torch.Generator
								A_       		=   self.get_dummy_components()
								A_       		=   SpectrogramDiffusionPipeline(**__SCREAMING_SNAKE_CASE)
								A_       		=   pipe.to(__SCREAMING_SNAKE_CASE)
								pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

								A_       		=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
								A_       		=   pipe(**__SCREAMING_SNAKE_CASE)
								A_       		=   output.audios

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

								assert mel_slice.shape == (3, 3)
								A_       		=   np.array(
								    [-11.51_29_25, -4.78_82_15, -0.46_17_29_05, -2.05_17_15, -10.53_91_47, -10.97_09_63, -9.09_16_34, 4.0, 4.0])
								assert np.abs(mel_slice.flatten() - expected_slice).max() < 1E-2



					@skip_mps
					def 						__snake_case	(  self    :	List[str])						->       int:
								return super().test_save_load_local()



					@skip_mps
					def 						__snake_case	(  self    :	Union[str, Any])						->       Any:
								return super().test_dict_tuple_outputs_equivalent()



					@skip_mps
					def 						__snake_case	(  self    :	Optional[int])						->       Union[str, Any]:
								return super().test_save_load_optional_components()



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



					def 						__snake_case	(  self    :	int)						->       Tuple:
								pass



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



					@skip_mps
					def 						__snake_case	(  self    :	List[str])						->       Optional[int]:
								return super().test_progress_bar()





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



					'''simple docstring'''





					def 						__snake_case	(  self    :	Optional[int])						->       Optional[Any]:
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()



					def 						__snake_case	(  self    :	str)						->       Dict:
								A_       		=   torch_device

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

								A_       		=   pipe.to(__SCREAMING_SNAKE_CASE)
								pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

								def callback(_lowercase    :	List[str]			,					_lowercase    :	int):
											# 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(__SCREAMING_SNAKE_CASE)

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



					def 						__snake_case	(  self    :	Dict)						->       str:
								A_       		=   torch_device

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

								A_       		=   processor(__SCREAMING_SNAKE_CASE)
								# just run two denoising loops
								A_       		=   input_tokens[:2]

								A_       		=   torch.manual_seed(0)
								A_       		=   pipe(__SCREAMING_SNAKE_CASE			,					num_inference_steps=2			,					generator=__SCREAMING_SNAKE_CASE)

								A_       		=   output.audios[0]

								assert abs(np.abs(__SCREAMING_SNAKE_CASE).sum() - 36_12.8_41) < 1E-1



					def 						__snake_case	(  self    :	Optional[int])						->       Union[str, Any]:
								A_       		=   torch_device

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

								A_       		=   MidiProcessor()

								A_       		=   processor(__SCREAMING_SNAKE_CASE)

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

								A_       		=   torch.manual_seed(0)
								A_       		=   pipe(__SCREAMING_SNAKE_CASE			,					num_inference_steps=100			,					generator=__SCREAMING_SNAKE_CASE)

								A_       		=   output.audios[0]
								assert abs(np.abs(__SCREAMING_SNAKE_CASE).sum() - 93_89.11_11) < 5E-2










'''simple docstring'''
import os
import pickle
import unittest
from collections import UserDict, namedtuple

import torch

from accelerate.test_utils.testing import require_cuda, require_torch_min_version
from accelerate.test_utils.training import RegressionModel
from accelerate.utils import (
    convert_outputs_to_fpaa,
    extract_model_from_parallel,
    find_device,
    listify,
    patch_environment,
    recursively_apply,
    send_to_device,
)


_SCREAMING_SNAKE_CASE =	namedtuple("ExampleNamedTuple", "a b c")
class      __UpperCAmelCase							(    unittest.TestCase							):



					'''simple docstring'''





					def 						__snake_case	(  self    :	List[Any])						->       Tuple:
								A_       		=   torch.randn(5			,					2)
								A_       		=   torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu')

								A_       		=   send_to_device(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta.cpu()			,					_lowercase))

								A_       		=   send_to_device((tensor, [tensor, tensor], 1)			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta[0].cpu()			,					_lowercase))
								self.assertIsInstance(resulta[1]			,					_lowercase)
								self.assertTrue(torch.equal(resulta[1][0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta[1][1].cpu()			,					_lowercase))
								self.assertEqual(resulta[2]			,					1)

								A_       		=   send_to_device({'a': tensor, 'b': [tensor, tensor], 'c': 1}			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta['a'].cpu()			,					_lowercase))
								self.assertIsInstance(resulta['b']			,					_lowercase)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta['b'][1].cpu()			,					_lowercase))
								self.assertEqual(resulta['c']			,					1)

								A_       		=   send_to_device(ExampleNamedTuple(a=_lowercase			,					b=[tensor, tensor]			,					c=1)			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta.a.cpu()			,					_lowercase))
								self.assertIsInstance(resulta.b			,					_lowercase)
								self.assertTrue(torch.equal(resulta.b[0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta.b[1].cpu()			,					_lowercase))
								self.assertEqual(resulta.c			,					1)

								A_       		=   send_to_device(UserDict({'a': tensor, 'b': [tensor, tensor], 'c': 1})			,					_lowercase)
								self.assertIsInstance(_lowercase			,					_lowercase)
								self.assertTrue(torch.equal(resulta['a'].cpu()			,					_lowercase))
								self.assertIsInstance(resulta['b']			,					_lowercase)
								self.assertTrue(torch.equal(resulta['b'][0].cpu()			,					_lowercase))
								self.assertTrue(torch.equal(resulta['b'][1].cpu()			,					_lowercase))
								self.assertEqual(resulta['c']			,					1)



					def 						__snake_case	(  self    :	Dict)						->       str:
								with self.assertRaises(_lowercase) as cm:
											A_       		=   recursively_apply(torch.tensor			,					(torch.tensor(1), 1)			,					error_on_other_type=_lowercase)
								self.assertEqual(
								    str(cm.exception)			,					'Unsupported types (<class \'int\'>) passed to `tensor`. Only nested list/tuple/dicts of objects that are valid for `is_torch_tensor` should be passed.'			,					)



					def 						__snake_case	(  self    :	Union[str, Any])						->       List[str]:
								A_       		=   torch.tensor([1, 2, 3, 4, 5])
								self.assertEqual(listify(_lowercase)			,					[1, 2, 3, 4, 5])

								A_       		=   torch.tensor([[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]])
								self.assertEqual(listify(_lowercase)			,					[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]])

								A_       		=   torch.tensor([[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]])
								self.assertEqual(
								    listify(_lowercase)			,					[[[1, 2, 3, 4, 5], [6, 7, 8, 9, 10]], [[11, 12, 13, 14, 15], [16, 17, 18, 19, 20]]])



					def 						__snake_case	(  self    :	List[Any])						->       Union[str, Any]:
								with patch_environment(aa=1			,					BB=2):
											self.assertEqual(os.environ.get('AA')			,					'1')
											self.assertEqual(os.environ.get('BB')			,					'2')

								self.assertNotIn('AA'			,					os.environ)
								self.assertNotIn('BB'			,					os.environ)



					def 						__snake_case	(  self    :	Any)						->       str:
								A_       		=   RegressionModel()
								A_       		=   model.forward
								A_       		=   convert_outputs_to_fpaa(model.forward)
								A_       		=   extract_model_from_parallel(_lowercase			,					keep_fpaa_wrapper=_lowercase)
								A_       		=   pickle.dumps(_lowercase)



					@require_cuda
					def 						__snake_case	(  self    :	Union[str, Any])						->       Union[str, Any]:
								A_       		=   RegressionModel()
								A_       		=   model.forward
								A_       		=   torch.cuda.amp.autocast(dtype=torch.floataa)(model.forward)
								A_       		=   convert_outputs_to_fpaa(model.forward)
								A_       		=   extract_model_from_parallel(_lowercase			,					keep_fpaa_wrapper=_lowercase)
								A_       		=   pickle.dumps(_lowercase)



					@require_cuda
					@require_torch_min_version(version='2.0')
					def 						__snake_case	(  self    :	Optional[Any])						->       Dict:
								A_       		=   RegressionModel()
								A_       		=   model.forward
								A_       		=   torch.cuda.amp.autocast(dtype=torch.floataa)(model.forward)
								A_       		=   convert_outputs_to_fpaa(model.forward)
								A_       		=   torch.compile(model.forward			,					backend='inductor')
								A_       		=   torch.randn(4			,					10).cuda()
								A_       		=   model(_lowercase)



					def 						__snake_case	(  self    :	Union[str, Any])						->       Optional[Any]:
								A_       		=   RegressionModel()
								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								A_       		=   torch.nn.parallel.DataParallel(_lowercase)
								A_       		=   extract_model_from_parallel(_lowercase)

								self.assertEqual(_lowercase			,					_lowercase)



					@require_torch_min_version(version='2.0')
					def 						__snake_case	(  self    :	str)						->       int:
								A_       		=   RegressionModel()
								A_       		=   torch.compile(_lowercase)

								# could also do a test with DistributedDataParallel, but difficult to run on CPU or single GPU
								A_       		=   torch.nn.parallel.DataParallel(_lowercase)
								A_       		=   torch.compile(_lowercase)
								A_       		=   extract_model_from_parallel(_lowercase)

								self.assertEqual(compiled_model._orig_mod			,					compiled_model_unwrapped._orig_mod)



					def 						__snake_case	(  self    :	List[str])						->       Optional[int]:
								self.assertEqual(find_device([1, 'a', torch.tensor([1, 2, 3])])			,					torch.device('cpu'))
								self.assertEqual(find_device({'a': 1, 'b': torch.tensor([1, 2, 3])})			,					torch.device('cpu'))
								self.assertIsNone(find_device([1, 'a']))




