



'''simple docstring'''





import argparse
import json
from pathlib import Path

import requests
import torch
from huggingface_hub import hf_hub_download
from PIL import Image

from transformers import YolosConfig, YolosForObjectDetection, YolosImageProcessor
from transformers.utils import logging


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



def       lowercase       (   SCREAMING_SNAKE_CASE__       :	str				)  ->     YolosConfig:
 _snake_case    :  str					=					YolosConfig()

 # size of the architecture
 if "yolos_ti" in yolos_name:
  _snake_case    :  str					=					192
  _snake_case    :  Optional[Any]					=					768
  _snake_case    :  Optional[Any]					=					12
  _snake_case    :  Any					=					3
  _snake_case    :  int					=					[800, 1_333]
  _snake_case    :  Tuple					=					False
 elif yolos_name == "yolos_s_dWr":
  _snake_case    :  Dict					=					330
  _snake_case    :  List[Any]					=					14
  _snake_case    :  int					=					6
  _snake_case    :  str					=					1_320
 elif "yolos_s" in yolos_name:
  _snake_case    :  Union[str, Any]					=					384
  _snake_case    :  str					=					1_536
  _snake_case    :  List[str]					=					12
  _snake_case    :  Optional[Any]					=					6
 elif "yolos_b" in yolos_name:
  _snake_case    :  Union[str, Any]					=					[800, 1_344]

 _snake_case    :  List[Any]					=					91
 _snake_case    :  Dict					=					"huggingface/label-files"
 _snake_case    :  str					=					"coco-detection-id2label.json"
 _snake_case    :  Any					=					json.load(open(hf_hub_download(SCREAMING_SNAKE_CASE_				,       SCREAMING_SNAKE_CASE_				,       repo_type="""dataset"""				)				,       """r"""				)				)
 _snake_case    :  int					=					{int(SCREAMING_SNAKE_CASE_				): v for k, v in idalabel.items()}
 _snake_case    :  Optional[Any]					=					idalabel
 _snake_case    :  Dict					=					{v: k for k, v in idalabel.items()}

 return config



def       lowercase       (   SCREAMING_SNAKE_CASE__       :	dict				,       SCREAMING_SNAKE_CASE__       :	YolosConfig				,       SCREAMING_SNAKE_CASE__       :	bool = False				)  ->     Optional[Any]:

 for i in range(config.num_hidden_layers				):
  # read in weights + bias of input projection layer (in timm, this is a single matrix + bias)
  _snake_case    :  Any					=					state_dict.pop(F'''blocks.{i}.attn.qkv.weight'''				)
  _snake_case    :  Union[str, Any]					=					state_dict.pop(F'''blocks.{i}.attn.qkv.bias'''				)
  # next, add query, keys and values (in that order) to the state dict
  _snake_case    :  Any					=					in_proj_weight[: config.hidden_size, :]
  _snake_case    :  Union[str, Any]					=					in_proj_bias[: config.hidden_size]
  _snake_case    :  Union[str, Any]					=					in_proj_weight[
      config.hidden_size : config.hidden_size * 2, :
  ]
  _snake_case    :  Any					=					in_proj_bias[
      config.hidden_size : config.hidden_size * 2
  ]
  _snake_case    :  int					=					in_proj_weight[-config.hidden_size :, :]
  _snake_case    :  List[Any]					=					in_proj_bias[-config.hidden_size :]



