
"""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 							__magic_name__     (		lowercase__      ,		unittest.TestCase   ):
					_SCREAMING_SNAKE_CASE       :					Any     							=   AudioLDMPipeline
					_SCREAMING_SNAKE_CASE       :					int     							=   TEXT_TO_AUDIO_PARAMS
					_SCREAMING_SNAKE_CASE       :					List[str]     							=   TEXT_TO_AUDIO_BATCH_PARAMS
					_SCREAMING_SNAKE_CASE       :					int     							=   frozenset(
					    [
					        'num_inference_steps',
					        'num_waveforms_per_prompt',
					        'generator',
					        'latents',
					        'output_type',
					        'return_dict',
					        'callback',
					        'callback_steps',
					    ]   )





					def 							lowerCAmelCase      (				self   :							Optional[int]    ):
												torch.manual_seed(0    )
												__snake_case     			=       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=snake_case_							,						)
												__snake_case     			=       DDIMScheduler(
												    beta_start=0.00085							,						beta_end=0.012							,						beta_schedule="scaled_linear"							,						clip_sample=snake_case_							,						set_alpha_to_one=snake_case_							,						)
												torch.manual_seed(0    )
												__snake_case     			=       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    )
												__snake_case     			=       ClapTextConfig(
												    bos_token_id=0							,						eos_token_id=2							,						hidden_size=32							,						intermediate_size=37							,						layer_norm_eps=1e-05							,						num_attention_heads=4							,						num_hidden_layers=5							,						pad_token_id=1							,						vocab_size=1000							,						projection_dim=32							,						)
												__snake_case     			=       ClapTextModelWithProjection(snake_case_    )
												__snake_case     			=       RobertaTokenizer.from_pretrained("hf-internal-testing/tiny-random-roberta"							,						model_max_length=77    )

												__snake_case     			=       SpeechTaHifiGanConfig(
												    model_in_dim=8							,						sampling_rate=16000							,						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=snake_case_							,						)

												__snake_case     			=       SpeechTaHifiGan(snake_case_    )

												__snake_case     			=       {
												    "unet": unet,
												    "scheduler": scheduler,
												    "vae": vae,
												    "text_encoder": text_encoder,
												    "tokenizer": tokenizer,
												    "vocoder": vocoder,
												}
												return components





					def 							lowerCAmelCase      (				self   :							Optional[Any]							,						snake_case_   :							Optional[Any]							,						snake_case_   :							int=0    ):
												if str(snake_case_    ).startswith("mps"    ):
																			__snake_case     			=       torch.manual_seed(snake_case_    )
												else:
																			__snake_case     			=       torch.Generator(device=snake_case_    ).manual_seed(snake_case_    )
												__snake_case     			=       {
												    "prompt": "A hammer hitting a wooden surface",
												    "generator": generator,
												    "num_inference_steps": 2,
												    "guidance_scale": 6.0,
												}
												return inputs





					def 							lowerCAmelCase      (				self   :							Any    ):
												__snake_case     			=       "cpu"  # ensure determinism for the device-dependent torch.Generator

												__snake_case     			=       self.get_dummy_components()
												__snake_case     			=       AudioLDMPipeline(**snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       self.get_dummy_inputs(snake_case_    )
												__snake_case     			=       audioldm_pipe(**snake_case_    )
												__snake_case     			=       output.audios[0]

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

												__snake_case     			=       audio[:10]
												__snake_case     			=       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 							lowerCAmelCase      (				self   :							Dict    ):
												__snake_case     			=       self.get_dummy_components()
												__snake_case     			=       AudioLDMPipeline(**snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       self.get_dummy_inputs(snake_case_    )
												__snake_case     			=       3 * [inputs["prompt"]]

												# forward
												__snake_case     			=       audioldm_pipe(**snake_case_    )
												__snake_case     			=       output.audios[0]

												__snake_case     			=       self.get_dummy_inputs(snake_case_    )
												__snake_case     			=       3 * [inputs.pop("prompt"    )]

												__snake_case     			=       audioldm_pipe.tokenizer(
												    snake_case_							,						padding="max_length"							,						max_length=audioldm_pipe.tokenizer.model_max_length							,						truncation=snake_case_							,						return_tensors="pt"							,						)
												__snake_case     			=       text_inputs["input_ids"].to(snake_case_    )

												__snake_case     			=       audioldm_pipe.text_encoder(
												    snake_case_							,						)
												__snake_case     			=       prompt_embeds.text_embeds
												# additional L_2 normalization over each hidden-state
												__snake_case     			=       F.normalize(snake_case_							,						dim=-1    )

												__snake_case     			=       prompt_embeds

												# forward
												__snake_case     			=       audioldm_pipe(**snake_case_    )
												__snake_case     			=       output.audios[0]

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





					def 							lowerCAmelCase      (				self   :							Optional[Any]    ):
												__snake_case     			=       self.get_dummy_components()
												__snake_case     			=       AudioLDMPipeline(**snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       self.get_dummy_inputs(snake_case_    )
												__snake_case     			=       3 * ["this is a negative prompt"]
												__snake_case     			=       negative_prompt
												__snake_case     			=       3 * [inputs["prompt"]]

												# forward
												__snake_case     			=       audioldm_pipe(**snake_case_    )
												__snake_case     			=       output.audios[0]

												__snake_case     			=       self.get_dummy_inputs(snake_case_    )
												__snake_case     			=       3 * [inputs.pop("prompt"    )]

												__snake_case     			=       []
												for p in [prompt, negative_prompt]:
																			__snake_case     			=       audioldm_pipe.tokenizer(
																			    snake_case_							,						padding="max_length"							,						max_length=audioldm_pipe.tokenizer.model_max_length							,						truncation=snake_case_							,						return_tensors="pt"							,						)
																			__snake_case     			=       text_inputs["input_ids"].to(snake_case_    )

																			__snake_case     			=       audioldm_pipe.text_encoder(
																			    snake_case_							,						)
																			__snake_case     			=       text_embeds.text_embeds
																			# additional L_2 normalization over each hidden-state
																			__snake_case     			=       F.normalize(snake_case_							,						dim=-1    )

																			embeds.append(snake_case_    )

												__snake_case			,   __snake_case     			=       embeds

												# forward
												__snake_case     			=       audioldm_pipe(**snake_case_    )
												__snake_case     			=       output.audios[0]

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





					def 							lowerCAmelCase      (				self   :							List[Any]    ):
												__snake_case     			=       "cpu"  # ensure determinism for the device-dependent torch.Generator
												__snake_case     			=       self.get_dummy_components()
												__snake_case     			=       PNDMScheduler(skip_prk_steps=snake_case_    )
												__snake_case     			=       AudioLDMPipeline(**snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       self.get_dummy_inputs(snake_case_    )
												__snake_case     			=       "egg cracking"
												__snake_case     			=       audioldm_pipe(**snake_case_							,						negative_prompt=snake_case_    )
												__snake_case     			=       output.audios[0]

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

												__snake_case     			=       audio[:10]
												__snake_case     			=       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 							lowerCAmelCase      (				self   :							List[str]    ):
												__snake_case     			=       "cpu"  # ensure determinism for the device-dependent torch.Generator
												__snake_case     			=       self.get_dummy_components()
												__snake_case     			=       PNDMScheduler(skip_prk_steps=snake_case_    )
												__snake_case     			=       AudioLDMPipeline(**snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       "A hammer hitting a wooden surface"

												# test num_waveforms_per_prompt=1 (default)
												__snake_case     			=       audioldm_pipe(snake_case_							,						num_inference_steps=2    ).audios

												assert audios.shape == (1, 256)

												# test num_waveforms_per_prompt=1 (default) for batch of prompts
												__snake_case     			=       2
												__snake_case     			=       audioldm_pipe([prompt] * batch_size							,						num_inference_steps=2    ).audios

												assert audios.shape == (batch_size, 256)

												# test num_waveforms_per_prompt for single prompt
												__snake_case     			=       2
												__snake_case     			=       audioldm_pipe(snake_case_							,						num_inference_steps=2							,						num_waveforms_per_prompt=snake_case_    ).audios

												assert audios.shape == (num_waveforms_per_prompt, 256)

												# test num_waveforms_per_prompt for batch of prompts
												__snake_case     			=       2
												__snake_case     			=       audioldm_pipe(
												    [prompt] * batch_size							,						num_inference_steps=2							,						num_waveforms_per_prompt=snake_case_    ).audios

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





					def 							lowerCAmelCase      (				self   :							Any    ):
												__snake_case     			=       "cpu"  # ensure determinism for the device-dependent torch.Generator
												__snake_case     			=       self.get_dummy_components()
												__snake_case     			=       AudioLDMPipeline(**snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )
												__snake_case     			=       audioldm_pipe.vocoder.config.sampling_rate

												__snake_case     			=       self.get_dummy_inputs(snake_case_    )
												__snake_case     			=       audioldm_pipe(audio_length_in_s=0.016							,						**snake_case_    )
												__snake_case     			=       output.audios[0]

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

												__snake_case     			=       audioldm_pipe(audio_length_in_s=0.032							,						**snake_case_    )
												__snake_case     			=       output.audios[0]

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





					def 							lowerCAmelCase      (				self   :							Optional[Any]    ):
												__snake_case     			=       self.get_dummy_components()
												__snake_case     			=       AudioLDMPipeline(**snake_case_    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       ["hey"]

												__snake_case     			=       audioldm_pipe(snake_case_							,						num_inference_steps=1    )
												__snake_case     			=       output.audios.shape
												assert audio_shape == (1, 256)

												__snake_case     			=       audioldm_pipe.vocoder.config
												config.model_in_dim *= 2
												__snake_case     			=       SpeechTaHifiGan(snake_case_    ).to(snake_case_    )
												__snake_case     			=       audioldm_pipe(snake_case_							,						num_inference_steps=1    )
												__snake_case     			=       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 							lowerCAmelCase      (				self   :							int    ):
												self._test_attention_slicing_forward_pass(test_mean_pixel_difference=snake_case_    )





					def 							lowerCAmelCase      (				self   :							Tuple    ):
												self._test_inference_batch_single_identical(test_mean_pixel_difference=snake_case_    )





					@unittest.skipIf(
					    torch_device != "cuda" or not is_xformers_available()							,						reason="XFormers attention is only available with CUDA and `xformers` installed"							,						)
					def 							lowerCAmelCase      (				self   :							Any    ):
												self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case_    )







@slow
class 							__magic_name__     (		unittest.TestCase   ):





					def 							lowerCAmelCase      (				self   :							Tuple    ):
												super().tearDown()
												gc.collect()
												torch.cuda.empty_cache()





					def 							lowerCAmelCase      (				self   :							Optional[int]							,						snake_case_   :							List[str]							,						snake_case_   :							Optional[Any]="cpu"							,						snake_case_   :							Tuple=torch.floataa							,						snake_case_   :							Any=0    ):
												__snake_case     			=       torch.Generator(device=snake_case_    ).manual_seed(snake_case_    )
												__snake_case     			=       np.random.RandomState(snake_case_    ).standard_normal((1, 8, 128, 16)    )
												__snake_case     			=       torch.from_numpy(snake_case_    ).to(device=snake_case_							,						dtype=snake_case_    )
												__snake_case     			=       {
												    "prompt": "A hammer hitting a wooden surface",
												    "latents": latents,
												    "generator": generator,
												    "num_inference_steps": 3,
												    "guidance_scale": 2.5,
												}
												return inputs





					def 							lowerCAmelCase      (				self   :							int    ):
												__snake_case     			=       AudioLDMPipeline.from_pretrained("cvssp/audioldm"    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       self.get_inputs(snake_case_    )
												__snake_case     			=       25
												__snake_case     			=       audioldm_pipe(**snake_case_    ).audios[0]

												assert audio.ndim == 1
												assert len(snake_case_    ) == 81920

												__snake_case     			=       audio[77230:77240]
												__snake_case     			=       np.array(
												    [-0.4884, -0.4607, 0.0023, 0.5007, 0.5896, 0.5151, 0.3813, -0.0208, -0.3687, -0.4315]    )
												__snake_case     			=       np.abs(expected_slice - audio_slice    ).max()
												assert max_diff < 1e-2





					def 							lowerCAmelCase      (				self   :							Tuple    ):
												__snake_case     			=       AudioLDMPipeline.from_pretrained("cvssp/audioldm"    )
												__snake_case     			=       LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config    )
												__snake_case     			=       audioldm_pipe.to(snake_case_    )
												audioldm_pipe.set_progress_bar_config(disable=snake_case_    )

												__snake_case     			=       self.get_inputs(snake_case_    )
												__snake_case     			=       audioldm_pipe(**snake_case_    ).audios[0]

												assert audio.ndim == 1
												assert len(snake_case_    ) == 81920

												__snake_case     			=       audio[27780:27790]
												__snake_case     			=       np.array([-0.2131, -0.0873, -0.0124, -0.0189, 0.0569, 0.1373, 0.1883, 0.2886, 0.3297, 0.2212]    )
												__snake_case     			=       np.abs(expected_slice - audio_slice    ).max()
												assert max_diff < 3e-2


"""simple docstring"""







import math

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_SCREAMING_SNAKE_CASE	       =      logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE	       =      {
    """facebook/data2vec-base-960h""": """https://huggingface.co/facebook/data2vec-audio-base-960h/resolve/main/config.json""",
    # See all Data2VecAudio models at https://huggingface.co/models?filter=data2vec-audio
}







