





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_					(			_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 _lowerCAmelCase      (self	:List[str] )->					Tuple:
         torch.manual_seed(0 )
         __A          =				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    ,   )
         __A          =				DDIMScheduler(
             beta_start=0.0_0_0_8_5    ,   beta_end=0.0_1_2    ,   beta_schedule='''scaled_linear'''    ,   clip_sample=__a    ,   set_alpha_to_one=__a    ,   )
         torch.manual_seed(0 )
         __A          =				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          =				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          =				ClapTextModelWithProjection(__a )
         __A          =				RobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-random-roberta'''    ,   model_max_length=77 )

         __A          =				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=__a    ,   )

         __A          =				SpeechTaHifiGan(__a )

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




   def _lowerCAmelCase      (self	:Union[str, Any]    ,   _UpperCamelCase	:List[Any]    ,   _UpperCamelCase	:int=0 )->					List[Any]:

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




   def _lowerCAmelCase      (self	:Any )->					Union[str, Any]:
         __A          =				"cpu"  # ensure determinism for the device-dependent torch.Generator

         __A          =				self.get_dummy_components()
         __A          =				AudioLDMPipeline(**__a )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )

         __A          =				self.get_dummy_inputs(__a )
         __A          =				audioldm_pipe(**__a )
         __A          =				output.audios[0]

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

         __A          =				audio[:10]
         __A          =				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 _lowerCAmelCase      (self	:int )->					Tuple:
         __A          =				self.get_dummy_components()
         __A          =				AudioLDMPipeline(**__a )
         __A          =				audioldm_pipe.to(__a )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )

         __A          =				self.get_dummy_inputs(__a )
         __A          =				3 * [inputs["prompt"]]

         # forward
         __A          =				audioldm_pipe(**__a )
         __A          =				output.audios[0]

         __A          =				self.get_dummy_inputs(__a )
         __A          =				3 * [inputs.pop('''prompt''' )]

         __A          =				audioldm_pipe.tokenizer(
             __a    ,   padding='''max_length'''    ,   max_length=audioldm_pipe.tokenizer.model_max_length    ,   truncation=__a    ,   return_tensors='''pt'''    ,   )
         __A          =				text_inputs["input_ids"].to(__a )

         __A          =				audioldm_pipe.text_encoder(
             __a    ,   )
         __A          =				prompt_embeds.text_embeds
         # additional L_2 normalization over each hidden-state
         __A          =				F.normalize(__a    ,   dim=-1 )

         __A          =				prompt_embeds

         # forward
         __A          =				audioldm_pipe(**__a )
         __A          =				output.audios[0]

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




   def _lowerCAmelCase      (self	:List[Any] )->					Tuple:
         __A          =				self.get_dummy_components()
         __A          =				AudioLDMPipeline(**__a )
         __A          =				audioldm_pipe.to(__a )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )

         __A          =				self.get_dummy_inputs(__a )
         __A          =				3 * ["this is a negative prompt"]
         __A          =				negative_prompt
         __A          =				3 * [inputs["prompt"]]

         # forward
         __A          =				audioldm_pipe(**__a )
         __A          =				output.audios[0]

         __A          =				self.get_dummy_inputs(__a )
         __A          =				3 * [inputs.pop('''prompt''' )]

         __A          =				[]
         for p in [prompt, negative_prompt]:
               __A          =				audioldm_pipe.tokenizer(
                   __a    ,   padding='''max_length'''    ,   max_length=audioldm_pipe.tokenizer.model_max_length    ,   truncation=__a    ,   return_tensors='''pt'''    ,   )
               __A          =				text_inputs["input_ids"].to(__a )

               __A          =				audioldm_pipe.text_encoder(
                   __a    ,   )
               __A          =				text_embeds.text_embeds
               # additional L_2 normalization over each hidden-state
               __A          =				F.normalize(__a    ,   dim=-1 )

               embeds.append(__a )

         __A          =				embeds

         # forward
         __A          =				audioldm_pipe(**__a )
         __A          =				output.audios[0]

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




   def _lowerCAmelCase      (self	:Tuple )->					str:
         __A          =				"cpu"  # ensure determinism for the device-dependent torch.Generator
         __A          =				self.get_dummy_components()
         __A          =				PNDMScheduler(skip_prk_steps=__a )
         __A          =				AudioLDMPipeline(**__a )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )

         __A          =				self.get_dummy_inputs(__a )
         __A          =				"egg cracking"
         __A          =				audioldm_pipe(**__a    ,   negative_prompt=__a )
         __A          =				output.audios[0]

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

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

         __A          =				"A hammer hitting a wooden surface"

         # test num_waveforms_per_prompt=1 (default)
         __A          =				audioldm_pipe(__a    ,   num_inference_steps=2 ).audios

         assert audios.shape == (1, 256)

         # test num_waveforms_per_prompt=1 (default) for batch of prompts
         __A          =				2
         __A          =				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          =				2
         __A          =				audioldm_pipe(__a    ,   num_inference_steps=2    ,   num_waveforms_per_prompt=__a ).audios

         assert audios.shape == (num_waveforms_per_prompt, 256)

         # test num_waveforms_per_prompt for batch of prompts
         __A          =				2
         __A          =				audioldm_pipe(
             [prompt] * batch_size    ,   num_inference_steps=2    ,   num_waveforms_per_prompt=__a ).audios

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




   def _lowerCAmelCase      (self	:Optional[Any] )->					Optional[Any]:
         __A          =				"cpu"  # ensure determinism for the device-dependent torch.Generator
         __A          =				self.get_dummy_components()
         __A          =				AudioLDMPipeline(**__a )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )
         __A          =				audioldm_pipe.vocoder.config.sampling_rate

         __A          =				self.get_dummy_inputs(__a )
         __A          =				audioldm_pipe(audio_length_in_s=0.0_1_6    ,   **__a )
         __A          =				output.audios[0]

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

         __A          =				audioldm_pipe(audio_length_in_s=0.0_3_2    ,   **__a )
         __A          =				output.audios[0]

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




   def _lowerCAmelCase      (self	:Optional[Any] )->					List[str]:
         __A          =				self.get_dummy_components()
         __A          =				AudioLDMPipeline(**__a )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )

         __A          =				["hey"]

         __A          =				audioldm_pipe(__a    ,   num_inference_steps=1 )
         __A          =				output.audios.shape
         assert audio_shape == (1, 256)

         __A          =				audioldm_pipe.vocoder.config
         config.model_in_dim *= 2
         __A          =				SpeechTaHifiGan(__a ).to(__a )
         __A          =				audioldm_pipe(__a    ,   num_inference_steps=1 )
         __A          =				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 _lowerCAmelCase      (self	:Union[str, Any] )->					Any:
         self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__a )




   def _lowerCAmelCase      (self	:Optional[int] )->					Union[str, 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 _lowerCAmelCase      (self	:Union[str, Any] )->					Optional[Any]:
         self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__a )





@slow
class A_					(			unittest.TestCase   ):




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




   def _lowerCAmelCase      (self	:Optional[int]    ,   _UpperCamelCase	:str    ,   _UpperCamelCase	:Optional[int]="cpu"    ,   _UpperCamelCase	:List[Any]=torch.floataa    ,   _UpperCamelCase	:Union[str, Any]=0 )->					List[str]:
         __A          =				torch.Generator(device=__a ).manual_seed(__a )
         __A          =				np.random.RandomState(__a ).standard_normal((1, 8, 128, 16) )
         __A          =				torch.from_numpy(__a ).to(device=__a    ,   dtype=__a )
         __A          =				{
             "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[int] )->					str:
         __A          =				AudioLDMPipeline.from_pretrained('''cvssp/audioldm''' )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )

         __A          =				self.get_inputs(__a )
         __A          =				25
         __A          =				audioldm_pipe(**__a ).audios[0]

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

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






   def _lowerCAmelCase      (self	:int )->					Any:
         __A          =				AudioLDMPipeline.from_pretrained('''cvssp/audioldm''' )
         __A          =				LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config )
         __A          =				audioldm_pipe.to(__a )
         audioldm_pipe.set_progress_bar_config(disable=__a )

         __A          =				self.get_inputs(__a )
         __A          =				audioldm_pipe(**__a ).audios[0]

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

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










import importlib.metadata
import operator
import re
import sys
from typing import Optional

from packaging import version


_snake_case		=  {
    "<": operator.lt,
    "<=": operator.le,
    "==": operator.eq,
    "!=": operator.ne,
    ">=": operator.ge,
    ">": operator.gt,
}






def 				A				(					_lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							if got_ver is None or want_ver is None:
														raise ValueError(
														    F"Unable to compare versions for {requirement}: need={want_ver} found={got_ver}. This is unusual. Consider"
														    F" reinstalling {pkg}."  )
							if not ops[op](version.parse(_lowerCamelCase  )				, version.parse(_lowerCamelCase  )  ):
														raise ImportError(
														    F"{requirement} is required for a normal functioning of this module, but found {pkg}=={got_ver}.{hint}"  )






def 				A				(					_lowerCamelCase				, _lowerCamelCase = None  ):





							'''simple docstring'''




							_lowerCAmelCase     :							List[str]    =    F"\n{hint}" if hint is not None else ""

							# non-versioned check
							if re.match(r"^[\w_\-\d]+$"				, _lowerCamelCase  ):
														_lowerCAmelCase   ,    _lowerCAmelCase   ,    _lowerCAmelCase     :							List[str]    =    requirement, None, None
							else:
														_lowerCAmelCase     :							Optional[int]    =    re.findall(r"^([^!=<>\s]+)([\s!=<>]{1,2}.+)"				, _lowerCamelCase  )
														if not match:
																					raise ValueError(
																					    "requirement needs to be in the pip package format, .e.g., package_a==1.23, or package_b>=1.23, but"
																					    F" got {requirement}"  )
														_lowerCAmelCase   ,    _lowerCAmelCase     :							Dict    =    match[0]
														_lowerCAmelCase     :							Any    =    want_full.split(","  )  # there could be multiple requirements
														_lowerCAmelCase     :							Optional[int]    =    {}
														for w in want_range:
																					_lowerCAmelCase     :							Any    =    re.findall(r"^([\s!=<>]{1,2})(.+)"				, _lowerCamelCase  )
																					if not match:
																												raise ValueError(
																												    "requirement needs to be in the pip package format, .e.g., package_a==1.23, or package_b>=1.23,"
																												    F" but got {requirement}"  )
																					_lowerCAmelCase   ,    _lowerCAmelCase     :							Tuple    =    match[0]
																					_lowerCAmelCase     :							Union[str, Any]    =    want_ver
																					if op not in ops:
																												raise ValueError(F"{requirement}: need one of {list(ops.keys()  )}, but got {op}"  )

    # special case
							if pkg == "python":
														_lowerCAmelCase     :							Tuple    =    ".".join([str(_lowerCamelCase  ) for x in sys.version_info[:3]]  )
														for op, want_ver in wanted.items():
																					_compare_versions(_lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase  )
														return

							# check if any version is installed
							try:
														_lowerCAmelCase     :							Any    =    importlib.metadata.version(_lowerCamelCase  )
							except importlib.metadata.PackageNotFoundError:
														raise importlib.metadata.PackageNotFoundError(
														    F"The '{requirement}' distribution was not found and is required by this application. {hint}"  )

							# check that the right version is installed if version number or a range was provided
							if want_ver is not None:
														for op, want_ver in wanted.items():
																					_compare_versions(_lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase  )






def 				A				(					_lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							List[str]    =    "Try: pip install transformers -U or pip install -e '.[dev]' if you're working with git main"
							return require_version(_lowerCamelCase				, _lowerCamelCase  )
