





import argparse

import gdown
import numpy as np
import torch
from huggingface_hub import hf_hub_download

from transformers import (
    CLIPTokenizer,
    CLIPTokenizerFast,
    VideoMAEImageProcessor,
    XCLIPConfig,
    XCLIPModel,
    XCLIPProcessor,
    XCLIPTextConfig,
    XCLIPVisionConfig,
)







def    A				(							UpperCAmelCase  ,  UpperCAmelCase				):
					_snake_case       :					int				       =   XCLIPTextConfig()

					# derive patch size from model name
					_snake_case       :					Dict				       =   model_name.find("patch"				)
					_snake_case       :					Optional[Any]				       =   int(model_name[start_idx + len("patch"				) : start_idx + len("patch"				) + 2]				)
					_snake_case       :					int				       =   XCLIPVisionConfig(patch_size=UpperCAmelCase  ,  num_frames=UpperCAmelCase				)

					if "large" in model_name:
										_snake_case       :					str				       =   768
										_snake_case       :					int				       =   3_072
										_snake_case       :					str				       =   12

										_snake_case       :					Optional[Any]				       =   1_024
										_snake_case       :					Tuple				       =   4_096
										_snake_case       :					Any				       =   16
										_snake_case       :					Any				       =   24
										_snake_case       :					Dict				       =   768
										_snake_case       :					Tuple				       =   3_072

					if model_name == "xclip-large-patch14-16-frames":
										_snake_case       :					List[str]				       =   336

					_snake_case       :					Any				       =   XCLIPConfig.from_text_vision_configs(UpperCAmelCase  ,  UpperCAmelCase				)

					if "large" in model_name:
										_snake_case       :					Optional[int]				       =   768

					return config







def    A				(							UpperCAmelCase				):
					# text encoder
					if name == "token_embedding.weight":
										_snake_case       :					List[str]				       =   name.replace("token_embedding.weight"  ,  "text_model.embeddings.token_embedding.weight"				)
					if name == "positional_embedding":
										_snake_case       :					Dict				       =   name.replace("positional_embedding"  ,  "text_model.embeddings.position_embedding.weight"				)
					if "ln_1" in name:
										_snake_case       :					Dict				       =   name.replace("ln_1"  ,  "layer_norm1"				)
					if "ln_2" in name:
										_snake_case       :					Optional[Any]				       =   name.replace("ln_2"  ,  "layer_norm2"				)
					if "c_fc" in name:
										_snake_case       :					Optional[int]				       =   name.replace("c_fc"  ,  "fc1"				)
					if "c_proj" in name:
										_snake_case       :					Optional[Any]				       =   name.replace("c_proj"  ,  "fc2"				)
					if name.startswith("transformer.resblocks"				):
										_snake_case       :					Optional[Any]				       =   name.replace("transformer.resblocks"  ,  "text_model.encoder.layers"				)
					if "attn.out_proj" in name and "message" not in name:
										_snake_case       :					List[Any]				       =   name.replace("attn.out_proj"  ,  "self_attn.out_proj"				)
					if "ln_final" in name:
										_snake_case       :					Optional[Any]				       =   name.replace("ln_final"  ,  "text_model.final_layer_norm"				)
					# visual encoder
					if name == "visual.class_embedding":
										_snake_case       :					List[Any]				       =   name.replace("visual.class_embedding"  ,  "vision_model.embeddings.class_embedding"				)
					if name == "visual.positional_embedding":
										_snake_case       :					Any				       =   name.replace("visual.positional_embedding"  ,  "vision_model.embeddings.position_embedding.weight"				)
					if name.startswith("visual.transformer.resblocks"				):
										_snake_case       :					str				       =   name.replace("visual.transformer.resblocks"  ,  "vision_model.encoder.layers"				)
					if "visual.conv1" in name:
										_snake_case       :					Dict				       =   name.replace("visual.conv1"  ,  "vision_model.embeddings.patch_embedding"				)
					if "visual.ln_pre" in name:
										_snake_case       :					Optional[int]				       =   name.replace("visual.ln_pre"  ,  "vision_model.pre_layernorm"				)
					if "visual.ln_post" in name:
										_snake_case       :					Union[str, Any]				       =   name.replace("visual.ln_post"  ,  "vision_model.post_layernorm"				)
					if "visual.proj" in name:
										_snake_case       :					Optional[Any]				       =   name.replace("visual.proj"  ,  "visual_projection.weight"				)
					if "text_projection" in name:
										_snake_case       :					Dict				       =   name.replace("text_projection"  ,  "text_projection.weight"				)
					# things on top
					if "prompts_visual_proj" in name:
										_snake_case       :					List[Any]				       =   name.replace("prompts_visual_proj"  ,  "prompts_visual_projection"				)
					if "prompts_visual_ln" in name:
										_snake_case       :					Optional[int]				       =   name.replace("prompts_visual_ln"  ,  "prompts_visual_layernorm"				)
					# mit
					if name == "mit.positional_embedding":
										_snake_case       :					Any				       =   name.replace("positional"  ,  "position"				)
					if name.startswith("mit.resblocks"				):
										_snake_case       :					int				       =   name.replace("mit.resblocks"  ,  "mit.encoder.layers"				)
					# prompts generator
					if name.startswith("prompts_generator.norm"				):
										_snake_case       :					Optional[Any]				       =   name.replace("prompts_generator.norm"  ,  "prompts_generator.layernorm"				)

					return name







