




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


__A								=       False







@skip_mps
class        __lowerCAmelCase					(			__magic_name__     ,							__magic_name__     ,							__magic_name__     ,							unittest.TestCase					):

			"""simple docstring"""







			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  lowercase_       ( cls				)       -> List[Any]:



							'''simple docstring'''
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCamelCase__				)


			@classmethod
			def  lowercase_       ( cls				)       -> List[str]:



							'''simple docstring'''
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCamelCase__				)


			def  lowercase_       ( self				)       -> Any:



							'''simple docstring'''
							torch.manual_seed(0				)
							__lowerCamelCase									=       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=lowerCamelCase__     ,		)
							__lowerCamelCase									=       DDIMScheduler(
							    beta_start=0.0_00_85     ,		beta_end=0.0_12     ,		beta_schedule='scaled_linear'     ,		clip_sample=lowerCamelCase__     ,		set_alpha_to_one=lowerCamelCase__     ,		)
							torch.manual_seed(0				)
							__lowerCamelCase									=       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									=       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     ,		hidden_act='gelu'     ,		projection_dim=512     ,		)
							__lowerCamelCase									=       CLIPTextModel(lowerCamelCase__				)
							__lowerCamelCase									=       CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'				)

							__lowerCamelCase									=       {
							    'unet': unet,
							    'scheduler': scheduler,
							    'vae': vae,
							    'text_encoder': text_encoder,
							    'tokenizer': tokenizer,
							    'safety_checker': None,
							    'feature_extractor': None,
							}

							return components


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



							'''simple docstring'''
							if str(lowerCamelCase__				).startswith('mps'				):
											__lowerCamelCase									=       torch.manual_seed(lowerCamelCase__				)
							else:
											__lowerCamelCase									=       torch.Generator(device=lowerCamelCase__				).manual_seed(lowerCamelCase__				)
							__lowerCamelCase									=       __lowerCamelCase									=       {
							    '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  lowercase_       ( self				)       -> Optional[int]:



							'''simple docstring'''
							__lowerCamelCase									=       'cpu'

							__lowerCamelCase									=       self.get_dummy_components()
							__lowerCamelCase									=       self.pipeline_class(**lowerCamelCase__				)
							pipe.to(lowerCamelCase__				)
							pipe.set_progress_bar_config(disable=lowerCamelCase__				)

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

							self.assertEqual(image.shape     ,		(1, 64, 64, 3)				)
							__lowerCamelCase									=       np.array(
							    [0.63_90_53_64, 0.62_89_73_07, 0.48_59_90_17, 0.5_13_36_24, 0.5_55_00_48, 0.45_76_95_16, 0.50_32_69_73, 0.5_02_31_39, 0.45_38_44_96]				)
							__lowerCamelCase									=       np.abs(image_slice.flatten() - expected_slice				).max()
							self.assertLessEqual(lowerCamelCase__     ,		1e-3				)


			def  lowercase_       ( self				)       -> int:



							'''simple docstring'''
							super().test_cpu_offload_forward_pass(expected_max_diff=5e-4				)


			def  lowercase_       ( self				)       -> List[str]:



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


			def  lowercase_       ( self				)       -> List[Any]:



							'''simple docstring'''
							self._test_inference_batch_single_identical(batch_size=2     ,		expected_max_diff=7e-4				)


			def  lowercase_       ( self				)       -> Dict:



							'''simple docstring'''
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3				)


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



							'''simple docstring'''
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4				)


			def  lowercase_       ( self				)       -> Optional[int]:



							'''simple docstring'''
							super().test_save_load_local(expected_max_difference=5e-4				)






			def  lowercase_       ( self				)       -> Tuple:



							'''simple docstring'''
							super().test_save_load_optional_components(expected_max_difference=4e-4				)







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

			"""simple docstring"""







			@classmethod
			def  lowercase_       ( cls				)       -> int:



							'''simple docstring'''
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCamelCase__				)


			@classmethod
			def  lowercase_       ( cls				)       -> str:



							'''simple docstring'''
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCamelCase__				)


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



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






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



							'''simple docstring'''
							__lowerCamelCase									=       torch.manual_seed(51				)

							__lowerCamelCase									=       StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    'CompVis/stable-diffusion-v1-4'     ,		safety_checker=lowerCamelCase__     ,		torch_dtype=torch.floataa				)
							pipe.to('cuda'				)

							__lowerCamelCase									=       'a painting of an elephant with glasses'
							__lowerCamelCase									=       [5, 7]

							__lowerCamelCase									=       pipe(
							    prompt=lowerCamelCase__     ,		token_indices=lowerCamelCase__     ,		guidance_scale=7.5     ,		generator=lowerCamelCase__     ,		num_inference_steps=5     ,		max_iter_to_alter=5     ,		output_type='numpy'     ,		).images[0]

							__lowerCamelCase									=       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








