



'''simple docstring'''






import gc
import unittest

import numpy as np
import torch
import torch.nn.functional as F
from transformers import (
    ClapTextConfig,
    ClapTextModelWithProjection,
    RobertaTokenizer,
    SpeechTaHifiGan,
    SpeechTaHifiGanConfig,
)

from diffusers import (
    AudioLDMPipeline,
    AutoencoderKL,
    DDIMScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_AUDIO_BATCH_PARAMS, TEXT_TO_AUDIO_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()

class 							a__( UpperCamelCase__		,			unittest.TestCase     ):
				lowercase__			=   AudioLDMPipeline
				lowercase__			=   TEXT_TO_AUDIO_PARAMS
				lowercase__			=   TEXT_TO_AUDIO_BATCH_PARAMS
				lowercase__			=   frozenset(
				    [
				        """num_inference_steps""",
				        """num_waveforms_per_prompt""",
				        """generator""",
				        """latents""",
				        """output_type""",
				        """return_dict""",
				        """callback""",
				        """callback_steps""",
				    ]     )







				def       lowercase_	(				self			:  List[str]     ):
						torch.manual_seed(0     )
						a						: Dict					       =					UNetaDConditionModel(
						    block_out_channels=(32, 64)						,      layers_per_block=2						,      sample_size=32						,      in_channels=4						,      out_channels=4						,      down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')						,      up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')						,      cross_attention_dim=(32, 64)						,      class_embed_type='simple_projection'						,      projection_class_embeddings_input_dim=32						,      class_embeddings_concat=__a						,      )
						a						: Any					       =					DDIMScheduler(
						    beta_start=0.00085						,      beta_end=0.012						,      beta_schedule='scaled_linear'						,      clip_sample=__a						,      set_alpha_to_one=__a						,      )
						torch.manual_seed(0     )
						a						: List[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     )
						a						: List[str]					       =					ClapTextConfig(
						    bos_token_id=0						,      eos_token_id=2						,      hidden_size=32						,      intermediate_size=37						,      layer_norm_eps=1e-0_5						,      num_attention_heads=4						,      num_hidden_layers=5						,      pad_token_id=1						,      vocab_size=10_00						,      projection_dim=32						,      )
						a						: Any					       =					ClapTextModelWithProjection(__a     )
						a						: int					       =					RobertaTokenizer.from_pretrained('hf-internal-testing/tiny-random-roberta'						,      model_max_length=77     )

						a						: Tuple					       =					SpeechTaHifiGanConfig(
						    model_in_dim=8						,      sampling_rate=1_60_00						,      upsample_initial_channel=16						,      upsample_rates=[2, 2]						,      upsample_kernel_sizes=[4, 4]						,      resblock_kernel_sizes=[3, 7]						,      resblock_dilation_sizes=[[1, 3, 5], [1, 3, 5]]						,      normalize_before=__a						,      )

						a						: Dict					       =					SpeechTaHifiGan(__a     )

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







				def       lowercase_	(				self			:  Dict						,      __snake_case			:  Any						,      __snake_case			:  Any=0     ):
						if str(__a     ).startswith('mps'     ):
								a						: List[str]					       =					torch.manual_seed(__a     )
						else:
								a						: int					       =					torch.Generator(device=__a     ).manual_seed(__a     )
						a						: Any					       =					{
						    'prompt': 'A hammer hitting a wooden surface',
						    'generator': generator,
						    'num_inference_steps': 2,
						    'guidance_scale': 6.0,
						}
						return inputs







				def       lowercase_	(				self			:  List[Any]     ):
						a						: Any					       =					'cpu'  # ensure determinism for the device-dependent torch.Generator

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

						a						: Tuple					       =					self.get_dummy_inputs(__a     )
						a						: Tuple					       =					audioldm_pipe(**__a     )
						a						: List[str]					       =					output.audios[0]

						assert audio.ndim == 1
						assert len(__a     ) == 2_56

						a						: int					       =					audio[:10]
						a						: Dict					       =					np.array(
						    [-0.0050, 0.0050, -0.0060, 0.0033, -0.0026, 0.0033, -0.0027, 0.0033, -0.0028, 0.0033]     )

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







				def       lowercase_	(				self			:  List[Any]     ):
						a						: Optional[int]					       =					self.get_dummy_components()
						a						: Union[str, Any]					       =					AudioLDMPipeline(**__a     )
						a						: Tuple					       =					audioldm_pipe.to(__a     )
						a						: Optional[int]					       =					audioldm_pipe.to(__a     )
						audioldm_pipe.set_progress_bar_config(disable=__a     )

						a						: int					       =					self.get_dummy_inputs(__a     )
						a						: int					       =					3 * [inputs['prompt']]

						# forward
						a						: Tuple					       =					audioldm_pipe(**__a     )
						a						: List[Any]					       =					output.audios[0]

						a						: Optional[Any]					       =					self.get_dummy_inputs(__a     )
						a						: Any					       =					3 * [inputs.pop('prompt'     )]

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

						a						: Union[str, Any]					       =					audioldm_pipe.text_encoder(
						    __a						,      )
						a						: Any					       =					prompt_embeds.text_embeds
						# additional L_2 normalization over each hidden-state
						a						: Dict					       =					F.normalize(__a						,      dim=-1     )

						a						: Union[str, Any]					       =					prompt_embeds

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

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







				def       lowercase_	(				self			:  Union[str, Any]     ):
						a						: Tuple					       =					self.get_dummy_components()
						a						: int					       =					AudioLDMPipeline(**__a     )
						a						: Tuple					       =					audioldm_pipe.to(__a     )
						a						: str					       =					audioldm_pipe.to(__a     )
						audioldm_pipe.set_progress_bar_config(disable=__a     )

						a						: Union[str, Any]					       =					self.get_dummy_inputs(__a     )
						a						: List[str]					       =					3 * ['this is a negative prompt']
						a						: List[Any]					       =					negative_prompt
						a						: Optional[Any]					       =					3 * [inputs['prompt']]

						# forward
						a						: List[Any]					       =					audioldm_pipe(**__a     )
						a						: Tuple					       =					output.audios[0]

						a						: Union[str, Any]					       =					self.get_dummy_inputs(__a     )
						a						: List[str]					       =					3 * [inputs.pop('prompt'     )]

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

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

								embeds.append(__a     )

						a       ,     a						: List[str]					       =					embeds

						# forward
						a						: List[str]					       =					audioldm_pipe(**__a     )
						a						: Dict					       =					output.audios[0]

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







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

						a						: Optional[int]					       =					self.get_dummy_inputs(__a     )
						a						: Dict					       =					'egg cracking'
						a						: Dict					       =					audioldm_pipe(**__a						,      negative_prompt=__a     )
						a						: Dict					       =					output.audios[0]

						assert audio.ndim == 1
						assert len(__a     ) == 2_56

						a						: Union[str, Any]					       =					audio[:10]
						a						: Optional[Any]					       =					np.array(
						    [-0.0051, 0.0050, -0.0060, 0.0034, -0.0026, 0.0033, -0.0027, 0.0033, -0.0028, 0.0032]     )

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







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

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

						# test num_waveforms_per_prompt=1 (default)
						a						: Any					       =					audioldm_pipe(__a						,      num_inference_steps=2     ).audios

						assert audios.shape == (1, 2_56)

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

						assert audios.shape == (batch_size, 2_56)

						# test num_waveforms_per_prompt for single prompt
						a						: Optional[Any]					       =					2
						a						: Tuple					       =					audioldm_pipe(__a						,      num_inference_steps=2						,      num_waveforms_per_prompt=__a     ).audios

						assert audios.shape == (num_waveforms_per_prompt, 2_56)

						# test num_waveforms_per_prompt for batch of prompts
						a						: List[str]					       =					2
						a						: str					       =					audioldm_pipe(
						    [prompt] * batch_size						,      num_inference_steps=2						,      num_waveforms_per_prompt=__a     ).audios

						assert audios.shape == (batch_size * num_waveforms_per_prompt, 2_56)







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

						a						: Tuple					       =					self.get_dummy_inputs(__a     )
						a						: Optional[Any]					       =					audioldm_pipe(audio_length_in_s=0.016						,      **__a     )
						a						: Union[str, Any]					       =					output.audios[0]

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

						a						: List[str]					       =					audioldm_pipe(audio_length_in_s=0.032						,      **__a     )
						a						: Tuple					       =					output.audios[0]

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







				def       lowercase_	(				self			:  Dict     ):
						a						: int					       =					self.get_dummy_components()
						a						: List[str]					       =					AudioLDMPipeline(**__a     )
						a						: List[Any]					       =					audioldm_pipe.to(__a     )
						audioldm_pipe.set_progress_bar_config(disable=__a     )

						a						: Optional[Any]					       =					['hey']

						a						: int					       =					audioldm_pipe(__a						,      num_inference_steps=1     )
						a						: Tuple					       =					output.audios.shape
						assert audio_shape == (1, 2_56)

						a						: str					       =					audioldm_pipe.vocoder.config
						config.model_in_dim *= 2
						a						: Any					       =					SpeechTaHifiGan(__a     ).to(__a     )
						a						: Optional[int]					       =					audioldm_pipe(__a						,      num_inference_steps=1     )
						a						: 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_56)







				def       lowercase_	(				self			:  Union[str, Any]     ):
						self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__a     )







				def       lowercase_	(				self			:  Optional[Any]     ):
						self._test_inference_batch_single_identical(test_mean_pixel_difference=__a     )







				@unittest.skipIf(
				    torch_device != 'cuda' or not is_xformers_available()						,      reason='XFormers attention is only available with CUDA and `xformers` installed'						,      )
				def       lowercase_	(				self			:  str     ):
						self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__a     )

