


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






@unittest.skip("The note-seq package currently throws an error on import"		)
class 		_snake_case    (				snake_case_  ,  unittest.TestCase		):






						'''simple docstring'''







						__snake_case   =      SpectrogramDiffusionPipeline
						__snake_case   =      PipelineTesterMixin.required_optional_params - {
						    "callback",
						    "latents",
						    "callback_steps",
						    "output_type",
						    "num_images_per_prompt",
						}
						__snake_case   =      False
						__snake_case   =      False
						__snake_case   =      TOKENS_TO_AUDIO_GENERATION_PARAMS
						__snake_case   =      TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS







						def lowerCAmelCase__ (  self:	Any     )							->	Dict:
										torch.manual_seed(0     )
										__magic_name__				:					Tuple						       =       SpectrogramNotesEncoder(
										    max_length=2048      ,							vocab_size=1536      ,							d_model=768      ,							dropout_rate=0.1      ,							num_layers=1      ,							num_heads=1      ,							d_kv=4      ,							d_ff=2048      ,							feed_forward_proj="gated-gelu"      ,							)

										__magic_name__				:					List[str]						       =       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=2048      ,							feed_forward_proj="gated-gelu"      ,							)

										__magic_name__				:					Optional[int]						       =       TaFilmDecoder(
										    input_dims=128      ,							targets_length=256      ,							max_decoder_noise_time=2_0_0_0_0.0      ,							d_model=768      ,							num_layers=1      ,							num_heads=1      ,							d_kv=4      ,							d_ff=2048      ,							dropout_rate=0.1      ,							)

										__magic_name__				:					Any						       =       DDPMScheduler()

										__magic_name__				:					Optional[int]						       =       {
										    "notes_encoder": notes_encoder.eval(),
										    "continuous_encoder": continuous_encoder.eval(),
										    "decoder": decoder.eval(),
										    "scheduler": scheduler,
										    "melgan": None,
										}
										return components







						def lowerCAmelCase__ (  self:	Optional[int]      ,							__UpperCamelCase:	List[str]      ,							__UpperCamelCase:	Union[str, Any]=0     )							->	Dict:
										if str(__UpperCamelCase     ).startswith("mps"     ):
														__magic_name__				:					List[str]						       =       torch.manual_seed(__UpperCamelCase     )
										else:
														__magic_name__				:					List[str]						       =       torch.Generator(device=__UpperCamelCase     ).manual_seed(__UpperCamelCase     )
										__magic_name__				:					Union[str, Any]						       =       {
										    "input_tokens": [
										        [1134, 90, 1135, 1133, 1080, 112, 1132, 1080, 1133, 1079, 133, 1132, 1079, 1133, 1] + [0] * 2033
										    ],
										    "generator": generator,
										    "num_inference_steps": 4,
										    "output_type": "mel",
										}
										return inputs







						def lowerCAmelCase__ (  self:	Optional[int]     )							->	List[str]:
										__magic_name__				:					List[Any]						       =       "cpu"  # ensure determinism for the device-dependent torch.Generator
										__magic_name__				:					List[str]						       =       self.get_dummy_components()
										__magic_name__				:					Optional[Any]						       =       SpectrogramDiffusionPipeline(**__UpperCamelCase     )
										__magic_name__				:					Any						       =       pipe.to(__UpperCamelCase     )
										pipe.set_progress_bar_config(disable=__UpperCamelCase     )

										__magic_name__				:					List[Any]						       =       self.get_dummy_inputs(__UpperCamelCase     )
										__magic_name__				:					Any						       =       pipe(**__UpperCamelCase     )
										__magic_name__				:					int						       =       output.audios

										__magic_name__				:					Union[str, Any]						       =       mel[0, -3:, -3:]

										assert mel_slice.shape == (3, 3)
										__magic_name__				:					Optional[Any]						       =       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 lowerCAmelCase__ (  self:	Optional[Any]     )							->	Tuple:
										return super().test_save_load_local()







						@skip_mps
						def lowerCAmelCase__ (  self:	List[Any]     )							->	Optional[int]:
										return super().test_dict_tuple_outputs_equivalent()







						@skip_mps
						def lowerCAmelCase__ (  self:	Tuple     )							->	List[str]:
										return super().test_save_load_optional_components()







						@skip_mps
						def lowerCAmelCase__ (  self:	List[Any]     )							->	Union[str, Any]:
										return super().test_attention_slicing_forward_pass()







						def lowerCAmelCase__ (  self:	Tuple     )							->	Dict:
										pass







						def lowerCAmelCase__ (  self:	Union[str, Any]     )							->	Dict:
										pass







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






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






						'''simple docstring'''







						def lowerCAmelCase__ (  self:	Dict     )							->	int:
										# clean up the VRAM after each test
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()







						def lowerCAmelCase__ (  self:	int     )							->	Optional[int]:
										# TODO - test that pipeline can decode tokens in a callback
										# so that music can be played live
										__magic_name__				:					Tuple						       =       torch_device

										__magic_name__				:					List[str]						       =       SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"     )
										__magic_name__				:					List[Any]						       =       pipe.melgan
										__magic_name__				:					Tuple						       =       None

										__magic_name__				:					Union[str, Any]						       =       pipe.to(__UpperCamelCase     )
										pipe.set_progress_bar_config(disable=__UpperCamelCase     )

										def callback(__UpperCamelCase:	Tuple      ,							__UpperCamelCase:	Tuple     ):
														# decode mel to audio
														__magic_name__				:					List[Any]						       =       melgan(input_features=mel_output.astype(np.floataa     )     )[0]
														assert len(audio[0]     ) == 8_1920 * (step + 1)
														# simulate that audio is played
														return audio

										__magic_name__				:					Optional[Any]						       =       MidiProcessor()
										__magic_name__				:					str						       =       processor(__UpperCamelCase     )

										__magic_name__				:					Dict						       =       input_tokens[:3]
										__magic_name__				:					Tuple						       =       torch.manual_seed(0     )
										pipe(__UpperCamelCase      ,							num_inference_steps=5      ,							generator=__UpperCamelCase      ,							callback=__UpperCamelCase      ,							output_type="mel"     )







						def lowerCAmelCase__ (  self:	List[str]     )							->	List[Any]:
										__magic_name__				:					Dict						       =       torch_device

										__magic_name__				:					Any						       =       SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"     )
										__magic_name__				:					Tuple						       =       pipe.to(__UpperCamelCase     )
										pipe.set_progress_bar_config(disable=__UpperCamelCase     )
										__magic_name__				:					List[Any]						       =       MidiProcessor()

										__magic_name__				:					Optional[Any]						       =       processor(__UpperCamelCase     )
										# just run two denoising loops
										__magic_name__				:					Tuple						       =       input_tokens[:2]

										__magic_name__				:					Optional[Any]						       =       torch.manual_seed(0     )
										__magic_name__				:					Optional[int]						       =       pipe(__UpperCamelCase      ,							num_inference_steps=2      ,							generator=__UpperCamelCase     )

										__magic_name__				:					str						       =       output.audios[0]

										assert abs(np.abs(__UpperCamelCase     ).sum() - 3_6_1_2.8_4_1     ) < 1E-1







						def lowerCAmelCase__ (  self:	List[Any]     )							->	Union[str, Any]:
										__magic_name__				:					List[Any]						       =       torch_device

										__magic_name__				:					Any						       =       SpectrogramDiffusionPipeline.from_pretrained("google/music-spectrogram-diffusion"     )
										__magic_name__				:					Dict						       =       pipe.to(__UpperCamelCase     )
										pipe.set_progress_bar_config(disable=__UpperCamelCase     )

										__magic_name__				:					str						       =       MidiProcessor()

										__magic_name__				:					int						       =       processor(__UpperCamelCase     )

										# just run 4 denoising loops
										__magic_name__				:					Optional[int]						       =       input_tokens[:4]

										__magic_name__				:					Dict						       =       torch.manual_seed(0     )
										__magic_name__				:					Tuple						       =       pipe(__UpperCamelCase      ,							num_inference_steps=100      ,							generator=__UpperCamelCase     )

										__magic_name__				:					List[Any]						       =       output.audios[0]
										assert abs(np.abs(__UpperCamelCase     ).sum() - 9_3_8_9.1_1_1_1     ) < 5E-2



