





'''simple docstring'''






import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


lowerCAmelCase									=   False
@skip_mps
class 		lowerCamelCase	(   _A   , _A   , _A   , unittest.TestCase							):
	snake_case_   = StableDiffusionAttendAndExcitePipeline
	snake_case_   = False
	snake_case_   = TEXT_TO_IMAGE_PARAMS
	snake_case_   = TEXT_TO_IMAGE_BATCH_PARAMS.union({"token_indices"}							)
	snake_case_   = TEXT_TO_IMAGE_IMAGE_PARAMS
	snake_case_   = TEXT_TO_IMAGE_IMAGE_PARAMS



	@classmethod
	def 	_lowerCamelCase     ( cls						):
			super().setUpClass()
			torch.use_deterministic_algorithms(a_						)



	@classmethod
	def 	_lowerCamelCase     ( cls						):
			super().tearDownClass()
			torch.use_deterministic_algorithms(a_						)



	def 	_lowerCamelCase     ( self						):
			torch.manual_seed(0						)
			lowerCAmelCase  :     int       						= UNetaDConditionModel(
			    block_out_channels=(32, 64)		,   layers_per_block=1		,   sample_size=32		,   in_channels=4		,   out_channels=4		,   down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")		,   up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")		,   cross_attention_dim=32		,   attention_head_dim=(2, 4)		,   use_linear_projection=a_		,   )
			lowerCAmelCase  :     Union[str, 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						)
			lowerCAmelCase  :     Tuple       						= 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		,   sample_size=128		,   )
			torch.manual_seed(0						)
			lowerCAmelCase  :     int       						= CLIPTextConfig(
			    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=1_000		,   hidden_act="gelu"		,   projection_dim=512		,   )
			lowerCAmelCase  :     Tuple       						= CLIPTextModel(a_						)
			lowerCAmelCase  :     Union[str, Any]       						= CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"						)

			lowerCAmelCase  :     Dict       						= {
			    "unet": unet,
			    "scheduler": scheduler,
			    "vae": vae,
			    "text_encoder": text_encoder,
			    "tokenizer": tokenizer,
			    "safety_checker": None,
			    "feature_extractor": None,
			}

			return components



	def 	_lowerCamelCase     ( self		,   a_		,   a_=0						):
			if str(a_						).startswith("mps"						):
					lowerCAmelCase  :     str       						= torch.manual_seed(a_						)
			else:
					lowerCAmelCase  :     int       						= torch.Generator(device=a_						).manual_seed(a_						)
			lowerCAmelCase  :     Any       						= {
			    "prompt": "a cat and a frog",
			    "token_indices": [2, 5],
			    "generator": generator,
			    "num_inference_steps": 1,
			    "guidance_scale": 6.0,
			    "output_type": "numpy",
			    "max_iter_to_alter": 2,
			    "thresholds": {0: 0.7},
			}
			return inputs



	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     str       						= "cpu"

			lowerCAmelCase  :     int       						= self.get_dummy_components()
			lowerCAmelCase  :     Union[str, Any]       						= self.pipeline_class(**a_						)
			pipe.to(a_						)
			pipe.set_progress_bar_config(disable=a_						)

			lowerCAmelCase  :     Optional[Any]       						= self.get_dummy_inputs(a_						)
			lowerCAmelCase  :     Optional[Any]       						= pipe(**a_						).images
			lowerCAmelCase  :     Tuple       						= image[0, -3:, -3:, -1]

			self.assertEqual(image.shape		,   (1, 64, 64, 3)						)
			lowerCAmelCase  :     Union[str, Any]       						= np.array(
			    [0.63905364, 0.62897307, 0.48599017, 0.5133624, 0.5550048, 0.45769516, 0.50326973, 0.5023139, 0.45384496]						)
			lowerCAmelCase  :     Tuple       						= np.abs(image_slice.flatten() - expected_slice						).max()
			self.assertLessEqual(a_		,   1e-3						)



	def 	_lowerCamelCase     ( self						):
			super().test_cpu_offload_forward_pass(expected_max_diff=5e-4						)



	def 	_lowerCamelCase     ( self						):
			# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
			self._test_inference_batch_consistent(batch_sizes=[1, 2]						)



	def 	_lowerCamelCase     ( self						):
			self._test_inference_batch_single_identical(batch_size=2		,   expected_max_diff=7e-4						)



	def 	_lowerCamelCase     ( self						):
			super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3						)



	def 	_lowerCamelCase     ( self						):
			super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4						)



	def 	_lowerCamelCase     ( self						):
			super().test_save_load_local(expected_max_difference=5e-4						)





	def 	_lowerCamelCase     ( self						):
			super().test_save_load_optional_components(expected_max_difference=4e-4						)






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



	@classmethod
	def 	_lowerCamelCase     ( cls						):
			super().setUpClass()
			torch.use_deterministic_algorithms(a_						)



	@classmethod
	def 	_lowerCamelCase     ( cls						):
			super().tearDownClass()
			torch.use_deterministic_algorithms(a_						)



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





	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     List[Any]       						= torch.manual_seed(51						)

			lowerCAmelCase  :     List[Any]       						= StableDiffusionAttendAndExcitePipeline.from_pretrained(
			    "CompVis/stable-diffusion-v1-4"		,   safety_checker=a_		,   torch_dtype=torch.floataa						)
			pipe.to("cuda"						)

			lowerCAmelCase  :     List[str]       						= "a painting of an elephant with glasses"
			lowerCAmelCase  :     Dict       						= [5, 7]

			lowerCAmelCase  :     List[str]       						= pipe(
			    prompt=a_		,   token_indices=a_		,   guidance_scale=7.5		,   generator=a_		,   num_inference_steps=5		,   max_iter_to_alter=5		,   output_type="numpy"		,   ).images[0]

			lowerCAmelCase  :     Any       						= load_numpy(
			    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy"						)
			assert np.abs((expected_image - image).max()						) < 5e-1










'''simple docstring'''






import itertools
import random
import unittest

import numpy as np

from transformers import WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST, WavaVecaConfig, WavaVecaFeatureExtractor
from transformers.testing_utils import require_torch, slow

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


lowerCAmelCase									=   random.Random()






def   __A    (				a_    :	Union[str, Any]   ,a_    :	Tuple=1.0   ,a_    :	Optional[int]=None   ,a_    :	Union[str, Any]=None						):
		if rng is None:
				lowerCAmelCase  :     str       						= global_rng

		lowerCAmelCase  :     Any       						= []
		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							):



	def __init__( self		,   a_		,   a_=7		,   a_=400		,   a_=2_000		,   a_=1		,   a_=0.0		,   a_=16_000		,   a_=True		,   a_=True		,   ):
			lowerCAmelCase  :     Dict       						= parent
			lowerCAmelCase  :     Optional[int]       						= batch_size
			lowerCAmelCase  :     Dict       						= min_seq_length
			lowerCAmelCase  :     Dict       						= max_seq_length
			lowerCAmelCase  :     Optional[Any]       						= (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
			lowerCAmelCase  :     Union[str, Any]       						= feature_size
			lowerCAmelCase  :     Tuple       						= padding_value
			lowerCAmelCase  :     Dict       						= sampling_rate
			lowerCAmelCase  :     int       						= return_attention_mask
			lowerCAmelCase  :     Optional[Any]       						= do_normalize



	def 	_lowerCamelCase     ( self						):
			return {
			    "feature_size": self.feature_size,
			    "padding_value": self.padding_value,
			    "sampling_rate": self.sampling_rate,
			    "return_attention_mask": self.return_attention_mask,
			    "do_normalize": self.do_normalize,
			}





	def 	_lowerCamelCase     ( self		,   a_=False		,   a_=False						):
			def _flatten(a_						):
					return list(itertools.chain(*a_						)						)

			if equal_length:
					lowerCAmelCase  :     List[Any]       						= floats_list((self.batch_size, self.max_seq_length)						)
			else:
					# make sure that inputs increase in size
					lowerCAmelCase  :     Any       						= [
					    _flatten(floats_list((x, self.feature_size)						)						)
					    for x in range(self.min_seq_length		,   self.max_seq_length		,   self.seq_length_diff						)
					]

			if numpify:
					lowerCAmelCase  :     List[str]       						= [np.asarray(a_						) for x in speech_inputs]

			return speech_inputs






class 		lowerCamelCase	(   _A   , unittest.TestCase							):
	snake_case_   = WavaVecaFeatureExtractor



	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     Tuple       						= WavaVecaFeatureExtractionTester(self						)



	def 	_lowerCamelCase     ( self		,   a_						):
			self.assertTrue(np.all(np.mean(a_		,   axis=0						) < 1e-3						)						)
			self.assertTrue(np.all(np.abs(np.var(a_		,   axis=0						) - 1						) < 1e-3						)						)



	def 	_lowerCamelCase     ( self						):
			# Tests that all call wrap to encode_plus and batch_encode_plus
			lowerCAmelCase  :     int       						= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
			# create three inputs of length 800, 1000, and 1200
			lowerCAmelCase  :     str       						= [floats_list((1, x)						)[0] for x in range(800		,   1_400		,   200						)]
			lowerCAmelCase  :     Dict       						= [np.asarray(a_						) for speech_input in speech_inputs]

			# Test not batched input
			lowerCAmelCase  :     Optional[int]       						= feat_extract(speech_inputs[0]		,   return_tensors="np"						).input_values
			lowerCAmelCase  :     Any       						= feat_extract(np_speech_inputs[0]		,   return_tensors="np"						).input_values
			self.assertTrue(np.allclose(a_		,   a_		,   atol=1e-3						)						)

			# Test batched
			lowerCAmelCase  :     Any       						= feat_extract(a_		,   return_tensors="np"						).input_values
			lowerCAmelCase  :     int       						= feat_extract(a_		,   return_tensors="np"						).input_values
			for enc_seq_a, enc_seq_a in zip(a_		,   a_						):
					self.assertTrue(np.allclose(a_		,   a_		,   atol=1e-3						)						)

			# Test 2-D numpy arrays are batched.
			lowerCAmelCase  :     Dict       						= [floats_list((1, x)						)[0] for x in (800, 800, 800)]
			lowerCAmelCase  :     Tuple       						= np.asarray(a_						)
			lowerCAmelCase  :     int       						= feat_extract(a_		,   return_tensors="np"						).input_values
			lowerCAmelCase  :     Union[str, Any]       						= feat_extract(a_		,   return_tensors="np"						).input_values
			for enc_seq_a, enc_seq_a in zip(a_		,   a_						):
					self.assertTrue(np.allclose(a_		,   a_		,   atol=1e-3						)						)



	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     int       						= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
			lowerCAmelCase  :     Any       						= [floats_list((1, x)						)[0] for x in range(800		,   1_400		,   200						)]

			lowerCAmelCase  :     Union[str, Any]       						= ["longest", "max_length", "do_not_pad"]
			lowerCAmelCase  :     Optional[int]       						= [None, 1_600, None]
			for max_length, padding in zip(a_		,   a_						):
					lowerCAmelCase  :     List[Any]       						= feat_extract(a_		,   padding=a_		,   max_length=a_		,   return_tensors="np"						)
					lowerCAmelCase  :     Any       						= processed.input_values

					self._check_zero_mean_unit_variance(input_values[0][:800]						)
					self.assertTrue(input_values[0][800:].sum() < 1e-6						)
					self._check_zero_mean_unit_variance(input_values[1][:1_000]						)
					self.assertTrue(input_values[0][1_000:].sum() < 1e-6						)
					self._check_zero_mean_unit_variance(input_values[2][:1_200]						)



	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     List[Any]       						= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
			lowerCAmelCase  :     Optional[int]       						= range(800		,   1_400		,   200						)
			lowerCAmelCase  :     Tuple       						= [floats_list((1, x)						)[0] for x in lengths]

			lowerCAmelCase  :     str       						= ["longest", "max_length", "do_not_pad"]
			lowerCAmelCase  :     Any       						= [None, 1_600, None]

			for max_length, padding in zip(a_		,   a_						):
					lowerCAmelCase  :     Dict       						= feat_extract(a_		,   max_length=a_		,   padding=a_						)
					lowerCAmelCase  :     int       						= processed.input_values

					self._check_zero_mean_unit_variance(input_values[0][:800]						)
					self._check_zero_mean_unit_variance(input_values[1][:1_000]						)
					self._check_zero_mean_unit_variance(input_values[2][:1_200]						)



	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     List[Any]       						= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
			lowerCAmelCase  :     Tuple       						= [floats_list((1, x)						)[0] for x in range(800		,   1_400		,   200						)]
			lowerCAmelCase  :     Dict       						= feat_extract(
			    a_		,   truncation=a_		,   max_length=1_000		,   padding="max_length"		,   return_tensors="np"						)
			lowerCAmelCase  :     Union[str, Any]       						= processed.input_values

			self._check_zero_mean_unit_variance(input_values[0, :800]						)
			self._check_zero_mean_unit_variance(input_values[1]						)
			self._check_zero_mean_unit_variance(input_values[2]						)



	def 	_lowerCamelCase     ( self						):
			lowerCAmelCase  :     Optional[int]       						= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
			lowerCAmelCase  :     Tuple       						= [floats_list((1, x)						)[0] for x in range(800		,   1_400		,   200						)]
			lowerCAmelCase  :     Union[str, Any]       						= feat_extract(
			    a_		,   truncation=a_		,   max_length=1_000		,   padding="longest"		,   return_tensors="np"						)
			lowerCAmelCase  :     Dict       						= processed.input_values

			self._check_zero_mean_unit_variance(input_values[0, :800]						)
			self._check_zero_mean_unit_variance(input_values[1, :1_000]						)
			self._check_zero_mean_unit_variance(input_values[2]						)

			# make sure that if max_length < longest -> then pad to max_length
			self.assertTrue(input_values.shape == (3, 1_000)						)

			lowerCAmelCase  :     List[str]       						= [floats_list((1, x)						)[0] for x in range(800		,   1_400		,   200						)]
			lowerCAmelCase  :     Union[str, Any]       						= feat_extract(
			    a_		,   truncation=a_		,   max_length=2_000		,   padding="longest"		,   return_tensors="np"						)
			lowerCAmelCase  :     Optional[Any]       						= processed.input_values

			self._check_zero_mean_unit_variance(input_values[0, :800]						)
			self._check_zero_mean_unit_variance(input_values[1, :1_000]						)
			self._check_zero_mean_unit_variance(input_values[2]						)

			# make sure that if max_length > longest -> then pad to longest
			self.assertTrue(input_values.shape == (3, 1_200)						)



	@require_torch
	def 	_lowerCamelCase     ( self						):
			import torch

			lowerCAmelCase  :     Optional[int]       						= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()						)
			lowerCAmelCase  :     List[Any]       						= np.random.rand(100						).astype(np.floataa						)
			lowerCAmelCase  :     List[Any]       						= np_speech_inputs.tolist()

			for inputs in [py_speech_inputs, np_speech_inputs]:
					lowerCAmelCase  :     Union[str, Any]       						= feature_extractor.pad([{"input_values": inputs}]		,   return_tensors="np"						)
					self.assertTrue(np_processed.input_values.dtype == np.floataa						)
					lowerCAmelCase  :     Tuple       						= feature_extractor.pad([{"input_values": inputs}]		,   return_tensors="pt"						)
					self.assertTrue(pt_processed.input_values.dtype == torch.floataa						)





	@slow
	@require_torch
	def 	_lowerCamelCase     ( self						):
			# this test makes sure that models that are using
			# group norm don't have their feature extractor return the
			# attention_mask
			for model_id in WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST:
					lowerCAmelCase  :     Any       						= WavaVecaConfig.from_pretrained(a_						)
					lowerCAmelCase  :     List[str]       						= WavaVecaFeatureExtractor.from_pretrained(a_						)

					# only "layer" feature extraction norm should make use of
					# attention_mask
					self.assertEqual(feat_extract.return_attention_mask		,   config.feat_extract_norm == "layer"						)