def    A				(							UpperCAmelCase  ,  UpperCAmelCase				):
					for key in orig_state_dict.copy().keys():
										_snake_case       :					List[Any]				       =   orig_state_dict.pop(UpperCAmelCase				)

										if "attn.in_proj" in key:
															_snake_case       :					int				       =   key.split("."				)
															if key.startswith("visual"				):
																				_snake_case       :					str				       =   key_split[3]
																				_snake_case       :					Any				       =   config.vision_config.hidden_size
																				if "message_attn" in key:
																									if "weight" in key:
																														_snake_case       :					Optional[Any]				       =   val[
																														    :dim, :
																														]
																														_snake_case       :					List[Any]				       =   val[
																														    dim : dim * 2, :
																														]
																														_snake_case       :					str				       =   val[
																														    -dim:, :
																														]
																									else:
																														_snake_case       :					str				       =   val[
																														    :dim
																														]
																														_snake_case       :					Dict				       =   val[
																														    dim : dim * 2
																														]
																														_snake_case       :					List[str]				       =   val[
																														    -dim:
																														]
																				else:
																									if "weight" in key:
																														_snake_case       :					List[Any]				       =   val[
																														    :dim, :
																														]
																														_snake_case       :					Union[str, Any]				       =   val[
																														    dim : dim * 2, :
																														]
																														_snake_case       :					Optional[int]				       =   val[
																														    -dim:, :
																														]
																									else:
																														_snake_case       :					List[str]				       =   val[:dim]
																														_snake_case       :					Any				       =   val[
																														    dim : dim * 2
																														]
																														_snake_case       :					int				       =   val[-dim:]
															elif key.startswith("mit"				):
																				_snake_case       :					Dict				       =   key_split[2]
																				_snake_case       :					str				       =   config.vision_config.mit_hidden_size
																				if "weight" in key:
																									_snake_case       :					int				       =   val[:dim, :]
																									_snake_case       :					List[Any]				       =   val[dim : dim * 2, :]
																									_snake_case       :					int				       =   val[-dim:, :]
																				else:
																									_snake_case       :					Any				       =   val[:dim]
																									_snake_case       :					int				       =   val[dim : dim * 2]
																									_snake_case       :					Dict				       =   val[-dim:]
															else:
																				_snake_case       :					Any				       =   key_split[2]
																				_snake_case       :					Any				       =   config.text_config.hidden_size
																				if "weight" in key:
																									_snake_case       :					List[str]				       =   val[:dim, :]
																									_snake_case       :					Optional[Any]				       =   val[
																									    dim : dim * 2, :
																									]
																									_snake_case       :					Dict				       =   val[-dim:, :]
																				else:
																									_snake_case       :					List[str]				       =   val[:dim]
																									_snake_case       :					Optional[Any]				       =   val[
																									    dim : dim * 2
																									]
																									_snake_case       :					Tuple				       =   val[-dim:]
										else:
															_snake_case       :					str				       =   rename_key(UpperCAmelCase				)
															if new_key_name in ["visual_projection.weight", "text_projection.weight"]:
																				_snake_case       :					Union[str, Any]				       =   val.T
															_snake_case       :					Optional[int]				       =   val

					return orig_state_dict