def       lowercase       (   SCREAMING_SNAKE_CASE__       :	str				)  ->     str:

 if "backbone" in name:
  _snake_case    :  Optional[int]					=					name.replace("""backbone"""				,       """vit"""				)
 if "cls_token" in name:
  _snake_case    :  str					=					name.replace("""cls_token"""				,       """embeddings.cls_token"""				)
 if "det_token" in name:
  _snake_case    :  str					=					name.replace("""det_token"""				,       """embeddings.detection_tokens"""				)
 if "mid_pos_embed" in name:
  _snake_case    :  List[str]					=					name.replace("""mid_pos_embed"""				,       """encoder.mid_position_embeddings"""				)
 if "pos_embed" in name:
  _snake_case    :  str					=					name.replace("""pos_embed"""				,       """embeddings.position_embeddings"""				)
 if "patch_embed.proj" in name:
  _snake_case    :  Optional[Any]					=					name.replace("""patch_embed.proj"""				,       """embeddings.patch_embeddings.projection"""				)
 if "blocks" in name:
  _snake_case    :  Dict					=					name.replace("""blocks"""				,       """encoder.layer"""				)
 if "attn.proj" in name:
  _snake_case    :  Tuple					=					name.replace("""attn.proj"""				,       """attention.output.dense"""				)
 if "attn" in name:
  _snake_case    :  str					=					name.replace("""attn"""				,       """attention.self"""				)
 if "norm1" in name:
  _snake_case    :  Any					=					name.replace("""norm1"""				,       """layernorm_before"""				)
 if "norm2" in name:
  _snake_case    :  Optional[int]					=					name.replace("""norm2"""				,       """layernorm_after"""				)
 if "mlp.fc1" in name:
  _snake_case    :  str					=					name.replace("""mlp.fc1"""				,       """intermediate.dense"""				)
 if "mlp.fc2" in name:
  _snake_case    :  Dict					=					name.replace("""mlp.fc2"""				,       """output.dense"""				)
 if "class_embed" in name:
  _snake_case    :  int					=					name.replace("""class_embed"""				,       """class_labels_classifier"""				)
 if "bbox_embed" in name:
  _snake_case    :  Any					=					name.replace("""bbox_embed"""				,       """bbox_predictor"""				)
 if "vit.norm" in name:
  _snake_case    :  Any					=					name.replace("""vit.norm"""				,       """vit.layernorm"""				)

 return name



def       lowercase       (   SCREAMING_SNAKE_CASE__       :	dict				,       SCREAMING_SNAKE_CASE__       :	YolosForObjectDetection				)  ->     dict:

 for key in orig_state_dict.copy().keys():
  _snake_case    :  List[Any]					=					orig_state_dict.pop(SCREAMING_SNAKE_CASE_				)

  if "qkv" in key:
   _snake_case    :  str					=					key.split("""."""				)
   _snake_case    :  str					=					int(key_split[2]				)
   _snake_case    :  Dict					=					model.vit.encoder.layer[layer_num].attention.attention.all_head_size
   if "weight" in key:
    _snake_case    :  Dict					=					val[:dim, :]
    _snake_case    :  Optional[int]					=					val[
        dim : dim * 2, :
    ]
    _snake_case    :  List[Any]					=					val[-dim:, :]
   else:
    _snake_case    :  str					=					val[:dim]
    _snake_case    :  Optional[Any]					=					val[dim : dim * 2]
    _snake_case    :  int					=					val[-dim:]
  else:
   _snake_case    :  Dict					=					val

 return orig_state_dict



def       lowercase       (   )  ->     torch.Tensor:
 _snake_case    :  Union[str, Any]					=					"http://images.cocodataset.org/val2017/000000039769.jpg"
 _snake_case    :  List[Any]					=					Image.open(requests.get(SCREAMING_SNAKE_CASE_				,       stream=SCREAMING_SNAKE_CASE_				).raw				)
 return im



