

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




	def        lowerCAmelCase_       (     self						:     List[Any]							):
		torch.manual_seed(0							)
		__A :			List[Any]		    =			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 :			List[str]		    =			DDIMScheduler(
		    beta_start=0.0_0_0_8_5					,      beta_end=0.0_1_2					,      beta_schedule="""scaled_linear"""					,      clip_sample=__A					,      set_alpha_to_one=__A					,      )
		torch.manual_seed(0							)
		__A :			Any		    =			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 :			str		    =			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 :			int		    =			ClapTextModelWithProjection(__A							)
		__A :			str		    =			RobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-random-roberta"""					,      model_max_length=77							)

		__A :			List[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 :			List[Any]		    =			SpeechTaHifiGan(__A							)

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




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




	def        lowerCAmelCase_       (     self						:     Optional[Any]							):
		__A :			Any		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator

		__A :			Tuple		    =			self.get_dummy_components()
		__A :			Dict		    =			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 :			Optional[Any]		    =			audioldm_pipe(**__A							)
		__A :			List[str]		    =			output.audios[0]

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

		__A :			str		    =			audio[:10]
		__A :			List[str]		    =			np.array(
		    [-0.0_0_5_0, 0.0_0_5_0, -0.0_0_6_0, 0.0_0_3_3, -0.0_0_2_6, 0.0_0_3_3, -0.0_0_2_7, 0.0_0_3_3, -0.0_0_2_8, 0.0_0_3_3]							)

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




	def        lowerCAmelCase_       (     self						:     Union[str, Any]							):
		__A :			str		    =			self.get_dummy_components()
		__A :			List[str]		    =			AudioLDMPipeline(**__A							)
		__A :			Any		    =			audioldm_pipe.to(__A							)
		__A :			Any		    =			audioldm_pipe.to(__A							)
		audioldm_pipe.set_progress_bar_config(disable=__A							)

		__A :			int		    =			self.get_dummy_inputs(__A							)
		__A :			List[str]		    =			3 * [inputs["""prompt"""]]

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

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

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

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

		__A :			List[str]		    =			prompt_embeds

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

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




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

		__A :			str		    =			self.get_dummy_inputs(__A							)
		__A :			Dict		    =			3 * ["""this is a negative prompt"""]
		__A :			List[Any]		    =			negative_prompt
		__A :			Tuple		    =			3 * [inputs["""prompt"""]]

		# forward
		__A :			Union[str, Any]		    =			audioldm_pipe(**__A							)
		__A :			Tuple		    =			output.audios[0]

		__A :			Tuple		    =			self.get_dummy_inputs(__A							)
		__A :			Optional[Any]		    =			3 * [inputs.pop("""prompt"""							)]

		__A :			Tuple		    =			[]
		for p in [prompt, negative_prompt]:
			__A :			Any		    =			audioldm_pipe.tokenizer(
			    __A					,      padding="""max_length"""					,      max_length=audioldm_pipe.tokenizer.model_max_length					,      truncation=__A					,      return_tensors="""pt"""					,      )
			__A :			Union[str, Any]		    =			text_inputs["""input_ids"""].to(__A							)

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

			embeds.append(__A							)

		__A    ,				__A :			Optional[Any]		    =			embeds

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

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




	def        lowerCAmelCase_       (     self						:     Any							):
		__A :			List[str]		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
		__A :			Union[str, Any]		    =			self.get_dummy_components()
		__A :			Dict		    =			PNDMScheduler(skip_prk_steps=__A							)
		__A :			Any		    =			AudioLDMPipeline(**__A							)
		__A :			List[Any]		    =			audioldm_pipe.to(__A							)
		audioldm_pipe.set_progress_bar_config(disable=__A							)

		__A :			Tuple		    =			self.get_dummy_inputs(__A							)
		__A :			int		    =			"""egg cracking"""
		__A :			List[str]		    =			audioldm_pipe(**__A					,      negative_prompt=__A							)
		__A :			str		    =			output.audios[0]

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

		__A :			List[Any]		    =			audio[:10]
		__A :			Union[str, Any]		    =			np.array(
		    [-0.0_0_5_1, 0.0_0_5_0, -0.0_0_6_0, 0.0_0_3_4, -0.0_0_2_6, 0.0_0_3_3, -0.0_0_2_7, 0.0_0_3_3, -0.0_0_2_8, 0.0_0_3_2]							)

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




	def        lowerCAmelCase_       (     self						:     Union[str, Any]							):
		__A :			int		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
		__A :			Optional[int]		    =			self.get_dummy_components()
		__A :			Any		    =			PNDMScheduler(skip_prk_steps=__A							)
		__A :			Union[str, Any]		    =			AudioLDMPipeline(**__A							)
		__A :			Optional[Any]		    =			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 :			Dict		    =			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 :			List[str]		    =			2
		__A :			str		    =			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 :			List[str]		    =			2
		__A :			int		    =			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 :			Union[str, Any]		    =			2
		__A :			Any		    =			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        lowerCAmelCase_       (     self						:     Any							):
		__A :			int		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
		__A :			str		    =			self.get_dummy_components()
		__A :			Union[str, Any]		    =			AudioLDMPipeline(**__A							)
		__A :			Optional[int]		    =			audioldm_pipe.to(__A							)
		audioldm_pipe.set_progress_bar_config(disable=__A							)
		__A :			int		    =			audioldm_pipe.vocoder.config.sampling_rate

		__A :			Tuple		    =			self.get_dummy_inputs(__A							)
		__A :			Optional[Any]		    =			audioldm_pipe(audio_length_in_s=0.0_1_6					,      **__A							)
		__A :			Union[str, Any]		    =			output.audios[0]

		assert audio.ndim == 1
		assert len(__A							) / vocoder_sampling_rate == 0.0_1_6

		__A :			Dict		    =			audioldm_pipe(audio_length_in_s=0.0_3_2					,      **__A							)
		__A :			Optional[int]		    =			output.audios[0]

		assert audio.ndim == 1
		assert len(__A							) / vocoder_sampling_rate == 0.0_3_2




	def        lowerCAmelCase_       (     self						:     int							):
		__A :			List[Any]		    =			self.get_dummy_components()
		__A :			Dict		    =			AudioLDMPipeline(**__A							)
		__A :			Union[str, Any]		    =			audioldm_pipe.to(__A							)
		audioldm_pipe.set_progress_bar_config(disable=__A							)

		__A :			int		    =			["""hey"""]

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

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




	def        lowerCAmelCase_       (     self						:     str							):
		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        lowerCAmelCase_       (     self						:     Optional[Any]							):
		self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__A							)






@slow
class      lowerCamelCase_      (	unittest.TestCase	):




	def        lowerCAmelCase_       (     self						:     Dict							):
		super().tearDown()
		gc.collect()
		torch.cuda.empty_cache()




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




	def        lowerCAmelCase_       (     self						:     Dict							):
		__A :			Any		    =			AudioLDMPipeline.from_pretrained("""cvssp/audioldm"""							)
		__A :			Optional[int]		    =			audioldm_pipe.to(__A							)
		audioldm_pipe.set_progress_bar_config(disable=__A							)

		__A :			Optional[Any]		    =			self.get_inputs(__A							)
		__A :			Dict		    =			25
		__A :			Tuple		    =			audioldm_pipe(**__A							).audios[0]

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

		__A :			Tuple		    =			audio[7_7230:7_7240]
		__A :			Optional[Any]		    =			np.array(
		    [-0.4_8_8_4, -0.4_6_0_7, 0.0_0_2_3, 0.5_0_0_7, 0.5_8_9_6, 0.5_1_5_1, 0.3_8_1_3, -0.0_2_0_8, -0.3_6_8_7, -0.4_3_1_5]							)
		__A :			Optional[Any]		    =			np.abs(expected_slice - audio_slice							).max()
		assert max_diff < 1e-2




	def        lowerCAmelCase_       (     self						:     List[Any]							):
		__A :			int		    =			AudioLDMPipeline.from_pretrained("""cvssp/audioldm"""							)
		__A :			Any		    =			LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config							)
		__A :			List[Any]		    =			audioldm_pipe.to(__A							)
		audioldm_pipe.set_progress_bar_config(disable=__A							)

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

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

		__A :			str		    =			audio[2_7780:2_7790]
		__A :			str		    =			np.array([-0.2_1_3_1, -0.0_8_7_3, -0.0_1_2_4, -0.0_1_8_9, 0.0_5_6_9, 0.1_3_7_3, 0.1_8_8_3, 0.2_8_8_6, 0.3_2_9_7, 0.2_2_1_2]							)
		__A :			Union[str, Any]		    =			np.abs(expected_slice - audio_slice							).max()
		assert max_diff < 3e-2











import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
					import torch

					from transformers.activations import gelu_new, gelu_python, get_activation







@require_torch
class        __lowerCAmelCase					(			unittest.TestCase					):

			"""simple docstring"""







			def  lowercase_       ( self				)       -> int:



							'''simple docstring'''
							__lowerCamelCase									=       torch.tensor([-100, -1, -0.1, 0, 0.1, 1.0, 100]				)
							__lowerCamelCase									=       get_activation('gelu'				)
							self.assertTrue(torch.allclose(gelu_python(lowerCamelCase__				)     ,		torch_builtin(lowerCamelCase__				)				)				)
							self.assertFalse(torch.allclose(gelu_python(lowerCamelCase__				)     ,		gelu_new(lowerCamelCase__				)				)				)


			def  lowercase_       ( self				)       -> Optional[int]:



							'''simple docstring'''
							__lowerCamelCase									=       torch.tensor([-100, -1, -0.1, 0, 0.1, 1.0, 100]				)
							__lowerCamelCase									=       get_activation('gelu'				)
							__lowerCamelCase									=       get_activation('gelu_10'				)

							__lowerCamelCase									=       torch_builtin(lowerCamelCase__				)
							__lowerCamelCase									=       geluaa(lowerCamelCase__				)

							__lowerCamelCase									=       torch.where(y_gelu_aa < 10.0     ,		1     ,		0				)

							self.assertTrue(torch.max(lowerCamelCase__				).item() == 10.0				)
							self.assertTrue(torch.allclose(y_gelu * clipped_mask     ,		y_gelu_aa * clipped_mask				)				)


			def  lowercase_       ( self				)       -> Any:



							'''simple docstring'''
							get_activation('gelu'				)
							get_activation('gelu_10'				)
							get_activation('gelu_fast'				)
							get_activation('gelu_new'				)
							get_activation('gelu_python'				)
							get_activation('gelu_pytorch_tanh'				)
							get_activation('linear'				)
							get_activation('mish'				)
							get_activation('quick_gelu'				)
							get_activation('relu'				)
							get_activation('sigmoid'				)
							get_activation('silu'				)
							get_activation('swish'				)
							get_activation('tanh'				)
							with self.assertRaises(lowerCamelCase__				):
											get_activation('bogus'				)
							with self.assertRaises(lowerCamelCase__				):
											get_activation(lowerCamelCase__				)






			def  lowercase_       ( self				)       -> Dict:



							'''simple docstring'''
							__lowerCamelCase									=       get_activation('gelu'				)
							__lowerCamelCase									=       1
							__lowerCamelCase									=       get_activation('gelu'				)
							self.assertEqual(acta.a     ,		1				)
							with self.assertRaises(lowerCamelCase__				):
											__lowerCamelCase									=       acta.a


