
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   __lowercase						(						__snake_case   , unittest.TestCase       ):
	_A       =   AudioLDMPipeline
	_A       =   TEXT_TO_AUDIO_PARAMS
	_A       =   TEXT_TO_AUDIO_BATCH_PARAMS
	_A       =   frozenset(
	    [
	        "num_inference_steps",
	        "num_waveforms_per_prompt",
	        "generator",
	        "latents",
	        "output_type",
	        "return_dict",
	        "callback",
	        "callback_steps",
	    ]       )






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

							_lowercase       :	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=snake_case      ,	)

							_lowercase       :	Optional[int]           =   SpeechTaHifiGan(snake_case    )

							_lowercase       :	Union[str, Any]           =   {
							    "unet": unet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "vocoder": vocoder,
							}
							return components






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






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

							_lowercase       :	int           =   self.get_dummy_components()
							_lowercase       :	Tuple           =   AudioLDMPipeline(**snake_case    )
							_lowercase       :	int           =   audioldm_pipe.to(snake_case    )
							audioldm_pipe.set_progress_bar_config(disable=snake_case    )

							_lowercase       :	Optional[int]           =   self.get_dummy_inputs(snake_case    )
							_lowercase       :	int           =   audioldm_pipe(**snake_case    )
							_lowercase       :	str           =   output.audios[0]

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

							_lowercase       :	List[str]           =   audio[:10]
							_lowercase       :	str           =   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       _a(self				: List[str]    ) ->			str:
							_lowercase       :	Tuple           =   self.get_dummy_components()
							_lowercase       :	Any           =   AudioLDMPipeline(**snake_case    )
							_lowercase       :	Optional[Any]           =   audioldm_pipe.to(snake_case    )
							_lowercase       :	Any           =   audioldm_pipe.to(snake_case    )
							audioldm_pipe.set_progress_bar_config(disable=snake_case    )

							_lowercase       :	List[str]           =   self.get_dummy_inputs(snake_case    )
							_lowercase       :	Any           =   3 * [inputs["prompt"]]

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

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

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

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

							_lowercase       :	List[Any]           =   prompt_embeds

							# forward
							_lowercase       :	int           =   audioldm_pipe(**snake_case    )
							_lowercase       :	Optional[int]           =   output.audios[0]

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






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

							_lowercase       :	int           =   self.get_dummy_inputs(snake_case    )
							_lowercase       :	Tuple           =   3 * ["this is a negative prompt"]
							_lowercase       :	Dict           =   negative_prompt
							_lowercase       :	Union[str, Any]           =   3 * [inputs["prompt"]]

							# forward
							_lowercase       :	Tuple           =   audioldm_pipe(**snake_case    )
							_lowercase       :	List[str]           =   output.audios[0]

							_lowercase       :	Union[str, Any]           =   self.get_dummy_inputs(snake_case    )
							_lowercase       :	List[str]           =   3 * [inputs.pop("prompt"    )]

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

													_lowercase       :	Dict           =   audioldm_pipe.text_encoder(
													    snake_case      ,	)
													_lowercase       :	List[Any]           =   text_embeds.text_embeds
													# additional L_2 normalization over each hidden-state
													_lowercase       :	Optional[Any]           =   F.normalize(snake_case      ,	dim=-1    )

													embeds.append(snake_case    )

							_lowercase      ,    _lowercase       :	int           =   embeds

							# forward
							_lowercase       :	str           =   audioldm_pipe(**snake_case    )
							_lowercase       :	List[str]           =   output.audios[0]

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






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

							_lowercase       :	str           =   self.get_dummy_inputs(snake_case    )
							_lowercase       :	Any           =   "egg cracking"
							_lowercase       :	List[Any]           =   audioldm_pipe(**snake_case      ,	negative_prompt=snake_case    )
							_lowercase       :	str           =   output.audios[0]

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

							_lowercase       :	List[Any]           =   audio[:10]
							_lowercase       :	str           =   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       _a(self				: Dict    ) ->			Optional[int]:
							_lowercase       :	Tuple           =   "cpu"  # ensure determinism for the device-dependent torch.Generator
							_lowercase       :	str           =   self.get_dummy_components()
							_lowercase       :	int           =   PNDMScheduler(skip_prk_steps=snake_case    )
							_lowercase       :	Any           =   AudioLDMPipeline(**snake_case    )
							_lowercase       :	Optional[int]           =   audioldm_pipe.to(snake_case    )
							audioldm_pipe.set_progress_bar_config(disable=snake_case    )

							_lowercase       :	Optional[Any]           =   "A hammer hitting a wooden surface"

							# test num_waveforms_per_prompt=1 (default)
							_lowercase       :	Union[str, Any]           =   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
							_lowercase       :	Union[str, Any]           =   2
							_lowercase       :	Tuple           =   audioldm_pipe([prompt] * batch_size      ,	num_inference_steps=2    ).audios

							assert audios.shape == (batch_size, 256)

							# test num_waveforms_per_prompt for single prompt
							_lowercase       :	int           =   2
							_lowercase       :	str           =   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
							_lowercase       :	int           =   2
							_lowercase       :	List[str]           =   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       _a(self				: Dict    ) ->			List[Any]:
							_lowercase       :	Optional[int]           =   "cpu"  # ensure determinism for the device-dependent torch.Generator
							_lowercase       :	Any           =   self.get_dummy_components()
							_lowercase       :	Dict           =   AudioLDMPipeline(**snake_case    )
							_lowercase       :	List[str]           =   audioldm_pipe.to(snake_case    )
							audioldm_pipe.set_progress_bar_config(disable=snake_case    )
							_lowercase       :	Union[str, Any]           =   audioldm_pipe.vocoder.config.sampling_rate

							_lowercase       :	Dict           =   self.get_dummy_inputs(snake_case    )
							_lowercase       :	Optional[Any]           =   audioldm_pipe(audio_length_in_s=0.0_16      ,	**snake_case    )
							_lowercase       :	List[Any]           =   output.audios[0]

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

							_lowercase       :	Tuple           =   audioldm_pipe(audio_length_in_s=0.0_32      ,	**snake_case    )
							_lowercase       :	str           =   output.audios[0]

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






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

							_lowercase       :	str           =   ["hey"]

							_lowercase       :	Optional[int]           =   audioldm_pipe(snake_case      ,	num_inference_steps=1    )
							_lowercase       :	List[str]           =   output.audios.shape
							assert audio_shape == (1, 256)

							_lowercase       :	Union[str, Any]           =   audioldm_pipe.vocoder.config
							config.model_in_dim *= 2
							_lowercase       :	Union[str, Any]           =   SpeechTaHifiGan(snake_case    ).to(snake_case    )
							_lowercase       :	Dict           =   audioldm_pipe(snake_case      ,	num_inference_steps=1    )
							_lowercase       :	int           =   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       _a(self				: Dict    ) ->			List[str]:
							self._test_attention_slicing_forward_pass(test_mean_pixel_difference=snake_case    )






	def       _a(self				: Dict    ) ->			Optional[Any]:
							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       _a(self				: Union[str, Any]    ) ->			List[Any]:
							self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case    )