@torch.no_grad()
def       lowercase       (   SCREAMING_SNAKE_CASE__       :	str				,       SCREAMING_SNAKE_CASE__       :	str				,       SCREAMING_SNAKE_CASE__       :	str				,       SCREAMING_SNAKE_CASE__       :	bool = False				)  ->     List[Any]:
 _snake_case    :  str					=					get_yolos_config(SCREAMING_SNAKE_CASE_				)

 # load original state_dict
 _snake_case    :  Union[str, Any]					=					torch.load(SCREAMING_SNAKE_CASE_				,       map_location="""cpu"""				)["model"]

 # load 🤗 model
 _snake_case    :  List[Any]					=					YolosForObjectDetection(SCREAMING_SNAKE_CASE_				)
 model.eval()
 _snake_case    :  int					=					convert_state_dict(SCREAMING_SNAKE_CASE_				,       SCREAMING_SNAKE_CASE_				)
 model.load_state_dict(SCREAMING_SNAKE_CASE_				)

 # Check outputs on an image, prepared by YolosImageProcessor
 _snake_case    :  int					=					800 if yolos_name != "yolos_ti" else 512
 _snake_case    :  Dict					=					YolosImageProcessor(format="""coco_detection"""				,       size=SCREAMING_SNAKE_CASE_				)
 _snake_case    :  Union[str, Any]					=					image_processor(images=prepare_img()				,       return_tensors="""pt"""				)
 _snake_case    :  Union[str, Any]					=					model(**SCREAMING_SNAKE_CASE_				)
 _snake_case    :  List[str]					=					outputs.logits, outputs.pred_boxes

 _snake_case    :  Optional[int]					=					None, None
 if yolos_name == "yolos_ti":
  _snake_case    :  Union[str, Any]					=					torch.tensor(
      [[-3_9.5_0_2_2, -1_1.9_8_2_0, -1_7.6_8_8_8], [-2_9.9_5_7_4, -9.9_7_6_9, -1_7.7_6_9_1], [-4_2.3_2_8_1, -2_0.7_2_0_0, -3_0.6_2_9_4]]				)
  _snake_case    :  Optional[Any]					=					torch.tensor(
      [[0.4_0_2_1, 0.0_8_3_6, 0.7_9_7_9], [0.0_1_8_4, 0.2_6_0_9, 0.0_3_6_4], [0.1_7_8_1, 0.2_0_0_4, 0.2_0_9_5]]				)
 elif yolos_name == "yolos_s_200_pre":
  _snake_case    :  List[Any]					=					torch.tensor(
      [[-2_4.0_2_4_8, -1_0.3_0_2_4, -1_4.8_2_9_0], [-4_2.0_3_9_2, -1_6.8_2_0_0, -2_7.4_3_3_4], [-2_7.2_7_4_3, -1_1.8_1_5_4, -1_8.7_1_4_8]]				)
  _snake_case    :  str					=					torch.tensor(
      [[0.2_5_5_9, 0.5_4_5_5, 0.4_7_0_6], [0.2_9_8_9, 0.7_2_7_9, 0.1_8_7_5], [0.7_7_3_2, 0.4_0_1_7, 0.4_4_6_2]]				)
 elif yolos_name == "yolos_s_300_pre":
  _snake_case    :  List[Any]					=					torch.tensor(
      [[-3_6.2_2_2_0, -1_4.4_3_8_5, -2_3.5_4_5_7], [-3_5.6_9_7_0, -1_4.7_5_8_3, -2_1.3_9_3_5], [-3_1.5_9_3_9, -1_3.6_0_4_2, -1_6.8_0_4_9]]				)
  _snake_case    :  Optional[Any]					=					torch.tensor(
      [[0.7_6_1_4, 0.2_3_1_6, 0.4_7_2_8], [0.7_1_6_8, 0.4_4_9_5, 0.3_8_5_5], [0.4_9_9_6, 0.1_4_6_6, 0.9_9_9_6]]				)
 elif yolos_name == "yolos_s_dWr":
  _snake_case    :  List[Any]					=					torch.tensor(
      [[-4_2.8_6_6_8, -2_4.1_0_4_9, -4_1.1_6_9_0], [-3_4.7_4_5_6, -1_4.1_2_7_4, -2_4.9_1_9_4], [-3_3.7_8_9_8, -1_2.1_9_4_6, -2_5.6_4_9_5]]				)
  _snake_case    :  Tuple					=					torch.tensor(
      [[0.5_5_8_7, 0.2_7_7_3, 0.0_6_0_5], [0.5_0_0_4, 0.3_0_1_4, 0.9_9_9_4], [0.4_9_9_9, 0.1_5_4_8, 0.9_9_9_4]]				)
 elif yolos_name == "yolos_base":
  _snake_case    :  List[str]					=					torch.tensor(
      [[-4_0.6_0_6_4, -2_4.3_0_8_4, -3_2.6_4_4_7], [-5_5.1_9_9_0, -3_0.7_7_1_9, -3_5.5_8_7_7], [-5_1.4_3_1_1, -3_3.3_5_0_7, -3_5.6_4_6_2]]				)
  _snake_case    :  List[Any]					=					torch.tensor(
      [[0.5_5_5_5, 0.2_7_9_4, 0.0_6_5_5], [0.9_0_4_9, 0.2_6_6_4, 0.1_8_9_4], [0.9_1_8_3, 0.1_9_8_4, 0.1_6_3_5]]				)
 else:
  raise ValueError(F'''Unknown yolos_name: {yolos_name}'''				)

 assert torch.allclose(logits[0, :3, :3]				,       SCREAMING_SNAKE_CASE_				,       atol=1e-4				)
 assert torch.allclose(pred_boxes[0, :3, :3]				,       SCREAMING_SNAKE_CASE_				,       atol=1e-4				)

 Path(SCREAMING_SNAKE_CASE_				).mkdir(exist_ok=SCREAMING_SNAKE_CASE_				)
 print(F'''Saving model {yolos_name} to {pytorch_dump_folder_path}'''				)
 model.save_pretrained(SCREAMING_SNAKE_CASE_				)
 print(F'''Saving image processor to {pytorch_dump_folder_path}'''				)
 image_processor.save_pretrained(SCREAMING_SNAKE_CASE_				)

 if push_to_hub:
  _snake_case    :  List[str]					=					{
      "yolos_ti": "yolos-tiny",
      "yolos_s_200_pre": "yolos-small",
      "yolos_s_300_pre": "yolos-small-300",
      "yolos_s_dWr": "yolos-small-dwr",
      "yolos_base": "yolos-base",
  }

  print("""Pushing to the hub..."""				)
  _snake_case    :  Union[str, Any]					=					model_mapping[yolos_name]
  image_processor.push_to_hub(SCREAMING_SNAKE_CASE_				,       organization="""hustvl"""				)
  model.push_to_hub(SCREAMING_SNAKE_CASE_				,       organization="""hustvl"""				)


