

import itertools
import json
import linecache
import os
import pickle
import re
import socket
import string
from collections import Counter
from logging import getLogger
from pathlib import Path
from typing import Callable, Dict, Iterable, List

import git
import torch
from torch.utils.data import Dataset

from transformers import BartTokenizer, RagTokenizer, TaTokenizer


def        __lowerCAmelCase    (       _UpperCamelCase		:  Any		,							_UpperCamelCase		:  Tuple		,							_UpperCamelCase		:  int		,							_UpperCamelCase		:  Optional[Any]		,							_UpperCamelCase		:  str=True		,							_UpperCamelCase		:  Optional[Any]="pt"	)		->							List[Any]:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			{"""add_prefix_space""": True} if isinstance(_UpperCamelCase		,							_UpperCamelCase	) and not line.startswith(' '	) else {}
 SCREAMING_SNAKE_CASE								=			padding_side
 return tokenizer(
     [line]		,							max_length=_UpperCamelCase		,							padding='max_length' if pad_to_max_length else None		,							truncation=_UpperCamelCase		,							return_tensors=_UpperCamelCase		,							add_special_tokens=_UpperCamelCase		,							**_UpperCamelCase		,							)


def        __lowerCAmelCase    (       _UpperCamelCase		:  Optional[int]		,							_UpperCamelCase		:  Union[str, Any]		,							_UpperCamelCase		:  List[Any]=None		,							)		->							List[Any]:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			input_ids.ne(_UpperCamelCase	).any(dim=0	)
 if attention_mask is None:
  return input_ids[:, keep_column_mask]
 else:
  return (input_ids[:, keep_column_mask], attention_mask[:, keep_column_mask])




