


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






       def      lowerCAmelCase						(     self)		->    Dict:




          '''simple docstring'''
          torch.manual_seed(0)
          _UpperCAmelCase  :  Union[str, Any]				       =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=snake_case    ,				)
          _UpperCAmelCase  :  Optional[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)
          _UpperCAmelCase  :  str				       =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  :  int				       =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  :  Optional[int]				       =ClapTextModelWithProjection(snake_case)
          _UpperCAmelCase  :  List[str]				       =RobertaTokenizer.from_pretrained('hf-internal-testing/tiny-random-roberta'    ,				model_max_length=7_7)

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

          _UpperCAmelCase  :  str				       =SpeechTaHifiGan(snake_case)

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






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case=0)		->    str:




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






       def      lowerCAmelCase						(     self)		->    str:




          '''simple docstring'''
          _UpperCAmelCase  :  Union[str, Any]				       ='cpu'  # ensure determinism for the device-dependent torch.Generator

          _UpperCAmelCase  :  Dict				       =self.get_dummy_components()
          _UpperCAmelCase  :  int				       =AudioLDMPipeline(**snake_case)
          _UpperCAmelCase  :  Dict				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)

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

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

          _UpperCAmelCase  :  int				       =audio[:1_0]
          _UpperCAmelCase  :  Tuple				       =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      lowerCAmelCase						(     self)		->    Any:




          '''simple docstring'''
          _UpperCAmelCase  :  List[str]				       =self.get_dummy_components()
          _UpperCAmelCase  :  Dict				       =AudioLDMPipeline(**snake_case)
          _UpperCAmelCase  :  str				       =audioldm_pipe.to(snake_case)
          _UpperCAmelCase  :  Tuple				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  Optional[Any]				       =self.get_dummy_inputs(snake_case)
          _UpperCAmelCase  :  List[Any]				       =3 * [inputs['prompt']]

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

          _UpperCAmelCase  :  Union[str, Any]				       =self.get_dummy_inputs(snake_case)
          _UpperCAmelCase  :  Tuple				       =3 * [inputs.pop('prompt')]

          _UpperCAmelCase  :  int				       =audioldm_pipe.tokenizer(
              snake_case    ,				padding='max_length'    ,				max_length=audioldm_pipe.tokenizer.model_max_length    ,				truncation=snake_case    ,				return_tensors='pt'    ,				)
          _UpperCAmelCase  :  Union[str, Any]				       =text_inputs['input_ids'].to(snake_case)

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

          _UpperCAmelCase  :  List[str]				       =prompt_embeds

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

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






       def      lowerCAmelCase						(     self)		->    Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[int]				       =self.get_dummy_components()
          _UpperCAmelCase  :  Optional[int]				       =AudioLDMPipeline(**snake_case)
          _UpperCAmelCase  :  Any				       =audioldm_pipe.to(snake_case)
          _UpperCAmelCase  :  Any				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  Optional[int]				       =self.get_dummy_inputs(snake_case)
          _UpperCAmelCase  :  Union[str, Any]				       =3 * ['this is a negative prompt']
          _UpperCAmelCase  :  Tuple				       =negative_prompt
          _UpperCAmelCase  :  List[str]				       =3 * [inputs['prompt']]

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

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

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

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

             embeds.append(snake_case)

          _UpperCAmelCase       ,			_UpperCAmelCase  :  Tuple				       =embeds

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

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






       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[Any]				       ='cpu'  # ensure determinism for the device-dependent torch.Generator
          _UpperCAmelCase  :  Optional[Any]				       =self.get_dummy_components()
          _UpperCAmelCase  :  Dict				       =PNDMScheduler(skip_prk_steps=snake_case)
          _UpperCAmelCase  :  Tuple				       =AudioLDMPipeline(**snake_case)
          _UpperCAmelCase  :  Any				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  Optional[int]				       =self.get_dummy_inputs(snake_case)
          _UpperCAmelCase  :  Optional[Any]				       ='egg cracking'
          _UpperCAmelCase  :  Optional[Any]				       =audioldm_pipe(**snake_case    ,				negative_prompt=snake_case)
          _UpperCAmelCase  :  Optional[int]				       =output.audios[0]

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

          _UpperCAmelCase  :  str				       =audio[:1_0]
          _UpperCAmelCase  :  int				       =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      lowerCAmelCase						(     self)		->    Dict:




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

          _UpperCAmelCase  :  Optional[Any]				       ='A hammer hitting a wooden surface'

          # test num_waveforms_per_prompt=1 (default)
          _UpperCAmelCase  :  Tuple				       =audioldm_pipe(snake_case    ,				num_inference_steps=2).audios

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

          # test num_waveforms_per_prompt=1 (default) for batch of prompts
          _UpperCAmelCase  :  Any				       =2
          _UpperCAmelCase  :  Optional[int]				       =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  :  Optional[int]				       =audioldm_pipe(snake_case    ,				num_inference_steps=2    ,				num_waveforms_per_prompt=snake_case).audios

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

          # test num_waveforms_per_prompt for batch of prompts
          _UpperCAmelCase  :  Tuple				       =2
          _UpperCAmelCase  :  Union[str, Any]				       =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, 2_5_6)






       def      lowerCAmelCase						(     self)		->    str:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[Any]				       ='cpu'  # ensure determinism for the device-dependent torch.Generator
          _UpperCAmelCase  :  str				       =self.get_dummy_components()
          _UpperCAmelCase  :  List[str]				       =AudioLDMPipeline(**snake_case)
          _UpperCAmelCase  :  Optional[int]				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)
          _UpperCAmelCase  :  List[Any]				       =audioldm_pipe.vocoder.config.sampling_rate

          _UpperCAmelCase  :  Union[str, Any]				       =self.get_dummy_inputs(snake_case)
          _UpperCAmelCase  :  str				       =audioldm_pipe(audio_length_in_s=0.0_16    ,				**snake_case)
          _UpperCAmelCase  :  Union[str, Any]				       =output.audios[0]

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

          _UpperCAmelCase  :  Any				       =audioldm_pipe(audio_length_in_s=0.0_32    ,				**snake_case)
          _UpperCAmelCase  :  Optional[Any]				       =output.audios[0]

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






       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  List[Any]				       =self.get_dummy_components()
          _UpperCAmelCase  :  str				       =AudioLDMPipeline(**snake_case)
          _UpperCAmelCase  :  Any				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  Any				       =['hey']

          _UpperCAmelCase  :  Dict				       =audioldm_pipe(snake_case    ,				num_inference_steps=1)
          _UpperCAmelCase  :  str				       =output.audios.shape
          assert audio_shape == (1, 2_5_6)

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






       def      lowerCAmelCase						(     self)		->    Tuple:




          '''simple docstring'''
          self._test_attention_slicing_forward_pass(test_mean_pixel_difference=snake_case)






       def      lowerCAmelCase						(     self)		->    str:




          '''simple docstring'''
          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      lowerCAmelCase						(     self)		->    str:




          '''simple docstring'''
          self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case)






