


'''simple docstring'''



import gc
import unittest

import numpy as np
import torch
import torch.nn.functional as F
from transformers import (
    ClapTextConfig,
    ClapTextModelWithProjection,
    RobertaTokenizer,
    SpeechTaHifiGan,
    SpeechTaHifiGanConfig,
)

from diffusers import (
    AudioLDMPipeline,
    AutoencoderKL,
    DDIMScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_AUDIO_BATCH_PARAMS, TEXT_TO_AUDIO_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





class 							__UpperCamelCase				(		lowerCAmelCase_		,      unittest.TestCase					):
	A_				=    AudioLDMPipeline
	A_				=    TEXT_TO_AUDIO_PARAMS
	A_				=    TEXT_TO_AUDIO_BATCH_PARAMS
	A_				=    frozenset(
	    [
	        "num_inference_steps",
	        "num_waveforms_per_prompt",
	        "generator",
	        "latents",
	        "output_type",
	        "return_dict",
	        "callback",
	        "callback_steps",
	    ]					)




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						torch.manual_seed(0							)
						__a			:       Dict					    =  UNetaDConditionModel(
						    block_out_channels=(32, 64)						,				layers_per_block=2						,				sample_size=32						,				in_channels=4						,				out_channels=4						,				down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')						,				up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')						,				cross_attention_dim=(32, 64)						,				class_embed_type='simple_projection'						,				projection_class_embeddings_input_dim=32						,				class_embeddings_concat=__a						,				)
						__a			:       Dict					    =  DDIMScheduler(
						    beta_start=0.00085						,				beta_end=0.012						,				beta_schedule='scaled_linear'						,				clip_sample=__a						,				set_alpha_to_one=__a						,				)
						torch.manual_seed(0							)
						__a			:       int					    =  AutoencoderKL(
						    block_out_channels=[32, 64]						,				in_channels=1						,				out_channels=1						,				down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']						,				up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']						,				latent_channels=4						,				)
						torch.manual_seed(0							)
						__a			:       Any					    =  ClapTextConfig(
						    bos_token_id=0						,				eos_token_id=2						,				hidden_size=32						,				intermediate_size=37						,				layer_norm_eps=1E-0_5						,				num_attention_heads=4						,				num_hidden_layers=5						,				pad_token_id=1						,				vocab_size=1000						,				projection_dim=32						,				)
						__a			:       List[str]					    =  ClapTextModelWithProjection(__a							)
						__a			:       Any					    =  RobertaTokenizer.from_pretrained('hf-internal-testing/tiny-random-roberta'						,				model_max_length=77							)

						__a			:       str					    =  SpeechTaHifiGanConfig(
						    model_in_dim=8						,				sampling_rate=1_6000						,				upsample_initial_channel=16						,				upsample_rates=[2, 2]						,				upsample_kernel_sizes=[4, 4]						,				resblock_kernel_sizes=[3, 7]						,				resblock_dilation_sizes=[[1, 3, 5], [1, 3, 5]]						,				normalize_before=__a						,				)

						__a			:       Optional[int]					    =  SpeechTaHifiGan(__a							)

						__a			:       Dict					    =  {
						    'unet': unet,
						    'scheduler': scheduler,
						    'vae': vae,
						    'text_encoder': text_encoder,
						    'tokenizer': tokenizer,
						    'vocoder': vocoder,
						}
						return components




	def       __UpperCAmelCase							(   self						,				__a						,				__a=0							):

						'''simple docstring'''





						if str(__a							).startswith('mps'							):
											__a			:       List[str]					    =  torch.manual_seed(__a							)
						else:
											__a			:       Union[str, Any]					    =  torch.Generator(device=__a							).manual_seed(__a							)
						__a			:       int					    =  {
						    'prompt': 'A hammer hitting a wooden surface',
						    'generator': generator,
						    'num_inference_steps': 2,
						    'guidance_scale': 6.0,
						}
						return inputs




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       List[str]					    =  'cpu'  # ensure determinism for the device-dependent torch.Generator

						__a			:       Tuple					    =  self.get_dummy_components()
						__a			:       List[str]					    =  AudioLDMPipeline(**__a							)
						__a			:       int					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       List[Any]					    =  self.get_dummy_inputs(__a							)
						__a			:       Any					    =  audioldm_pipe(**__a							)
						__a			:       Optional[Any]					    =  output.audios[0]

						assert audio.ndim == 1
						assert len(__a							) == 256

						__a			:       Dict					    =  audio[:10]
						__a			:       Union[str, Any]					    =  np.array(
						    [-0.0050, 0.0050, -0.0060, 0.0033, -0.0026, 0.0033, -0.0027, 0.0033, -0.0028, 0.0033]							)

						assert np.abs(audio_slice - expected_slice							).max() < 1E-2




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       str					    =  self.get_dummy_components()
						__a			:       Optional[Any]					    =  AudioLDMPipeline(**__a							)
						__a			:       Any					    =  audioldm_pipe.to(__a							)
						__a			:       List[Any]					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       Any					    =  self.get_dummy_inputs(__a							)
						__a			:       Optional[int]					    =  3 * [inputs['prompt']]

						# forward
						__a			:       List[Any]					    =  audioldm_pipe(**__a							)
						__a			:       str					    =  output.audios[0]

						__a			:       Any					    =  self.get_dummy_inputs(__a							)
						__a			:       Optional[Any]					    =  3 * [inputs.pop('prompt'							)]

						__a			:       int					    =  audioldm_pipe.tokenizer(
						    __a						,				padding='max_length'						,				max_length=audioldm_pipe.tokenizer.model_max_length						,				truncation=__a						,				return_tensors='pt'						,				)
						__a			:       str					    =  text_inputs['input_ids'].to(__a							)

						__a			:       Optional[int]					    =  audioldm_pipe.text_encoder(
						    __a						,				)
						__a			:       int					    =  prompt_embeds.text_embeds
						# additional L_2 normalization over each hidden-state
						__a			:       Union[str, Any]					    =  F.normalize(__a						,				dim=-1							)

						__a			:       Optional[int]					    =  prompt_embeds

						# forward
						__a			:       Union[str, Any]					    =  audioldm_pipe(**__a							)
						__a			:       Optional[int]					    =  output.audios[0]

						assert np.abs(audio_a - audio_a							).max() < 1E-2




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       Any					    =  self.get_dummy_components()
						__a			:       Optional[int]					    =  AudioLDMPipeline(**__a							)
						__a			:       Union[str, Any]					    =  audioldm_pipe.to(__a							)
						__a			:       Union[str, Any]					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       str					    =  self.get_dummy_inputs(__a							)
						__a			:       List[str]					    =  3 * ['this is a negative prompt']
						__a			:       int					    =  negative_prompt
						__a			:       List[Any]					    =  3 * [inputs['prompt']]

						# forward
						__a			:       Dict					    =  audioldm_pipe(**__a							)
						__a			:       str					    =  output.audios[0]

						__a			:       List[Any]					    =  self.get_dummy_inputs(__a							)
						__a			:       List[str]					    =  3 * [inputs.pop('prompt'							)]

						__a			:       int					    =  []
						for p in [prompt, negative_prompt]:
											__a			:       List[str]					    =  audioldm_pipe.tokenizer(
											    __a						,				padding='max_length'						,				max_length=audioldm_pipe.tokenizer.model_max_length						,				truncation=__a						,				return_tensors='pt'						,				)
											__a			:       Dict					    =  text_inputs['input_ids'].to(__a							)

											__a			:       List[Any]					    =  audioldm_pipe.text_encoder(
											    __a						,				)
											__a			:       Tuple					    =  text_embeds.text_embeds
											# additional L_2 normalization over each hidden-state
											__a			:       Optional[int]					    =  F.normalize(__a						,				dim=-1							)

											embeds.append(__a							)

						__a    ,				__a			:       Union[str, Any]					    =  embeds

						# forward
						__a			:       List[str]					    =  audioldm_pipe(**__a							)
						__a			:       Optional[Any]					    =  output.audios[0]

						assert np.abs(audio_a - audio_a							).max() < 1E-2




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       Dict					    =  'cpu'  # ensure determinism for the device-dependent torch.Generator
						__a			:       Tuple					    =  self.get_dummy_components()
						__a			:       Optional[int]					    =  PNDMScheduler(skip_prk_steps=__a							)
						__a			:       List[Any]					    =  AudioLDMPipeline(**__a							)
						__a			:       int					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       List[str]					    =  self.get_dummy_inputs(__a							)
						__a			:       str					    =  'egg cracking'
						__a			:       Union[str, Any]					    =  audioldm_pipe(**__a						,				negative_prompt=__a							)
						__a			:       List[str]					    =  output.audios[0]

						assert audio.ndim == 1
						assert len(__a							) == 256

						__a			:       Tuple					    =  audio[:10]
						__a			:       List[Any]					    =  np.array(
						    [-0.0051, 0.0050, -0.0060, 0.0034, -0.0026, 0.0033, -0.0027, 0.0033, -0.0028, 0.0032]							)

						assert np.abs(audio_slice - expected_slice							).max() < 1E-2




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       int					    =  'cpu'  # ensure determinism for the device-dependent torch.Generator
						__a			:       int					    =  self.get_dummy_components()
						__a			:       Dict					    =  PNDMScheduler(skip_prk_steps=__a							)
						__a			:       Any					    =  AudioLDMPipeline(**__a							)
						__a			:       Dict					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       List[str]					    =  'A hammer hitting a wooden surface'

						# test num_waveforms_per_prompt=1 (default)
						__a			:       str					    =  audioldm_pipe(__a						,				num_inference_steps=2							).audios

						assert audios.shape == (1, 256)

						# test num_waveforms_per_prompt=1 (default) for batch of prompts
						__a			:       Tuple					    =  2
						__a			:       Optional[Any]					    =  audioldm_pipe([prompt] * batch_size						,				num_inference_steps=2							).audios

						assert audios.shape == (batch_size, 256)

						# test num_waveforms_per_prompt for single prompt
						__a			:       Tuple					    =  2
						__a			:       Dict					    =  audioldm_pipe(__a						,				num_inference_steps=2						,				num_waveforms_per_prompt=__a							).audios

						assert audios.shape == (num_waveforms_per_prompt, 256)

						# test num_waveforms_per_prompt for batch of prompts
						__a			:       Optional[Any]					    =  2
						__a			:       List[str]					    =  audioldm_pipe(
						    [prompt] * batch_size						,				num_inference_steps=2						,				num_waveforms_per_prompt=__a							).audios

						assert audios.shape == (batch_size * num_waveforms_per_prompt, 256)




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       List[Any]					    =  'cpu'  # ensure determinism for the device-dependent torch.Generator
						__a			:       int					    =  self.get_dummy_components()
						__a			:       Optional[Any]					    =  AudioLDMPipeline(**__a							)
						__a			:       Any					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)
						__a			:       Dict					    =  audioldm_pipe.vocoder.config.sampling_rate

						__a			:       Union[str, Any]					    =  self.get_dummy_inputs(__a							)
						__a			:       str					    =  audioldm_pipe(audio_length_in_s=0.016						,				**__a							)
						__a			:       Optional[Any]					    =  output.audios[0]

						assert audio.ndim == 1
						assert len(__a							) / vocoder_sampling_rate == 0.016

						__a			:       Tuple					    =  audioldm_pipe(audio_length_in_s=0.032						,				**__a							)
						__a			:       Optional[Any]					    =  output.audios[0]

						assert audio.ndim == 1
						assert len(__a							) / vocoder_sampling_rate == 0.032




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       Optional[Any]					    =  self.get_dummy_components()
						__a			:       int					    =  AudioLDMPipeline(**__a							)
						__a			:       Optional[Any]					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       Optional[Any]					    =  ['hey']

						__a			:       Optional[Any]					    =  audioldm_pipe(__a						,				num_inference_steps=1							)
						__a			:       int					    =  output.audios.shape
						assert audio_shape == (1, 256)

						__a			:       str					    =  audioldm_pipe.vocoder.config
						config.model_in_dim *= 2
						__a			:       Any					    =  SpeechTaHifiGan(__a							).to(__a							)
						__a			:       str					    =  audioldm_pipe(__a						,				num_inference_steps=1							)
						__a			:       int					    =  output.audios.shape
						# waveform shape is unchanged, we just have 2x the number of mel channels in the spectrogram
						assert audio_shape == (1, 256)




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__a							)




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						self._test_inference_batch_single_identical(test_mean_pixel_difference=__a							)





	@unittest.skipIf(
	    torch_device != 'cuda' or not is_xformers_available()						,				reason='XFormers attention is only available with CUDA and `xformers` installed'						,				)
	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__a							)





