




import unittest

import numpy as np
import torch
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



class       a_       (     SCREAMING_SNAKE_CASE__  ,    unittest.TestCase     ):
	A    							= KandinskyVaaPriorPipeline
	A    							= ['''prompt''']
	A    							= ['''prompt''', '''negative_prompt''']
	A    							= [
	    '''num_images_per_prompt''',
	    '''generator''',
	    '''num_inference_steps''',
	    '''latents''',
	    '''negative_prompt''',
	    '''guidance_scale''',
	    '''output_type''',
	    '''return_dict''',
	]
	A    							= False





	@property
	def 						A_(							self      )  ->					str:




				"""simple docstring"""



				return 32





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




				"""simple docstring"""



				return 32





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




				"""simple docstring"""



				return self.time_input_dim





	@property
	def 						A_(							self      )  ->					str:




				"""simple docstring"""



				return self.time_input_dim * 4





	@property
	def 						A_(							self      )  ->					Optional[Any]:




				"""simple docstring"""



				return 100





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




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )
				return tokenizer





	@property
	def 						A_(							self      )  ->					Optional[int]:




				"""simple docstring"""



				torch.manual_seed(0      )
				SCREAMING_SNAKE_CASE_								=						CLIPTextConfig(
				    bos_token_id=0      ,     eos_token_id=2      ,     hidden_size=self.text_embedder_hidden_size      ,     projection_dim=self.text_embedder_hidden_size      ,     intermediate_size=37      ,     layer_norm_eps=1e-05      ,     num_attention_heads=4      ,     num_hidden_layers=5      ,     pad_token_id=1      ,     vocab_size=1000      ,     )
				return CLIPTextModelWithProjection(SCREAMING_SNAKE_CASE      )





	@property
	def 						A_(							self      )  ->					int:




				"""simple docstring"""



				torch.manual_seed(0      )

				SCREAMING_SNAKE_CASE_								=						{
				    'num_attention_heads': 2,
				    'attention_head_dim': 12,
				    'embedding_dim': self.text_embedder_hidden_size,
				    'num_layers': 1,
				}

				SCREAMING_SNAKE_CASE_								=						PriorTransformer(**SCREAMING_SNAKE_CASE      )
				# clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
				SCREAMING_SNAKE_CASE_								=						nn.Parameter(torch.ones(model.clip_std.shape      )      )
				return model





	@property
	def 						A_(							self      )  ->					Optional[Any]:




				"""simple docstring"""



				torch.manual_seed(0      )
				SCREAMING_SNAKE_CASE_								=						CLIPVisionConfig(
				    hidden_size=self.text_embedder_hidden_size      ,     image_size=224      ,     projection_dim=self.text_embedder_hidden_size      ,     intermediate_size=37      ,     num_attention_heads=4      ,     num_channels=3      ,     num_hidden_layers=5      ,     patch_size=14      ,     )

				SCREAMING_SNAKE_CASE_								=						CLIPVisionModelWithProjection(SCREAMING_SNAKE_CASE      )
				return model





	@property
	def 						A_(							self      )  ->					Optional[int]:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						CLIPImageProcessor(
				    crop_size=224      ,     do_center_crop=SCREAMING_SNAKE_CASE      ,     do_normalize=SCREAMING_SNAKE_CASE      ,     do_resize=SCREAMING_SNAKE_CASE      ,     image_mean=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]      ,     image_std=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]      ,     resample=3      ,     size=224      ,     )

				return image_processor





	def 						A_(							self      )  ->					Any:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						self.dummy_prior
				SCREAMING_SNAKE_CASE_								=						self.dummy_image_encoder
				SCREAMING_SNAKE_CASE_								=						self.dummy_text_encoder
				SCREAMING_SNAKE_CASE_								=						self.dummy_tokenizer
				SCREAMING_SNAKE_CASE_								=						self.dummy_image_processor

				SCREAMING_SNAKE_CASE_								=						UnCLIPScheduler(
				    variance_type='fixed_small_log'      ,     prediction_type='sample'      ,     num_train_timesteps=1000      ,     clip_sample=SCREAMING_SNAKE_CASE      ,     clip_sample_range=1_0.0      ,     )

				SCREAMING_SNAKE_CASE_								=						{
				    'prior': prior,
				    'image_encoder': image_encoder,
				    'text_encoder': text_encoder,
				    'tokenizer': tokenizer,
				    'scheduler': scheduler,
				    'image_processor': image_processor,
				}

				return components





	def 						A_(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE=0      )  ->					str:




				"""simple docstring"""



				if str(SCREAMING_SNAKE_CASE      ).startswith('mps'      ):
							SCREAMING_SNAKE_CASE_								=						torch.manual_seed(SCREAMING_SNAKE_CASE      )
				else:
							SCREAMING_SNAKE_CASE_								=						torch.Generator(device=SCREAMING_SNAKE_CASE      ).manual_seed(SCREAMING_SNAKE_CASE      )
				SCREAMING_SNAKE_CASE_								=						{
				    'prompt': 'horse',
				    'generator': generator,
				    'guidance_scale': 4.0,
				    'num_inference_steps': 2,
				    'output_type': 'np',
				}
				return inputs





	def 						A_(							self      )  ->					Tuple:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						'cpu'

				SCREAMING_SNAKE_CASE_								=						self.get_dummy_components()

				SCREAMING_SNAKE_CASE_								=						self.pipeline_class(**SCREAMING_SNAKE_CASE      )
				SCREAMING_SNAKE_CASE_								=						pipe.to(SCREAMING_SNAKE_CASE      )

				pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )

				SCREAMING_SNAKE_CASE_								=						pipe(**self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )      )
				SCREAMING_SNAKE_CASE_								=						output.image_embeds

				SCREAMING_SNAKE_CASE_								=						pipe(
				    **self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )      ,     return_dict=SCREAMING_SNAKE_CASE      ,     )[0]

				SCREAMING_SNAKE_CASE_								=						image[0, -10:]
				SCREAMING_SNAKE_CASE_								=						image_from_tuple[0, -10:]

				assert image.shape == (1, 32)

				SCREAMING_SNAKE_CASE_								=						np.array(
				    [-0.0_5_3_2, 1.7_1_2_0, 0.3_6_5_6, -1.0_8_5_2, -0.8_9_4_6, -1.1_7_5_6, 0.4_3_4_8, 0.2_4_8_2, 0.5_1_4_6, -0.1_1_5_6]      )

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





	@skip_mps
	def 						A_(							self      )  ->					Any:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						torch_device == 'cpu'
				SCREAMING_SNAKE_CASE_								=						True
				SCREAMING_SNAKE_CASE_								=						False

				self._test_inference_batch_single_identical(
				    test_max_difference=SCREAMING_SNAKE_CASE      ,     relax_max_difference=SCREAMING_SNAKE_CASE      ,     test_mean_pixel_difference=SCREAMING_SNAKE_CASE      ,     )





	@skip_mps
	def 						A_(							self      )  ->					Tuple:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						torch_device == 'cpu'
				SCREAMING_SNAKE_CASE_								=						False

				self._test_attention_slicing_forward_pass(
				    test_max_difference=SCREAMING_SNAKE_CASE      ,     test_mean_pixel_difference=SCREAMING_SNAKE_CASE      ,     )