def    A				(							UpperCAmelCase				):
					if num_frames == 8:
										_snake_case       :					Dict				       =   "eating_spaghetti_8_frames.npy"
					elif num_frames == 16:
										_snake_case       :					Dict				       =   "eating_spaghetti.npy"
					elif num_frames == 32:
										_snake_case       :					str				       =   "eating_spaghetti_32_frames.npy"
					_snake_case       :					Optional[int]				       =   hf_hub_download(
					    repo_id="hf-internal-testing/spaghetti-video"  ,  filename=UpperCAmelCase  ,  repo_type="dataset"  ,  )
					_snake_case       :					List[str]				       =   np.load(UpperCAmelCase				)
					return list(UpperCAmelCase				)







def    A				(							UpperCAmelCase  ,  UpperCAmelCase=None  ,  UpperCAmelCase=False				):
					_snake_case       :					List[Any]				       =   {
					    # fully supervised kinetics-400 checkpoints
					    "xclip-base-patch32": "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/k400_32_8.pth",
					    "xclip-base-patch32-16-frames": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/k400_32_16.pth"
					    ),
					    "xclip-base-patch16": "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/k400_16_8.pth",
					    "xclip-base-patch16-16-frames": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/k400_16_16.pth"
					    ),
					    "xclip-large-patch14": "https://drive.google.com/u/0/uc?id=1NUOImq0o5DlQTST17iIP3vG7DgmHQuCx&amp;export=download&amp;confirm=t&amp;uuid=b26caedc-88e2-473e-830a-9d158b653cdb",
					    "xclip-large-patch14-16-frames": "https://drive.google.com/u/0/uc?id=1FOYgnJc097OJ4lGwtRCCydQyVPJEOH7d&amp;export=download&amp;confirm=t&amp;uuid=538fa810-e671-4050-b385-9a623f89804f",
					    # fully supervised kinetics-600 checkpoints
					    "xclip-base-patch16-kinetics-600": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/k600_16_8.pth"
					    ),
					    "xclip-base-patch16-kinetics-600-16-frames": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/k600_16_16.pth"
					    ),
					    "xclip-large-patch14-kinetics-600": "https://drive.google.com/u/0/uc?id=1FV8C1INuM91sLAN4ImjzePLIlpMSihwV&amp;export=download&amp;confirm=t&amp;uuid=141d4977-4a65-44ae-864f-4b0c19f838be",
					    # few shot
					    "xclip-base-patch16-hmdb-2-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_hmdb_2.pth"
					    ),
					    "xclip-base-patch16-hmdb-4-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_hmdb_4.pth"
					    ),
					    "xclip-base-patch16-hmdb-8-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_hmdb_8.pth"
					    ),
					    "xclip-base-patch16-hmdb-16-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_hmdb_16.pth"
					    ),
					    "xclip-base-patch16-ucf-2-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_ucf_2.pth"
					    ),
					    "xclip-base-patch16-ucf-4-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_ucf_4.pth"
					    ),
					    "xclip-base-patch16-ucf-8-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_ucf_8.pth"
					    ),
					    "xclip-base-patch16-ucf-16-shot": (
					        "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/few_ucf_16.pth"
					    ),
					    # zero shot
					    "xclip-base-patch16-zero-shot": "https://github.com/nbl97/X-CLIP_Model_Zoo/releases/download/v1.0/zero.pth",
					}

					_snake_case       :					str				       =   model_to_url[model_name]
					_snake_case       :					List[Any]				       =   8
					if "16-frames" in model_name:
										_snake_case       :					Optional[int]				       =   16
					elif "shot" in model_name:
										_snake_case       :					List[str]				       =   32

					_snake_case       :					int				       =   get_xclip_config(UpperCAmelCase  ,  UpperCAmelCase				)
					_snake_case       :					Optional[int]				       =   XCLIPModel(UpperCAmelCase				)
					model.eval()

					if "drive" in checkpoint_url:
										_snake_case       :					Union[str, Any]				       =   "pytorch_model.bin"
										gdown.cached_download(UpperCAmelCase  ,  UpperCAmelCase  ,  quiet=UpperCAmelCase				)
										_snake_case       :					Tuple				       =   torch.load(UpperCAmelCase  ,  map_location="cpu"				)["model"]
					else:
										_snake_case       :					Dict				       =   torch.hub.load_state_dict_from_url(UpperCAmelCase				)["model"]

					_snake_case       :					Dict				       =   convert_state_dict(UpperCAmelCase  ,  UpperCAmelCase				)

					_snake_case       :					Optional[int]				       =   XCLIPModel(UpperCAmelCase				)
					_snake_case       :					Tuple				       =   model.load_state_dict(UpperCAmelCase  ,  strict=UpperCAmelCase				)
					assert missing_keys == ["text_model.embeddings.position_ids", "vision_model.embeddings.position_ids"]
					model.eval()

					_snake_case       :					str				       =   336 if model_name == "xclip-large-patch14-16-frames" else 224
					_snake_case       :					Optional[Any]				       =   VideoMAEImageProcessor(size=UpperCAmelCase				)
					_snake_case       :					Dict				       =   CLIPTokenizer.from_pretrained("openai/clip-vit-base-patch32"				)
					_snake_case       :					Optional[int]				       =   CLIPTokenizerFast.from_pretrained("openai/clip-vit-base-patch32"				)
					_snake_case       :					Optional[Any]				       =   XCLIPProcessor(image_processor=UpperCAmelCase  ,  tokenizer=UpperCAmelCase				)

					_snake_case       :					Tuple				       =   prepare_video(UpperCAmelCase				)
					_snake_case       :					str				       =   processor(
					    text=["playing sports", "eating spaghetti", "go shopping"]  ,  videos=UpperCAmelCase  ,  return_tensors="pt"  ,  padding=UpperCAmelCase				)

					print("Shape of pixel values:"  ,  inputs.pixel_values.shape				)

					with torch.no_grad():
										_snake_case       :					Tuple				       =   model(**UpperCAmelCase				)

					# Verify outputs
					_snake_case       :					List[Any]				       =   outputs.logits_per_video
					_snake_case       :					Any				       =   logits_per_video.softmax(dim=1				)
					print("Probs:"  ,  UpperCAmelCase				)
					# kinetics-400
					if model_name == "xclip-base-patch32":
										_snake_case       :					Union[str, Any]				       =   torch.tensor([[0.0_0_1_9, 0.9_9_5_1, 0.0_0_3_0]]				)
					elif model_name == "xclip-base-patch32-16-frames":
										_snake_case       :					Tuple				       =   torch.tensor([[7.09_99e-04, 9.98_83e-01, 4.55_80e-04]]				)
					elif model_name == "xclip-base-patch16":
										_snake_case       :					str				       =   torch.tensor([[0.0_0_8_3, 0.9_6_8_1, 0.0_2_3_6]]				)
					elif model_name == "xclip-base-patch16-16-frames":
										_snake_case       :					List[Any]				       =   torch.tensor([[7.69_37e-04, 9.97_28e-01, 1.94_73e-03]]				)
					elif model_name == "xclip-large-patch14":
										_snake_case       :					Dict				       =   torch.tensor([[0.0_0_6_2, 0.9_8_6_4, 0.0_0_7_5]]				)
					elif model_name == "xclip-large-patch14-16-frames":
										_snake_case       :					Tuple				       =   torch.tensor([[3.38_77e-04, 9.99_37e-01, 2.88_88e-04]]				)
					# kinetics-600
					elif model_name == "xclip-base-patch16-kinetics-600":
										_snake_case       :					Union[str, Any]				       =   torch.tensor([[0.0_5_5_5, 0.8_9_1_4, 0.0_5_3_1]]				)
					elif model_name == "xclip-base-patch16-kinetics-600-16-frames":
										_snake_case       :					List[Any]				       =   torch.tensor([[3.85_54e-04, 9.99_29e-01, 3.27_54e-04]]				)
					elif model_name == "xclip-large-patch14-kinetics-600":
										_snake_case       :					Tuple				       =   torch.tensor([[0.0_0_3_6, 0.9_9_2_0, 0.0_0_4_5]]				)
					# few shot
					elif model_name == "xclip-base-patch16-hmdb-2-shot":
										_snake_case       :					Tuple				       =   torch.tensor([[7.18_90e-06, 9.99_94e-01, 5.65_59e-05]]				)
					elif model_name == "xclip-base-patch16-hmdb-4-shot":
										_snake_case       :					Any				       =   torch.tensor([[1.03_20e-05, 9.99_93e-01, 6.24_35e-05]]				)
					elif model_name == "xclip-base-patch16-hmdb-8-shot":
										_snake_case       :					List[str]				       =   torch.tensor([[4.13_77e-06, 9.99_90e-01, 9.83_86e-05]]				)
					elif model_name == "xclip-base-patch16-hmdb-16-shot":
										_snake_case       :					Union[str, Any]				       =   torch.tensor([[4.13_47e-05, 9.99_62e-01, 3.34_11e-04]]				)
					elif model_name == "xclip-base-patch16-ucf-2-shot":
										_snake_case       :					Union[str, Any]				       =   torch.tensor([[8.58_57e-05, 9.99_28e-01, 6.32_91e-04]]				)
					elif model_name == "xclip-base-patch16-ucf-4-shot":
										_snake_case       :					List[str]				       =   torch.tensor([[8.58_57e-05, 9.99_28e-01, 6.32_91e-04]]				)
					elif model_name == "xclip-base-patch16-ucf-8-shot":
										_snake_case       :					Optional[Any]				       =   torch.tensor([[0.0_0_2_7, 0.9_9_0_4, 0.0_0_7_0]]				)
					elif model_name == "xclip-base-patch16-ucf-16-shot":
										_snake_case       :					Optional[Any]				       =   torch.tensor([[9.82_19e-04, 9.95_93e-01, 3.08_63e-03]]				)
					# zero shot
					elif model_name == "xclip-base-patch16-zero-shot":
										_snake_case       :					Any				       =   torch.tensor([[3.50_82e-04, 9.97_85e-01, 1.79_66e-03]]				)
					else:
										raise ValueError(F"""Model name {model_name} not supported"""				)
					assert torch.allclose(UpperCAmelCase  ,  UpperCAmelCase  ,  atol=1e-3				)
					print("Looks ok!"				)

					if pytorch_dump_folder_path is not None:
										print(F"""Saving model {model_name} to {pytorch_dump_folder_path}"""				)
										model.save_pretrained(UpperCAmelCase				)

					if push_to_hub:
										print("Pushing model, processor and slow tokenizer files to the hub..."				)
										model.push_to_hub(UpperCAmelCase  ,  organization="nielsr"				)
										processor.push_to_hub(UpperCAmelCase  ,  organization="nielsr"				)
										slow_tokenizer.push_to_hub(UpperCAmelCase  ,  organization="nielsr"				)


