




"""simple docstring"""






import argparse

import fairseq
import torch
from torch import nn

from transformers import (
    MBartaaTokenizer,
    MBartConfig,
    MBartForCausalLM,
    SpeechEncoderDecoderConfig,
    SpeechEncoderDecoderModel,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaModel,
    logging,
)


logging.set_verbosity_info()
UpperCamelCase__						=			logging.get_logger(__name__)

UpperCamelCase__						=			{
    '''post_extract_proj''': '''feature_projection.projection''',
    '''encoder.pos_conv.0''': '''encoder.pos_conv_embed.conv''',
    '''self_attn.k_proj''': '''encoder.layers.*.attention.k_proj''',
    '''self_attn.v_proj''': '''encoder.layers.*.attention.v_proj''',
    '''self_attn.q_proj''': '''encoder.layers.*.attention.q_proj''',
    '''self_attn.out_proj''': '''encoder.layers.*.attention.out_proj''',
    '''self_attn_layer_norm''': '''encoder.layers.*.layer_norm''',
    '''fc1''': '''encoder.layers.*.feed_forward.intermediate_dense''',
    '''fc2''': '''encoder.layers.*.feed_forward.output_dense''',
    '''final_layer_norm''': '''encoder.layers.*.final_layer_norm''',
    '''encoder.layer_norm''': '''encoder.layer_norm''',
    '''w2v_model.layer_norm''': '''feature_projection.layer_norm''',
    '''quantizer.weight_proj''': '''quantizer.weight_proj''',
    '''quantizer.vars''': '''quantizer.codevectors''',
    '''project_q''': '''project_q''',
    '''final_proj''': '''project_hid''',
    '''w2v_encoder.proj''': '''lm_head''',
    '''mask_emb''': '''masked_spec_embed''',
}
UpperCamelCase__						=			[
    '''lm_head''',
    '''quantizer.weight_proj''',
    '''quantizer.codevectors''',
    '''project_q''',
    '''project_hid''',
]





def 							UpperCAmelCase		(		snake_case			:					str      ,       snake_case			:					List[str]      ,       snake_case			:					Optional[int]      ,       snake_case			:					Any      ,       snake_case			:					Optional[Any]    ):
					for attribute in key.split('''.'''    ):
										_lowerCAmelCase:List[str]									=  getattr(snake_case      ,       snake_case    )

					if weight_type is not None:
										_lowerCAmelCase:List[str]									=  getattr(snake_case      ,       snake_case    ).shape
					else:
										_lowerCAmelCase:Union[str, Any]									=  hf_pointer.shape

					assert hf_shape == value.shape, (
					    F'Shape of hf {key + "." + weight_type if weight_type is not None else ""} is {hf_shape}, but should be'
					    F' {value.shape} for {full_name}'
					)

					if weight_type == "weight":
										_lowerCAmelCase:Any									=  value
					elif weight_type == "weight_g":
										_lowerCAmelCase:Optional[int]									=  value
					elif weight_type == "weight_v":
										_lowerCAmelCase:Union[str, Any]									=  value
					elif weight_type == "bias":
										_lowerCAmelCase:Optional[int]									=  value
					else:
										_lowerCAmelCase:List[Any]									=  value

					logger.info(F'{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}.'    )





def 							UpperCAmelCase		(		snake_case			:					Dict      ,       snake_case			:					Tuple    ):
					_lowerCAmelCase:Union[str, Any]									=  []
					_lowerCAmelCase:str									=  fairseq_model.state_dict()

					_lowerCAmelCase:Union[str, Any]									=  hf_model.feature_extractor
					_lowerCAmelCase:Tuple									=  hf_model.adapter

					for name, value in fairseq_dict.items():
										_lowerCAmelCase:Dict									=  False
										if "conv_layers" in name:
															load_conv_layer(
															    snake_case      ,       snake_case      ,       snake_case      ,       snake_case      ,       hf_model.config.feat_extract_norm == '''group'''      ,       )
															_lowerCAmelCase:Any									=  True
										elif any(x in name for x in ['''adaptor''', '''w2v_encoder.proj.''', '''w2v_proj_ln.''']    ):
															load_adapter(snake_case      ,       snake_case      ,       snake_case      ,       snake_case    )
															_lowerCAmelCase:Tuple									=  True
										else:
															for key, mapped_key in MAPPING.items():
																				if key in name or key.split('''w2v_model.'''    )[-1] == name.split('''.'''    )[0]:
																									_lowerCAmelCase:int									=  True
																									if "*" in mapped_key:
																														_lowerCAmelCase:int									=  name.split(snake_case    )[0].split('''.'''    )[-2]
																														_lowerCAmelCase:Dict									=  mapped_key.replace('''*'''      ,       snake_case    )
																									if "weight_g" in name:
																														_lowerCAmelCase:Dict									=  '''weight_g'''
																									elif "weight_v" in name:
																														_lowerCAmelCase:Tuple									=  '''weight_v'''
																									elif "bias" in name:
																														_lowerCAmelCase:int									=  '''bias'''
																									elif "weight" in name:
																														_lowerCAmelCase:Tuple									=  '''weight'''
																									else:
																														_lowerCAmelCase:int									=  None
																									set_recursively(snake_case      ,       snake_case      ,       snake_case      ,       snake_case      ,       snake_case    )
																				continue
										if not is_used:
															unused_weights.append(snake_case    )

					logger.warning(F'Unused weights: {unused_weights}'    )