class      UpperCamelCase (     __SCREAMING_SNAKE_CASE   ):

    def __init__(			self :      List[Any]							,    snake_case__ :      Optional[Any]							,    snake_case__ :      Tuple							,    snake_case__ :      Dict							,    snake_case__ :      Tuple							,    snake_case__ :      Tuple="train"							,    snake_case__ :      Optional[int]=None							,    snake_case__ :      Any=None							,    snake_case__ :      int=None							,    snake_case__ :      Union[str, Any]=""							,    ):

     """simple docstring"""






     super().__init__()
     SCREAMING_SNAKE_CASE								=			Path(_a    ).joinpath(type_path + '.source'    )
     SCREAMING_SNAKE_CASE								=			Path(_a    ).joinpath(type_path + '.target'    )
     SCREAMING_SNAKE_CASE								=			self.get_char_lens(self.src_file    )
     SCREAMING_SNAKE_CASE								=			max_source_length
     SCREAMING_SNAKE_CASE								=			max_target_length
     assert min(self.src_lens    ) > 0, F"""found empty line in {self.src_file}"""
     SCREAMING_SNAKE_CASE								=			tokenizer
     SCREAMING_SNAKE_CASE								=			prefix
     if n_obs is not None:
      SCREAMING_SNAKE_CASE								=			self.src_lens[:n_obs]
     SCREAMING_SNAKE_CASE								=			src_lang
     SCREAMING_SNAKE_CASE								=			tgt_lang

    def __len__(			self :      Tuple    ):

     """simple docstring"""






     return len(self.src_lens    )

    def __getitem__(			self :      List[str]							,    snake_case__ :      Tuple    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			index + 1  # linecache starts at 1
     SCREAMING_SNAKE_CASE								=			self.prefix + linecache.getline(str(self.src_file    )							,    _a    ).rstrip('\n'    )
     SCREAMING_SNAKE_CASE								=			linecache.getline(str(self.tgt_file    )							,    _a    ).rstrip('\n'    )
     assert source_line, F"""empty source line for index {index}"""
     assert tgt_line, F"""empty tgt line for index {index}"""

     # Need to add eos token manually for T5
     if isinstance(self.tokenizer							,    _a    ):
      source_line += self.tokenizer.eos_token
      tgt_line += self.tokenizer.eos_token

     # Pad source and target to the right
     SCREAMING_SNAKE_CASE								=			(
         self.tokenizer.question_encoder if isinstance(self.tokenizer							,    _a    ) else self.tokenizer
     )
     SCREAMING_SNAKE_CASE								=			self.tokenizer.generator if isinstance(self.tokenizer							,    _a    ) else self.tokenizer

     SCREAMING_SNAKE_CASE								=			encode_line(_a							,    _a							,    self.max_source_length							,    'right'    )
     SCREAMING_SNAKE_CASE								=			encode_line(_a							,    _a							,    self.max_target_length							,    'right'    )

     SCREAMING_SNAKE_CASE								=			source_inputs["""input_ids"""].squeeze()
     SCREAMING_SNAKE_CASE								=			target_inputs["""input_ids"""].squeeze()
     SCREAMING_SNAKE_CASE								=			source_inputs["""attention_mask"""].squeeze()
     return {
         "input_ids": source_ids,
         "attention_mask": src_mask,
         "decoder_input_ids": target_ids,
     }

    @staticmethod
    def UpperCamelCase			(			snake_case__ :      int    ):

     """simple docstring"""






     return [len(_a    ) for x in Path(_a    ).open().readlines()]




    def UpperCamelCase			(			self :      Optional[int]							,    snake_case__ :      Dict    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			torch.stack([x['input_ids'] for x in batch]    )
     SCREAMING_SNAKE_CASE								=			torch.stack([x['attention_mask'] for x in batch]    )
     SCREAMING_SNAKE_CASE								=			torch.stack([x['decoder_input_ids'] for x in batch]    )
     SCREAMING_SNAKE_CASE								=			(
         self.tokenizer.generator.pad_token_id
         if isinstance(self.tokenizer							,    _a    )
         else self.tokenizer.pad_token_id
     )
     SCREAMING_SNAKE_CASE								=			(
         self.tokenizer.question_encoder.pad_token_id
         if isinstance(self.tokenizer							,    _a    )
         else self.tokenizer.pad_token_id
     )
     SCREAMING_SNAKE_CASE								=			trim_batch(_a							,    _a    )
     SCREAMING_SNAKE_CASE								=			trim_batch(_a							,    _a							,    attention_mask=_a    )
     SCREAMING_SNAKE_CASE								=			{
         """input_ids""": source_ids,
         """attention_mask""": source_mask,
         """decoder_input_ids""": y,
     }
     return batch


a_    :						str    		=      getLogger(__name__)


def        __lowerCAmelCase    (       _UpperCamelCase		:  List[List]	)		->							Any:
 '''simple docstring'''


 return list(itertools.chain.from_iterable(_UpperCamelCase	)	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  str	)		->							Any:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			get_git_info()
 save_json(_UpperCamelCase		,							os.path.join(_UpperCamelCase		,							'git_log.json'	)	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  List[str]		,							_UpperCamelCase		:  List[Any]		,							_UpperCamelCase		:  List[str]=4		,							**_UpperCamelCase		:  List[str]	)		->							Any:
 '''simple docstring'''


 with open(_UpperCamelCase		,							'w'	) as f:
  json.dump(_UpperCamelCase		,							_UpperCamelCase		,							indent=_UpperCamelCase		,							**_UpperCamelCase	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  Any	)		->							List[Any]:
 '''simple docstring'''


 with open(_UpperCamelCase	) as f:
  return json.load(_UpperCamelCase	)


def        __lowerCAmelCase    (       )		->							str:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			git.Repo(search_parent_directories=_UpperCamelCase	)
 SCREAMING_SNAKE_CASE								=			{
     """repo_id""": str(_UpperCamelCase	),
     """repo_sha""": str(repo.head.object.hexsha	),
     """repo_branch""": str(repo.active_branch	),
     """hostname""": str(socket.gethostname()	),
 }
 return repo_infos


def        __lowerCAmelCase    (       _UpperCamelCase		:  Callable		,							_UpperCamelCase		:  Iterable	)		->							Tuple:
 '''simple docstring'''


 return list(map(_UpperCamelCase		,							_UpperCamelCase	)	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  int		,							_UpperCamelCase		:  Union[str, Any]	)		->							Union[str, Any]:
 '''simple docstring'''


 with open(_UpperCamelCase		,							'wb'	) as f:
  return pickle.dump(_UpperCamelCase		,							_UpperCamelCase	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  List[str]	)		->							str:
 '''simple docstring'''


 def remove_articles(_UpperCamelCase		:  Any	):
  return re.sub(R'\b(a|an|the)\b'		,							' '		,							_UpperCamelCase	)

 def white_space_fix(_UpperCamelCase		:  List[Any]	):
  return " ".join(text.split()	)

 def remove_punc(_UpperCamelCase		:  Union[str, Any]	):
  SCREAMING_SNAKE_CASE								=			set(string.punctuation	)
  return "".join(ch for ch in text if ch not in exclude	)

 def lower(_UpperCamelCase		:  List[str]	):
  return text.lower()

 return white_space_fix(remove_articles(remove_punc(lower(_UpperCamelCase	)	)	)	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  int		,							_UpperCamelCase		:  int	)		->							Any:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			normalize_answer(_UpperCamelCase	).split()
 SCREAMING_SNAKE_CASE								=			normalize_answer(_UpperCamelCase	).split()
 SCREAMING_SNAKE_CASE								=			Counter(_UpperCamelCase	) & Counter(_UpperCamelCase	)
 SCREAMING_SNAKE_CASE								=			sum(common.values()	)
 if num_same == 0:
  return 0
 SCREAMING_SNAKE_CASE								=			1.0 * num_same / len(_UpperCamelCase	)
 SCREAMING_SNAKE_CASE								=			1.0 * num_same / len(_UpperCamelCase	)
 SCREAMING_SNAKE_CASE								=			(2 * precision * recall) / (precision + recall)
 return fa


def        __lowerCAmelCase    (       _UpperCamelCase		:  Any		,							_UpperCamelCase		:  Any	)		->							Dict:
 '''simple docstring'''


 return normalize_answer(_UpperCamelCase	) == normalize_answer(_UpperCamelCase	)


def        __lowerCAmelCase    (       _UpperCamelCase		:  List[str]		,							_UpperCamelCase		:  List[str]	)		->							Any:
 '''simple docstring'''


 assert len(_UpperCamelCase	) == len(_UpperCamelCase	)
 SCREAMING_SNAKE_CASE								=			0
 for hypo, pred in zip(_UpperCamelCase		,							_UpperCamelCase	):
  em += exact_match_score(_UpperCamelCase		,							_UpperCamelCase	)
 if len(_UpperCamelCase	) > 0:
  em /= len(_UpperCamelCase	)
 return {"em": em}


def        __lowerCAmelCase    (       _UpperCamelCase		:  Union[str, Any]	)		->							Any:
 '''simple docstring'''


 return model_prefix.startswith('rag'	)




def        __lowerCAmelCase    (       _UpperCamelCase		:  Optional[Any]		,							_UpperCamelCase		:  int		,							_UpperCamelCase		:  Union[str, Any]	)		->							List[Any]:
 '''simple docstring'''


 SCREAMING_SNAKE_CASE								=			{p: p for p in extra_params}
 # T5 models don't have `dropout` param, they have `dropout_rate` instead
 SCREAMING_SNAKE_CASE								=			"""dropout_rate"""
 for p in extra_params:
  if getattr(_UpperCamelCase		,							_UpperCamelCase		,							_UpperCamelCase	):
   if not hasattr(_UpperCamelCase		,							_UpperCamelCase	) and not hasattr(_UpperCamelCase		,							equivalent_param[p]	):
    logger.info('config doesn\'t have a `{}` attribute'.format(_UpperCamelCase	)	)
    delattr(_UpperCamelCase		,							_UpperCamelCase	)
    continue
   SCREAMING_SNAKE_CASE								=			p if hasattr(_UpperCamelCase		,							_UpperCamelCase	) else equivalent_param[p]
   setattr(_UpperCamelCase		,							_UpperCamelCase		,							getattr(_UpperCamelCase		,							_UpperCamelCase	)	)
   delattr(_UpperCamelCase		,							_UpperCamelCase	)
 return hparams, config









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

    def UpperCamelCase			(			self :      Optional[Any]    ):

     """simple docstring"""






     torch.manual_seed(0    )
     SCREAMING_SNAKE_CASE								=			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__							,    )
     SCREAMING_SNAKE_CASE								=			DDIMScheduler(
         beta_start=0.00_085							,    beta_end=0.012							,    beta_schedule='scaled_linear'							,    clip_sample=snake_case__							,    set_alpha_to_one=snake_case__							,    )
     torch.manual_seed(0    )
     SCREAMING_SNAKE_CASE								=			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    )
     SCREAMING_SNAKE_CASE								=			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							,    )
     SCREAMING_SNAKE_CASE								=			ClapTextModelWithProjection(snake_case__    )
     SCREAMING_SNAKE_CASE								=			RobertaTokenizer.from_pretrained('hf-internal-testing/tiny-random-roberta'							,    model_max_length=7_7    )

     SCREAMING_SNAKE_CASE								=			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__							,    )

     SCREAMING_SNAKE_CASE								=			SpeechTaHifiGan(snake_case__    )

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

    def UpperCamelCase			(			self :      Optional[int]							,    snake_case__ :      int							,    snake_case__ :      int=0    ):

     """simple docstring"""






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

    def UpperCamelCase			(			self :      Union[str, Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'cpu'  # ensure determinism for the device-dependent torch.Generator

     SCREAMING_SNAKE_CASE								=			self.get_dummy_components()
     SCREAMING_SNAKE_CASE								=			AudioLDMPipeline(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     audioldm_pipe.set_progress_bar_config(disable=snake_case__    )

     SCREAMING_SNAKE_CASE								=			self.get_dummy_inputs(snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			output.audios[0]

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

     SCREAMING_SNAKE_CASE								=			audio[:1_0]
     SCREAMING_SNAKE_CASE								=			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 :      Dict    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			self.get_dummy_components()
     SCREAMING_SNAKE_CASE								=			AudioLDMPipeline(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     audioldm_pipe.set_progress_bar_config(disable=snake_case__    )

     SCREAMING_SNAKE_CASE								=			self.get_dummy_inputs(snake_case__    )
     SCREAMING_SNAKE_CASE								=			3 * [inputs['prompt']]

     # forward
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			output.audios[0]

     SCREAMING_SNAKE_CASE								=			self.get_dummy_inputs(snake_case__    )
     SCREAMING_SNAKE_CASE								=			3 * [inputs.pop('prompt'    )]

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

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

     SCREAMING_SNAKE_CASE								=			prompt_embeds

     # forward
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			output.audios[0]

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

    def UpperCamelCase			(			self :      int    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			self.get_dummy_components()
     SCREAMING_SNAKE_CASE								=			AudioLDMPipeline(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     audioldm_pipe.set_progress_bar_config(disable=snake_case__    )

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

     # forward
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			output.audios[0]

     SCREAMING_SNAKE_CASE								=			self.get_dummy_inputs(snake_case__    )
     SCREAMING_SNAKE_CASE								=			3 * [inputs.pop('prompt'    )]

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

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

      embeds.append(snake_case__    )

     SCREAMING_SNAKE_CASE						,	SCREAMING_SNAKE_CASE								=			embeds

     # forward
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			output.audios[0]

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

    def UpperCamelCase			(			self :      int    ):

     """simple docstring"""






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

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

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

     SCREAMING_SNAKE_CASE								=			audio[:1_0]
     SCREAMING_SNAKE_CASE								=			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 :      str    ):

     """simple docstring"""






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

     SCREAMING_SNAKE_CASE								=			'A hammer hitting a wooden surface'

     # test num_waveforms_per_prompt=1 (default)
     SCREAMING_SNAKE_CASE								=			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
     SCREAMING_SNAKE_CASE								=			2
     SCREAMING_SNAKE_CASE								=			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
     SCREAMING_SNAKE_CASE								=			2
     SCREAMING_SNAKE_CASE								=			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
     SCREAMING_SNAKE_CASE								=			2
     SCREAMING_SNAKE_CASE								=			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 UpperCamelCase			(			self :      Union[str, Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			'cpu'  # ensure determinism for the device-dependent torch.Generator
     SCREAMING_SNAKE_CASE								=			self.get_dummy_components()
     SCREAMING_SNAKE_CASE								=			AudioLDMPipeline(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     audioldm_pipe.set_progress_bar_config(disable=snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.vocoder.config.sampling_rate

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

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

     SCREAMING_SNAKE_CASE								=			audioldm_pipe(audio_length_in_s=0.032							,    **snake_case__    )
     SCREAMING_SNAKE_CASE								=			output.audios[0]

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

    def UpperCamelCase			(			self :      Optional[Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			self.get_dummy_components()
     SCREAMING_SNAKE_CASE								=			AudioLDMPipeline(**snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     audioldm_pipe.set_progress_bar_config(disable=snake_case__    )

     SCREAMING_SNAKE_CASE								=			['hey']

     SCREAMING_SNAKE_CASE								=			audioldm_pipe(snake_case__							,    num_inference_steps=1    )
     SCREAMING_SNAKE_CASE								=			output.audios.shape
     assert audio_shape == (1, 2_5_6)

     SCREAMING_SNAKE_CASE								=			audioldm_pipe.vocoder.config
     config.model_in_dim *= 2
     SCREAMING_SNAKE_CASE								=			SpeechTaHifiGan(snake_case__    ).to(snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(snake_case__							,    num_inference_steps=1    )
     SCREAMING_SNAKE_CASE								=			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 UpperCamelCase			(			self :      Tuple    ):

     """simple docstring"""






     self._test_attention_slicing_forward_pass(test_mean_pixel_difference=snake_case__    )

    def UpperCamelCase			(			self :      int    ):

     """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 UpperCamelCase			(			self :      Dict    ):

     """simple docstring"""






     self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case__    )




@slow
class      UpperCamelCase (     unittest.TestCase   ):

    def UpperCamelCase			(			self :      Any    ):

     """simple docstring"""






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

    def UpperCamelCase			(			self :      int							,    snake_case__ :      int							,    snake_case__ :      Tuple="cpu"							,    snake_case__ :      List[str]=torch.floataa							,    snake_case__ :      Optional[Any]=0    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			torch.Generator(device=snake_case__    ).manual_seed(snake_case__    )
     SCREAMING_SNAKE_CASE								=			np.random.RandomState(snake_case__    ).standard_normal((1, 8, 1_2_8, 1_6)    )
     SCREAMING_SNAKE_CASE								=			torch.from_numpy(snake_case__    ).to(device=snake_case__							,    dtype=snake_case__    )
     SCREAMING_SNAKE_CASE								=			{
         'prompt': 'A hammer hitting a wooden surface',
         'latents': latents,
         'generator': generator,
         'num_inference_steps': 3,
         'guidance_scale': 2.5,
     }
     return inputs

    def UpperCamelCase			(			self :      int    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			AudioLDMPipeline.from_pretrained('cvssp/audioldm'    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe.to(snake_case__    )
     audioldm_pipe.set_progress_bar_config(disable=snake_case__    )

     SCREAMING_SNAKE_CASE								=			self.get_inputs(snake_case__    )
     SCREAMING_SNAKE_CASE								=			2_5
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(**snake_case__    ).audios[0]

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

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




    def UpperCamelCase			(			self :      Optional[Any]    ):

     """simple docstring"""






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

     SCREAMING_SNAKE_CASE								=			self.get_inputs(snake_case__    )
     SCREAMING_SNAKE_CASE								=			audioldm_pipe(**snake_case__    ).audios[0]

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

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