class 							__magic_name__     (		lowercase__   ):
					_SCREAMING_SNAKE_CASE       :					Any     							=   'data2vec-audio'





					def __init__(				self   :							Any							,						snake_case_   :							Optional[Any]=32							,						snake_case_   :							Tuple=768							,						snake_case_   :							Optional[int]=12							,						snake_case_   :							str=12							,						snake_case_   :							int=3072							,						snake_case_   :							Tuple="gelu"							,						snake_case_   :							Optional[Any]=0.1							,						snake_case_   :							str=0.1							,						snake_case_   :							int=0.1							,						snake_case_   :							int=0.0							,						snake_case_   :							Optional[int]=0.1							,						snake_case_   :							Dict=0.1							,						snake_case_   :							Dict=0.02							,						snake_case_   :							Optional[int]=1e-5							,						snake_case_   :							Optional[Any]="gelu"							,						snake_case_   :							Tuple=(512, 512, 512, 512, 512, 512, 512)							,						snake_case_   :							int=(5, 2, 2, 2, 2, 2, 2)							,						snake_case_   :							Union[str, Any]=(10, 3, 3, 3, 3, 2, 2)							,						snake_case_   :							Optional[int]=False							,						snake_case_   :							int=16							,						snake_case_   :							List[Any]=19							,						snake_case_   :							str=5							,						snake_case_   :							List[Any]=0.05							,						snake_case_   :							Dict=10							,						snake_case_   :							Dict=2							,						snake_case_   :							Dict=0.0							,						snake_case_   :							Union[str, Any]=10							,						snake_case_   :							Optional[Any]=0							,						snake_case_   :							Optional[Any]="sum"							,						snake_case_   :							List[str]=False							,						snake_case_   :							List[Any]=False							,						snake_case_   :							Optional[int]=256							,						snake_case_   :							int=(512, 512, 512, 512, 1500)							,						snake_case_   :							List[Any]=(5, 3, 3, 1, 1)							,						snake_case_   :							Any=(1, 2, 3, 1, 1)							,						snake_case_   :							List[Any]=512							,						snake_case_   :							Optional[int]=0							,						snake_case_   :							Dict=1							,						snake_case_   :							List[str]=2							,						snake_case_   :							Optional[Any]=False							,						snake_case_   :							Optional[Any]=3							,						snake_case_   :							Tuple=2							,						snake_case_   :							Any=3							,						snake_case_   :							Union[str, Any]=None							,						**snake_case_   :							int							,						):
												super().__init__(**snake_case_							,						pad_token_id=snake_case_							,						bos_token_id=snake_case_							,						eos_token_id=snake_case_    )
												__snake_case     			=       hidden_size
												__snake_case     			=       feat_extract_activation
												__snake_case     			=       list(snake_case_    )
												__snake_case     			=       list(snake_case_    )
												__snake_case     			=       list(snake_case_    )
												__snake_case     			=       conv_bias
												__snake_case     			=       num_conv_pos_embeddings
												__snake_case     			=       num_conv_pos_embedding_groups
												__snake_case     			=       conv_pos_kernel_size
												__snake_case     			=       len(self.conv_dim    )
												__snake_case     			=       num_hidden_layers
												__snake_case     			=       intermediate_size
												__snake_case     			=       hidden_act
												__snake_case     			=       num_attention_heads
												__snake_case     			=       hidden_dropout
												__snake_case     			=       attention_dropout
												__snake_case     			=       activation_dropout
												__snake_case     			=       feat_proj_dropout
												__snake_case     			=       final_dropout
												__snake_case     			=       layerdrop
												__snake_case     			=       layer_norm_eps
												__snake_case     			=       initializer_range
												__snake_case     			=       vocab_size
												__snake_case     			=       use_weighted_layer_sum

												if (
												    (len(self.conv_stride    ) != self.num_feat_extract_layers)
												    or (len(self.conv_kernel    ) != self.num_feat_extract_layers)
												    or (len(self.conv_dim    ) != self.num_feat_extract_layers)
												):
																			raise ValueError(
																			    "Configuration for convolutional layers is incorrect. It is required that `len(config.conv_dim)` =="
																			    " `len(config.conv_stride)` == `len(config.conv_kernel)`, but is `len(config.conv_dim) ="
																			    F''' {len(self.conv_dim    )}`, `len(config.conv_stride) = {len(self.conv_stride    )}`,'''
																			    F''' `len(config.conv_kernel) = {len(self.conv_kernel    )}`.'''    )

												# fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
												__snake_case     			=       mask_time_prob
												__snake_case     			=       mask_time_length
												__snake_case     			=       mask_time_min_masks
												__snake_case     			=       mask_feature_prob
												__snake_case     			=       mask_feature_length
												__snake_case     			=       mask_feature_min_masks

												# ctc loss
												__snake_case     			=       ctc_loss_reduction
												__snake_case     			=       ctc_zero_infinity

												# adapter
												__snake_case     			=       add_adapter
												__snake_case     			=       adapter_kernel_size
												__snake_case     			=       adapter_stride
												__snake_case     			=       num_adapter_layers
												__snake_case     			=       output_hidden_size or hidden_size

												# SequenceClassification-specific parameter. Feel free to ignore for other classes.
												__snake_case     			=       classifier_proj_size

												# XVector-specific parameters. Feel free to ignore for other classes.
												__snake_case     			=       list(snake_case_    )
												__snake_case     			=       list(snake_case_    )
												__snake_case     			=       list(snake_case_    )
												__snake_case     			=       xvector_output_dim





					@property
					def 							lowerCAmelCase      (				self   :							List[str]    ):
												return math.prod(self.conv_stride    )
