






'''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   __SCREAMING_SNAKE_CASE       (__A							,       unittest.TestCase       ):


  """simple docstring"""


  _a  : Optional[Any]      			=     AudioLDMPipeline
  _a  : List[Any]      			=     TEXT_TO_AUDIO_PARAMS
  _a  : Dict      			=     TEXT_TO_AUDIO_BATCH_PARAMS
  _a  : int      			=     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_						     =			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_						     =			DDIMScheduler(
             beta_start=0.00_085   ,	beta_end=0.012   ,	beta_schedule='scaled_linear'   ,	clip_sample=UpperCamelCase__   ,	set_alpha_to_one=UpperCamelCase__   ,	)
         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=1_000   ,	projection_dim=32   ,	)
         a_						     =			ClapTextModelWithProjection(UpperCamelCase__	)
         a_						     =			RobertaTokenizer.from_pretrained('hf-internal-testing/tiny-random-roberta'   ,	model_max_length=77	)

         a_						     =			SpeechTaHifiGanConfig(
             model_in_dim=8   ,	sampling_rate=16_000   ,	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_						     =			SpeechTaHifiGan(UpperCamelCase__	)

         a_						     =			{
             '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_						     =			torch.manual_seed(UpperCamelCase__	)
         else:
                a_						     =			torch.Generator(device=UpperCamelCase__	).manual_seed(UpperCamelCase__	)
         a_						     =			{
             '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_						     =			'cpu'  # ensure determinism for the device-dependent torch.Generator

         a_						     =			self.get_dummy_components()
         a_						     =			AudioLDMPipeline(**UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			self.get_dummy_inputs(UpperCamelCase__	)
         a_						     =			audioldm_pipe(**UpperCamelCase__	)
         a_						     =			output.audios[0]

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

         a_						     =			audio[:10]
         a_						     =			np.array(
             [-0.0_050, 0.0_050, -0.0_060, 0.0_033, -0.0_026, 0.0_033, -0.0_027, 0.0_033, -0.0_028, 0.0_033]	)

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







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			self.get_dummy_components()
         a_						     =			AudioLDMPipeline(**UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			self.get_dummy_inputs(UpperCamelCase__	)
         a_						     =			3 * [inputs['prompt']]

         # forward
         a_						     =			audioldm_pipe(**UpperCamelCase__	)
         a_						     =			output.audios[0]

         a_						     =			self.get_dummy_inputs(UpperCamelCase__	)
         a_						     =			3 * [inputs.pop('prompt'	)]

         a_						     =			audioldm_pipe.tokenizer(
             UpperCamelCase__   ,	padding='max_length'   ,	max_length=audioldm_pipe.tokenizer.model_max_length   ,	truncation=UpperCamelCase__   ,	return_tensors='pt'   ,	)
         a_						     =			text_inputs['input_ids'].to(UpperCamelCase__	)

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

         a_						     =			prompt_embeds

         # forward
         a_						     =			audioldm_pipe(**UpperCamelCase__	)
         a_						     =			output.audios[0]

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







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			self.get_dummy_components()
         a_						     =			AudioLDMPipeline(**UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			self.get_dummy_inputs(UpperCamelCase__	)
         a_						     =			3 * ['this is a negative prompt']
         a_						     =			negative_prompt
         a_						     =			3 * [inputs['prompt']]

         # forward
         a_						     =			audioldm_pipe(**UpperCamelCase__	)
         a_						     =			output.audios[0]

         a_						     =			self.get_dummy_inputs(UpperCamelCase__	)
         a_						     =			3 * [inputs.pop('prompt'	)]

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

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

                embeds.append(UpperCamelCase__	)

         a_     ,	a_						     =			embeds

         # forward
         a_						     =			audioldm_pipe(**UpperCamelCase__	)
         a_						     =			output.audios[0]

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







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			'cpu'  # ensure determinism for the device-dependent torch.Generator
         a_						     =			self.get_dummy_components()
         a_						     =			PNDMScheduler(skip_prk_steps=UpperCamelCase__	)
         a_						     =			AudioLDMPipeline(**UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			self.get_dummy_inputs(UpperCamelCase__	)
         a_						     =			'egg cracking'
         a_						     =			audioldm_pipe(**UpperCamelCase__   ,	negative_prompt=UpperCamelCase__	)
         a_						     =			output.audios[0]

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

         a_						     =			audio[:10]
         a_						     =			np.array(
             [-0.0_051, 0.0_050, -0.0_060, 0.0_034, -0.0_026, 0.0_033, -0.0_027, 0.0_033, -0.0_028, 0.0_032]	)

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







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			'cpu'  # ensure determinism for the device-dependent torch.Generator
         a_						     =			self.get_dummy_components()
         a_						     =			PNDMScheduler(skip_prk_steps=UpperCamelCase__	)
         a_						     =			AudioLDMPipeline(**UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			'A hammer hitting a wooden surface'

         # test num_waveforms_per_prompt=1 (default)
         a_						     =			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_						     =			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(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_						     =			2
         a_						     =			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_						     =			'cpu'  # ensure determinism for the device-dependent torch.Generator
         a_						     =			self.get_dummy_components()
         a_						     =			AudioLDMPipeline(**UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)
         a_						     =			audioldm_pipe.vocoder.config.sampling_rate

         a_						     =			self.get_dummy_inputs(UpperCamelCase__	)
         a_						     =			audioldm_pipe(audio_length_in_s=0.016   ,	**UpperCamelCase__	)
         a_						     =			output.audios[0]

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

         a_						     =			audioldm_pipe(audio_length_in_s=0.032   ,	**UpperCamelCase__	)
         a_						     =			output.audios[0]

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







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			self.get_dummy_components()
         a_						     =			AudioLDMPipeline(**UpperCamelCase__	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			['hey']

         a_						     =			audioldm_pipe(UpperCamelCase__   ,	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(UpperCamelCase__	).to(UpperCamelCase__	)
         a_						     =			audioldm_pipe(UpperCamelCase__   ,	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    _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   __SCREAMING_SNAKE_CASE       (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_						     =			torch.Generator(device=UpperCamelCase__	).manual_seed(UpperCamelCase__	)
         a_						     =			np.random.RandomState(UpperCamelCase__	).standard_normal((1, 8, 128, 16)	)
         a_						     =			torch.from_numpy(UpperCamelCase__	).to(device=UpperCamelCase__   ,	dtype=UpperCamelCase__	)
         a_						     =			{
             '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_						     =			AudioLDMPipeline.from_pretrained('cvssp/audioldm'	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			self.get_inputs(UpperCamelCase__	)
         a_						     =			25
         a_						     =			audioldm_pipe(**UpperCamelCase__	).audios[0]

         assert audio.ndim == 1
         assert len(UpperCamelCase__	) == 81_920

         a_						     =			audio[77_230:77_240]
         a_						     =			np.array(
             [-0.4_884, -0.4_607, 0.0_023, 0.5_007, 0.5_896, 0.5_151, 0.3_813, -0.0_208, -0.3_687, -0.4_315]	)
         a_						     =			np.abs(expected_slice - audio_slice	).max()
         assert max_diff < 1e-2







  def    _a       (			self	):






         """simple docstring"""


         a_						     =			AudioLDMPipeline.from_pretrained('cvssp/audioldm'	)
         a_						     =			LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config	)
         a_						     =			audioldm_pipe.to(UpperCamelCase__	)
         audioldm_pipe.set_progress_bar_config(disable=UpperCamelCase__	)

         a_						     =			self.get_inputs(UpperCamelCase__	)
         a_						     =			audioldm_pipe(**UpperCamelCase__	).audios[0]

         assert audio.ndim == 1
         assert len(UpperCamelCase__	) == 81_920

         a_						     =			audio[27_780:27_790]
         a_						     =			np.array([-0.2_131, -0.0_873, -0.0_124, -0.0_189, 0.0_569, 0.1_373, 0.1_883, 0.2_886, 0.3_297, 0.2_212]	)
         a_						     =			np.abs(expected_slice - audio_slice	).max()
         assert max_diff < 3e-2












'''simple docstring'''







import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding




class   __SCREAMING_SNAKE_CASE       (__A       ):


  """simple docstring"""


  _a  : List[Any]      			=     ['''image_processor''', '''tokenizer''']
  _a  : List[Any]      			=     '''ViTImageProcessor'''
  _a  : int      			=     ('''CLIPTokenizer''', '''CLIPTokenizerFast''')







  def __init__(			self   ,	UpperCamelCase__=None   ,	UpperCamelCase__=None   ,	**UpperCamelCase__	):






         """simple docstring"""


         a_						     =			None
         if "feature_extractor" in kwargs:
                warnings.warn(
                    'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
                    ' instead.'   ,	UpperCamelCase__   ,	)
                a_						     =			kwargs.pop('feature_extractor'	)

         a_						     =			image_processor if image_processor is not None else feature_extractor
         if image_processor is None:
                raise ValueError('You need to specify an `image_processor`.'	)
         if tokenizer is None:
                raise ValueError('You need to specify a `tokenizer`.'	)

         super().__init__(UpperCamelCase__   ,	UpperCamelCase__	)







  def __call__(			self   ,	UpperCamelCase__=None   ,	UpperCamelCase__=None   ,	UpperCamelCase__=None   ,	UpperCamelCase__=None   ,	**UpperCamelCase__	):






         """simple docstring"""


         if text is None and visual_prompt is None and images is None:
                raise ValueError('You have to specify either text, visual prompt or images.'	)

         if text is not None and visual_prompt is not None:
                raise ValueError('You have to specify exactly one type of prompt. Either text or visual prompt.'	)

         if text is not None:
                a_						     =			self.tokenizer(UpperCamelCase__   ,	return_tensors=UpperCamelCase__   ,	**UpperCamelCase__	)

         if visual_prompt is not None:
                a_						     =			self.image_processor(UpperCamelCase__   ,	return_tensors=UpperCamelCase__   ,	**UpperCamelCase__	)

         if images is not None:
                a_						     =			self.image_processor(UpperCamelCase__   ,	return_tensors=UpperCamelCase__   ,	**UpperCamelCase__	)

         if visual_prompt is not None and images is not None:
                a_						     =			{
                    'pixel_values': image_features.pixel_values,
                    'conditional_pixel_values': prompt_features.pixel_values,
                }
                return encoding
         elif text is not None and images is not None:
                a_						     =			image_features.pixel_values
                return encoding
         elif text is not None:
                return encoding
         elif visual_prompt is not None:
                a_						     =			{
                    'conditional_pixel_values': prompt_features.pixel_values,
                }
                return encoding
         else:
                return BatchEncoding(data=dict(**UpperCamelCase__	)   ,	tensor_type=UpperCamelCase__	)







  def    _a       (			self   ,	*UpperCamelCase__   ,	**UpperCamelCase__	):






         """simple docstring"""


         return self.tokenizer.batch_decode(*UpperCamelCase__   ,	**UpperCamelCase__	)







  def    _a       (			self   ,	*UpperCamelCase__   ,	**UpperCamelCase__	):






         """simple docstring"""


         return self.tokenizer.decode(*UpperCamelCase__   ,	**UpperCamelCase__	)







  @property
  def    _a       (			self	):






         """simple docstring"""


         warnings.warn(
             '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'   ,	UpperCamelCase__   ,	)
         return self.image_processor_class







  @property
  def    _a       (			self	):






         """simple docstring"""


         warnings.warn(
             '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'   ,	UpperCamelCase__   ,	)
         return self.image_processor