@slow
class 							__UpperCamelCase				(		unittest.TestCase					):




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





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




	def       __UpperCAmelCase							(   self						,				__a						,				__a="cpu"						,				__a=torch.floataa						,				__a=0							):

						'''simple docstring'''





						__a			:       Tuple					    =  torch.Generator(device=__a							).manual_seed(__a							)
						__a			:       List[str]					    =  np.random.RandomState(__a							).standard_normal((1, 8, 128, 16)							)
						__a			:       Tuple					    =  torch.from_numpy(__a							).to(device=__a						,				dtype=__a							)
						__a			:       Optional[Any]					    =  {
						    'prompt': 'A hammer hitting a wooden surface',
						    'latents': latents,
						    'generator': generator,
						    'num_inference_steps': 3,
						    'guidance_scale': 2.5,
						}
						return inputs




	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       List[str]					    =  AudioLDMPipeline.from_pretrained('cvssp/audioldm'							)
						__a			:       int					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       Optional[Any]					    =  self.get_inputs(__a							)
						__a			:       Optional[Any]					    =  25
						__a			:       List[Any]					    =  audioldm_pipe(**__a							).audios[0]

						assert audio.ndim == 1
						assert len(__a							) == 8_1920

						__a			:       List[str]					    =  audio[7_7230:7_7240]
						__a			:       str					    =  np.array(
						    [-0.4884, -0.4607, 0.0023, 0.5007, 0.5896, 0.5151, 0.3813, -0.0208, -0.3687, -0.4315]							)
						__a			:       Dict					    =  np.abs(expected_slice - audio_slice							).max()
						assert max_diff < 1E-2





	def       __UpperCAmelCase							(   self							):

						'''simple docstring'''





						__a			:       Any					    =  AudioLDMPipeline.from_pretrained('cvssp/audioldm'							)
						__a			:       Optional[int]					    =  LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config							)
						__a			:       Optional[Any]					    =  audioldm_pipe.to(__a							)
						audioldm_pipe.set_progress_bar_config(disable=__a							)

						__a			:       Dict					    =  self.get_inputs(__a							)
						__a			:       Tuple					    =  audioldm_pipe(**__a							).audios[0]

						assert audio.ndim == 1
						assert len(__a							) == 8_1920

						__a			:       int					    =  audio[2_7780:2_7790]
						__a			:       str					    =  np.array([-0.2131, -0.0873, -0.0124, -0.0189, 0.0569, 0.1373, 0.1883, 0.2886, 0.3297, 0.2212]							)
						__a			:       Optional[Any]					    =  np.abs(expected_slice - audio_slice							).max()
						assert max_diff < 3E-2







