"""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      (   lowercase				,     unittest.TestCase						):
				UpperCAmelCase    :				str   =		AudioLDMPipeline
				UpperCAmelCase    :				str   =		TEXT_TO_AUDIO_PARAMS
				UpperCAmelCase    :				Union[str, Any]   =		TEXT_TO_AUDIO_BATCH_PARAMS
				UpperCAmelCase    :				str   =		frozenset(
				    [
				        """num_inference_steps""",
				        """num_waveforms_per_prompt""",
				        """generator""",
				        """latents""",
				        """output_type""",
				        """return_dict""",
				        """callback""",
				        """callback_steps""",
				    ]						)



				def  _lowercase					(self    :				int)   ->     Union[str, Any]:
					torch.manual_seed(0)
					__snake_case      :							str						=			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     ,  )
					__snake_case      :							int						=			DDIMScheduler(
					    beta_start=0.00_085     ,  beta_end=0.012     ,  beta_schedule='scaled_linear'     ,  clip_sample=_A     ,  set_alpha_to_one=_A     ,  )
					torch.manual_seed(0)
					__snake_case      :							Optional[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)
					__snake_case      :							Tuple						=			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=10_00     ,  projection_dim=32     ,  )
					__snake_case      :							Tuple						=			ClapTextModelWithProjection(_A)
					__snake_case      :							Optional[Any]						=			RobertaTokenizer.from_pretrained('hf-internal-testing/tiny-random-roberta'     ,  model_max_length=77)

					__snake_case      :							Union[str, Any]						=			SpeechTaHifiGanConfig(
					    model_in_dim=8     ,  sampling_rate=1_60_00     ,  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     ,  )

					__snake_case      :							List[Any]						=			SpeechTaHifiGan(_A)

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



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



				def  _lowercase					(self    :				Optional[Any])   ->     Dict:
					__snake_case      :							Optional[int]						=			'cpu'  # ensure determinism for the device-dependent torch.Generator

					__snake_case      :							Dict						=			self.get_dummy_components()
					__snake_case      :							Union[str, Any]						=			AudioLDMPipeline(**_A)
					__snake_case      :							Any						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							List[str]						=			self.get_dummy_inputs(_A)
					__snake_case      :							str						=			audioldm_pipe(**_A)
					__snake_case      :							Tuple						=			output.audios[0]

					assert audio.ndim == 1
					assert len(_A) == 2_56

					__snake_case      :							Any						=			audio[:10]
					__snake_case      :							Optional[Any]						=			np.array(
					    [-0.0_050, 0.0_050, -0.0_060, 0.0_033, -0.0_026, 0.0_033, -0.0_027, 0.0_033, -0.0_028, 0.0_033])

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



				def  _lowercase					(self    :				Optional[Any])   ->     Union[str, Any]:
					__snake_case      :							Any						=			self.get_dummy_components()
					__snake_case      :							Optional[int]						=			AudioLDMPipeline(**_A)
					__snake_case      :							List[Any]						=			audioldm_pipe.to(_A)
					__snake_case      :							List[str]						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Union[str, Any]						=			self.get_dummy_inputs(_A)
					__snake_case      :							Dict						=			3 * [inputs['prompt']]

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

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

					__snake_case      :							Dict						=			audioldm_pipe.tokenizer(
					    _A     ,  padding='max_length'     ,  max_length=audioldm_pipe.tokenizer.model_max_length     ,  truncation=_A     ,  return_tensors='pt'     ,  )
					__snake_case      :							Tuple						=			text_inputs['input_ids'].to(_A)

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

					__snake_case      :							Any						=			prompt_embeds

					# forward
					__snake_case      :							Tuple						=			audioldm_pipe(**_A)
					__snake_case      :							Optional[Any]						=			output.audios[0]

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



				def  _lowercase					(self    :				Tuple)   ->     int:
					__snake_case      :							Optional[Any]						=			self.get_dummy_components()
					__snake_case      :							str						=			AudioLDMPipeline(**_A)
					__snake_case      :							List[str]						=			audioldm_pipe.to(_A)
					__snake_case      :							Dict						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

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

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

					__snake_case      :							Dict						=			self.get_dummy_inputs(_A)
					__snake_case      :							Any						=			3 * [inputs.pop('prompt')]

					__snake_case      :							Dict						=			[]
					for p in [prompt, negative_prompt]:
						__snake_case      :							int						=			audioldm_pipe.tokenizer(
						    _A     ,  padding='max_length'     ,  max_length=audioldm_pipe.tokenizer.model_max_length     ,  truncation=_A     ,  return_tensors='pt'     ,  )
						__snake_case      :							Optional[Any]						=			text_inputs['input_ids'].to(_A)

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

						embeds.append(_A)

					__snake_case		,    __snake_case      :							Tuple						=			embeds

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

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



				def  _lowercase					(self    :				Optional[Any])   ->     Union[str, Any]:
					__snake_case      :							Tuple						=			'cpu'  # ensure determinism for the device-dependent torch.Generator
					__snake_case      :							Any						=			self.get_dummy_components()
					__snake_case      :							int						=			PNDMScheduler(skip_prk_steps=_A)
					__snake_case      :							Any						=			AudioLDMPipeline(**_A)
					__snake_case      :							Dict						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Tuple						=			self.get_dummy_inputs(_A)
					__snake_case      :							Tuple						=			'egg cracking'
					__snake_case      :							Union[str, Any]						=			audioldm_pipe(**_A     ,  negative_prompt=_A)
					__snake_case      :							Dict						=			output.audios[0]

					assert audio.ndim == 1
					assert len(_A) == 2_56

					__snake_case      :							Any						=			audio[:10]
					__snake_case      :							Tuple						=			np.array(
					    [-0.0_051, 0.0_050, -0.0_060, 0.0_034, -0.0_026, 0.0_033, -0.0_027, 0.0_033, -0.0_028, 0.0_032])

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



				def  _lowercase					(self    :				List[str])   ->     str:
					__snake_case      :							Optional[Any]						=			'cpu'  # ensure determinism for the device-dependent torch.Generator
					__snake_case      :							Union[str, Any]						=			self.get_dummy_components()
					__snake_case      :							Union[str, Any]						=			PNDMScheduler(skip_prk_steps=_A)
					__snake_case      :							List[str]						=			AudioLDMPipeline(**_A)
					__snake_case      :							List[Any]						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							List[Any]						=			'A hammer hitting a wooden surface'

					# test num_waveforms_per_prompt=1 (default)
					__snake_case      :							Dict						=			audioldm_pipe(_A     ,  num_inference_steps=2).audios

					assert audios.shape == (1, 2_56)

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

					assert audios.shape == (batch_size, 2_56)

					# test num_waveforms_per_prompt for single prompt
					__snake_case      :							List[Any]						=			2
					__snake_case      :							List[Any]						=			audioldm_pipe(_A     ,  num_inference_steps=2     ,  num_waveforms_per_prompt=_A).audios

					assert audios.shape == (num_waveforms_per_prompt, 2_56)

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

					assert audios.shape == (batch_size * num_waveforms_per_prompt, 2_56)



				def  _lowercase					(self    :				List[Any])   ->     Tuple:
					__snake_case      :							List[Any]						=			'cpu'  # ensure determinism for the device-dependent torch.Generator
					__snake_case      :							int						=			self.get_dummy_components()
					__snake_case      :							Union[str, Any]						=			AudioLDMPipeline(**_A)
					__snake_case      :							Union[str, Any]						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)
					__snake_case      :							Optional[Any]						=			audioldm_pipe.vocoder.config.sampling_rate

					__snake_case      :							Any						=			self.get_dummy_inputs(_A)
					__snake_case      :							Tuple						=			audioldm_pipe(audio_length_in_s=0.016     ,  **_A)
					__snake_case      :							Dict						=			output.audios[0]

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

					__snake_case      :							Any						=			audioldm_pipe(audio_length_in_s=0.032     ,  **_A)
					__snake_case      :							int						=			output.audios[0]

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



				def  _lowercase					(self    :				Tuple)   ->     Optional[Any]:
					__snake_case      :							Optional[Any]						=			self.get_dummy_components()
					__snake_case      :							Dict						=			AudioLDMPipeline(**_A)
					__snake_case      :							int						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Any						=			['hey']

					__snake_case      :							Any						=			audioldm_pipe(_A     ,  num_inference_steps=1)
					__snake_case      :							Optional[Any]						=			output.audios.shape
					assert audio_shape == (1, 2_56)

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



				def  _lowercase					(self    :				Optional[Any])   ->     Tuple:
					self._test_attention_slicing_forward_pass(test_mean_pixel_difference=_A)



				def  _lowercase					(self    :				Tuple)   ->     Optional[Any]:
					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  _lowercase					(self    :				List[Any])   ->     Any:
					self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_A)







@slow
class 		UpperCamelCase      (   unittest.TestCase						):



				def  _lowercase					(self    :				int)   ->     Optional[int]:
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()



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



				def  _lowercase					(self    :				Optional[Any])   ->     int:
					__snake_case      :							Any						=			AudioLDMPipeline.from_pretrained('cvssp/audioldm')
					__snake_case      :							List[str]						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							int						=			self.get_inputs(_A)
					__snake_case      :							Tuple						=			25
					__snake_case      :							Dict						=			audioldm_pipe(**_A).audios[0]

					assert audio.ndim == 1
					assert len(_A) == 8_19_20

					__snake_case      :							Union[str, Any]						=			audio[7_72_30:7_72_40]
					__snake_case      :							List[str]						=			np.array(
					    [-0.4_884, -0.4_607, 0.0_023, 0.5_007, 0.5_896, 0.5_151, 0.3_813, -0.0_208, -0.3_687, -0.4_315])
					__snake_case      :							List[str]						=			np.abs(expected_slice - audio_slice).max()
					assert max_diff < 1E-2






				def  _lowercase					(self    :				Dict)   ->     str:
					__snake_case      :							Dict						=			AudioLDMPipeline.from_pretrained('cvssp/audioldm')
					__snake_case      :							Union[str, Any]						=			LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config)
					__snake_case      :							str						=			audioldm_pipe.to(_A)
					audioldm_pipe.set_progress_bar_config(disable=_A)

					__snake_case      :							Dict						=			self.get_inputs(_A)
					__snake_case      :							Optional[int]						=			audioldm_pipe(**_A).audios[0]

					assert audio.ndim == 1
					assert len(_A) == 8_19_20

					__snake_case      :							List[str]						=			audio[2_77_80:2_77_90]
					__snake_case      :							List[str]						=			np.array([-0.2_131, -0.0_873, -0.0_124, -0.0_189, 0.0_569, 0.1_373, 0.1_883, 0.2_886, 0.3_297, 0.2_212])
					__snake_case      :							Optional[int]						=			np.abs(expected_slice - audio_slice).max()
					assert max_diff < 3E-2






"""simple docstring"""







def   __UpperCAmelCase    (   UpperCAmelCase_   :    list      ,							UpperCAmelCase_   :    int      ,							UpperCAmelCase_   :    int = 0      ,							UpperCAmelCase_   :    int = 0  )							->      int:

	'''simple docstring'''





	__snake_case      :							str						=			right or len(UpperCAmelCase_  ) - 1
	if left > right:
		return -1
	elif list_data[left] == key:
		return left
	elif list_data[right] == key:
		return right
	else:
		return search(UpperCAmelCase_      ,							UpperCAmelCase_      ,							left + 1      ,							right - 1  )


if __name__ == "__main__":
	import doctest

	doctest.testmod()





