
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( a  ,   unittest.TestCase			):
				snake_case_						   =    KandinskyVaaPriorPipeline
				snake_case_						   =    ['''prompt''']
				snake_case_						   =    ['''prompt''', '''negative_prompt''']
				snake_case_						   =    [
				    '''num_images_per_prompt''',
				    '''generator''',
				    '''num_inference_steps''',
				    '''latents''',
				    '''negative_prompt''',
				    '''guidance_scale''',
				    '''output_type''',
				    '''return_dict''',
				]
				snake_case_						   =    False




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      str:


								'''simple docstring'''




								return 32




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      List[str]:


								'''simple docstring'''




								return 32




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Any:


								'''simple docstring'''




								return self.time_input_dim




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Tuple:


								'''simple docstring'''




								return self.time_input_dim * 4




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


								'''simple docstring'''




								return 100




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      str:


								'''simple docstring'''




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




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      int:


								'''simple docstring'''




								torch.manual_seed(0				)
								__a										=    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=1_000		,     )
								return CLIPTextModelWithProjection(_snake_case				)




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


								'''simple docstring'''




								torch.manual_seed(0				)

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

								__a										=    PriorTransformer(**_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
								__a										=    nn.Parameter(torch.ones(model.clip_std.shape				)				)
								return model




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


								'''simple docstring'''




								torch.manual_seed(0				)
								__a										=    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		,     )

								__a										=    CLIPVisionModelWithProjection(_snake_case				)
								return model




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


								'''simple docstring'''




								__a										=    CLIPImageProcessor(
								    crop_size=224		,     do_center_crop=_snake_case		,     do_normalize=_snake_case		,     do_resize=_snake_case		,     image_mean=[0.4814_5466, 0.457_8275, 0.4082_1073]		,     image_std=[0.2686_2954, 0.2613_0258, 0.2757_7711]		,     resample=3		,     size=224		,     )

								return image_processor




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


								'''simple docstring'''




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

								__a										=    UnCLIPScheduler(
								    variance_type='''fixed_small_log'''		,     prediction_type='''sample'''		,     num_train_timesteps=1_000		,     clip_sample=_snake_case		,     clip_sample_range=10.0		,     )

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

								return components




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case=0				)    ->      Dict:


								'''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''': '''horse''',
								    '''generator''': generator,
								    '''guidance_scale''': 4.0,
								    '''num_inference_steps''': 2,
								    '''output_type''': '''np''',
								}
								return inputs




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Any:


								'''simple docstring'''




								__a										=    '''cpu'''

								__a										=    self.get_dummy_components()

								__a										=    self.pipeline_class(**_snake_case				)
								__a										=    pipe.to(_snake_case				)

								pipe.set_progress_bar_config(disable=_snake_case				)

								__a										=    pipe(**self.get_dummy_inputs(_snake_case				)				)
								__a										=    output.image_embeds

								__a										=    pipe(
								    **self.get_dummy_inputs(_snake_case				)		,     return_dict=_snake_case		,     )[0]

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

								assert image.shape == (1, 32)

								__a										=    np.array(
								    [-0.0532, 1.7120, 0.3656, -1.0852, -0.8946, -1.1756, 0.4348, 0.2482, 0.5146, -0.1156]				)

								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 				SCREAMING_SNAKE_CASE_		(     self				)    ->      List[str]:


								'''simple docstring'''




								__a										=    torch_device == '''cpu'''
								__a										=    True
								__a										=    False

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




				@skip_mps
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Tuple:


								'''simple docstring'''




								__a										=    torch_device == '''cpu'''
								__a										=    False

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

import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding







class 			__A( a			):
				snake_case_						   =    ['''image_processor''', '''tokenizer''']
				snake_case_						   =    '''ChineseCLIPImageProcessor'''
				snake_case_						   =    ('''BertTokenizer''', '''BertTokenizerFast''')




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


								'''simple docstring'''




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

								__a										=    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__(_snake_case		,     _snake_case				)
								__a										=    self.image_processor




				def __call__(     self		,     _snake_case=None		,     _snake_case=None		,     _snake_case=None		,     **_snake_case				)    ->      Optional[Any]:


								'''simple docstring'''




								if text is None and images is None:
												raise ValueError('''You have to specify either text or images. Both cannot be none.'''				)

								if text is not None:
												__a										=    self.tokenizer(_snake_case		,     return_tensors=_snake_case		,     **_snake_case				)

								if images is not None:
												__a										=    self.image_processor(_snake_case		,     return_tensors=_snake_case		,     **_snake_case				)

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




				def 				SCREAMING_SNAKE_CASE_		(     self		,     *_snake_case		,     **_snake_case				)    ->      str:


								'''simple docstring'''




								return self.tokenizer.batch_decode(*_snake_case		,     **_snake_case				)




				def 				SCREAMING_SNAKE_CASE_		(     self		,     *_snake_case		,     **_snake_case				)    ->      Dict:


								'''simple docstring'''




								return self.tokenizer.decode(*_snake_case		,     **_snake_case				)




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Any:


								'''simple docstring'''




								__a										=    self.tokenizer.model_input_names
								__a										=    self.image_processor.model_input_names
								return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names				)				)




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


								'''simple docstring'''




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