
import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, LDMTextToImagePipeline, UNetaDConditionModel
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    load_numpy,
    nightly,
    require_torch_gpu,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



class __lowerCAmelCase				(					_lowercase     ,				unittest.TestCase      ):




			"""simple docstring"""







			snake_case	      =				LDMTextToImagePipeline
			snake_case	      =				TEXT_TO_IMAGE_PARAMS - {
			    "negative_prompt",
			    "negative_prompt_embeds",
			    "cross_attention_kwargs",
			    "prompt_embeds",
			}
			snake_case	      =				PipelineTesterMixin.required_optional_params - {
			    "num_images_per_prompt",
			    "callback",
			    "callback_steps",
			}
			snake_case	      =				TEXT_TO_IMAGE_BATCH_PARAMS
			snake_case	      =				False




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







								"""simple docstring"""
								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       ,   )
								A_				 =						DDIMScheduler(
								    beta_start=0.0_0_0_8_5       ,   beta_end=0.0_1_2       ,   beta_schedule="scaled_linear"       ,   clip_sample=_snake_case       ,   set_alpha_to_one=_snake_case       ,   )
								torch.manual_seed(0	)
								A_				 =						AutoencoderKL(
								    block_out_channels=(32, 64)       ,   in_channels=3       ,   out_channels=3       ,   down_block_types=("DownEncoderBlock2D", "DownEncoderBlock2D")       ,   up_block_types=("UpDecoderBlock2D", "UpDecoderBlock2D")       ,   latent_channels=4       ,   )
								torch.manual_seed(0	)
								A_				 =						CLIPTextConfig(
								    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       ,   )
								A_				 =						CLIPTextModel(_snake_case	)
								A_				 =						CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"	)

								A_				 =						{
								    "unet": unet,
								    "scheduler": scheduler,
								    "vqvae": vae,
								    "bert": text_encoder,
								    "tokenizer": tokenizer,
								}
								return components




			def      lowerCamelCase__     (				self						:					List[Any]       ,   _snake_case						:					Optional[int]       ,   _snake_case						:					Union[str, Any]=0	)      -> Optional[Any]:







								"""simple docstring"""
								if str(_snake_case	).startswith("mps"	):
													A_				 =						torch.manual_seed(_snake_case	)
								else:
													A_				 =						torch.Generator(device=_snake_case	).manual_seed(_snake_case	)
								A_				 =						{
								    "prompt": "A painting of a squirrel eating a burger",
								    "generator": generator,
								    "num_inference_steps": 2,
								    "guidance_scale": 6.0,
								    "output_type": "numpy",
								}
								return inputs






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







								"""simple docstring"""
								A_				 =						"cpu"  # ensure determinism for the device-dependent torch.Generator

								A_				 =						self.get_dummy_components()
								A_				 =						LDMTextToImagePipeline(**_snake_case	)
								pipe.to(_snake_case	)
								pipe.set_progress_bar_config(disable=_snake_case	)

								A_				 =						self.get_dummy_inputs(_snake_case	)
								A_				 =						pipe(**_snake_case	).images
								A_				 =						image[0, -3:, -3:, -1]

								assert image.shape == (1, 16, 16, 3)
								A_				 =						np.array([0.6_1_0_1, 0.6_1_5_6, 0.5_6_2_2, 0.4_8_9_5, 0.6_6_6_1, 0.3_8_0_4, 0.5_7_4_8, 0.6_1_3_6, 0.5_0_1_4]	)

								assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3





@slow
@require_torch_gpu
class __lowerCAmelCase				(					unittest.TestCase      ):




			"""simple docstring"""







			def      lowerCamelCase__     (				self						:					str	)      -> Dict:







								"""simple docstring"""
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()




			def      lowerCamelCase__     (				self						:					Dict       ,   _snake_case						:					Optional[int]       ,   _snake_case						:					Optional[Any]=torch.floataa       ,   _snake_case						:					List[str]=0	)      -> List[str]:







								"""simple docstring"""
								A_				 =						torch.manual_seed(_snake_case	)
								A_				 =						np.random.RandomState(_snake_case	).standard_normal((1, 4, 32, 32)	)
								A_				 =						torch.from_numpy(_snake_case	).to(device=_snake_case       ,   dtype=_snake_case	)
								A_				 =						{
								    "prompt": "A painting of a squirrel eating a burger",
								    "latents": latents,
								    "generator": generator,
								    "num_inference_steps": 3,
								    "guidance_scale": 6.0,
								    "output_type": "numpy",
								}
								return inputs






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







								"""simple docstring"""
								A_				 =						LDMTextToImagePipeline.from_pretrained("CompVis/ldm-text2im-large-256"	).to(_snake_case	)
								pipe.set_progress_bar_config(disable=_snake_case	)

								A_				 =						self.get_inputs(_snake_case	)
								A_				 =						pipe(**_snake_case	).images
								A_				 =						image[0, -3:, -3:, -1].flatten()

								assert image.shape == (1, 256, 256, 3)
								A_				 =						np.array([0.5_1_8_2_5, 0.5_2_8_5_0, 0.5_2_5_4_3, 0.5_4_2_5_8, 0.5_2_3_0_4, 0.5_2_5_6_9, 0.5_4_3_6_3, 0.5_5_2_7_6, 0.5_6_8_7_8]	)
								A_				 =						np.abs(expected_slice - image_slice	).max()
								assert max_diff < 1e-3