'''simple docstring'''







from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


_SCREAMING_SNAKE_CASE       :   Any        = {
    "configuration_deberta_v2": ["DEBERTA_V2_PRETRAINED_CONFIG_ARCHIVE_MAP", "DebertaV2Config", "DebertaV2OnnxConfig"],
    "tokenization_deberta_v2": ["DebertaV2Tokenizer"],
}

try:
		if not is_tokenizers_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_SCREAMING_SNAKE_CASE       :   List[Any]        = ["DebertaV2TokenizerFast"]

try:
		if not is_tf_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_SCREAMING_SNAKE_CASE       :   int        = [
		    "TF_DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST",
		    "TFDebertaV2ForMaskedLM",
		    "TFDebertaV2ForQuestionAnswering",
		    "TFDebertaV2ForSequenceClassification",
		    "TFDebertaV2ForTokenClassification",
		    "TFDebertaV2Model",
		    "TFDebertaV2PreTrainedModel",
		]

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_SCREAMING_SNAKE_CASE       :   Union[str, Any]        = [
		    "DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST",
		    "DebertaV2ForMaskedLM",
		    "DebertaV2ForMultipleChoice",
		    "DebertaV2ForQuestionAnswering",
		    "DebertaV2ForSequenceClassification",
		    "DebertaV2ForTokenClassification",
		    "DebertaV2Model",
		    "DebertaV2PreTrainedModel",
		]


if TYPE_CHECKING:
		from .configuration_deberta_va import (
		    DEBERTA_V2_PRETRAINED_CONFIG_ARCHIVE_MAP,
		    DebertaVaConfig,
		    DebertaVaOnnxConfig,
		)
		from .tokenization_deberta_va import DebertaVaTokenizer

		try:
				if not is_tokenizers_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .tokenization_deberta_va_fast import DebertaVaTokenizerFast

		try:
				if not is_tf_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_tf_deberta_va import (
				    TF_DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST,
				    TFDebertaVaForMaskedLM,
				    TFDebertaVaForQuestionAnswering,
				    TFDebertaVaForSequenceClassification,
				    TFDebertaVaForTokenClassification,
				    TFDebertaVaModel,
				    TFDebertaVaPreTrainedModel,
				)

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_deberta_va import (
				    DEBERTA_V2_PRETRAINED_MODEL_ARCHIVE_LIST,
				    DebertaVaForMaskedLM,
				    DebertaVaForMultipleChoice,
				    DebertaVaForQuestionAnswering,
				    DebertaVaForSequenceClassification,
				    DebertaVaForTokenClassification,
				    DebertaVaModel,
				    DebertaVaPreTrainedModel,
				)

else:
		import sys

		_SCREAMING_SNAKE_CASE       :   Dict        = _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)