



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 				_a						(       SCREAMING_SNAKE_CASE       ,  unittest.TestCase			):





							'''simple docstring'''

							A				:Optional[Any] =       AudioLDMPipeline
							A				:Dict =       TEXT_TO_AUDIO_PARAMS
							A				:Optional[int] =       TEXT_TO_AUDIO_BATCH_PARAMS
							A				:str =       frozenset(
							    [
							        "num_inference_steps",
							        "num_waveforms_per_prompt",
							        "generator",
							        "latents",
							        "output_type",
							        "return_dict",
							        "callback",
							        "callback_steps",
							    ]			)



							def 		_A	(							self				):







											"""simple docstring"""



											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=__UpperCAmelCase ,					)
											a__	:							Tuple      						= DDIMScheduler(
											    beta_start=0.0_0_0_8_5 ,					beta_end=0.0_1_2 ,					beta_schedule="scaled_linear" ,					clip_sample=__UpperCAmelCase ,					set_alpha_to_one=__UpperCAmelCase ,					)
											torch.manual_seed(0				)
											a__	:							Optional[int]      						= 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__	:							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=1000 ,					projection_dim=32 ,					)
											a__	:							Optional[Any]      						= ClapTextModelWithProjection(__UpperCAmelCase				)
											a__	:							Any      						= RobertaTokenizer.from_pretrained("hf-internal-testing/tiny-random-roberta" ,					model_max_length=77				)

											a__	:							Optional[int]      						= 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=__UpperCAmelCase ,					)

											a__	:							str      						= SpeechTaHifiGan(__UpperCAmelCase				)

											a__	:							str      						= {
											    "unet": unet,
											    "scheduler": scheduler,
											    "vae": vae,
											    "text_encoder": text_encoder,
											    "tokenizer": tokenizer,
											    "vocoder": vocoder,
											}
											return components



							def 		_A	(							self ,					__UpperCAmelCase ,					__UpperCAmelCase=0				):







											"""simple docstring"""



											if str(__UpperCAmelCase				).startswith("mps"				):
															a__	:							Any      						= torch.manual_seed(__UpperCAmelCase				)
											else:
															a__	:							str      						= torch.Generator(device=__UpperCAmelCase				).manual_seed(__UpperCAmelCase				)
											a__	:							Tuple      						= {
											    "prompt": "A hammer hitting a wooden surface",
											    "generator": generator,
											    "num_inference_steps": 2,
											    "guidance_scale": 6.0,
											}
											return inputs



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							str      						= "cpu"  # ensure determinism for the device-dependent torch.Generator

											a__	:							Union[str, Any]      						= self.get_dummy_components()
											a__	:							Optional[int]      						= AudioLDMPipeline(**__UpperCAmelCase				)
											a__	:							Any      						= audioldm_pipe.to(__UpperCAmelCase				)
											audioldm_pipe.set_progress_bar_config(disable=__UpperCAmelCase				)

											a__	:							Tuple      						= self.get_dummy_inputs(__UpperCAmelCase				)
											a__	:							Tuple      						= audioldm_pipe(**__UpperCAmelCase				)
											a__	:							List[Any]      						= output.audios[0]

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

											a__	:							Optional[int]      						= 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 		_A	(							self				):







											"""simple docstring"""



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

											a__	:							Any      						= self.get_dummy_inputs(__UpperCAmelCase				)
											a__	:							Dict      						= 3 * [inputs["prompt"]]

											# forward
											a__	:							Dict      						= audioldm_pipe(**__UpperCAmelCase				)
											a__	:							str      						= output.audios[0]

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

											a__	:							str      						= audioldm_pipe.tokenizer(
											    __UpperCAmelCase ,					padding="max_length" ,					max_length=audioldm_pipe.tokenizer.model_max_length ,					truncation=__UpperCAmelCase ,					return_tensors="pt" ,					)
											a__	:							List[str]      						= text_inputs["input_ids"].to(__UpperCAmelCase				)

											a__	:							Any      						= audioldm_pipe.text_encoder(
											    __UpperCAmelCase ,					)
											a__	:							int      						= prompt_embeds.text_embeds
											# additional L_2 normalization over each hidden-state
											a__	:							Dict      						= F.normalize(__UpperCAmelCase ,					dim=-1				)

											a__	:							Union[str, Any]      						= prompt_embeds

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

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



							def 		_A	(							self				):







											"""simple docstring"""



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

											a__	:							Dict      						= self.get_dummy_inputs(__UpperCAmelCase				)
											a__	:							Optional[Any]      						= 3 * ["this is a negative prompt"]
											a__	:							str      						= negative_prompt
											a__	:							int      						= 3 * [inputs["prompt"]]

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

											a__	:							Dict      						= self.get_dummy_inputs(__UpperCAmelCase				)
											a__	:							Dict      						= 3 * [inputs.pop("prompt"				)]

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

															a__	:							Dict      						= audioldm_pipe.text_encoder(
															    __UpperCAmelCase ,					)
															a__	:							Dict      						= text_embeds.text_embeds
															# additional L_2 normalization over each hidden-state
															a__	:							Any      						= F.normalize(__UpperCAmelCase ,					dim=-1				)

															embeds.append(__UpperCAmelCase				)

											a__						,   a__	:							Any      						= embeds

											# forward
											a__	:							Any      						= audioldm_pipe(**__UpperCAmelCase				)
											a__	:							Optional[Any]      						= output.audios[0]

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



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							List[str]      						= "cpu"  # ensure determinism for the device-dependent torch.Generator
											a__	:							str      						= self.get_dummy_components()
											a__	:							Any      						= PNDMScheduler(skip_prk_steps=__UpperCAmelCase				)
											a__	:							Tuple      						= AudioLDMPipeline(**__UpperCAmelCase				)
											a__	:							List[Any]      						= audioldm_pipe.to(__UpperCAmelCase				)
											audioldm_pipe.set_progress_bar_config(disable=__UpperCAmelCase				)

											a__	:							Optional[Any]      						= self.get_dummy_inputs(__UpperCAmelCase				)
											a__	:							Optional[Any]      						= "egg cracking"
											a__	:							Union[str, Any]      						= audioldm_pipe(**__UpperCAmelCase ,					negative_prompt=__UpperCAmelCase				)
											a__	:							List[Any]      						= output.audios[0]

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

											a__	:							Optional[int]      						= audio[:10]
											a__	:							Tuple      						= 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 		_A	(							self				):







											"""simple docstring"""



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

											a__	:							Optional[int]      						= "A hammer hitting a wooden surface"

											# test num_waveforms_per_prompt=1 (default)
											a__	:							Optional[int]      						= audioldm_pipe(__UpperCAmelCase ,					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__	:							Dict      						= 2
											a__	:							str      						= audioldm_pipe(__UpperCAmelCase ,					num_inference_steps=2 ,					num_waveforms_per_prompt=__UpperCAmelCase				).audios

											assert audios.shape == (num_waveforms_per_prompt, 256)

											# test num_waveforms_per_prompt for batch of prompts
											a__	:							int      						= 2
											a__	:							Tuple      						= audioldm_pipe(
											    [prompt] * batch_size ,					num_inference_steps=2 ,					num_waveforms_per_prompt=__UpperCAmelCase				).audios

											assert audios.shape == (batch_size * num_waveforms_per_prompt, 256)



							def 		_A	(							self				):







											"""simple docstring"""



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

											a__	:							Any      						= self.get_dummy_inputs(__UpperCAmelCase				)
											a__	:							Dict      						= audioldm_pipe(audio_length_in_s=0.0_1_6 ,					**__UpperCAmelCase				)
											a__	:							List[Any]      						= output.audios[0]

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

											a__	:							Any      						= audioldm_pipe(audio_length_in_s=0.0_3_2 ,					**__UpperCAmelCase				)
											a__	:							Dict      						= output.audios[0]

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



							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							Tuple      						= self.get_dummy_components()
											a__	:							Dict      						= AudioLDMPipeline(**__UpperCAmelCase				)
											a__	:							str      						= audioldm_pipe.to(__UpperCAmelCase				)
											audioldm_pipe.set_progress_bar_config(disable=__UpperCAmelCase				)

											a__	:							int      						= ["hey"]

											a__	:							Optional[Any]      						= audioldm_pipe(__UpperCAmelCase ,					num_inference_steps=1				)
											a__	:							int      						= output.audios.shape
											assert audio_shape == (1, 256)

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







											"""simple docstring"""



											self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__UpperCAmelCase				)



							def 		_A	(							self				):







											"""simple docstring"""



											self._test_inference_batch_single_identical(test_mean_pixel_difference=__UpperCAmelCase				)




							@unittest.skipIf(
							    torch_device != "cuda" or not is_xformers_available() ,					reason="XFormers attention is only available with CUDA and `xformers` installed" ,					)
							def 		_A	(							self				):







											"""simple docstring"""



											self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__UpperCAmelCase				)