import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available







class        __lowerCAmelCase					(			__magic_name__					):

			"""simple docstring"""







			snake_case_  						=			['''image_processor''', '''tokenizer''']
			snake_case_  						=			'''OwlViTImageProcessor'''
			snake_case_  						=			('''CLIPTokenizer''', '''CLIPTokenizerFast''')


			def __init__( self     ,		lowerCamelCase__=None     ,		lowerCamelCase__=None     ,		**lowerCamelCase__				)       -> Tuple:



							'''simple docstring'''
							__lowerCamelCase									=       None
							if "feature_extractor" in kwargs:
											warnings.warn(
											    'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
											    ' instead.'     ,		lowerCamelCase__     ,		)
											__lowerCamelCase									=       kwargs.pop('feature_extractor'				)

							__lowerCamelCase									=       image_processor if image_processor is not None else feature_extractor
							if image_processor is None:
											raise ValueError('You need to specify an `image_processor`.'				)
							if tokenizer is None:
											raise ValueError('You need to specify a `tokenizer`.'				)

							super().__init__(lowerCamelCase__     ,		lowerCamelCase__				)


			def __call__( self     ,		lowerCamelCase__=None     ,		lowerCamelCase__=None     ,		lowerCamelCase__=None     ,		lowerCamelCase__="max_length"     ,		lowerCamelCase__="np"     ,		**lowerCamelCase__				)       -> Optional[int]:



							'''simple docstring'''

							if text is None and query_images is None and images is None:
											raise ValueError(
											    'You have to specify at least one text or query image or image. All three cannot be none.'				)

							if text is not None:
											if isinstance(lowerCamelCase__     ,		lowerCamelCase__				) or (isinstance(lowerCamelCase__     ,		lowerCamelCase__				) and not isinstance(text[0]     ,		lowerCamelCase__				)):
															__lowerCamelCase									=       [self.tokenizer(lowerCamelCase__     ,		padding=lowerCamelCase__     ,		return_tensors=lowerCamelCase__     ,		**lowerCamelCase__				)]

											elif isinstance(lowerCamelCase__     ,		lowerCamelCase__				) and isinstance(text[0]     ,		lowerCamelCase__				):
															__lowerCamelCase									=       []

															# Maximum number of queries across batch
															__lowerCamelCase									=       max([len(lowerCamelCase__				) for t in text]				)

															# Pad all batch samples to max number of text queries
															for t in text:
																			if len(lowerCamelCase__				) != max_num_queries:
																							__lowerCamelCase									=       t + [' '] * (max_num_queries - len(lowerCamelCase__				))

																			__lowerCamelCase									=       self.tokenizer(lowerCamelCase__     ,		padding=lowerCamelCase__     ,		return_tensors=lowerCamelCase__     ,		**lowerCamelCase__				)
																			encodings.append(lowerCamelCase__				)
											else:
															raise TypeError('Input text should be a string, a list of strings or a nested list of strings'				)

											if return_tensors == "np":
															__lowerCamelCase									=       np.concatenate([encoding['input_ids'] for encoding in encodings]     ,		axis=0				)
															__lowerCamelCase									=       np.concatenate([encoding['attention_mask'] for encoding in encodings]     ,		axis=0				)

											elif return_tensors == "jax" and is_flax_available():
															import jax.numpy as jnp

															__lowerCamelCase									=       jnp.concatenate([encoding['input_ids'] for encoding in encodings]     ,		axis=0				)
															__lowerCamelCase									=       jnp.concatenate([encoding['attention_mask'] for encoding in encodings]     ,		axis=0				)

											elif return_tensors == "pt" and is_torch_available():
															import torch

															__lowerCamelCase									=       torch.cat([encoding['input_ids'] for encoding in encodings]     ,		dim=0				)
															__lowerCamelCase									=       torch.cat([encoding['attention_mask'] for encoding in encodings]     ,		dim=0				)

											elif return_tensors == "tf" and is_tf_available():
															import tensorflow as tf

															__lowerCamelCase									=       tf.stack([encoding['input_ids'] for encoding in encodings]     ,		axis=0				)
															__lowerCamelCase									=       tf.stack([encoding['attention_mask'] for encoding in encodings]     ,		axis=0				)

											else:
															raise ValueError('Target return tensor type could not be returned'				)

											__lowerCamelCase									=       BatchEncoding()
											__lowerCamelCase									=       input_ids
											__lowerCamelCase									=       attention_mask

							if query_images is not None:
											__lowerCamelCase									=       BatchEncoding()
											__lowerCamelCase									=       self.image_processor(
											    lowerCamelCase__     ,		return_tensors=lowerCamelCase__     ,		**lowerCamelCase__				).pixel_values
											__lowerCamelCase									=       query_pixel_values

							if images is not None:
											__lowerCamelCase									=       self.image_processor(lowerCamelCase__     ,		return_tensors=lowerCamelCase__     ,		**lowerCamelCase__				)

							if text is not None and images is not None:
											__lowerCamelCase									=       image_features.pixel_values
											return encoding
							elif query_images is not None and images is not None:
											__lowerCamelCase									=       image_features.pixel_values
											return encoding
							elif text is not None or query_images is not None:
											return encoding
							else:
											return BatchEncoding(data=dict(**lowerCamelCase__				)     ,		tensor_type=lowerCamelCase__				)


			def  lowercase_       ( self     ,		*lowerCamelCase__     ,		**lowerCamelCase__				)       -> Tuple:



							'''simple docstring'''
							return self.image_processor.post_process(*lowerCamelCase__     ,		**lowerCamelCase__				)


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



							'''simple docstring'''
							return self.image_processor.post_process_object_detection(*lowerCamelCase__     ,		**lowerCamelCase__				)


			def  lowercase_       ( self     ,		*lowerCamelCase__     ,		**lowerCamelCase__				)       -> Optional[int]:



							'''simple docstring'''
							return self.image_processor.post_process_image_guided_detection(*lowerCamelCase__     ,		**lowerCamelCase__				)


			def  lowercase_       ( self     ,		*lowerCamelCase__     ,		**lowerCamelCase__				)       -> Optional[int]:



							'''simple docstring'''
							return self.tokenizer.batch_decode(*lowerCamelCase__     ,		**lowerCamelCase__				)


			def  lowercase_       ( self     ,		*lowerCamelCase__     ,		**lowerCamelCase__				)       -> Optional[int]:



							'''simple docstring'''
							return self.tokenizer.decode(*lowerCamelCase__     ,		**lowerCamelCase__				)


			@property
			def  lowercase_       ( self				)       -> Dict:



							'''simple docstring'''
							warnings.warn(
							    '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'     ,		lowerCamelCase__     ,		)
							return self.image_processor_class






			@property
			def  lowercase_       ( self				)       -> Union[str, Any]:



							'''simple docstring'''
							warnings.warn(
							    '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'     ,		lowerCamelCase__     ,		)
							return self.image_processor


