



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 					_A (			__UpperCAmelCase   ,unittest.TestCase					):
       UpperCamelCase__						:					Any		      =		AudioLDMPipeline
       UpperCamelCase__						:					Tuple		      =		TEXT_TO_AUDIO_PARAMS
       UpperCamelCase__						:					str		      =		TEXT_TO_AUDIO_BATCH_PARAMS
       UpperCamelCase__						:					Any		      =		frozenset(
           [
               '''num_inference_steps''',
               '''num_waveforms_per_prompt''',
               '''generator''',
               '''latents''',
               '''output_type''',
               '''return_dict''',
               '''callback''',
               '''callback_steps''',
           ]					)



       def 	_lowerCamelCase    (	self							:  Any):


           '''simple docstring'''

           torch.manual_seed(0)
           __a		 =       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=__SCREAMING_SNAKE_CASE  ,	)
           __a		 =       DDIMScheduler(
               beta_start=0.0_00_85  ,	beta_end=0.0_12  ,	beta_schedule='''scaled_linear'''  ,	clip_sample=__SCREAMING_SNAKE_CASE  ,	set_alpha_to_one=__SCREAMING_SNAKE_CASE  ,	)
           torch.manual_seed(0)
           __a		 =       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		 =       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=1_000  ,	projection_dim=32  ,	)
           __a		 =       ClapTextModelWithProjection(__SCREAMING_SNAKE_CASE)
           __a		 =       RobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-random-roberta'''  ,	model_max_length=77)

           __a		 =       SpeechTaHifiGanConfig(
               model_in_dim=8  ,	sampling_rate=16_000  ,	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=__SCREAMING_SNAKE_CASE  ,	)

           __a		 =       SpeechTaHifiGan(__SCREAMING_SNAKE_CASE)

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



       def 	_lowerCamelCase    (	self							:  Tuple  ,	__SCREAMING_SNAKE_CASE							:  Optional[Any]  ,	__SCREAMING_SNAKE_CASE							:  str=0):


           '''simple docstring'''

           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		 =       {
               '''prompt''': '''A hammer hitting a wooden surface''',
               '''generator''': generator,
               '''num_inference_steps''': 2,
               '''guidance_scale''': 6.0,
           }
           return inputs



       def 	_lowerCamelCase    (	self							:  Optional[int]):


           '''simple docstring'''

           __a		 =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator

           __a		 =       self.get_dummy_components()
           __a		 =       AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

           __a		 =       self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

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

           __a		 =       audio[:10]
           __a		 =       np.array(
               [-0.00_50, 0.00_50, -0.00_60, 0.00_33, -0.00_26, 0.00_33, -0.00_27, 0.00_33, -0.00_28, 0.00_33])

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



       def 	_lowerCamelCase    (	self							:  Union[str, Any]):


           '''simple docstring'''

           __a		 =       self.get_dummy_components()
           __a		 =       AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

           __a		 =       self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       3 * [inputs['''prompt''']]

           # forward
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

           __a		 =       self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       3 * [inputs.pop('''prompt''')]

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

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

           __a		 =       prompt_embeds

           # forward
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

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



       def 	_lowerCamelCase    (	self							:  Optional[int]):


           '''simple docstring'''

           __a		 =       self.get_dummy_components()
           __a		 =       AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

           __a		 =       self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       3 * ['''this is a negative prompt''']
           __a		 =       negative_prompt
           __a		 =       3 * [inputs['''prompt''']]

           # forward
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

           __a		 =       self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       3 * [inputs.pop('''prompt''')]

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

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

               embeds.append(__SCREAMING_SNAKE_CASE)

           __a ,     __a		 =       embeds

           # forward
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

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



       def 	_lowerCamelCase    (	self							:  List[str]):


           '''simple docstring'''

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

           __a		 =       self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       '''egg cracking'''
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE  ,	negative_prompt=__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

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

           __a		 =       audio[:10]
           __a		 =       np.array(
               [-0.00_51, 0.00_50, -0.00_60, 0.00_34, -0.00_26, 0.00_33, -0.00_27, 0.00_33, -0.00_28, 0.00_32])

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



       def 	_lowerCamelCase    (	self							:  List[str]):


           '''simple docstring'''

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

           __a		 =       '''A hammer hitting a wooden surface'''

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

           assert audios.shape == (1, 256)

           # test num_waveforms_per_prompt=1 (default) for batch of prompts
           __a		 =       2
           __a		 =       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		 =       2
           __a		 =       audioldm_pipe(__SCREAMING_SNAKE_CASE  ,	num_inference_steps=2  ,	num_waveforms_per_prompt=__SCREAMING_SNAKE_CASE).audios

           assert audios.shape == (num_waveforms_per_prompt, 256)

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

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



       def 	_lowerCamelCase    (	self							:  Tuple):


           '''simple docstring'''

           __a		 =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
           __a		 =       self.get_dummy_components()
           __a		 =       AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.vocoder.config.sampling_rate

           __a		 =       self.get_dummy_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe(audio_length_in_s=0.0_16  ,	**__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

           assert audio.ndim == 1
           assert len(__SCREAMING_SNAKE_CASE) / vocoder_sampling_rate == 0.0_16

           __a		 =       audioldm_pipe(audio_length_in_s=0.0_32  ,	**__SCREAMING_SNAKE_CASE)
           __a		 =       output.audios[0]

           assert audio.ndim == 1
           assert len(__SCREAMING_SNAKE_CASE) / vocoder_sampling_rate == 0.0_32



       def 	_lowerCamelCase    (	self							:  Union[str, Any]):


           '''simple docstring'''

           __a		 =       self.get_dummy_components()
           __a		 =       AudioLDMPipeline(**__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

           __a		 =       ['''hey''']

           __a		 =       audioldm_pipe(__SCREAMING_SNAKE_CASE  ,	num_inference_steps=1)
           __a		 =       output.audios.shape
           assert audio_shape == (1, 256)

           __a		 =       audioldm_pipe.vocoder.config
           config.model_in_dim *= 2
           __a		 =       SpeechTaHifiGan(__SCREAMING_SNAKE_CASE).to(__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe(__SCREAMING_SNAKE_CASE  ,	num_inference_steps=1)
           __a		 =       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							:  Optional[Any]):


           '''simple docstring'''

           self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE)



       def 	_lowerCamelCase    (	self							:  str):


           '''simple docstring'''

           self._test_inference_batch_single_identical(test_mean_pixel_difference=__SCREAMING_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):


           '''simple docstring'''

           self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE)


@slow
class 					_A (			unittest.TestCase					):



       def 	_lowerCamelCase    (	self							:  List[str]):


           '''simple docstring'''

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



       def 	_lowerCamelCase    (	self							:  List[Any]  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]  ,	__SCREAMING_SNAKE_CASE							:  Optional[int]="cpu"  ,	__SCREAMING_SNAKE_CASE							:  List[Any]=torch.floataa  ,	__SCREAMING_SNAKE_CASE							:  Optional[Any]=0):


           '''simple docstring'''

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



       def 	_lowerCamelCase    (	self							:  Optional[Any]):


           '''simple docstring'''

           __a		 =       AudioLDMPipeline.from_pretrained('''cvssp/audioldm''')
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

           __a		 =       self.get_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       25
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE).audios[0]

           assert audio.ndim == 1
           assert len(__SCREAMING_SNAKE_CASE) == 81_920

           __a		 =       audio[77_230:77_240]
           __a		 =       np.array(
               [-0.48_84, -0.46_07, 0.00_23, 0.50_07, 0.58_96, 0.51_51, 0.38_13, -0.02_08, -0.36_87, -0.43_15])
           __a		 =       np.abs(expected_slice - audio_slice).max()
           assert max_diff < 1E-2





       def 	_lowerCamelCase    (	self							:  Union[str, Any]):


           '''simple docstring'''

           __a		 =       AudioLDMPipeline.from_pretrained('''cvssp/audioldm''')
           __a		 =       LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config)
           __a		 =       audioldm_pipe.to(__SCREAMING_SNAKE_CASE)
           audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE)

           __a		 =       self.get_inputs(__SCREAMING_SNAKE_CASE)
           __a		 =       audioldm_pipe(**__SCREAMING_SNAKE_CASE).audios[0]

           assert audio.ndim == 1
           assert len(__SCREAMING_SNAKE_CASE) == 81_920

           __a		 =       audio[27_780:27_790]
           __a		 =       np.array([-0.21_31, -0.08_73, -0.01_24, -0.01_89, 0.05_69, 0.13_73, 0.18_83, 0.28_86, 0.32_97, 0.22_12])
           __a		 =       np.abs(expected_slice - audio_slice).max()
           assert max_diff < 3E-2










def   __snake_case			(    _UpperCAmelCase     ,				_UpperCAmelCase = False       ):

    if not isinstance(_UpperCAmelCase     ,				_UpperCAmelCase       ):
        __a		 =       f'Expected string as input, found {type(_UpperCAmelCase       )}'
        raise ValueError(_UpperCAmelCase       )
    if not isinstance(_UpperCAmelCase     ,				_UpperCAmelCase       ):
        __a		 =       f'Expected boolean as use_pascal parameter, found {type(_UpperCAmelCase       )}'
        raise ValueError(_UpperCAmelCase       )

    __a		 =       input_str.split('''_'''       )

    __a		 =       0 if use_pascal else 1

    __a		 =       words[start_index:]

    __a		 =       [word[0].upper() + word[1:] for word in words_to_capitalize]

    __a		 =       '''''' if use_pascal else words[0]

    return "".join([initial_word, *capitalized_words]       )


if __name__ == "__main__":
 from doctest import testmod

 testmod()