def 							UpperCAmelCase		(		snake_case			:					List[str]      ,       snake_case			:					Tuple      ,       snake_case			:					List[str]      ,       snake_case			:					Optional[int]      ,       snake_case			:					Tuple    ):
					_lowerCAmelCase:int									=  full_name.split('''conv_layers.'''    )[-1]
					_lowerCAmelCase:Any									=  name.split('''.'''    )
					_lowerCAmelCase:List[str]									=  int(items[0]    )
					_lowerCAmelCase:List[str]									=  int(items[1]    )

					if type_id == 0:
										if "bias" in name:
															assert value.shape == feature_extractor.conv_layers[layer_id].conv.bias.data.shape, (
															    F'{full_name} has size {value.shape}, but'
															    F' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.'
															)
															_lowerCAmelCase:List[Any]									=  value
															logger.info(F'Feat extract conv layer {layer_id} was initialized from {full_name}.'    )
										elif "weight" in name:
															assert value.shape == feature_extractor.conv_layers[layer_id].conv.weight.data.shape, (
															    F'{full_name} has size {value.shape}, but'
															    F' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.'
															)
															_lowerCAmelCase:Optional[Any]									=  value
															logger.info(F'Feat extract conv layer {layer_id} was initialized from {full_name}.'    )
					elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
										if "bias" in name:
															assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape, (
															    F'{full_name} has size {value.shape}, but {feature_extractor[layer_id].layer_norm.bias.data.shape} was'
															    " found."
															)
															_lowerCAmelCase:Optional[int]									=  value
															logger.info(F'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'    )
										elif "weight" in name:
															assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape, (
															    F'{full_name} has size {value.shape}, but'
															    F' {feature_extractor[layer_id].layer_norm.weight.data.shape} was found.'
															)
															_lowerCAmelCase:Optional[int]									=  value
															logger.info(F'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'    )
					else:
										unused_weights.append(snake_case    )





def 							UpperCAmelCase		(		snake_case			:					Union[str, Any]      ,       snake_case			:					Optional[Any]      ,       snake_case			:					Any      ,       snake_case			:					Union[str, Any]    ):
					_lowerCAmelCase:int									=  full_name.split('''adaptor.'''    )[-1]
					_lowerCAmelCase:Tuple									=  name.split('''.'''    )

					if items[1].isdigit():
										_lowerCAmelCase:List[str]									=  int(items[1]    )
					else:
										_lowerCAmelCase:List[str]									=  None

					if "adaptor" not in full_name:
										if "proj_ln" in full_name:
															# has to be layer norm
															if "bias" in name:
																				assert (
																				    value.shape == adapter.proj_layer_norm.bias.data.shape
																				), F'{full_name} has size {value.shape}, but {adapter.proj_layer_norm.bias.data.shape} was found.'
																				_lowerCAmelCase:Optional[int]									=  value
																				logger.info(F'Adapter proj layer norm bias was initialized from {full_name}.'    )
															if "weight" in name:
																				assert (
																				    value.shape == adapter.proj_layer_norm.weight.data.shape
																				), F'{full_name} has size {value.shape}, but {adapter.proj_layer_norm.weight.data.shape} was found.'
																				_lowerCAmelCase:List[Any]									=  value
										else:
															# has to be projection layer
															if "bias" in name:
																				assert (
																				    value.shape == adapter.proj.bias.data.shape
																				), F'{full_name} has size {value.shape}, but {adapter.proj.bias.data.shape} was found.'
																				_lowerCAmelCase:List[Any]									=  value
																				logger.info(F'Adapter proj layer bias was initialized from {full_name}.'    )
															if "weight" in name:
																				assert (
																				    value.shape == adapter.proj.weight.data.shape
																				), F'{full_name} has size {value.shape}, but {adapter.proj.weight.data.shape} was found.'
																				_lowerCAmelCase:List[Any]									=  value
																				logger.info(F'Adapter proj layer weight was initialized from {full_name}.'    )
					elif isinstance(snake_case      ,       snake_case    ):
										if "bias" in name:
															assert (
															    value.shape == adapter.layers[layer_id].conv.bias.data.shape
															), F'{full_name} has size {value.shape}, but {adapter.layers[layer_id].conv.bias.data.shape} was found.'
															_lowerCAmelCase:Tuple									=  value
															logger.info(F'Adapter layer {layer_id} bias was initialized from {full_name}.'    )
										elif "weight" in name:
															assert (
															    value.shape == adapter.layers[layer_id].conv.weight.data.shape
															), F'{full_name} has size {value.shape}, but {adapter.layers[layer_id].conv.weight.data.shape} was found.'
															_lowerCAmelCase:Optional[Any]									=  value
															logger.info(F'Adapter layer {layer_id} bias was initialized from {full_name}.'    )
					else:
										unused_weights.append(snake_case    )