'''simple docstring'''



from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


__lowercase	:						str       =	{
    'configuration_data2vec_audio': ['DATA2VEC_AUDIO_PRETRAINED_CONFIG_ARCHIVE_MAP', 'Data2VecAudioConfig'],
    'configuration_data2vec_text': [
        'DATA2VEC_TEXT_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'Data2VecTextConfig',
        'Data2VecTextOnnxConfig',
    ],
    'configuration_data2vec_vision': [
        'DATA2VEC_VISION_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'Data2VecVisionConfig',
        'Data2VecVisionOnnxConfig',
    ],
}

try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	__lowercase	:						str       =	[
	    'DATA2VEC_AUDIO_PRETRAINED_MODEL_ARCHIVE_LIST',
	    'Data2VecAudioForAudioFrameClassification',
	    'Data2VecAudioForCTC',
	    'Data2VecAudioForSequenceClassification',
	    'Data2VecAudioForXVector',
	    'Data2VecAudioModel',
	    'Data2VecAudioPreTrainedModel',
	]
	__lowercase	:						Tuple       =	[
	    'DATA2VEC_TEXT_PRETRAINED_MODEL_ARCHIVE_LIST',
	    'Data2VecTextForCausalLM',
	    'Data2VecTextForMaskedLM',
	    'Data2VecTextForMultipleChoice',
	    'Data2VecTextForQuestionAnswering',
	    'Data2VecTextForSequenceClassification',
	    'Data2VecTextForTokenClassification',
	    'Data2VecTextModel',
	    'Data2VecTextPreTrainedModel',
	]
	__lowercase	:						Dict       =	[
	    'DATA2VEC_VISION_PRETRAINED_MODEL_ARCHIVE_LIST',
	    'Data2VecVisionForImageClassification',
	    'Data2VecVisionForMaskedImageModeling',
	    'Data2VecVisionForSemanticSegmentation',
	    'Data2VecVisionModel',
	    'Data2VecVisionPreTrainedModel',
	]

