






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






			def       SCREAMING_SNAKE_CASE   (			self   :      str       )      ->      int:
							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=__SCREAMING_SNAKE_CASE	,    )
							a_		:	Optional[int]									=   DDIMScheduler(
							    beta_start=0.0_0085	,    beta_end=0.012	,    beta_schedule='''scaled_linear'''	,    clip_sample=__SCREAMING_SNAKE_CASE	,    set_alpha_to_one=__SCREAMING_SNAKE_CASE	,    )
							torch.manual_seed(0       )
							a_		:	Tuple									=   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_		:	Optional[Any]									=   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(__SCREAMING_SNAKE_CASE       )
							a_		:	Union[str, Any]									=   RobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-random-roberta'''	,    model_max_length=77       )

							a_		:	Union[str, Any]									=   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=__SCREAMING_SNAKE_CASE	,    )

							a_		:	Dict									=   SpeechTaHifiGan(__SCREAMING_SNAKE_CASE       )

							a_		:	str									=   {
							    '''unet''': unet,
							    '''scheduler''': scheduler,
							    '''vae''': vae,
							    '''text_encoder''': text_encoder,
							    '''tokenizer''': tokenizer,
							    '''vocoder''': vocoder,
							}
							return components






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






			def       SCREAMING_SNAKE_CASE   (			self   :      Union[str, Any]       )      ->      Dict:
							a_		:	Optional[int]									=   '''cpu'''  # ensure determinism for the device-dependent torch.Generator

							a_		:	Any									=   self.get_dummy_components()
							a_		:	str									=   AudioLDMPipeline(**__SCREAMING_SNAKE_CASE       )
							a_		:	Tuple									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	str									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	List[Any]									=   audioldm_pipe(**__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[int]									=   output.audios[0]

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

							a_		:	int									=   audio[:10]
							a_		:	Optional[Any]									=   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       SCREAMING_SNAKE_CASE   (			self   :      Dict       )      ->      Optional[int]:
							a_		:	str									=   self.get_dummy_components()
							a_		:	Optional[Any]									=   AudioLDMPipeline(**__SCREAMING_SNAKE_CASE       )
							a_		:	Tuple									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							a_		:	List[Any]									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	Any									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	Union[str, Any]									=   3 * [inputs['''prompt''']]

							# forward
							a_		:	Optional[int]									=   audioldm_pipe(**__SCREAMING_SNAKE_CASE       )
							a_		:	List[Any]									=   output.audios[0]

							a_		:	List[Any]									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	Dict									=   3 * [inputs.pop('''prompt'''       )]

							a_		:	Dict									=   audioldm_pipe.tokenizer(
							    __SCREAMING_SNAKE_CASE	,    padding='''max_length'''	,    max_length=audioldm_pipe.tokenizer.model_max_length	,    truncation=__SCREAMING_SNAKE_CASE	,    return_tensors='''pt'''	,    )
							a_		:	List[str]									=   text_inputs['''input_ids'''].to(__SCREAMING_SNAKE_CASE       )

							a_		:	Dict									=   audioldm_pipe.text_encoder(
							    __SCREAMING_SNAKE_CASE	,    )
							a_		:	List[str]									=   prompt_embeds.text_embeds
							# additional L_2 normalization over each hidden-state
							a_		:	int									=   F.normalize(__SCREAMING_SNAKE_CASE	,    dim=-1       )

							a_		:	Optional[Any]									=   prompt_embeds

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

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






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[Any]       )      ->      Tuple:
							a_		:	Any									=   self.get_dummy_components()
							a_		:	int									=   AudioLDMPipeline(**__SCREAMING_SNAKE_CASE       )
							a_		:	int									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							a_		:	List[str]									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	Any									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	int									=   3 * ['''this is a negative prompt''']
							a_		:	Dict									=   negative_prompt
							a_		:	int									=   3 * [inputs['''prompt''']]

							# forward
							a_		:	Optional[int]									=   audioldm_pipe(**__SCREAMING_SNAKE_CASE       )
							a_		:	Any									=   output.audios[0]

							a_		:	str									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[Any]									=   3 * [inputs.pop('''prompt'''       )]

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

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

											embeds.append(__SCREAMING_SNAKE_CASE       )

							a_					,      a_		:	Optional[Any]									=   embeds

							# forward
							a_		:	Optional[Any]									=   audioldm_pipe(**__SCREAMING_SNAKE_CASE       )
							a_		:	List[Any]									=   output.audios[0]

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






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[Any]       )      ->      Optional[int]:
							a_		:	Optional[Any]									=   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
							a_		:	str									=   self.get_dummy_components()
							a_		:	Any									=   PNDMScheduler(skip_prk_steps=__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[Any]									=   AudioLDMPipeline(**__SCREAMING_SNAKE_CASE       )
							a_		:	Any									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	Any									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	Any									=   '''egg cracking'''
							a_		:	str									=   audioldm_pipe(**__SCREAMING_SNAKE_CASE	,    negative_prompt=__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[Any]									=   output.audios[0]

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

							a_		:	Any									=   audio[:10]
							a_		:	Optional[Any]									=   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       SCREAMING_SNAKE_CASE   (			self   :      Dict       )      ->      Any:
							a_		:	Dict									=   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
							a_		:	List[Any]									=   self.get_dummy_components()
							a_		:	Any									=   PNDMScheduler(skip_prk_steps=__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[Any]									=   AudioLDMPipeline(**__SCREAMING_SNAKE_CASE       )
							a_		:	Any									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	Optional[int]									=   '''A hammer hitting a wooden surface'''

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

							assert audios.shape == (1, 256)

							# test num_waveforms_per_prompt=1 (default) for batch of prompts
							a_		:	List[Any]									=   2
							a_		:	Dict									=   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_		:	Tuple									=   2
							a_		:	Union[str, Any]									=   audioldm_pipe(__SCREAMING_SNAKE_CASE	,    num_inference_steps=2	,    num_waveforms_per_prompt=__SCREAMING_SNAKE_CASE       ).audios

							assert audios.shape == (num_waveforms_per_prompt, 256)

							# test num_waveforms_per_prompt for batch of prompts
							a_		:	str									=   2
							a_		:	Any									=   audioldm_pipe(
							    [prompt] * batch_size	,    num_inference_steps=2	,    num_waveforms_per_prompt=__SCREAMING_SNAKE_CASE       ).audios

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






			def       SCREAMING_SNAKE_CASE   (			self   :      str       )      ->      Tuple:
							a_		:	Optional[Any]									=   '''cpu'''  # ensure determinism for the device-dependent torch.Generator
							a_		:	List[Any]									=   self.get_dummy_components()
							a_		:	str									=   AudioLDMPipeline(**__SCREAMING_SNAKE_CASE       )
							a_		:	int									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )
							a_		:	str									=   audioldm_pipe.vocoder.config.sampling_rate

							a_		:	Any									=   self.get_dummy_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	List[str]									=   audioldm_pipe(audio_length_in_s=0.016	,    **__SCREAMING_SNAKE_CASE       )
							a_		:	List[str]									=   output.audios[0]

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

							a_		:	Optional[int]									=   audioldm_pipe(audio_length_in_s=0.032	,    **__SCREAMING_SNAKE_CASE       )
							a_		:	Tuple									=   output.audios[0]

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






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[Any]       )      ->      List[str]:
							a_		:	Tuple									=   self.get_dummy_components()
							a_		:	int									=   AudioLDMPipeline(**__SCREAMING_SNAKE_CASE       )
							a_		:	Union[str, Any]									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	List[str]									=   ['''hey''']

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

							a_		:	str									=   audioldm_pipe.vocoder.config
							config.model_in_dim *= 2
							a_		:	Any									=   SpeechTaHifiGan(__SCREAMING_SNAKE_CASE       ).to(__SCREAMING_SNAKE_CASE       )
							a_		:	Dict									=   audioldm_pipe(__SCREAMING_SNAKE_CASE	,    num_inference_steps=1       )
							a_		:	Optional[Any]									=   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       SCREAMING_SNAKE_CASE   (			self   :      Any       )      ->      Optional[Any]:
							self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE       )






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[int]       )      ->      Any:
							self._test_inference_batch_single_identical(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE       )







			@unittest.skipIf(
			    torch_device != '''cuda''' or not is_xformers_available()	,    reason='''XFormers attention is only available with CUDA and `xformers` installed'''	,    )
			def       SCREAMING_SNAKE_CASE   (			self   :      List[Any]       )      ->      List[str]:
							self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__SCREAMING_SNAKE_CASE       )





@slow
class 						SCREAMING_SNAKE_CASE       (       unittest.TestCase							):






			def       SCREAMING_SNAKE_CASE   (			self   :      Dict       )      ->      Optional[Any]:
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()






			def       SCREAMING_SNAKE_CASE   (			self   :      Tuple	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]="cpu"	,    __SCREAMING_SNAKE_CASE   :      Any=torch.floataa	,    __SCREAMING_SNAKE_CASE   :      Union[str, Any]=0       )      ->      List[str]:
							a_		:	Dict									=   torch.Generator(device=__SCREAMING_SNAKE_CASE       ).manual_seed(__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[int]									=   np.random.RandomState(__SCREAMING_SNAKE_CASE       ).standard_normal((1, 8, 128, 16)       )
							a_		:	List[str]									=   torch.from_numpy(__SCREAMING_SNAKE_CASE       ).to(device=__SCREAMING_SNAKE_CASE	,    dtype=__SCREAMING_SNAKE_CASE       )
							a_		:	List[Any]									=   {
							    '''prompt''': '''A hammer hitting a wooden surface''',
							    '''latents''': latents,
							    '''generator''': generator,
							    '''num_inference_steps''': 3,
							    '''guidance_scale''': 2.5,
							}
							return inputs






			def       SCREAMING_SNAKE_CASE   (			self   :      Optional[Any]       )      ->      str:
							a_		:	Union[str, Any]									=   AudioLDMPipeline.from_pretrained('''cvssp/audioldm'''       )
							a_		:	str									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	Optional[int]									=   self.get_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	Optional[int]									=   25
							a_		:	Dict									=   audioldm_pipe(**__SCREAMING_SNAKE_CASE       ).audios[0]

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

							a_		:	Any									=   audio[7_7230:7_7240]
							a_		:	List[Any]									=   np.array(
							    [-0.4884, -0.4607, 0.0023, 0.5007, 0.5896, 0.5151, 0.3813, -0.0208, -0.3687, -0.4315]       )
							a_		:	Optional[int]									=   np.abs(expected_slice - audio_slice       ).max()
							assert max_diff < 1e-2







			def       SCREAMING_SNAKE_CASE   (			self   :      Tuple       )      ->      Tuple:
							a_		:	Tuple									=   AudioLDMPipeline.from_pretrained('''cvssp/audioldm'''       )
							a_		:	Tuple									=   LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config       )
							a_		:	int									=   audioldm_pipe.to(__SCREAMING_SNAKE_CASE       )
							audioldm_pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE       )

							a_		:	Optional[Any]									=   self.get_inputs(__SCREAMING_SNAKE_CASE       )
							a_		:	Any									=   audioldm_pipe(**__SCREAMING_SNAKE_CASE       ).audios[0]

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

							a_		:	int									=   audio[2_7780:2_7790]
							a_		:	str									=   np.array([-0.2131, -0.0873, -0.0124, -0.0189, 0.0569, 0.1373, 0.1883, 0.2886, 0.3297, 0.2212]       )
							a_		:	Optional[Any]									=   np.abs(expected_slice - audio_slice       ).max()
							assert max_diff < 3e-2












'''simple docstring'''







# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool





class 						SCREAMING_SNAKE_CASE       (       SCREAMING_SNAKE_CASE_							):
			snake_case__	=					"openai/whisper-base"
			snake_case__	=					(
			    "This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the "
			    "transcribed text."
			)
			snake_case__	=					"transcriber"
			snake_case__	=					WhisperProcessor
			snake_case__	=					WhisperForConditionalGeneration

			snake_case__	=					["audio"]
			snake_case__	=					["text"]






			def       SCREAMING_SNAKE_CASE   (			self   :      str	,    __SCREAMING_SNAKE_CASE   :      Tuple       )      ->      str:
							return self.pre_processor(__SCREAMING_SNAKE_CASE	,    return_tensors='''pt'''       ).input_features






			def       SCREAMING_SNAKE_CASE   (			self   :      List[Any]	,    __SCREAMING_SNAKE_CASE   :      Optional[Any]       )      ->      List[str]:
							return self.model.generate(inputs=__SCREAMING_SNAKE_CASE       )







			def       SCREAMING_SNAKE_CASE   (			self   :      Dict	,    __SCREAMING_SNAKE_CASE   :      Optional[int]       )      ->      Optional[Any]:
							return self.pre_processor.batch_decode(__SCREAMING_SNAKE_CASE	,    skip_special_tokens=__SCREAMING_SNAKE_CASE       )[0]