def 							UpperCAmelCase		(		snake_case			:					List[Any]    ):
					_lowerCAmelCase							,   _lowerCAmelCase:Dict									=  emb.weight.shape
					_lowerCAmelCase:List[Any]									=  nn.Linear(snake_case      ,       snake_case      ,       bias=snake_case    )
					_lowerCAmelCase:Union[str, Any]									=  emb.weight.data
					return lin_layer





@torch.no_grad()
def 							UpperCAmelCase		(		snake_case			:					str      ,       snake_case			:					Dict      ,       snake_case			:					int      ,       snake_case			:					int      ,       snake_case			:					Optional[Any]      ,       snake_case			:					Optional[Any]      ,       snake_case			:					Any      ,       snake_case			:					Optional[Any]      ,       snake_case			:					List[Any]      ,       snake_case			:					str      ,       snake_case			:					int      ,       ):
					_lowerCAmelCase:List[Any]									=  WavaVecaConfig.from_pretrained(
					    snake_case      ,       add_adapter=snake_case      ,       adapter_stride=snake_case      ,       adapter_kernel_size=snake_case      ,       use_auth_token=snake_case      ,       output_hidden_size=snake_case      ,       )
					_lowerCAmelCase:Tuple									=  MBartConfig.from_pretrained(snake_case    )

					# load model
					_lowerCAmelCase							,   _lowerCAmelCase							,   _lowerCAmelCase:Any									=  fairseq.checkpoint_utils.load_model_ensemble_and_task(
					    [checkpoint_path]      ,       arg_overrides={
					        '''config_yaml''': config_yaml_path,
					        '''data''': '''/'''.join(dict_path.split('''/'''    )[:-1]    ),
					        '''w2v_path''': checkpoint_path,
					        '''load_pretrained_decoder_from''': None,
					    }      ,       )
					_lowerCAmelCase:Union[str, Any]									=  model[0].eval()

					# load feature extractor
					_lowerCAmelCase:List[str]									=  WavaVecaFeatureExtractor.from_pretrained(snake_case      ,       use_auth_token=snake_case    )

					# set weights for wav2vec2 encoder
					_lowerCAmelCase:Optional[Any]									=  WavaVecaModel(snake_case    )

					recursively_load_weights_wavaveca(model.encoder      ,       snake_case    )

					# load decoder weights
					_lowerCAmelCase:List[Any]									=  MBartForCausalLM(snake_case    )
					_lowerCAmelCase							,   _lowerCAmelCase:str									=  hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict()      ,       strict=snake_case    )
					logger.warning(F'The following keys are missing when loading the decoder weights: {missing_keys}'    )
					logger.warning(F'The following keys are unexpected when loading the decoder weights: {unexpected_keys}'    )

					_lowerCAmelCase:Union[str, Any]									=  SpeechEncoderDecoderModel(encoder=snake_case      ,       decoder=snake_case    )
					_lowerCAmelCase:int									=  False

					_lowerCAmelCase:Tuple									=  MBartaaTokenizer(snake_case    )
					tokenizer.save_pretrained(snake_case    )

					_lowerCAmelCase:Any									=  hf_wavavec.config.to_dict()
					_lowerCAmelCase:Any									=  tokenizer.pad_token_id
					_lowerCAmelCase:Dict									=  tokenizer.bos_token_id
					_lowerCAmelCase:List[Any]									=  tokenizer.eos_token_id
					_lowerCAmelCase:str									=  '''mbart50'''
					_lowerCAmelCase:Tuple									=  '''wav2vec2'''

					_lowerCAmelCase:List[str]									=  tokenizer.eos_token_id
					_lowerCAmelCase:Any									=  250004
					_lowerCAmelCase:int									=  tokenizer.eos_token_id

					_lowerCAmelCase:List[Any]									=  SpeechEncoderDecoderConfig.from_dict(snake_case    )

					hf_wavavec.save_pretrained(snake_case    )
					feature_extractor.save_pretrained(snake_case    )