@slow
class 							a__( unittest.TestCase     ):







				def       lowercase_	(				self			:  Tuple     ):
						super().tearDown()
						gc.collect()
						torch.cuda.empty_cache()







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







				def       lowercase_	(				self			:  str     ):
						a						: List[Any]					       =					AudioLDMPipeline.from_pretrained('cvssp/audioldm'     )
						a						: Tuple					       =					audioldm_pipe.to(__a     )
						audioldm_pipe.set_progress_bar_config(disable=__a     )

						a						: int					       =					self.get_inputs(__a     )
						a						: Optional[Any]					       =					25
						a						: str					       =					audioldm_pipe(**__a     ).audios[0]

						assert audio.ndim == 1
						assert len(__a     ) == 8_19_20

						a						: Tuple					       =					audio[7_72_30:7_72_40]
						a						: Any					       =					np.array(
						    [-0.4884, -0.4607, 0.0023, 0.5007, 0.5896, 0.5151, 0.3813, -0.0208, -0.3687, -0.4315]     )
						a						: int					       =					np.abs(expected_slice - audio_slice     ).max()
						assert max_diff < 1e-2







				def       lowercase_	(				self			:  Tuple     ):
						a						: int					       =					AudioLDMPipeline.from_pretrained('cvssp/audioldm'     )
						a						: List[str]					       =					LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config     )
						a						: int					       =					audioldm_pipe.to(__a     )
						audioldm_pipe.set_progress_bar_config(disable=__a     )

						a						: Tuple					       =					self.get_inputs(__a     )
						a						: Dict					       =					audioldm_pipe(**__a     ).audios[0]

						assert audio.ndim == 1
						assert len(__a     ) == 8_19_20

						a						: List[str]					       =					audio[2_77_80:2_77_90]
						a						: Dict					       =					np.array([-0.2131, -0.0873, -0.0124, -0.0189, 0.0569, 0.1373, 0.1883, 0.2886, 0.3297, 0.2212]     )
						a						: Tuple					       =					np.abs(expected_slice - audio_slice     ).max()
						assert max_diff < 3e-2




