


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







						def __a						(							self   :    str		):
								torch.manual_seed(0		)
								A					       = 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=_lowercase			,			)
								A					       = DDIMScheduler(
								    beta_start=0.0_0_0_8_5			,			beta_end=0.0_1_2			,			beta_schedule='scaled_linear'			,			clip_sample=_lowercase			,			set_alpha_to_one=_lowercase			,			)
								torch.manual_seed(0		)
								A					       = 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					       = 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=1_000			,			projection_dim=32			,			)
								A					       = ClapTextModelWithProjection(_lowercase		)
								A					       = RobertaTokenizer.from_pretrained('hf-internal-testing/tiny-random-roberta'			,			model_max_length=77		)

								A					       = SpeechTaHifiGanConfig(
								    model_in_dim=8			,			sampling_rate=16_000			,			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=_lowercase			,			)

								A					       = SpeechTaHifiGan(_lowercase		)

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







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







						def __a						(							self   :    int		):
								A					       = 'cpu'  # ensure determinism for the device-dependent torch.Generator

								A					       = self.get_dummy_components()
								A					       = AudioLDMPipeline(**_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)

								A					       = self.get_dummy_inputs(_lowercase		)
								A					       = audioldm_pipe(**_lowercase		)
								A					       = output.audios[0]

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

								A					       = audio[:10]
								A					       = np.array(
								    [-0.0_0_5_0, 0.0_0_5_0, -0.0_0_6_0, 0.0_0_3_3, -0.0_0_2_6, 0.0_0_3_3, -0.0_0_2_7, 0.0_0_3_3, -0.0_0_2_8, 0.0_0_3_3]		)

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







						def __a						(							self   :    Optional[Any]		):
								A					       = self.get_dummy_components()
								A					       = AudioLDMPipeline(**_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)

								A					       = self.get_dummy_inputs(_lowercase		)
								A					       = 3 * [inputs['prompt']]

								# forward
								A					       = audioldm_pipe(**_lowercase		)
								A					       = output.audios[0]

								A					       = self.get_dummy_inputs(_lowercase		)
								A					       = 3 * [inputs.pop('prompt'		)]

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

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

								A					       = prompt_embeds

								# forward
								A					       = audioldm_pipe(**_lowercase		)
								A					       = output.audios[0]

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







						def __a						(							self   :    Tuple		):
								A					       = self.get_dummy_components()
								A					       = AudioLDMPipeline(**_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)

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

								# forward
								A					       = audioldm_pipe(**_lowercase		)
								A					       = output.audios[0]

								A					       = self.get_dummy_inputs(_lowercase		)
								A					       = 3 * [inputs.pop('prompt'		)]

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

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

										embeds.append(_lowercase		)

								A       , A					       = embeds

								# forward
								A					       = audioldm_pipe(**_lowercase		)
								A					       = output.audios[0]

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







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

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

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

								A					       = audio[:10]
								A					       = np.array(
								    [-0.0_0_5_1, 0.0_0_5_0, -0.0_0_6_0, 0.0_0_3_4, -0.0_0_2_6, 0.0_0_3_3, -0.0_0_2_7, 0.0_0_3_3, -0.0_0_2_8, 0.0_0_3_2]		)

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







						def __a						(							self   :    str		):
								A					       = 'cpu'  # ensure determinism for the device-dependent torch.Generator
								A					       = self.get_dummy_components()
								A					       = PNDMScheduler(skip_prk_steps=_lowercase		)
								A					       = AudioLDMPipeline(**_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)

								A					       = 'A hammer hitting a wooden surface'

								# test num_waveforms_per_prompt=1 (default)
								A					       = audioldm_pipe(_lowercase			,			num_inference_steps=2		).audios

								assert audios.shape == (1, 256)

								# test num_waveforms_per_prompt=1 (default) for batch of prompts
								A					       = 2
								A					       = audioldm_pipe([prompt] * batch_size			,			num_inference_steps=2		).audios

								assert audios.shape == (batch_size, 256)

								# test num_waveforms_per_prompt for single prompt
								A					       = 2
								A					       = audioldm_pipe(_lowercase			,			num_inference_steps=2			,			num_waveforms_per_prompt=_lowercase		).audios

								assert audios.shape == (num_waveforms_per_prompt, 256)

								# test num_waveforms_per_prompt for batch of prompts
								A					       = 2
								A					       = audioldm_pipe(
								    [prompt] * batch_size			,			num_inference_steps=2			,			num_waveforms_per_prompt=_lowercase		).audios

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







						def __a						(							self   :    Any		):
								A					       = 'cpu'  # ensure determinism for the device-dependent torch.Generator
								A					       = self.get_dummy_components()
								A					       = AudioLDMPipeline(**_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)
								A					       = audioldm_pipe.vocoder.config.sampling_rate

								A					       = self.get_dummy_inputs(_lowercase		)
								A					       = audioldm_pipe(audio_length_in_s=0.0_1_6			,			**_lowercase		)
								A					       = output.audios[0]

								assert audio.ndim == 1
								assert len(_lowercase		) / vocoder_sampling_rate == 0.0_1_6

								A					       = audioldm_pipe(audio_length_in_s=0.0_3_2			,			**_lowercase		)
								A					       = output.audios[0]

								assert audio.ndim == 1
								assert len(_lowercase		) / vocoder_sampling_rate == 0.0_3_2







						def __a						(							self   :    Tuple		):
								A					       = self.get_dummy_components()
								A					       = AudioLDMPipeline(**_lowercase		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)

								A					       = ['hey']

								A					       = audioldm_pipe(_lowercase			,			num_inference_steps=1		)
								A					       = output.audios.shape
								assert audio_shape == (1, 256)

								A					       = audioldm_pipe.vocoder.config
								config.model_in_dim *= 2
								A					       = SpeechTaHifiGan(_lowercase		).to(_lowercase		)
								A					       = audioldm_pipe(_lowercase			,			num_inference_steps=1		)
								A					       = 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 __a						(							self   :    Optional[int]		):
								self._test_attention_slicing_forward_pass(test_mean_pixel_difference=_lowercase		)







						def __a						(							self   :    Tuple		):
								self._test_inference_batch_single_identical(test_mean_pixel_difference=_lowercase		)







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