@slow
class    __magic_name__				(						unittest.TestCase							):






       def      lowerCAmelCase						(     self)		->    Optional[int]:




          '''simple docstring'''
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case="cpu"    ,				snake_case=torch.floataa    ,				snake_case=0)		->    Any:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       =torch.Generator(device=snake_case).manual_seed(snake_case)
          _UpperCAmelCase  :  List[Any]				       =np.random.RandomState(snake_case).standard_normal((1, 8, 1_2_8, 1_6))
          _UpperCAmelCase  :  Tuple				       =torch.from_numpy(snake_case).to(device=snake_case    ,				dtype=snake_case)
          _UpperCAmelCase  :  Optional[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)		->    List[str]:




          '''simple docstring'''
          _UpperCAmelCase  :  Optional[Any]				       =AudioLDMPipeline.from_pretrained('cvssp/audioldm')
          _UpperCAmelCase  :  str				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)

          _UpperCAmelCase  :  Dict				       =self.get_inputs(snake_case)
          _UpperCAmelCase  :  Optional[Any]				       =2_5
          _UpperCAmelCase  :  Union[str, Any]				       =audioldm_pipe(**snake_case).audios[0]

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

          _UpperCAmelCase  :  str				       =audio[7_7_2_3_0:7_7_2_4_0]
          _UpperCAmelCase  :  Any				       =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])
          _UpperCAmelCase  :  Optional[Any]				       =np.abs(expected_slice - audio_slice).max()
          assert max_diff < 1E-2






       def      lowerCAmelCase						(     self)		->    Dict:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       =AudioLDMPipeline.from_pretrained('cvssp/audioldm')
          _UpperCAmelCase  :  Optional[Any]				       =LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config)
          _UpperCAmelCase  :  Optional[Any]				       =audioldm_pipe.to(snake_case)
          audioldm_pipe.set_progress_bar_config(disable=snake_case)

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

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

          _UpperCAmelCase  :  str				       =audio[2_7_7_8_0:2_7_7_9_0]
          _UpperCAmelCase  :  List[Any]				       =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])
          _UpperCAmelCase  :  Any				       =np.abs(expected_slice - audio_slice).max()
          assert max_diff < 3E-2