if __name__ == "__main__":
				UpperCamelCase__						=			argparse.ArgumentParser()
				parser.add_argument('''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the output PyTorch model.''')
				parser.add_argument('''--checkpoint_path''', default=None, type=str, help='''Path to fairseq checkpoint''')
				parser.add_argument('''--dict_path''', default=None, type=str, help='''Path to dict of fine-tuned model''')
				parser.add_argument('''--config_yaml_path''', default=None, type=str, help='''Path to yaml file of fine-tuned model''')
				parser.add_argument(
				    '''--encoder_config_path''',
				    default='''facebook/wav2vec2-xls-r-1b''',
				    type=str,
				    help='''Path to hf encoder wav2vec2 checkpoint config''',
				)
				parser.add_argument(
				    '''--decoder_config_path''',
				    default='''facebook/mbart-large-50-one-to-many-mmt''',
				    type=str,
				    help='''Path to hf decoder checkpoint config''',
				)
				parser.add_argument('''--add_adapter''', default=True, type=bool, help='''whethere to add model adapter layers''')
				parser.add_argument('''--adapter_stride''', default=2, type=int, help='''stride of adapter layers''')
				parser.add_argument('''--adapter_kernel_size''', default=3, type=int, help='''kernel size of adapter layers''')
				parser.add_argument('''--encoder_output_dim''', default=1_0_2_4, type=int, help='''encoder output dim''')
				parser.add_argument('''--start_token_id''', default=2_5_0_0_0_4, type=int, help='''`decoder_start_token_id` of model config''')

				UpperCamelCase__						=			parser.parse_args()
				convert_wavaveca_checkpoint(
				    args.checkpoint_path,
				    args.pytorch_dump_folder_path,
				    args.dict_path,
				    args.config_yaml_path,
				    encoder_config_path=args.encoder_config_path,
				    decoder_config_path=args.decoder_config_path,
				    add_adapter=args.add_adapter,
				    adapter_kernel_size=args.adapter_kernel_size,
				    adapter_stride=args.adapter_stride,
				    decoder_start_token_id=args.start_token_id,
				    encoder_output_dim=args.encoder_output_dim,
				)