@slow
class 			lowerCamelCase__						(   unittest.TestCase			):







						def __a						(							self   :    str		):
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()







						def __a						(							self   :    Tuple			,			_lowercase   :    str			,			_lowercase   :    Dict="cpu"			,			_lowercase   :    Any=torch.floataa			,			_lowercase   :    Any=0		):
								A					       = torch.Generator(device=_lowercase		).manual_seed(_lowercase		)
								A					       = np.random.RandomState(_lowercase		).standard_normal((1, 8, 128, 16)		)
								A					       = torch.from_numpy(_lowercase		).to(device=_lowercase			,			dtype=_lowercase		)
								A					       = {
								    'prompt': 'A hammer hitting a wooden surface',
								    'latents': latents,
								    'generator': generator,
								    'num_inference_steps': 3,
								    'guidance_scale': 2.5,
								}
								return inputs







						def __a						(							self   :    Optional[int]		):
								A					       = AudioLDMPipeline.from_pretrained('cvssp/audioldm'		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)

								A					       = self.get_inputs(_lowercase		)
								A					       = 25
								A					       = audioldm_pipe(**_lowercase		).audios[0]

								assert audio.ndim == 1
								assert len(_lowercase		) == 81_920

								A					       = audio[77_230:77_240]
								A					       = np.array(
								    [-0.4_8_8_4, -0.4_6_0_7, 0.0_0_2_3, 0.5_0_0_7, 0.5_8_9_6, 0.5_1_5_1, 0.3_8_1_3, -0.0_2_0_8, -0.3_6_8_7, -0.4_3_1_5]		)
								A					       = np.abs(expected_slice - audio_slice		).max()
								assert max_diff < 1e-2







						def __a						(							self   :    str		):
								A					       = AudioLDMPipeline.from_pretrained('cvssp/audioldm'		)
								A					       = LMSDiscreteScheduler.from_config(audioldm_pipe.scheduler.config		)
								A					       = audioldm_pipe.to(_lowercase		)
								audioldm_pipe.set_progress_bar_config(disable=_lowercase		)

								A					       = self.get_inputs(_lowercase		)
								A					       = audioldm_pipe(**_lowercase		).audios[0]

								assert audio.ndim == 1
								assert len(_lowercase		) == 81_920

								A					       = audio[27_780:27_790]
								A					       = np.array([-0.2_1_3_1, -0.0_8_7_3, -0.0_1_2_4, -0.0_1_8_9, 0.0_5_6_9, 0.1_3_7_3, 0.1_8_8_3, 0.2_8_8_6, 0.3_2_9_7, 0.2_2_1_2]		)
								A					       = np.abs(expected_slice - audio_slice		).max()
								assert max_diff < 3e-2