@nightly
@require_torch_gpu
class __lowerCAmelCase				(					unittest.TestCase      ):




			"""simple docstring"""







			def      lowerCamelCase__     (				self						:					Any	)      -> Dict:







								"""simple docstring"""
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()




			def      lowerCamelCase__     (				self						:					Union[str, Any]       ,   _snake_case						:					Optional[int]       ,   _snake_case						:					Optional[int]=torch.floataa       ,   _snake_case						:					List[Any]=0	)      -> Optional[int]:







								"""simple docstring"""
								A_				 =						torch.manual_seed(_snake_case	)
								A_				 =						np.random.RandomState(_snake_case	).standard_normal((1, 4, 32, 32)	)
								A_				 =						torch.from_numpy(_snake_case	).to(device=_snake_case       ,   dtype=_snake_case	)
								A_				 =						{
								    "prompt": "A painting of a squirrel eating a burger",
								    "latents": latents,
								    "generator": generator,
								    "num_inference_steps": 50,
								    "guidance_scale": 6.0,
								    "output_type": "numpy",
								}
								return inputs






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







								"""simple docstring"""
								A_				 =						LDMTextToImagePipeline.from_pretrained("CompVis/ldm-text2im-large-256"	).to(_snake_case	)
								pipe.set_progress_bar_config(disable=_snake_case	)

								A_				 =						self.get_inputs(_snake_case	)
								A_				 =						pipe(**_snake_case	).images[0]

								A_				 =						load_numpy(
								    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/ldm_text2img/ldm_large_256_ddim.npy"	)
								A_				 =						np.abs(expected_image - image	).max()
								assert max_diff < 1e-3






"""simple docstring"""





import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
		import torch

if is_datasets_available():
		from datasets import load_dataset

UpperCamelCase_					:						Union[str, Any]				= random.Random()
def 							A_     (__a				,   __a=1.0				,   __a=None				,   __a=None				):

					'''simple docstring'''





					if rng is None:
										A_				 =						global_rng

					A_				 =						[]
					for batch_idx in range(shape[0]				):
										values.append([]				)
										for _ in range(shape[1]				):
															values[-1].append(rng.random() * scale				)

					return values



class __lowerCAmelCase				(					unittest.TestCase      ):




			"""simple docstring"""







			def __init__(				self						:					Optional[Any]       ,   _snake_case						:					str       ,   _snake_case						:					Union[str, Any]=7       ,   _snake_case						:					Any=400       ,   _snake_case						:					Tuple=2_000       ,   _snake_case						:					str=2_048       ,   _snake_case						:					List[Any]=128       ,   _snake_case						:					List[Any]=1       ,   _snake_case						:					Union[str, Any]=512       ,   _snake_case						:					List[str]=30       ,   _snake_case						:					List[str]=44_100       ,   )      -> Optional[int]:







								"""simple docstring"""
								A_				 =						parent
								A_				 =						batch_size
								A_				 =						min_seq_length
								A_				 =						max_seq_length
								A_				 =						(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
								A_				 =						spectrogram_length
								A_				 =						feature_size
								A_				 =						num_audio_channels
								A_				 =						hop_length
								A_				 =						chunk_length
								A_				 =						sampling_rate




			def      lowerCamelCase__     (				self						:					Any	)      -> Optional[int]:







								"""simple docstring"""
								return {
								    "spectrogram_length": self.spectrogram_length,
								    "feature_size": self.feature_size,
								    "num_audio_channels": self.num_audio_channels,
								    "hop_length": self.hop_length,
								    "chunk_length": self.chunk_length,
								    "sampling_rate": self.sampling_rate,
								}






			def      lowerCamelCase__     (				self						:					Dict       ,   _snake_case						:					str=False       ,   _snake_case						:					Optional[Any]=False	)      -> List[str]:







								"""simple docstring"""
								def _flatten(_snake_case						:					List[Any]	):
													return list(itertools.chain(*_snake_case	)	)

								if equal_length:
													A_				 =						[floats_list((self.max_seq_length, self.feature_size)	) for _ in range(self.batch_size	)]
								else:
													# make sure that inputs increase in size
													A_				 =						[
													    floats_list((x, self.feature_size)	)
													    for x in range(self.min_seq_length       ,   self.max_seq_length       ,   self.seq_length_diff	)
													]
								if numpify:
													A_				 =						[np.asarray(_snake_case	) for x in speech_inputs]
								return speech_inputs





@require_torch
@require_torchaudio
class __lowerCAmelCase				(					_lowercase     ,				unittest.TestCase      ):




			"""simple docstring"""







			snake_case	      =				TvltFeatureExtractor




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







								"""simple docstring"""
								A_				 =						TvltFeatureExtractionTester(self	)




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







								"""simple docstring"""
								A_				 =						self.feature_extraction_class(**self.feat_extract_dict	)
								self.assertTrue(hasattr(_snake_case       ,   "spectrogram_length"	)	)
								self.assertTrue(hasattr(_snake_case       ,   "feature_size"	)	)
								self.assertTrue(hasattr(_snake_case       ,   "num_audio_channels"	)	)
								self.assertTrue(hasattr(_snake_case       ,   "hop_length"	)	)
								self.assertTrue(hasattr(_snake_case       ,   "chunk_length"	)	)
								self.assertTrue(hasattr(_snake_case       ,   "sampling_rate"	)	)




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







								"""simple docstring"""
								A_				 =						self.feature_extraction_class(**self.feat_extract_dict	)

								with tempfile.TemporaryDirectory() as tmpdirname:
													A_				 =						feat_extract_first.save_pretrained(_snake_case	)[0]
													check_json_file_has_correct_format(_snake_case	)
													A_				 =						self.feature_extraction_class.from_pretrained(_snake_case	)

								A_				 =						feat_extract_first.to_dict()
								A_				 =						feat_extract_second.to_dict()
								A_				 =						dict_first.pop("mel_filters"	)
								A_				 =						dict_second.pop("mel_filters"	)
								self.assertTrue(np.allclose(_snake_case       ,   _snake_case	)	)
								self.assertEqual(_snake_case       ,   _snake_case	)




			def      lowerCamelCase__     (				self						:					Optional[int]	)      -> Tuple:







								"""simple docstring"""
								A_				 =						self.feature_extraction_class(**self.feat_extract_dict	)

								with tempfile.TemporaryDirectory() as tmpdirname:
													A_				 =						os.path.join(_snake_case       ,   "feat_extract.json"	)
													feat_extract_first.to_json_file(_snake_case	)
													A_				 =						self.feature_extraction_class.from_json_file(_snake_case	)

								A_				 =						feat_extract_first.to_dict()
								A_				 =						feat_extract_second.to_dict()
								A_				 =						dict_first.pop("mel_filters"	)
								A_				 =						dict_second.pop("mel_filters"	)
								self.assertTrue(np.allclose(_snake_case       ,   _snake_case	)	)
								self.assertEqual(_snake_case       ,   _snake_case	)




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







								"""simple docstring"""
								# Initialize feature_extractor
								A_				 =						self.feature_extraction_class(**self.feat_extract_dict	)

								# create three inputs of length 800, 1000, and 1200
								A_				 =						[floats_list((1, x)	)[0] for x in range(800       ,   1_400       ,   200	)]
								A_				 =						[np.asarray(_snake_case	) for speech_input in speech_inputs]

								# Test not batched input
								A_				 =						feature_extractor(np_speech_inputs[0]       ,   return_tensors="np"       ,   sampling_rate=44_100	).audio_values

								self.assertTrue(encoded_audios.ndim == 4	)
								self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size	)
								self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length	)
								self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels	)

								# Test batched
								A_				 =						feature_extractor(_snake_case       ,   return_tensors="np"       ,   sampling_rate=44_100	).audio_values

								self.assertTrue(encoded_audios.ndim == 4	)
								self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size	)
								self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length	)
								self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels	)

								# Test audio masking
								A_				 =						feature_extractor(
								    _snake_case       ,   return_tensors="np"       ,   sampling_rate=44_100       ,   mask_audio=_snake_case	).audio_values

								self.assertTrue(encoded_audios.ndim == 4	)
								self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size	)
								self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length	)
								self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels	)

								# Test 2-D numpy arrays are batched.
								A_				 =						[floats_list((1, x)	)[0] for x in (800, 800, 800)]
								A_				 =						np.asarray(_snake_case	)
								A_				 =						feature_extractor(_snake_case       ,   return_tensors="np"       ,   sampling_rate=44_100	).audio_values
								self.assertTrue(encoded_audios.ndim == 4	)
								self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size	)
								self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length	)
								self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels	)




			def      lowerCamelCase__     (				self						:					Dict       ,   _snake_case						:					int	)      -> List[str]:







								"""simple docstring"""
								A_				 =						load_dataset("hf-internal-testing/librispeech_asr_dummy"       ,   "clean"       ,   split="validation"	)
								# automatic decoding with librispeech
								A_				 =						ds.sort("id"	).select(range(_snake_case	)	)[:num_samples]["audio"]

								return [x["array"] for x in speech_samples]






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







								"""simple docstring"""
								A_				 =						self._load_datasamples(1	)
								A_				 =						TvltFeatureExtractor()
								A_				 =						feature_extractor(_snake_case       ,   return_tensors="pt"	).audio_values

								self.assertEquals(audio_values.shape       ,   (1, 1, 192, 128)	)

								A_				 =						torch.tensor([[-0.3_0_3_2, -0.2_7_0_8], [-0.4_4_3_4, -0.4_0_0_7]]	)
								self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]       ,   _snake_case       ,   atol=1e-4	)	)