if __name__ == "__main__":
							__lowerCAmelCase :Optional[Any]     =    argparse.ArgumentParser()
							# Required parameters
							parser.add_argument(
							    '--model_name',
							    default='xclip-base-patch32',
							    type=str,
							    help='Name of the model.',
							)
							parser.add_argument(
							    '--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model directory.'
							)
							parser.add_argument(
							    '--push_to_hub', action='store_true', help='Whether or not to push the converted model to the 🤗 hub.'
							)

							__lowerCAmelCase :str     =    parser.parse_args()
							convert_xclip_checkpoint(args.model_name, args.pytorch_dump_folder_path, args.push_to_hub)









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(    __A	,   unittest.TestCase    ):
		lowerCamelCase__     :Optional[int]   				=		AudioLDMPipeline
		lowerCamelCase__     :List[Any]   				=		TEXT_TO_AUDIO_PARAMS
		lowerCamelCase__     :List[Any]   				=		TEXT_TO_AUDIO_BATCH_PARAMS
		lowerCamelCase__     :Dict   				=		frozenset(
		    [
		        'num_inference_steps',
		        'num_waveforms_per_prompt',
		        'generator',
		        'latents',
		        'output_type',
		        'return_dict',
		        'callback',
		        'callback_steps',
		    ]    )







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


							'''simple docstring'''

							torch.manual_seed(0				)
							_snake_case       :					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 ,						)
							_snake_case       :					int				       =   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				)
							_snake_case       :					Dict				       =   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				)
							_snake_case       :					Tuple				       =   ClapTextConfig(
							    bos_token_id=0 ,						eos_token_id=2 ,						hidden_size=3_2 ,						intermediate_size=3_7 ,						layer_norm_eps=1E-05 ,						num_attention_heads=4 ,						num_hidden_layers=5 ,						pad_token_id=1 ,						vocab_size=1_0_0_0 ,						projection_dim=3_2 ,						)
							_snake_case       :					Optional[int]				       =   ClapTextModelWithProjection(__snake_case				)
							_snake_case       :					Optional[Any]				       =   RobertaTokenizer.from_pretrained("hf-internal-testing/tiny-random-roberta" ,						model_max_length=7_7				)

							_snake_case       :					Dict				       =   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 ,						)

							_snake_case       :					Dict				       =   SpeechTaHifiGan(__snake_case				)

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







		def        lowercase       (    self ,						__snake_case ,						__snake_case=0				)      ->      Union[str, Any]:


							'''simple docstring'''

							if str(__snake_case				).startswith("mps"				):
												_snake_case       :					List[Any]				       =   torch.manual_seed(__snake_case				)
							else:
												_snake_case       :					Tuple				       =   torch.Generator(device=__snake_case				).manual_seed(__snake_case				)
							_snake_case       :					List[str]				       =   {
							    "prompt": "A hammer hitting a wooden surface",
							    "generator": generator,
							    "num_inference_steps": 2,
							    "guidance_scale": 6.0,
							}
							return inputs







		def        lowercase       (    self				)      ->      Any:


							'''simple docstring'''

							_snake_case       :					Tuple				       =   "cpu"  # ensure determinism for the device-dependent torch.Generator

							_snake_case       :					Union[str, Any]				       =   self.get_dummy_components()
							_snake_case       :					Any				       =   AudioLDMPipeline(**__snake_case				)
							_snake_case       :					List[str]				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)

							_snake_case       :					List[str]				       =   self.get_dummy_inputs(__snake_case				)
							_snake_case       :					Any				       =   audioldm_pipe(**__snake_case				)
							_snake_case       :					Tuple				       =   output.audios[0]

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

							_snake_case       :					Any				       =   audio[:1_0]
							_snake_case       :					Dict				       =   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        lowercase       (    self				)      ->      Tuple:


							'''simple docstring'''

							_snake_case       :					Tuple				       =   self.get_dummy_components()
							_snake_case       :					Any				       =   AudioLDMPipeline(**__snake_case				)
							_snake_case       :					Union[str, Any]				       =   audioldm_pipe.to(__snake_case				)
							_snake_case       :					List[str]				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)

							_snake_case       :					Optional[int]				       =   self.get_dummy_inputs(__snake_case				)
							_snake_case       :					List[str]				       =   3 * [inputs["prompt"]]

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

							_snake_case       :					List[str]				       =   self.get_dummy_inputs(__snake_case				)
							_snake_case       :					Tuple				       =   3 * [inputs.pop("prompt"				)]

							_snake_case       :					int				       =   audioldm_pipe.tokenizer(
							    __snake_case ,						padding="max_length" ,						max_length=audioldm_pipe.tokenizer.model_max_length ,						truncation=__snake_case ,						return_tensors="pt" ,						)
							_snake_case       :					Tuple				       =   text_inputs["input_ids"].to(__snake_case				)

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

							_snake_case       :					Union[str, Any]				       =   prompt_embeds

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

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







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


							'''simple docstring'''

							_snake_case       :					int				       =   self.get_dummy_components()
							_snake_case       :					Optional[int]				       =   AudioLDMPipeline(**__snake_case				)
							_snake_case       :					Tuple				       =   audioldm_pipe.to(__snake_case				)
							_snake_case       :					Any				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)

							_snake_case       :					Tuple				       =   self.get_dummy_inputs(__snake_case				)
							_snake_case       :					Union[str, Any]				       =   3 * ["this is a negative prompt"]
							_snake_case       :					int				       =   negative_prompt
							_snake_case       :					Tuple				       =   3 * [inputs["prompt"]]

							# forward
							_snake_case       :					Tuple				       =   audioldm_pipe(**__snake_case				)
							_snake_case       :					Optional[Any]				       =   output.audios[0]

							_snake_case       :					List[str]				       =   self.get_dummy_inputs(__snake_case				)
							_snake_case       :					Dict				       =   3 * [inputs.pop("prompt"				)]

							_snake_case       :					List[Any]				       =   []
							for p in [prompt, negative_prompt]:
												_snake_case       :					Optional[Any]				       =   audioldm_pipe.tokenizer(
												    __snake_case ,						padding="max_length" ,						max_length=audioldm_pipe.tokenizer.model_max_length ,						truncation=__snake_case ,						return_tensors="pt" ,						)
												_snake_case       :					List[Any]				       =   text_inputs["input_ids"].to(__snake_case				)

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

												embeds.append(__snake_case				)

							_snake_case       , _snake_case       :					int				       =   embeds

							# forward
							_snake_case       :					Any				       =   audioldm_pipe(**__snake_case				)
							_snake_case       :					Dict				       =   output.audios[0]

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







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


							'''simple docstring'''

							_snake_case       :					Union[str, Any]				       =   "cpu"  # ensure determinism for the device-dependent torch.Generator
							_snake_case       :					Tuple				       =   self.get_dummy_components()
							_snake_case       :					str				       =   PNDMScheduler(skip_prk_steps=__snake_case				)
							_snake_case       :					Optional[int]				       =   AudioLDMPipeline(**__snake_case				)
							_snake_case       :					Tuple				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)

							_snake_case       :					Tuple				       =   self.get_dummy_inputs(__snake_case				)
							_snake_case       :					str				       =   "egg cracking"
							_snake_case       :					int				       =   audioldm_pipe(**__snake_case ,						negative_prompt=__snake_case				)
							_snake_case       :					Union[str, Any]				       =   output.audios[0]

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

							_snake_case       :					List[str]				       =   audio[:1_0]
							_snake_case       :					Optional[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        lowercase       (    self				)      ->      Dict:


							'''simple docstring'''

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

							_snake_case       :					List[Any]				       =   "A hammer hitting a wooden surface"

							# test num_waveforms_per_prompt=1 (default)
							_snake_case       :					Dict				       =   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
							_snake_case       :					int				       =   2
							_snake_case       :					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
							_snake_case       :					List[Any]				       =   2
							_snake_case       :					List[str]				       =   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
							_snake_case       :					Optional[int]				       =   2
							_snake_case       :					str				       =   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        lowercase       (    self				)      ->      int:


							'''simple docstring'''

							_snake_case       :					List[str]				       =   "cpu"  # ensure determinism for the device-dependent torch.Generator
							_snake_case       :					int				       =   self.get_dummy_components()
							_snake_case       :					Union[str, Any]				       =   AudioLDMPipeline(**__snake_case				)
							_snake_case       :					Any				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)
							_snake_case       :					Tuple				       =   audioldm_pipe.vocoder.config.sampling_rate

							_snake_case       :					str				       =   self.get_dummy_inputs(__snake_case				)
							_snake_case       :					List[Any]				       =   audioldm_pipe(audio_length_in_s=0.0_16 ,						**__snake_case				)
							_snake_case       :					List[str]				       =   output.audios[0]

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

							_snake_case       :					Tuple				       =   audioldm_pipe(audio_length_in_s=0.0_32 ,						**__snake_case				)
							_snake_case       :					Union[str, Any]				       =   output.audios[0]

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







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


							'''simple docstring'''

							_snake_case       :					Union[str, Any]				       =   self.get_dummy_components()
							_snake_case       :					Any				       =   AudioLDMPipeline(**__snake_case				)
							_snake_case       :					List[str]				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)

							_snake_case       :					int				       =   ["hey"]

							_snake_case       :					int				       =   audioldm_pipe(__snake_case ,						num_inference_steps=1				)
							_snake_case       :					List[Any]				       =   output.audios.shape
							assert audio_shape == (1, 2_5_6)

							_snake_case       :					Optional[Any]				       =   audioldm_pipe.vocoder.config
							config.model_in_dim *= 2
							_snake_case       :					str				       =   SpeechTaHifiGan(__snake_case				).to(__snake_case				)
							_snake_case       :					Any				       =   audioldm_pipe(__snake_case ,						num_inference_steps=1				)
							_snake_case       :					Tuple				       =   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        lowercase       (    self				)      ->      Any:


							'''simple docstring'''

							self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__snake_case				)







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


							'''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        lowercase       (    self				)      ->      Dict:


							'''simple docstring'''

							self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__snake_case				)






@slow
class        _a(    unittest.TestCase    ):







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


							'''simple docstring'''

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







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


							'''simple docstring'''

							_snake_case       :					int				       =   torch.Generator(device=__snake_case				).manual_seed(__snake_case				)
							_snake_case       :					Dict				       =   np.random.RandomState(__snake_case				).standard_normal((1, 8, 1_2_8, 1_6)				)
							_snake_case       :					Any				       =   torch.from_numpy(__snake_case				).to(device=__snake_case ,						dtype=__snake_case				)
							_snake_case       :					List[Any]				       =   {
							    "prompt": "A hammer hitting a wooden surface",
							    "latents": latents,
							    "generator": generator,
							    "num_inference_steps": 3,
							    "guidance_scale": 2.5,
							}
							return inputs







		def        lowercase       (    self				)      ->      Dict:


							'''simple docstring'''

							_snake_case       :					List[Any]				       =   AudioLDMPipeline.from_pretrained("cvssp/audioldm"				)
							_snake_case       :					str				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)

							_snake_case       :					Tuple				       =   self.get_inputs(__snake_case				)
							_snake_case       :					Optional[Any]				       =   2_5
							_snake_case       :					str				       =   audioldm_pipe(**__snake_case				).audios[0]

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

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







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


							'''simple docstring'''

							_snake_case       :					Tuple				       =   AudioLDMPipeline.from_pretrained("cvssp/audioldm"				)
							_snake_case       :					Any				       =   LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config				)
							_snake_case       :					int				       =   audioldm_pipe.to(__snake_case				)
							audioldm_pipe.set_progress_bar_config(disable=__snake_case				)

							_snake_case       :					Any				       =   self.get_inputs(__snake_case				)
							_snake_case       :					Optional[Any]				       =   audioldm_pipe(**__snake_case				).audios[0]

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

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