if is_tf_available():
	__lowercase	:						Optional[Any]       =	[
	    'TFData2VecVisionForImageClassification',
	    'TFData2VecVisionForSemanticSegmentation',
	    'TFData2VecVisionModel',
	    'TFData2VecVisionPreTrainedModel',
	]

if TYPE_CHECKING:
	from .configuration_dataavec_audio import DATA2VEC_AUDIO_PRETRAINED_CONFIG_ARCHIVE_MAP, DataaVecAudioConfig
	from .configuration_dataavec_text import (
	    DATA2VEC_TEXT_PRETRAINED_CONFIG_ARCHIVE_MAP,
	    DataaVecTextConfig,
	    DataaVecTextOnnxConfig,
	)
	from .configuration_dataavec_vision import (
	    DATA2VEC_VISION_PRETRAINED_CONFIG_ARCHIVE_MAP,
	    DataaVecVisionConfig,
	    DataaVecVisionOnnxConfig,
	)

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_dataavec_audio import (
		    DATA2VEC_AUDIO_PRETRAINED_MODEL_ARCHIVE_LIST,
		    DataaVecAudioForAudioFrameClassification,
		    DataaVecAudioForCTC,
		    DataaVecAudioForSequenceClassification,
		    DataaVecAudioForXVector,
		    DataaVecAudioModel,
		    DataaVecAudioPreTrainedModel,
		)
		from .modeling_dataavec_text import (
		    DATA2VEC_TEXT_PRETRAINED_MODEL_ARCHIVE_LIST,
		    DataaVecTextForCausalLM,
		    DataaVecTextForMaskedLM,
		    DataaVecTextForMultipleChoice,
		    DataaVecTextForQuestionAnswering,
		    DataaVecTextForSequenceClassification,
		    DataaVecTextForTokenClassification,
		    DataaVecTextModel,
		    DataaVecTextPreTrainedModel,
		)
		from .modeling_dataavec_vision import (
		    DATA2VEC_VISION_PRETRAINED_MODEL_ARCHIVE_LIST,
		    DataaVecVisionForImageClassification,
		    DataaVecVisionForMaskedImageModeling,
		    DataaVecVisionForSemanticSegmentation,
		    DataaVecVisionModel,
		    DataaVecVisionPreTrainedModel,
		)
	if is_tf_available():
		from .modeling_tf_dataavec_vision import (
		    TFDataaVecVisionForImageClassification,
		    TFDataaVecVisionForSemanticSegmentation,
		    TFDataaVecVisionModel,
		    TFDataaVecVisionPreTrainedModel,
		)

else:
	import sys

	__lowercase	:						int       =	_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)