if __name__ == "__main__":
  a__    =	argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      """--yolos_name""",
      default="""yolos_s_200_pre""",
      type=str,
      help=(
          """Name of the YOLOS model you\'d like to convert. Should be one of \'yolos_ti\', \'yolos_s_200_pre\',"""
          """ \'yolos_s_300_pre\', \'yolos_s_dWr\', \'yolos_base\'."""
      ),
  )
  parser.add_argument(
      """--checkpoint_path""", default=None, type=str, help="""Path to the original state dict (.pth file)."""
  )
  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."""
  )

  a__    =	parser.parse_args()
  convert_yolos_checkpoint(args.yolos_name, args.checkpoint_path, 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 		snake_case						(	SCREAMING_SNAKE_CASE_					,unittest.TestCase  ):



   '''simple docstring'''






   snake_case_					: str          =		AudioLDMPipeline
   snake_case_					: Any          =		TEXT_TO_AUDIO_PARAMS
   snake_case_					: Tuple          =		TEXT_TO_AUDIO_BATCH_PARAMS
   snake_case_					: Dict          =		frozenset(
       [
           """num_inference_steps""",
           """num_waveforms_per_prompt""",
           """generator""",
           """latents""",
           """output_type""",
           """return_dict""",
           """callback""",
           """callback_steps""",
       ]  )

   def      UpperCamelCase_  (					self							:			Any)				->				Any:

    """simple docstring"""







    torch.manual_seed(0)
    _snake_case    :  List[str]					=					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=lowerCAmelCase				,		)
    _snake_case    :  List[Any]					=					DDIMScheduler(
        beta_start=0.00_085				,		beta_end=0.012				,		beta_schedule="""scaled_linear"""				,		clip_sample=lowerCAmelCase				,		set_alpha_to_one=lowerCAmelCase				,		)
    torch.manual_seed(0)
    _snake_case    :  str					=					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)
    _snake_case    :  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				,		)
    _snake_case    :  Any					=					ClapTextModelWithProjection(lowerCAmelCase)
    _snake_case    :  List[str]					=					RobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-random-roberta"""				,		model_max_length=77)

    _snake_case    :  List[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=lowerCAmelCase				,		)

    _snake_case    :  Tuple					=					SpeechTaHifiGan(lowerCAmelCase)

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

   def      UpperCamelCase_  (					self							:			Optional[Any]				,		lowerCAmelCase							:			Tuple				,		lowerCAmelCase							:			Dict=0)				->				int:

    """simple docstring"""







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

   def      UpperCamelCase_  (					self							:			List[Any])				->				str:

    """simple docstring"""







    _snake_case    :  Optional[Any]					=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator

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

    _snake_case    :  str					=					self.get_dummy_inputs(lowerCAmelCase)
    _snake_case    :  Tuple					=					audioldm_pipe(**lowerCAmelCase)
    _snake_case    :  int					=					output.audios[0]

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

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

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

   def      UpperCamelCase_  (					self							:			List[str])				->				Optional[int]:

    """simple docstring"""







    _snake_case    :  Dict					=					self.get_dummy_components()
    _snake_case    :  Any					=					AudioLDMPipeline(**lowerCAmelCase)
    _snake_case    :  int					=					audioldm_pipe.to(lowerCAmelCase)
    _snake_case    :  str					=					audioldm_pipe.to(lowerCAmelCase)
    audioldm_pipe.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case    :  str					=					self.get_dummy_inputs(lowerCAmelCase)
    _snake_case    :  List[Any]					=					3 * [inputs["""prompt"""]]

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

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

    _snake_case    :  List[Any]					=					audioldm_pipe.tokenizer(
        lowerCAmelCase				,		padding="""max_length"""				,		max_length=audioldm_pipe.tokenizer.model_max_length				,		truncation=lowerCAmelCase				,		return_tensors="""pt"""				,		)
    _snake_case    :  List[Any]					=					text_inputs["""input_ids"""].to(lowerCAmelCase)

    _snake_case    :  str					=					audioldm_pipe.text_encoder(
        lowerCAmelCase				,		)
    _snake_case    :  Optional[int]					=					prompt_embeds.text_embeds
    # additional L_2 normalization over each hidden-state
    _snake_case    :  Tuple					=					F.normalize(lowerCAmelCase				,		dim=-1)

    _snake_case    :  List[Any]					=					prompt_embeds

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

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

   def      UpperCamelCase_  (					self							:			Optional[Any])				->				Dict:

    """simple docstring"""







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

    _snake_case    :  Optional[int]					=					self.get_dummy_inputs(lowerCAmelCase)
    _snake_case    :  Optional[Any]					=					3 * ["""this is a negative prompt"""]
    _snake_case    :  int					=					negative_prompt
    _snake_case    :  Dict					=					3 * [inputs["""prompt"""]]

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

    _snake_case    :  str					=					self.get_dummy_inputs(lowerCAmelCase)
    _snake_case    :  Union[str, Any]					=					3 * [inputs.pop("""prompt""")]

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

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

     embeds.append(lowerCAmelCase)

    _snake_case	,    _snake_case    :  int					=					embeds

    # forward
    _snake_case    :  List[str]					=					audioldm_pipe(**lowerCAmelCase)
    _snake_case    :  Optional[Any]					=					output.audios[0]

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

   def      UpperCamelCase_  (					self							:			Optional[Any])				->				List[Any]:

    """simple docstring"""







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

    _snake_case    :  List[str]					=					self.get_dummy_inputs(lowerCAmelCase)
    _snake_case    :  List[Any]					=					"""egg cracking"""
    _snake_case    :  Optional[Any]					=					audioldm_pipe(**lowerCAmelCase				,		negative_prompt=lowerCAmelCase)
    _snake_case    :  List[str]					=					output.audios[0]

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

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

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

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

    """simple docstring"""







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

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

    # test num_waveforms_per_prompt=1 (default)
    _snake_case    :  List[Any]					=					audioldm_pipe(lowerCAmelCase				,		num_inference_steps=2).audios

    assert audios.shape == (1, 256)

    # test num_waveforms_per_prompt=1 (default) for batch of prompts
    _snake_case    :  Optional[Any]					=					2
    _snake_case    :  Optional[int]					=					audioldm_pipe([prompt] * batch_size				,		num_inference_steps=2).audios

    assert audios.shape == (batch_size, 256)

    # test num_waveforms_per_prompt for single prompt
    _snake_case    :  List[Any]					=					2
    _snake_case    :  Tuple					=					audioldm_pipe(lowerCAmelCase				,		num_inference_steps=2				,		num_waveforms_per_prompt=lowerCAmelCase).audios

    assert audios.shape == (num_waveforms_per_prompt, 256)

    # test num_waveforms_per_prompt for batch of prompts
    _snake_case    :  Dict					=					2
    _snake_case    :  int					=					audioldm_pipe(
        [prompt] * batch_size				,		num_inference_steps=2				,		num_waveforms_per_prompt=lowerCAmelCase).audios

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

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

    """simple docstring"""







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

    _snake_case    :  List[str]					=					self.get_dummy_inputs(lowerCAmelCase)
    _snake_case    :  Optional[int]					=					audioldm_pipe(audio_length_in_s=0.016				,		**lowerCAmelCase)
    _snake_case    :  str					=					output.audios[0]

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

    _snake_case    :  int					=					audioldm_pipe(audio_length_in_s=0.032				,		**lowerCAmelCase)
    _snake_case    :  List[Any]					=					output.audios[0]

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

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

    """simple docstring"""







    _snake_case    :  str					=					self.get_dummy_components()
    _snake_case    :  List[str]					=					AudioLDMPipeline(**lowerCAmelCase)
    _snake_case    :  str					=					audioldm_pipe.to(lowerCAmelCase)
    audioldm_pipe.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case    :  List[str]					=					["""hey"""]

    _snake_case    :  int					=					audioldm_pipe(lowerCAmelCase				,		num_inference_steps=1)
    _snake_case    :  Optional[int]					=					output.audios.shape
    assert audio_shape == (1, 256)

    _snake_case    :  Dict					=					audioldm_pipe.vocoder.config
    config.model_in_dim *= 2
    _snake_case    :  str					=					SpeechTaHifiGan(lowerCAmelCase).to(lowerCAmelCase)
    _snake_case    :  List[str]					=					audioldm_pipe(lowerCAmelCase				,		num_inference_steps=1)
    _snake_case    :  Dict					=					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							:			Dict)				->				Dict:

    """simple docstring"""







    self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCAmelCase)

   def      UpperCamelCase_  (					self							:			str)				->				Optional[int]:

    """simple docstring"""







    self._test_inference_batch_single_identical(test_mean_pixel_difference=lowerCAmelCase)


   @unittest.skipIf(
       torch_device != """cuda""" or not is_xformers_available()				,		reason="""XFormers attention is only available with CUDA and `xformers` installed"""				,		)
   def      UpperCamelCase_  (					self							:			Dict)				->				Union[str, Any]:

    """simple docstring"""







    self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCAmelCase)







