
'''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 			lowerCAmelCase__       (  lowerCamelCase_   ,						unittest.TestCase					):
					lowerCAmelCase_					       =					AudioLDMPipeline
					lowerCAmelCase_					       =					TEXT_TO_AUDIO_PARAMS
					lowerCAmelCase_					       =					TEXT_TO_AUDIO_BATCH_PARAMS
					lowerCAmelCase_					       =					frozenset(
					    [
					        '''num_inference_steps''',
					        '''num_waveforms_per_prompt''',
					        '''generator''',
					        '''latents''',
					        '''output_type''',
					        '''return_dict''',
					        '''callback''',
					        '''callback_steps''',
					    ]					)






					def      _snake_case   (       self		):


											"""simple docstring"""





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

											lowercase_     :					Tuple													=	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=__SCREAMING_SNAKE_CASE    ,							)

											lowercase_     :					str													=	SpeechTaHifiGan(__SCREAMING_SNAKE_CASE		)

											lowercase_     :					List[str]													=	{
											    '''unet''': unet,
											    '''scheduler''': scheduler,
											    '''vae''': vae,
											    '''text_encoder''': text_encoder,
											    '''tokenizer''': tokenizer,
											    '''vocoder''': vocoder,
											}
											return components






					def      _snake_case   (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE=0		):


											"""simple docstring"""





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






					def      _snake_case   (       self		):


											"""simple docstring"""





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

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

											lowercase_     :					List[Any]													=	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Optional[int]													=	audioldm_pipe(**__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Tuple													=	output.audios[0]

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

											lowercase_     :					Optional[int]													=	audio[:10]
											lowercase_     :					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      _snake_case   (       self		):


											"""simple docstring"""





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

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

											# forward
											lowercase_     :					Optional[int]													=	audioldm_pipe(**__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Tuple													=	output.audios[0]

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

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

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

											lowercase_     :					Dict													=	prompt_embeds

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

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






					def      _snake_case   (       self		):


											"""simple docstring"""





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

											lowercase_     :					Tuple													=	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Optional[Any]													=	3 * ['''this is a negative prompt''']
											lowercase_     :					Optional[Any]													=	negative_prompt
											lowercase_     :					str													=	3 * [inputs['''prompt''']]

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

											lowercase_     :					int													=	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					List[Any]													=	3 * [inputs.pop('''prompt'''		)]

											lowercase_     :					List[str]													=	[]
											for p in [prompt, negative_prompt]:
																	lowercase_     :					Tuple													=	audioldm_pipe.tokenizer(
																	    __SCREAMING_SNAKE_CASE    ,							padding='''max_length'''    ,							max_length=audioldm_pipe.tokenizer.model_max_length    ,							truncation=__SCREAMING_SNAKE_CASE    ,							return_tensors='''pt'''    ,							)
																	lowercase_     :					Optional[Any]													=	text_inputs['''input_ids'''].to(__SCREAMING_SNAKE_CASE		)

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

																	embeds.append(__SCREAMING_SNAKE_CASE		)

											lowercase_						,			lowercase_     :					int													=	embeds

											# forward
											lowercase_     :					Optional[int]													=	audioldm_pipe(**__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Optional[Any]													=	output.audios[0]

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






					def      _snake_case   (       self		):


											"""simple docstring"""





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

											lowercase_     :					Tuple													=	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Optional[Any]													=	'''egg cracking'''
											lowercase_     :					Any													=	audioldm_pipe(**__SCREAMING_SNAKE_CASE    ,							negative_prompt=__SCREAMING_SNAKE_CASE		)
											lowercase_     :					str													=	output.audios[0]

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

											lowercase_     :					Dict													=	audio[:10]
											lowercase_     :					Any													=	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      _snake_case   (       self		):


											"""simple docstring"""





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

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

											# test num_waveforms_per_prompt=1 (default)
											lowercase_     :					Optional[Any]													=	audioldm_pipe(__SCREAMING_SNAKE_CASE    ,							num_inference_steps=2		).audios

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

											# test num_waveforms_per_prompt=1 (default) for batch of prompts
											lowercase_     :					List[Any]													=	2
											lowercase_     :					Dict													=	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
											lowercase_     :					List[str]													=	2
											lowercase_     :					str													=	audioldm_pipe(__SCREAMING_SNAKE_CASE    ,							num_inference_steps=2    ,							num_waveforms_per_prompt=__SCREAMING_SNAKE_CASE		).audios

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

											# test num_waveforms_per_prompt for batch of prompts
											lowercase_     :					Tuple													=	2
											lowercase_     :					Optional[int]													=	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, 2_56)






					def      _snake_case   (       self		):


											"""simple docstring"""





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

											lowercase_     :					Dict													=	self.get_dummy_inputs(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Optional[int]													=	audioldm_pipe(audio_length_in_s=0.016    ,							**__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Optional[int]													=	output.audios[0]

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

											lowercase_     :					List[str]													=	audioldm_pipe(audio_length_in_s=0.032    ,							**__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Tuple													=	output.audios[0]

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






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					Dict													=	self.get_dummy_components()
											lowercase_     :					List[Any]													=	AudioLDMPipeline(**__SCREAMING_SNAKE_CASE		)
											lowercase_     :					List[Any]													=	audioldm_pipe.to(__SCREAMING_SNAKE_CASE		)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE		)

											lowercase_     :					Union[str, Any]													=	['''hey''']

											lowercase_     :					int													=	audioldm_pipe(__SCREAMING_SNAKE_CASE    ,							num_inference_steps=1		)
											lowercase_     :					Union[str, Any]													=	output.audios.shape
											assert audio_shape == (1, 2_56)

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


											"""simple docstring"""





											self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE		)






					def      _snake_case   (       self		):


											"""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      _snake_case   (       self		):


											"""simple docstring"""





											self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE		)







@slow
class 			lowerCAmelCase__       (  unittest.TestCase					):






					def      _snake_case   (       self		):


											"""simple docstring"""





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






					def      _snake_case   (       self    ,							__SCREAMING_SNAKE_CASE    ,							__SCREAMING_SNAKE_CASE="cpu"    ,							__SCREAMING_SNAKE_CASE=torch.floataa    ,							__SCREAMING_SNAKE_CASE=0		):


											"""simple docstring"""





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






					def      _snake_case   (       self		):


											"""simple docstring"""





											lowercase_     :					Any													=	AudioLDMPipeline.from_pretrained('''cvssp/audioldm'''		)
											lowercase_     :					Dict													=	audioldm_pipe.to(__SCREAMING_SNAKE_CASE		)
											audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE		)

											lowercase_     :					Any													=	self.get_inputs(__SCREAMING_SNAKE_CASE		)
											lowercase_     :					Any													=	25
											lowercase_     :					Union[str, Any]													=	audioldm_pipe(**__SCREAMING_SNAKE_CASE		).audios[0]

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

											lowercase_     :					int													=	audio[7_72_30:7_72_40]
											lowercase_     :					Any													=	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]		)
											lowercase_     :					Dict													=	np.abs(expected_slice - audio_slice		).max()
											assert max_diff < 1E-2






					def      _snake_case   (       self		):


											"""simple docstring"""





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

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

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

											lowercase_     :					Tuple													=	audio[2_77_80:2_77_90]
											lowercase_     :					Any													=	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]		)
											lowercase_     :					str													=	np.abs(expected_slice - audio_slice		).max()
											assert max_diff < 3E-2





'''simple docstring'''







from PIL import Image
def  snake_case_		(    __SCREAMING_SNAKE_CASE :			Image				,				__SCREAMING_SNAKE_CASE :			int							):



						"""simple docstring"""





						lowercase_     :					Optional[int]													=	(259 * (level + 255)) / (255 * (259 - level))

						def contrast(__SCREAMING_SNAKE_CASE :			int							) -> int:
												return int(128 + factor * (c - 128)							)

						return img.point(__SCREAMING_SNAKE_CASE							)


if __name__ == "__main__":
							# Load image
							with Image.open("image_data/lena.jpg") as img:
														# Change contrast to 170
														_lowercase			:  Union[str, Any]   							=	change_contrast(img, 1_7_0)
														cont_img.save("image_data/lena_high_contrast.png", format="png")