'''simple docstring'''






import torch

from transformers import AutoModel

class 							a__( torch.nn.Module     ):







    def __init__(				self			:  Any						,      __snake_case			:  str="sayef/fsner-bert-base-uncased"     ):
      super(__snake_case						,      self     ).__init__()

      a						: List[str]					       =					AutoModel.from_pretrained(__snake_case						,      return_dict=__snake_case     )
      a						: Optional[Any]					       =					torch.nn.CosineSimilarity(3						,      1e-0_8     )
      a						: Tuple					       =					torch.nn.Softmax(dim=1     )







    def       lowercase_	(				self			:  List[str]						,      **__snake_case			:  int     ):
      return self.bert(**__snake_case     ).last_hidden_state







    def       lowercase_	(				self			:  Optional[int]						,      __snake_case			:  Union[str, Any]     ):
      return token_embeddings.sum(2						,      keepdim=__snake_case     )







    def       lowercase_	(				self			:  str						,      __snake_case			:  Tuple						,      __snake_case			:  List[Any]						,      __snake_case			:  Optional[Any]=1     ):
      return self.softmax(T * self.cos(__snake_case						,      __snake_case     )     )







    def       lowercase_	(				self			:  int						,      __snake_case			:  int						,      __snake_case			:  Tuple     ):
      a						: List[Any]					       =					W_supports['sizes'].tolist()
      a						: Any					       =					W_supports['start_token_id'].item()
      a						: int					       =					W_supports['end_token_id'].item()

      del W_supports["sizes"]
      del W_supports["start_token_id"]
      del W_supports["end_token_id"]

      a						: Optional[Any]					       =					self.BERT(**__snake_case     )
      a						: Tuple					       =					self.BERT(**__snake_case     )

      a						: Dict					       =					None
      a						: Optional[Any]					       =					None

      a						: Union[str, Any]					       =					W_supports['input_ids'] == start_token_id
      a						: str					       =					W_supports['input_ids'] == end_token_id

      for i, size in enumerate(__snake_case     ):
        if i == 0:
          a						: Optional[int]					       =					0
        else:
          a						: Tuple					       =					support_sizes[i - 1]

        a						: Tuple					       =					S[s : s + size][start_token_masks[s : s + size]]
        a						: int					       =					S[s : s + size][end_token_masks[s : s + size]]

        a						: Any					       =					torch.matmul(q[i]						,      s_start.T     ).sum(1     ).softmax(0     )
        a						: List[str]					       =					torch.matmul(q[i]						,      s_end.T     ).sum(1     ).softmax(0     )

        if p_starts is not None:
          a						: List[Any]					       =					torch.vstack((p_starts, p_start)     )
          a						: List[Any]					       =					torch.vstack((p_ends, p_end)     )
        else:
          a						: List[Any]					       =					p_start
          a						: Optional[int]					       =					p_end

      return p_starts, p_ends