from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_torch_available, is_torch_tpu_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_torch_available():
				import torch

if is_torch_tpu_available(check_device=False):
				import torch_xla.core.xla_model as xm


SCREAMING_SNAKE_CASE__     :     int		    =       logging.get_logger(__name__)



@dataclass
class       a_       (     SCREAMING_SNAKE_CASE__     ):
	A    							= [
	    '''no_inference''',
	    '''no_cuda''',
	    '''no_tpu''',
	    '''no_speed''',
	    '''no_memory''',
	    '''no_env_print''',
	    '''no_multi_process''',
	]





	def __init__(							self      ,     **SCREAMING_SNAKE_CASE      )  ->					Dict:




				"""simple docstring"""



				for deprecated_arg in self.deprecated_args:
							if deprecated_arg in kwargs:
										SCREAMING_SNAKE_CASE_								=						deprecated_arg[3:]
										setattr(self      ,     SCREAMING_SNAKE_CASE      ,     not kwargs.pop(SCREAMING_SNAKE_CASE      )      )
										logger.warning(
										    f'{deprecated_arg} is depreciated. Please use --no_{positive_arg} or'
										    f' {positive_arg}={kwargs[positive_arg]}'      )

				SCREAMING_SNAKE_CASE_								=						kwargs.pop('torchscript'      ,     self.torchscript      )
				SCREAMING_SNAKE_CASE_								=						kwargs.pop('torch_xla_tpu_print_metrics'      ,     self.torch_xla_tpu_print_metrics      )
				SCREAMING_SNAKE_CASE_								=						kwargs.pop('fp16_opt_level'      ,     self.fpaa_opt_level      )
				super().__init__(**SCREAMING_SNAKE_CASE      )

	A    							= field(default=SCREAMING_SNAKE_CASE__  ,    metadata={'''help''': '''Trace the models using torchscript'''}     )
	A    							= field(default=SCREAMING_SNAKE_CASE__  ,    metadata={'''help''': '''Print Xla/PyTorch tpu metrics'''}     )
	A    							= field(
	    default='''O1'''  ,    metadata={
	        '''help''': (
	            '''For fp16: Apex AMP optimization level selected in [\'O0\', \'O1\', \'O2\', and \'O3\']. '''
	            '''See details at https://nvidia.github.io/apex/amp.html'''
	        )
	    }  ,    )





	@cached_property
	def 						A_(							self      )  ->					Tuple["torch.device", int]:




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				logger.info('PyTorch: setting up devices'      )
				if not self.cuda:
							SCREAMING_SNAKE_CASE_								=						torch.device('cpu'      )
							SCREAMING_SNAKE_CASE_								=						0
				elif is_torch_tpu_available():
							SCREAMING_SNAKE_CASE_								=						xm.xla_device()
							SCREAMING_SNAKE_CASE_								=						0
				else:
							SCREAMING_SNAKE_CASE_								=						torch.device('cuda' if torch.cuda.is_available() else 'cpu'      )
							SCREAMING_SNAKE_CASE_								=						torch.cuda.device_count()
				return device, n_gpu





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




				"""simple docstring"""



				return is_torch_tpu_available() and self.tpu





	@property
	def 						A_(							self      )  ->					int:




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				# TODO(PVP): currently only single GPU is supported
				return torch.cuda.current_device()





	@property
	def 						A_(							self      )  ->					"torch.device":




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				return self._setup_devices[0]





	@property
	def 						A_(							self      )  ->					Optional[Any]:




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				return self._setup_devices[1]





	@property
	def 						A_(							self      )  ->					Dict:




				"""simple docstring"""



				return self.n_gpu > 0