"""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   a__			(       UpperCamelCase_					,	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       __UpperCamelCase     (							self      :	int)						->       Any:




											"""simple docstring"""







											torch.manual_seed(0)
											_lowerCAmelCase:int									=  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__     ,)
											_lowerCAmelCase:Optional[int]									=  DDIMScheduler(
											    beta_start=0.00085     ,beta_end=0.012     ,beta_schedule='''scaled_linear'''     ,clip_sample=a__     ,set_alpha_to_one=a__     ,)
											torch.manual_seed(0)
											_lowerCAmelCase:Optional[Any]									=  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)
											_lowerCAmelCase:Dict									=  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     ,)
											_lowerCAmelCase:str									=  ClapTextModelWithProjection(a__)
											_lowerCAmelCase:Dict									=  RobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-random-roberta'''     ,model_max_length=77)

											_lowerCAmelCase: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=a__     ,)

											_lowerCAmelCase:List[Any]									=  SpeechTaHifiGan(a__)

											_lowerCAmelCase:Optional[Any]									=  {
											    '''unet''': unet,
											    '''scheduler''': scheduler,
											    '''vae''': vae,
											    '''text_encoder''': text_encoder,
											    '''tokenizer''': tokenizer,
											    '''vocoder''': vocoder,
											}
											return components





						def       __UpperCamelCase     (							self      :	List[Any]     ,a__      :	int     ,a__      :	Tuple=0)						->       Optional[int]:




											"""simple docstring"""







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





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




											"""simple docstring"""







											_lowerCAmelCase:str									=  '''cpu'''  # ensure determinism for the device-dependent torch.Generator

											_lowerCAmelCase:Dict									=  self.get_dummy_components()
											_lowerCAmelCase:List[str]									=  AudioLDMPipeline(**a__)
											_lowerCAmelCase:Optional[int]									=  audioldm_pipe.to(a__)
											audioldm_pipe.set_progress_bar_config(disable=a__)

											_lowerCAmelCase:str									=  self.get_dummy_inputs(a__)
											_lowerCAmelCase:Optional[Any]									=  audioldm_pipe(**a__)
											_lowerCAmelCase:List[str]									=  output.audios[0]

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

											_lowerCAmelCase:List[str]									=  audio[:10]
											_lowerCAmelCase:Dict									=  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       __UpperCamelCase     (							self      :	Tuple)						->       Tuple:




											"""simple docstring"""







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

											_lowerCAmelCase:Dict									=  self.get_dummy_inputs(a__)
											_lowerCAmelCase:List[Any]									=  3 * [inputs['''prompt''']]

											# forward
											_lowerCAmelCase:Dict									=  audioldm_pipe(**a__)
											_lowerCAmelCase:Union[str, Any]									=  output.audios[0]

											_lowerCAmelCase:Tuple									=  self.get_dummy_inputs(a__)
											_lowerCAmelCase:Optional[Any]									=  3 * [inputs.pop('''prompt''')]

											_lowerCAmelCase:Tuple									=  audioldm_pipe.tokenizer(
											    a__     ,padding='''max_length'''     ,max_length=audioldm_pipe.tokenizer.model_max_length     ,truncation=a__     ,return_tensors='''pt'''     ,)
											_lowerCAmelCase:Optional[int]									=  text_inputs['''input_ids'''].to(a__)

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

											_lowerCAmelCase:int									=  prompt_embeds

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

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





						def       __UpperCamelCase     (							self      :	Union[str, Any])						->       List[str]:




											"""simple docstring"""







											_lowerCAmelCase:Any									=  self.get_dummy_components()
											_lowerCAmelCase:str									=  AudioLDMPipeline(**a__)
											_lowerCAmelCase:int									=  audioldm_pipe.to(a__)
											_lowerCAmelCase:Any									=  audioldm_pipe.to(a__)
											audioldm_pipe.set_progress_bar_config(disable=a__)

											_lowerCAmelCase:Any									=  self.get_dummy_inputs(a__)
											_lowerCAmelCase:Tuple									=  3 * ['''this is a negative prompt''']
											_lowerCAmelCase:str									=  negative_prompt
											_lowerCAmelCase:List[Any]									=  3 * [inputs['''prompt''']]

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

											_lowerCAmelCase:Tuple									=  self.get_dummy_inputs(a__)
											_lowerCAmelCase:Tuple									=  3 * [inputs.pop('''prompt''')]

											_lowerCAmelCase:Tuple									=  []
											for p in [prompt, negative_prompt]:
																_lowerCAmelCase:str									=  audioldm_pipe.tokenizer(
																    a__     ,padding='''max_length'''     ,max_length=audioldm_pipe.tokenizer.model_max_length     ,truncation=a__     ,return_tensors='''pt'''     ,)
																_lowerCAmelCase:Dict									=  text_inputs['''input_ids'''].to(a__)

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

																embeds.append(a__)

											_lowerCAmelCase							,   _lowerCAmelCase:Optional[Any]									=  embeds

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

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





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




											"""simple docstring"""







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

											_lowerCAmelCase:Tuple									=  self.get_dummy_inputs(a__)
											_lowerCAmelCase:int									=  '''egg cracking'''
											_lowerCAmelCase:Union[str, Any]									=  audioldm_pipe(**a__     ,negative_prompt=a__)
											_lowerCAmelCase:Dict									=  output.audios[0]

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

											_lowerCAmelCase:Optional[int]									=  audio[:10]
											_lowerCAmelCase: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       __UpperCamelCase     (							self      :	Optional[Any])						->       List[Any]:




											"""simple docstring"""







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

											_lowerCAmelCase:Tuple									=  '''A hammer hitting a wooden surface'''

											# test num_waveforms_per_prompt=1 (default)
											_lowerCAmelCase:str									=  audioldm_pipe(a__     ,num_inference_steps=2).audios

											assert audios.shape == (1, 256)

											# test num_waveforms_per_prompt=1 (default) for batch of prompts
											_lowerCAmelCase:Dict									=  2
											_lowerCAmelCase:List[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
											_lowerCAmelCase:Any									=  2
											_lowerCAmelCase:Tuple									=  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
											_lowerCAmelCase:str									=  2
											_lowerCAmelCase:List[Any]									=  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       __UpperCamelCase     (							self      :	Optional[Any])						->       Dict:




											"""simple docstring"""







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

											_lowerCAmelCase:Tuple									=  self.get_dummy_inputs(a__)
											_lowerCAmelCase:Optional[Any]									=  audioldm_pipe(audio_length_in_s=0.016     ,**a__)
											_lowerCAmelCase:List[Any]									=  output.audios[0]

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

											_lowerCAmelCase:List[str]									=  audioldm_pipe(audio_length_in_s=0.032     ,**a__)
											_lowerCAmelCase:Optional[int]									=  output.audios[0]

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





						def       __UpperCamelCase     (							self      :	List[Any])						->       int:




											"""simple docstring"""







											_lowerCAmelCase:Optional[Any]									=  self.get_dummy_components()
											_lowerCAmelCase:Tuple									=  AudioLDMPipeline(**a__)
											_lowerCAmelCase:Dict									=  audioldm_pipe.to(a__)
											audioldm_pipe.set_progress_bar_config(disable=a__)

											_lowerCAmelCase:Optional[int]									=  ['''hey''']

											_lowerCAmelCase:List[str]									=  audioldm_pipe(a__     ,num_inference_steps=1)
											_lowerCAmelCase:Tuple									=  output.audios.shape
											assert audio_shape == (1, 256)

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




											"""simple docstring"""







											self._test_attention_slicing_forward_pass(test_mean_pixel_difference=a__)





						def       __UpperCamelCase     (							self      :	Dict)						->       Optional[int]:




											"""simple docstring"""







											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       __UpperCamelCase     (							self      :	Any)						->       Tuple:




											"""simple docstring"""







											self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=a__)




@slow
class   a__			(       unittest.TestCase  ):





						def       __UpperCamelCase     (							self      :	Tuple)						->       Optional[Any]:




											"""simple docstring"""







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





						def       __UpperCamelCase     (							self      :	Union[str, Any]     ,a__      :	Optional[Any]     ,a__      :	int="cpu"     ,a__      :	str=torch.floataa     ,a__      :	Union[str, Any]=0)						->       int:




											"""simple docstring"""







											_lowerCAmelCase:Optional[int]									=  torch.Generator(device=a__).manual_seed(a__)
											_lowerCAmelCase:Tuple									=  np.random.RandomState(a__).standard_normal((1, 8, 128, 16))
											_lowerCAmelCase:List[str]									=  torch.from_numpy(a__).to(device=a__     ,dtype=a__)
											_lowerCAmelCase:Dict									=  {
											    '''prompt''': '''A hammer hitting a wooden surface''',
											    '''latents''': latents,
											    '''generator''': generator,
											    '''num_inference_steps''': 3,
											    '''guidance_scale''': 2.5,
											}
											return inputs





						def       __UpperCamelCase     (							self      :	Dict)						->       int:




											"""simple docstring"""







											_lowerCAmelCase:str									=  AudioLDMPipeline.from_pretrained('''cvssp/audioldm''')
											_lowerCAmelCase:Optional[Any]									=  audioldm_pipe.to(a__)
											audioldm_pipe.set_progress_bar_config(disable=a__)

											_lowerCAmelCase:Optional[int]									=  self.get_inputs(a__)
											_lowerCAmelCase:Optional[Any]									=  25
											_lowerCAmelCase:int									=  audioldm_pipe(**a__).audios[0]

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

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







						def       __UpperCamelCase     (							self      :	List[str])						->       str:




											"""simple docstring"""







											_lowerCAmelCase:Union[str, Any]									=  AudioLDMPipeline.from_pretrained('''cvssp/audioldm''')
											_lowerCAmelCase:List[Any]									=  LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config)
											_lowerCAmelCase:Union[str, Any]									=  audioldm_pipe.to(a__)
											audioldm_pipe.set_progress_bar_config(disable=a__)

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

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

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