@slow
class 		snake_case						(	unittest.TestCase  ):



   '''simple docstring'''






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

    """simple docstring"""







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

   def      UpperCamelCase_  (					self							:			Any				,		lowerCAmelCase							:			Dict				,		lowerCAmelCase							:			Optional[Any]="cpu"				,		lowerCAmelCase							:			Any=torch.floataa				,		lowerCAmelCase							:			Optional[Any]=0)				->				Tuple:

    """simple docstring"""







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

   def      UpperCamelCase_  (					self							:			Any)				->				List[str]:

    """simple docstring"""







    _snake_case    :  Dict					=					AudioLDMPipeline.from_pretrained("""cvssp/audioldm""")
    _snake_case    :  Optional[int]					=					audioldm_pipe.to(lowerCAmelCase)
    audioldm_pipe.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case    :  Tuple					=					self.get_inputs(lowerCAmelCase)
    _snake_case    :  int					=					25
    _snake_case    :  Optional[Any]					=					audioldm_pipe(**lowerCAmelCase).audios[0]

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

    _snake_case    :  Any					=					audio[7_7230:7_7240]
    _snake_case    :  str					=					np.array(
        [-0.4_884, -0.4_607, 0.0_023, 0.5_007, 0.5_896, 0.5_151, 0.3_813, -0.0_208, -0.3_687, -0.4_315])
    _snake_case    :  List[str]					=					np.abs(expected_slice - audio_slice).max()
    assert max_diff < 1E-2


   def      UpperCamelCase_  (					self							:			Dict)				->				List[Any]:

    """simple docstring"""







    _snake_case    :  Dict					=					AudioLDMPipeline.from_pretrained("""cvssp/audioldm""")
    _snake_case    :  Any					=					LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config)
    _snake_case    :  Optional[int]					=					audioldm_pipe.to(lowerCAmelCase)
    audioldm_pipe.set_progress_bar_config(disable=lowerCAmelCase)

    _snake_case    :  Union[str, Any]					=					self.get_inputs(lowerCAmelCase)
    _snake_case    :  List[str]					=					audioldm_pipe(**lowerCAmelCase).audios[0]

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

    _snake_case    :  Optional[Any]					=					audio[2_7780:2_7790]
    _snake_case    :  int					=					np.array([-0.2_131, -0.0_873, -0.0_124, -0.0_189, 0.0_569, 0.1_373, 0.1_883, 0.2_886, 0.3_297, 0.2_212])
    _snake_case    :  List[Any]					=					np.abs(expected_slice - audio_slice).max()
    assert max_diff < 3E-2