"""simple docstring"""


import os
import unittest

from transformers.models.transfo_xl.tokenization_transfo_xl import VOCAB_FILES_NAMES, TransfoXLTokenizer

from ...test_tokenization_common import TokenizerTesterMixin



class 			lowerCamelCase__						(   UpperCAmelCase_							,    unittest.TestCase			):
						lowerCAmelCase       =     TransfoXLTokenizer
						lowerCAmelCase       =     False
						lowerCAmelCase       =     False







						def __a						(							self   :    Union[str, Any]		):
								super().setUp()

								A					       = [
								    '<unk>',
								    '[CLS]',
								    '[SEP]',
								    'want',
								    'unwanted',
								    'wa',
								    'un',
								    'running',
								    ',',
								    'low',
								    'l',
								]
								A					       = os.path.join(self.tmpdirname			,			VOCAB_FILES_NAMES['vocab_file']		)
								with open(self.vocab_file			,			'w'			,			encoding='utf-8'		) as vocab_writer:
										vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]		)		)







						def __a						(							self   :    int			,			**_lowercase   :    List[str]		):
								A					       = True
								return TransfoXLTokenizer.from_pretrained(self.tmpdirname			,			**_lowercase		)







						def __a						(							self   :    int			,			_lowercase   :    List[Any]		):
								A					       = '<unk> UNwanted , running'
								A					       = '<unk> unwanted, running'
								return input_text, output_text







						def __a						(							self   :    List[str]		):
								A					       = TransfoXLTokenizer(vocab_file=self.vocab_file			,			lower_case=_lowercase		)

								A					       = tokenizer.tokenize('<unk> UNwanted , running'		)
								self.assertListEqual(_lowercase			,			['<unk>', 'unwanted', ',', 'running']		)

								self.assertListEqual(tokenizer.convert_tokens_to_ids(_lowercase		)			,			[0, 4, 8, 7]		)







						def __a						(							self   :    int		):
								A					       = TransfoXLTokenizer(lower_case=_lowercase		)

								self.assertListEqual(
								    tokenizer.tokenize(' \tHeLLo ! how  \n Are yoU ?  '		)			,			['hello', '!', 'how', 'are', 'you', '?']		)







						def __a						(							self   :    Optional[Any]		):
								A					       = TransfoXLTokenizer(lower_case=_lowercase		)

								self.assertListEqual(
								    tokenizer.tokenize(' \tHeLLo ! how  \n Are yoU ?  '		)			,			['HeLLo', '!', 'how', 'Are', 'yoU', '?']		)







						def __a						(							self   :    str		):
								A					       = TransfoXLTokenizer(lower_case=_lowercase		)
								A					       = 'Hello (bracket) and side-scrolled [and] Henry\'s $5,000 with 3.34 m. What\'s up!?'
								A					       = [
								    'Hello',
								    '(',
								    'bracket',
								    ')',
								    'and',
								    'side',
								    '@-@',
								    'scrolled',
								    '[',
								    'and',
								    ']',
								    'Henry',
								    '\'s',
								    '$',
								    '5',
								    '@,@',
								    '000',
								    'with',
								    '3',
								    '@.@',
								    '34',
								    'm',
								    '.',
								    'What',
								    '\'s',
								    'up',
								    '!',
								    '?',
								]

								self.assertListEqual(tokenizer.tokenize(_lowercase		)			,			_lowercase		)

								self.assertEqual(tokenizer.convert_tokens_to_string(_lowercase		)			,			_lowercase		)







						def __a						(							self   :    Dict		):
								A					       = self.get_tokenizer()
								A					       = len(_lowercase		)

								tokenizer.add_tokens(['new1', 'new2']		)
								tokenizer.move_added_token('new1'			,			1		)

								# Check that moved token is not copied (duplicate)
								self.assertEqual(len(_lowercase		)			,			original_len + 2		)
								# Check that token is moved to specified id
								self.assertEqual(tokenizer.encode('new1'		)			,			[1]		)
								self.assertEqual(tokenizer.decode([1]		)			,			'new1'		)





