
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):




	'''simple docstring'''




	_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      _lowerCamelCase       (      self	:int		)	->				int:
								torch.manual_seed(0		)
								__UpperCamelCase			:     Tuple         =      UNetaDConditionModel(
								    block_out_channels=(3_2, 6_4)  ,     layers_per_block=2  ,     sample_size=3_2  ,     in_channels=4  ,     out_channels=4  ,     down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")  ,     up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")  ,     cross_attention_dim=(3_2, 6_4)  ,     class_embed_type="simple_projection"  ,     projection_class_embeddings_input_dim=3_2  ,     class_embeddings_concat=a  ,     )
								__UpperCamelCase			:     str         =      DDIMScheduler(
								    beta_start=0.00085  ,     beta_end=0.012  ,     beta_schedule="scaled_linear"  ,     clip_sample=a  ,     set_alpha_to_one=a  ,     )
								torch.manual_seed(0		)
								__UpperCamelCase			:     Union[str, Any]         =      AutoencoderKL(
								    block_out_channels=[3_2, 6_4]  ,     in_channels=1  ,     out_channels=1  ,     down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]  ,     up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]  ,     latent_channels=4  ,     )
								torch.manual_seed(0		)
								__UpperCamelCase			:     Optional[Any]         =      ClapTextConfig(
								    bos_token_id=0  ,     eos_token_id=2  ,     hidden_size=3_2  ,     intermediate_size=3_7  ,     layer_norm_eps=1E-0_5  ,     num_attention_heads=4  ,     num_hidden_layers=5  ,     pad_token_id=1  ,     vocab_size=1_0_0_0  ,     projection_dim=3_2  ,     )
								__UpperCamelCase			:     Dict         =      ClapTextModelWithProjection(a		)
								__UpperCamelCase			:     Any         =      RobertaTokenizer.from_pretrained("hf-internal-testing/tiny-random-roberta"  ,     model_max_length=7_7		)

								__UpperCamelCase			:     Union[str, Any]         =      SpeechTaHifiGanConfig(
								    model_in_dim=8  ,     sampling_rate=1_6_0_0_0  ,     upsample_initial_channel=1_6  ,     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  ,     )

								__UpperCamelCase			:     List[Any]         =      SpeechTaHifiGan(a		)

								__UpperCamelCase			:     Tuple         =      {
								    "unet": unet,
								    "scheduler": scheduler,
								    "vae": vae,
								    "text_encoder": text_encoder,
								    "tokenizer": tokenizer,
								    "vocoder": vocoder,
								}
								return components







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







	def      _lowerCamelCase       (      self	:int		)	->				Any:
								__UpperCamelCase			:     Tuple         =      "cpu"  # ensure determinism for the device-dependent torch.Generator

								__UpperCamelCase			:     Tuple         =      self.get_dummy_components()
								__UpperCamelCase			:     Union[str, Any]         =      AudioLDMPipeline(**a		)
								__UpperCamelCase			:     Union[str, Any]         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)

								__UpperCamelCase			:     Optional[int]         =      self.get_dummy_inputs(a		)
								__UpperCamelCase			:     Union[str, Any]         =      audioldm_pipe(**a		)
								__UpperCamelCase			:     Tuple         =      output.audios[0]

								assert audio.ndim == 1
								assert len(a		) == 2_5_6

								__UpperCamelCase			:     Optional[Any]         =      audio[:1_0]
								__UpperCamelCase			:     Tuple         =      np.array(
								    [-0.0050, 0.0050, -0.0060, 0.0033, -0.0026, 0.0033, -0.0027, 0.0033, -0.0028, 0.0033]		)

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







	def      _lowerCamelCase       (      self	:Dict		)	->				str:
								__UpperCamelCase			:     Union[str, Any]         =      self.get_dummy_components()
								__UpperCamelCase			:     str         =      AudioLDMPipeline(**a		)
								__UpperCamelCase			:     Union[str, Any]         =      audioldm_pipe.to(a		)
								__UpperCamelCase			:     Optional[int]         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)

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

								# forward
								__UpperCamelCase			:     Any         =      audioldm_pipe(**a		)
								__UpperCamelCase			:     str         =      output.audios[0]

								__UpperCamelCase			:     List[str]         =      self.get_dummy_inputs(a		)
								__UpperCamelCase			:     Dict         =      3 * [inputs.pop("prompt"		)]

								__UpperCamelCase			:     Optional[Any]         =      audioldm_pipe.tokenizer(
								    a  ,     padding="max_length"  ,     max_length=audioldm_pipe.tokenizer.model_max_length  ,     truncation=a  ,     return_tensors="pt"  ,     )
								__UpperCamelCase			:     Optional[Any]         =      text_inputs["input_ids"].to(a		)

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

								__UpperCamelCase			:     Optional[int]         =      prompt_embeds

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

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







	def      _lowerCamelCase       (      self	:Optional[Any]		)	->				Tuple:
								__UpperCamelCase			:     str         =      self.get_dummy_components()
								__UpperCamelCase			:     str         =      AudioLDMPipeline(**a		)
								__UpperCamelCase			:     Dict         =      audioldm_pipe.to(a		)
								__UpperCamelCase			:     Optional[int]         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)

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

								# forward
								__UpperCamelCase			:     Optional[Any]         =      audioldm_pipe(**a		)
								__UpperCamelCase			:     List[str]         =      output.audios[0]

								__UpperCamelCase			:     List[Any]         =      self.get_dummy_inputs(a		)
								__UpperCamelCase			:     Optional[int]         =      3 * [inputs.pop("prompt"		)]

								__UpperCamelCase			:     List[str]         =      []
								for p in [prompt, negative_prompt]:
															__UpperCamelCase			:     str         =      audioldm_pipe.tokenizer(
															    a  ,     padding="max_length"  ,     max_length=audioldm_pipe.tokenizer.model_max_length  ,     truncation=a  ,     return_tensors="pt"  ,     )
															__UpperCamelCase			:     List[str]         =      text_inputs["input_ids"].to(a		)

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

															embeds.append(a		)

								__UpperCamelCase     ,		__UpperCamelCase			:     Tuple         =      embeds

								# forward
								__UpperCamelCase			:     Any         =      audioldm_pipe(**a		)
								__UpperCamelCase			:     Any         =      output.audios[0]

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







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

								__UpperCamelCase			:     str         =      self.get_dummy_inputs(a		)
								__UpperCamelCase			:     Optional[int]         =      "egg cracking"
								__UpperCamelCase			:     Optional[int]         =      audioldm_pipe(**a  ,     negative_prompt=a		)
								__UpperCamelCase			:     Any         =      output.audios[0]

								assert audio.ndim == 1
								assert len(a		) == 2_5_6

								__UpperCamelCase			:     int         =      audio[:1_0]
								__UpperCamelCase			:     Optional[int]         =      np.array(
								    [-0.0051, 0.0050, -0.0060, 0.0034, -0.0026, 0.0033, -0.0027, 0.0033, -0.0028, 0.0032]		)

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







	def      _lowerCamelCase       (      self	:List[Any]		)	->				List[str]:
								__UpperCamelCase			:     Tuple         =      "cpu"  # ensure determinism for the device-dependent torch.Generator
								__UpperCamelCase			:     Optional[int]         =      self.get_dummy_components()
								__UpperCamelCase			:     int         =      PNDMScheduler(skip_prk_steps=a		)
								__UpperCamelCase			:     str         =      AudioLDMPipeline(**a		)
								__UpperCamelCase			:     Dict         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)

								__UpperCamelCase			:     int         =      "A hammer hitting a wooden surface"

								# test num_waveforms_per_prompt=1 (default)
								__UpperCamelCase			:     Optional[int]         =      audioldm_pipe(a  ,     num_inference_steps=2		).audios

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

								# test num_waveforms_per_prompt=1 (default) for batch of prompts
								__UpperCamelCase			:     List[Any]         =      2
								__UpperCamelCase			:     Union[str, Any]         =      audioldm_pipe([prompt] * batch_size  ,     num_inference_steps=2		).audios

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

								# test num_waveforms_per_prompt for single prompt
								__UpperCamelCase			:     Dict         =      2
								__UpperCamelCase			:     str         =      audioldm_pipe(a  ,     num_inference_steps=2  ,     num_waveforms_per_prompt=a		).audios

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

								# test num_waveforms_per_prompt for batch of prompts
								__UpperCamelCase			:     Any         =      2
								__UpperCamelCase			:     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_5_6)







	def      _lowerCamelCase       (      self	:List[Any]		)	->				str:
								__UpperCamelCase			:     Tuple         =      "cpu"  # ensure determinism for the device-dependent torch.Generator
								__UpperCamelCase			:     str         =      self.get_dummy_components()
								__UpperCamelCase			:     Any         =      AudioLDMPipeline(**a		)
								__UpperCamelCase			:     Any         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)
								__UpperCamelCase			:     List[str]         =      audioldm_pipe.vocoder.config.sampling_rate

								__UpperCamelCase			:     Union[str, Any]         =      self.get_dummy_inputs(a		)
								__UpperCamelCase			:     List[str]         =      audioldm_pipe(audio_length_in_s=0.016  ,     **a		)
								__UpperCamelCase			:     Optional[int]         =      output.audios[0]

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

								__UpperCamelCase			:     Optional[Any]         =      audioldm_pipe(audio_length_in_s=0.032  ,     **a		)
								__UpperCamelCase			:     Tuple         =      output.audios[0]

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







	def      _lowerCamelCase       (      self	:Optional[Any]		)	->				Any:
								__UpperCamelCase			:     List[Any]         =      self.get_dummy_components()
								__UpperCamelCase			:     Union[str, Any]         =      AudioLDMPipeline(**a		)
								__UpperCamelCase			:     Dict         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)

								__UpperCamelCase			:     List[Any]         =      ["hey"]

								__UpperCamelCase			:     Any         =      audioldm_pipe(a  ,     num_inference_steps=1		)
								__UpperCamelCase			:     Any         =      output.audios.shape
								assert audio_shape == (1, 2_5_6)

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







	def      _lowerCamelCase       (      self	:Union[str, Any]		)	->				List[str]:
								self._test_attention_slicing_forward_pass(test_mean_pixel_difference=a		)







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