@slow
class   __lowercase						(						unittest.TestCase       ):






	def       _a(self				: Optional[Any]    ) ->			List[str]:
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()






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






	def       _a(self				: Optional[int]    ) ->			Any:
							_lowercase       :	Any           =   AudioLDMPipeline.from_pretrained("cvssp/audioldm"    )
							_lowercase       :	str           =   audioldm_pipe.to(snake_case    )
							audioldm_pipe.set_progress_bar_config(disable=snake_case    )

							_lowercase       :	Optional[int]           =   self.get_inputs(snake_case    )
							_lowercase       :	Tuple           =   25
							_lowercase       :	Optional[Any]           =   audioldm_pipe(**snake_case    ).audios[0]

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

							_lowercase       :	int           =   audio[7_7230:7_7240]
							_lowercase       :	str           =   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]    )
							_lowercase       :	Dict           =   np.abs(expected_slice - audio_slice    ).max()
							assert max_diff < 1e-2






	def       _a(self				: str    ) ->			Dict:
							_lowercase       :	str           =   AudioLDMPipeline.from_pretrained("cvssp/audioldm"    )
							_lowercase       :	Tuple           =   LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config    )
							_lowercase       :	int           =   audioldm_pipe.to(snake_case    )
							audioldm_pipe.set_progress_bar_config(disable=snake_case    )

							_lowercase       :	Any           =   self.get_inputs(snake_case    )
							_lowercase       :	Union[str, Any]           =   audioldm_pipe(**snake_case    ).audios[0]

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

							_lowercase       :	Tuple           =   audio[2_7780:2_7790]
							_lowercase       :	Tuple           =   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]    )
							_lowercase       :	Dict           =   np.abs(expected_slice - audio_slice    ).max()
							assert max_diff < 3e-2






import copy
from dataclasses import dataclass
from pathlib import Path
from typing import Dict, Optional, Union




@dataclass
class   __lowercase						:
	_A       =   None
	_A       =   False
	_A       =   False
	_A       =   False
	_A       =   None
	_A       =   None
	_A       =   False
	_A       =   False
	_A       =   False
	_A       =   True
	_A       =   None
	_A       =   1
	_A       =   None
	_A       =   False
	_A       =   None
	_A       =   None






	def       _a(self				: str    ) ->			"DownloadConfig":
							return self.__class__(**{k: copy.deepcopy(snake_case    ) for k, v in self.__dict__.items()}    )