@slow
class 				_a						(       unittest.TestCase			):





							'''simple docstring'''

							def 		_A	(							self				):







											"""simple docstring"""



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



							def 		_A	(							self ,					__UpperCAmelCase ,					__UpperCAmelCase="cpu" ,					__UpperCAmelCase=torch.floataa ,					__UpperCAmelCase=0				):







											"""simple docstring"""



											a__	:							Tuple      						= torch.Generator(device=__UpperCAmelCase				).manual_seed(__UpperCAmelCase				)
											a__	:							int      						= np.random.RandomState(__UpperCAmelCase				).standard_normal((1, 8, 128, 16)				)
											a__	:							Dict      						= torch.from_numpy(__UpperCAmelCase				).to(device=__UpperCAmelCase ,					dtype=__UpperCAmelCase				)
											a__	:							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				):







											"""simple docstring"""



											a__	:							List[str]      						= AudioLDMPipeline.from_pretrained("cvssp/audioldm"				)
											a__	:							Dict      						= audioldm_pipe.to(__UpperCAmelCase				)
											audioldm_pipe.set_progress_bar_config(disable=__UpperCAmelCase				)

											a__	:							List[str]      						= self.get_inputs(__UpperCAmelCase				)
											a__	:							Union[str, Any]      						= 25
											a__	:							Optional[int]      						= audioldm_pipe(**__UpperCAmelCase				).audios[0]

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

											a__	:							Dict      						= audio[7_7230:7_7240]
											a__	:							Union[str, 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__	:							Dict      						= np.abs(expected_slice - audio_slice				).max()
											assert max_diff < 1E-2




							def 		_A	(							self				):







											"""simple docstring"""



											a__	:							List[str]      						= AudioLDMPipeline.from_pretrained("cvssp/audioldm"				)
											a__	:							Tuple      						= LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config				)
											a__	:							Tuple      						= audioldm_pipe.to(__UpperCAmelCase				)
											audioldm_pipe.set_progress_bar_config(disable=__UpperCAmelCase				)

											a__	:							Tuple      						= self.get_inputs(__UpperCAmelCase				)
											a__	:							str      						= audioldm_pipe(**__UpperCAmelCase				).audios[0]

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

											a__	:							List[str]      						= audio[2_7780:2_7790]
											a__	:							Optional[int]      						= 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__	:							List[Any]      						= np.abs(expected_slice - audio_slice				).max()
											assert max_diff < 3E-2







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowerCamelCase		       =      {
    """configuration_instructblip""": [
        """INSTRUCTBLIP_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """InstructBlipConfig""",
        """InstructBlipQFormerConfig""",
        """InstructBlipVisionConfig""",
    ],
    """processing_instructblip""": ["""InstructBlipProcessor"""],
}

try:
						if not is_torch_available():
												raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
						pass
else:
						lowerCamelCase		       =      [
						    """INSTRUCTBLIP_PRETRAINED_MODEL_ARCHIVE_LIST""",
						    """InstructBlipQFormerModel""",
						    """InstructBlipPreTrainedModel""",
						    """InstructBlipForConditionalGeneration""",
						    """InstructBlipVisionModel""",
						]

if TYPE_CHECKING:
						from .configuration_instructblip import (
						    INSTRUCTBLIP_PRETRAINED_CONFIG_ARCHIVE_MAP,
						    InstructBlipConfig,
						    InstructBlipQFormerConfig,
						    InstructBlipVisionConfig,
						)
						from .processing_instructblip import InstructBlipProcessor

						try:
												if not is_torch_available():
																		raise OptionalDependencyNotAvailable()
						except OptionalDependencyNotAvailable:
												pass
						else:
												from .modeling_instructblip import (
												    INSTRUCTBLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
												    InstructBlipForConditionalGeneration,
												    InstructBlipPreTrainedModel,
												    InstructBlipQFormerModel,
												    InstructBlipVisionModel,
												)

else:
						import sys

						lowerCamelCase		       =      _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)