@slow
class 				lowerCamelCase__							(       unittest.TestCase):




	'''simple docstring'''




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







	def      _lowerCamelCase       (      self	:List[str]  ,     a	:Optional[Any]  ,     a	:Dict="cpu"  ,     a	:str=torch.floataa  ,     a	:Union[str, Any]=0		)	->				List[str]:
								__UpperCamelCase			:     List[Any]         =      torch.Generator(device=a		).manual_seed(a		)
								__UpperCamelCase			:     Union[str, Any]         =      np.random.RandomState(a		).standard_normal((1, 8, 1_2_8, 1_6)		)
								__UpperCamelCase			:     Tuple         =      torch.from_numpy(a		).to(device=a  ,     dtype=a		)
								__UpperCamelCase			:     int         =      {
								    "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]		)	->				Optional[Any]:
								__UpperCamelCase			:     Union[str, Any]         =      AudioLDMPipeline.from_pretrained("cvssp/audioldm"		)
								__UpperCamelCase			:     Union[str, Any]         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)

								__UpperCamelCase			:     Dict         =      self.get_inputs(a		)
								__UpperCamelCase			:     List[str]         =      2_5
								__UpperCamelCase			:     Dict         =      audioldm_pipe(**a		).audios[0]

								assert audio.ndim == 1
								assert len(a		) == 8_1_9_2_0

								__UpperCamelCase			:     Any         =      audio[7_7_2_3_0:7_7_2_4_0]
								__UpperCamelCase			:     Optional[int]         =      np.array(
								    [-0.4884, -0.4607, 0.0023, 0.5007, 0.5896, 0.5151, 0.3813, -0.0208, -0.3687, -0.4315]		)
								__UpperCamelCase			:     Optional[int]         =      np.abs(expected_slice - audio_slice		).max()
								assert max_diff < 1E-2







	def      _lowerCamelCase       (      self	:Optional[Any]		)	->				Union[str, Any]:
								__UpperCamelCase			:     List[str]         =      AudioLDMPipeline.from_pretrained("cvssp/audioldm"		)
								__UpperCamelCase			:     Optional[int]         =      LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config		)
								__UpperCamelCase			:     Union[str, Any]         =      audioldm_pipe.to(a		)
								audioldm_pipe.set_progress_bar_config(disable=a		)

								__UpperCamelCase			:     int         =      self.get_inputs(a		)
								__UpperCamelCase			:     List[str]         =      audioldm_pipe(**a		).audios[0]

								assert audio.ndim == 1
								assert len(a		) == 8_1_9_2_0

								__UpperCamelCase			:     Any         =      audio[2_7_7_8_0:2_7_7_9_0]
								__UpperCamelCase			:     Tuple         =      np.array([-0.2131, -0.0873, -0.0124, -0.0189, 0.0569, 0.1373, 0.1883, 0.2886, 0.3297, 0.2212]		)
								__UpperCamelCase			:     Optional[Any]         =      np.abs(expected_slice - audio_slice		).max()
								assert max_diff < 3E-2

from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_torch_available,
)


lowercase						:      str									= {
    'configuration_vision_encoder_decoder': ['VisionEncoderDecoderConfig', 'VisionEncoderDecoderOnnxConfig']
}

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowercase						:      str									= ['VisionEncoderDecoderModel']

try:
		if not is_tf_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowercase						:      List[Any]									= ['TFVisionEncoderDecoderModel']

try:
		if not is_flax_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		lowercase						:      int									= ['FlaxVisionEncoderDecoderModel']

if TYPE_CHECKING:
		from .configuration_vision_encoder_decoder import VisionEncoderDecoderConfig, VisionEncoderDecoderOnnxConfig

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_vision_encoder_decoder import VisionEncoderDecoderModel

		try:
				if not is_tf_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_tf_vision_encoder_decoder import TFVisionEncoderDecoderModel

		try:
				if not is_flax_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_flax_vision_encoder_decoder import FlaxVisionEncoderDecoderModel

else:
		import sys

		lowercase						:      Dict									= _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)