


"""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_											=			"./tests/fixtures/elise_format0.mid"
@unittest.skip('The note-seq package currently throws an error on import')
class snake_case					(     _UpperCamelCase   , 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        a_       (    self					:				str						)				->  Union[str, Any]:





						'''simple docstring'''







						torch.manual_seed(0						)
						_A         =	SpectrogramNotesEncoder(
						    max_length=20_48			,      vocab_size=15_36			,      d_model=7_68			,      dropout_rate=0.1			,      num_layers=1			,      num_heads=1			,      d_kv=4			,      d_ff=20_48			,      feed_forward_proj="gated-gelu"			,      )

						_A         =	SpectrogramContEncoder(
						    input_dims=1_28			,      targets_context_length=2_56			,      d_model=7_68			,      dropout_rate=0.1			,      num_layers=1			,      num_heads=1			,      d_kv=4			,      d_ff=20_48			,      feed_forward_proj="gated-gelu"			,      )

						_A         =	TaFilmDecoder(
						    input_dims=1_28			,      targets_length=2_56			,      max_decoder_noise_time=2_0_0_0_0.0			,      d_model=7_68			,      num_layers=1			,      num_heads=1			,      d_kv=4			,      d_ff=20_48			,      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        a_       (    self					:				List[str]			,      a__					:				List[Any]			,      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": [
						        [11_34, 90, 11_35, 11_33, 10_80, 1_12, 11_32, 10_80, 11_33, 10_79, 1_33, 11_32, 10_79, 11_33, 1] + [0] * 20_33
						    ],
						    "generator": generator,
						    "num_inference_steps": 4,
						    "output_type": "mel",
						}
						return inputs







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





						'''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(
						    [-1_1.5_1_2_9_2_5, -4.7_8_8_2_1_5, -0.4_6_1_7_2_9_0_5, -2.0_5_1_7_1_5, -1_0.5_3_9_1_4_7, -1_0.9_7_0_9_6_3, -9.0_9_1_6_3_4, 4.0, 4.0]						)
						assert np.abs(mel_slice.flatten() - expected_slice						).max() < 1E-2







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





						'''simple docstring'''







						return super().test_save_load_local()







		@skip_mps
		def        a_       (    self					:				List[Any]						)				->  Optional[Any]:





						'''simple docstring'''







						return super().test_dict_tuple_outputs_equivalent()







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





						'''simple docstring'''







						return super().test_save_load_optional_components()







		@skip_mps
		def        a_       (    self					:				List[Any]						)				->  Tuple:





						'''simple docstring'''







						return super().test_attention_slicing_forward_pass()







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





						'''simple docstring'''







						pass







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





						'''simple docstring'''







						pass







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





						'''simple docstring'''







						return super().test_progress_bar()





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







		def        a_       (    self					:				Dict						)				->  Tuple:





						'''simple docstring'''







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







		def        a_       (    self					:				Dict						)				->  Tuple:





						'''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__					:				Tuple			,      a__					:				List[Any]						):
										# decode mel to audio
										_A         =	melgan(input_features=mel_output.astype(np.floataa						)						)[0]
										assert len(audio[0]						) == 8_19_20 * (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        a_       (    self					:				str						)				->  Any:





						'''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_6_1_2.8_4_1						) < 1E-1







		def        a_       (    self					:				Optional[int]						)				->  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=1_00			,      generator=a__						)

						_A         =	output.audios[0]
						assert abs(np.abs(a__						).sum() - 9_3_8_9.1_1_1_1						) < 5E-2



"""simple docstring"""


import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device


if is_torch_available():
		import torch

		from transformers import CamembertModel
@require_torch
@require_sentencepiece
@require_tokenizers
class snake_case					(     unittest.TestCase):







		@slow
		def        a_       (    self					:				int						)				->  Tuple:





						'''simple docstring'''







						_A         =	CamembertModel.from_pretrained("camembert-base"						)
						model.to(a__						)

						_A         =	torch.tensor(
						    [[5, 1_21, 11, 6_60, 16, 7_30, 2_55_43, 1_10, 83, 6]]			,      device=a__			,      dtype=torch.long			,      )  # J'aime le camembert !
						with torch.no_grad():
										_A         =	model(a__						)["last_hidden_state"]
						_A         =	torch.Size((1, 10, 7_68)						)
						self.assertEqual(output.shape			,      a__						)
						# compare the actual values for a slice.
						_A         =	torch.tensor(
						    [[[-0.0_2_5_4, 0.0_2_3_5, 0.1_0_2_7], [0.0_6_0_6, -0.1_8_1_1, -0.0_4_1_8], [-0.1_5_6_1, -0.1_1_2_7, 0.2_6_8_7]]]			,      device=a__			,      dtype=torch.float			,      )
						# camembert = torch.hub.load('pytorch/fairseq', 'camembert.v0')
						# camembert.eval()
						# expected_slice = roberta.model.forward(input_ids)[0][:, :3, :3].detach()

						self.assertTrue(torch.allclose(output[:, :3, :3]			,      a__			,      atol=1E-4						)						)