'''simple docstring'''






from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
 from PIL import Image

 from ..image_utils import load_image

if is_tf_available():
 import tensorflow as tf

 from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
 from ..tf_utils import stable_softmax

if is_torch_available():
 from ..models.auto.modeling_auto import MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING

lowercase				=logging.get_logger(__name__)






@add_end_docstrings(lowerCAmelCase							)
class    __magic_name__				(						lowerCAmelCase							):






       def __init__(     self    ,				*snake_case    ,				**snake_case)		->    Dict:




          '''simple docstring'''
          super().__init__(*snake_case    ,				**snake_case)
          requires_backends(self    ,				'vision')
          self.check_model_type(
              TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
              if self.framework == 'tf'
              else MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING)






       def      lowerCAmelCase						(     self    ,				snake_case=None)		->    str:




          '''simple docstring'''
          _UpperCAmelCase  :  Any				       ={}
          if top_k is not None:
             _UpperCAmelCase  :  Optional[int]				       =top_k
          return {}, {}, postprocess_params






       def __call__(     self    ,				snake_case    ,				**snake_case)		->    List[str]:




          '''simple docstring'''
          return super().__call__(snake_case    ,				**snake_case)






       def      lowerCAmelCase						(     self    ,				snake_case)		->    List[str]:




          '''simple docstring'''
          _UpperCAmelCase  :  int				       =load_image(snake_case)
          _UpperCAmelCase  :  Tuple				       =self.image_processor(images=snake_case    ,				return_tensors=self.framework)
          return model_inputs






       def      lowerCAmelCase						(     self    ,				snake_case)		->    Optional[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  List[str]				       =self.model(**snake_case)
          return model_outputs






       def      lowerCAmelCase						(     self    ,				snake_case    ,				snake_case=5)		->    Any:




          '''simple docstring'''
          if top_k > self.model.config.num_labels:
             _UpperCAmelCase  :  Optional[Any]				       =self.model.config.num_labels

          if self.framework == "pt":
             _UpperCAmelCase  :  List[Any]				       =model_outputs.logits.softmax(-1)[0]
             _UpperCAmelCase       ,			_UpperCAmelCase  :  Optional[Any]				       =probs.topk(snake_case)
          elif self.framework == "tf":
             _UpperCAmelCase  :  int				       =stable_softmax(model_outputs.logits    ,				axis=-1)[0]
             _UpperCAmelCase  :  List[Any]				       =tf.math.top_k(snake_case    ,				k=snake_case)
             _UpperCAmelCase       ,			_UpperCAmelCase  :  List[str]				       =topk.values.numpy(), topk.indices.numpy()
          else:
             raise ValueError(f"Unsupported framework: {self.framework}")

          _UpperCAmelCase  :  int				       =scores.tolist()
          _UpperCAmelCase  :  Tuple				       =ids.tolist()
          return [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(snake_case    ,				snake_case)]

