






"""simple docstring"""
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 _lowerCAmelCase						(    __UpperCAmelCase   ,  unittest.TestCase       ):
						SCREAMING_SNAKE_CASE_:					str 					=				KandinskyVaaPriorPipeline
						SCREAMING_SNAKE_CASE_:					str 					=				['prompt']
						SCREAMING_SNAKE_CASE_:					Optional[int] 					=				['prompt', 'negative_prompt']
						SCREAMING_SNAKE_CASE_:					Optional[int] 					=				[
						    'num_images_per_prompt',
						    'generator',
						    'num_inference_steps',
						    'latents',
						    'negative_prompt',
						    'guidance_scale',
						    'output_type',
						    'return_dict',
						]
						SCREAMING_SNAKE_CASE_:					Optional[Any] 					=				False







						@property
						def 			A      (    self							)		->							List[str]:
										return 3_2







						@property
						def 			A      (    self							)		->							Dict:
										return 3_2







						@property
						def 			A      (    self							)		->							str:
										return self.time_input_dim







						@property
						def 			A      (    self							)		->							Optional[Any]:
										return self.time_input_dim * 4







						@property
						def 			A      (    self							)		->							Any:
										return 1_0_0







						@property
						def 			A      (    self							)		->							Any:
										_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'							)
										return tokenizer







						@property
						def 			A      (    self							)		->							str:
										torch.manual_seed(0							)
										_SCREAMING_SNAKE_CASE   :		Dict       		=    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=3_7   ,				layer_norm_eps=1e-05   ,				num_attention_heads=4   ,				num_hidden_layers=5   ,				pad_token_id=1   ,				vocab_size=1_0_0_0   ,				)
										return CLIPTextModelWithProjection(lowerCAmelCase_							)







						@property
						def 			A      (    self							)		->							Optional[int]:
										torch.manual_seed(0							)

										_SCREAMING_SNAKE_CASE   :		List[str]       		=    {
										    'num_attention_heads': 2,
										    'attention_head_dim': 1_2,
										    'embedding_dim': self.text_embedder_hidden_size,
										    'num_layers': 1,
										}

										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    PriorTransformer(**lowerCAmelCase_							)
										# 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   :		Tuple       		=    nn.Parameter(torch.ones(model.clip_std.shape							)							)
										return model







						@property
						def 			A      (    self							)		->							Dict:
										torch.manual_seed(0							)
										_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    CLIPVisionConfig(
										    hidden_size=self.text_embedder_hidden_size   ,				image_size=2_2_4   ,				projection_dim=self.text_embedder_hidden_size   ,				intermediate_size=3_7   ,				num_attention_heads=4   ,				num_channels=3   ,				num_hidden_layers=5   ,				patch_size=1_4   ,				)

										_SCREAMING_SNAKE_CASE   :		int       		=    CLIPVisionModelWithProjection(lowerCAmelCase_							)
										return model







						@property
						def 			A      (    self							)		->							Any:
										_SCREAMING_SNAKE_CASE   :		List[str]       		=    CLIPImageProcessor(
										    crop_size=2_2_4   ,				do_center_crop=lowerCAmelCase_   ,				do_normalize=lowerCAmelCase_   ,				do_resize=lowerCAmelCase_   ,				image_mean=[0.48_145_466, 0.4_578_275, 0.40_821_073]   ,				image_std=[0.26_862_954, 0.26_130_258, 0.27_577_711]   ,				resample=3   ,				size=2_2_4   ,				)

										return image_processor







						def 			A      (    self							)		->							str:
										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    self.dummy_prior
										_SCREAMING_SNAKE_CASE   :		Tuple       		=    self.dummy_image_encoder
										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    self.dummy_text_encoder
										_SCREAMING_SNAKE_CASE   :		int       		=    self.dummy_tokenizer
										_SCREAMING_SNAKE_CASE   :		Dict       		=    self.dummy_image_processor

										_SCREAMING_SNAKE_CASE   :		List[str]       		=    UnCLIPScheduler(
										    variance_type='fixed_small_log'   ,				prediction_type='sample'   ,				num_train_timesteps=1_0_0_0   ,				clip_sample=lowerCAmelCase_   ,				clip_sample_range=10.0   ,				)

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

										return components







						def 			A      (    self   ,				lowerCAmelCase_   ,				lowerCAmelCase_=0							)		->							Dict:
										if str(lowerCAmelCase_							).startswith('mps'							):
														_SCREAMING_SNAKE_CASE   :		List[Any]       		=    torch.manual_seed(lowerCAmelCase_							)
										else:
														_SCREAMING_SNAKE_CASE   :		Any       		=    torch.Generator(device=lowerCAmelCase_							).manual_seed(lowerCAmelCase_							)
										_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    {
										    'prompt': 'horse',
										    'generator': generator,
										    'guidance_scale': 4.0,
										    'num_inference_steps': 2,
										    'output_type': 'np',
										}
										return inputs







						def 			A      (    self							)		->							str:
										_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    'cpu'

										_SCREAMING_SNAKE_CASE   :		int       		=    self.get_dummy_components()

										_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    self.pipeline_class(**lowerCAmelCase_							)
										_SCREAMING_SNAKE_CASE   :		Dict       		=    pipe.to(lowerCAmelCase_							)

										pipe.set_progress_bar_config(disable=lowerCAmelCase_							)

										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    pipe(**self.get_dummy_inputs(lowerCAmelCase_							)							)
										_SCREAMING_SNAKE_CASE   :		Tuple       		=    output.image_embeds

										_SCREAMING_SNAKE_CASE   :		int       		=    pipe(
										    **self.get_dummy_inputs(lowerCAmelCase_							)   ,				return_dict=lowerCAmelCase_   ,				)[0]

										_SCREAMING_SNAKE_CASE   :		List[str]       		=    image[0, -1_0:]
										_SCREAMING_SNAKE_CASE   :		Dict       		=    image_from_tuple[0, -1_0:]

										assert image.shape == (1, 3_2)

										_SCREAMING_SNAKE_CASE   :		str       		=    np.array(
										    [-0.0_532, 1.7_120, 0.3_656, -1.0_852, -0.8_946, -1.1_756, 0.4_348, 0.2_482, 0.5_146, -0.1_156]							)

										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							)		->							List[str]:
										_SCREAMING_SNAKE_CASE   :		int       		=    torch_device == 'cpu'
										_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    True
										_SCREAMING_SNAKE_CASE   :		Tuple       		=    False

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







						@skip_mps
						def 			A      (    self							)		->							int:
										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    torch_device == 'cpu'
										_SCREAMING_SNAKE_CASE   :		List[str]       		=    False

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









"""simple docstring"""
import functools
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

import huggingface_hub.utils as hf_hub_utils
from tqdm import auto as tqdm_lib


lowerCAmelCase__    						=			threading.Lock()
lowerCAmelCase__    						=			None

lowerCAmelCase__    						=			{
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
}

lowerCAmelCase__    						=			logging.WARNING

lowerCAmelCase__    						=			True





def 				lowercase__		(					):
				_SCREAMING_SNAKE_CASE   :		int       		=    os.getenv('TRANSFORMERS_VERBOSITY',						lowerCamelCase       )
				if env_level_str:
								if env_level_str in log_levels:
												return log_levels[env_level_str]
								else:
												logging.getLogger().warning(
												    f"""Unknown option TRANSFORMERS_VERBOSITY={env_level_str}, """
												    f"""has to be one of: { ", ".join(log_levels.keys()       ) }"""       )
				return _default_log_level





def 				lowercase__		(					):
				return __name__.split('.'       )[0]





def 				lowercase__		(					):
				return logging.getLogger(_get_library_name()       )





def 				lowercase__		(					):
				global _default_handler

				with _lock:
								if _default_handler:
												# This library has already configured the library root logger.
												return
								_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    logging.StreamHandler()  # Set sys.stderr as stream.
								_SCREAMING_SNAKE_CASE   :		Tuple       		=    sys.stderr.flush

								# Apply our default configuration to the library root logger.
								_SCREAMING_SNAKE_CASE   :		Union[str, Any]       		=    _get_library_root_logger()
								library_root_logger.addHandler(_default_handler       )
								library_root_logger.setLevel(_get_default_logging_level()       )
								_SCREAMING_SNAKE_CASE   :		int       		=    False





def 				lowercase__		(					):
				global _default_handler

				with _lock:
								if not _default_handler:
												return

								_SCREAMING_SNAKE_CASE   :		Any       		=    _get_library_root_logger()
								library_root_logger.removeHandler(_default_handler       )
								library_root_logger.setLevel(logging.NOTSET       )
								_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    None





def 				lowercase__		(					):
				return log_levels





def 				lowercase__		(					lowerCamelCase = None       ):

				if name is None:
								_SCREAMING_SNAKE_CASE   :		Tuple       		=    _get_library_name()

				_configure_library_root_logger()
				return logging.getLogger(lowerCamelCase       )





def 				lowercase__		(					):
				_configure_library_root_logger()
				return _get_library_root_logger().getEffectiveLevel()





def 				lowercase__		(					lowerCamelCase       ):
				_configure_library_root_logger()
				_get_library_root_logger().setLevel(lowerCamelCase       )





def 				lowercase__		(					):
				return set_verbosity(lowerCamelCase       )





def 				lowercase__		(					):
				return set_verbosity(lowerCamelCase       )





def 				lowercase__		(					):
				return set_verbosity(lowerCamelCase       )





def 				lowercase__		(					):
				return set_verbosity(lowerCamelCase       )





def 				lowercase__		(					):
				_configure_library_root_logger()

				assert _default_handler is not None
				_get_library_root_logger().removeHandler(_default_handler       )





def 				lowercase__		(					):
				_configure_library_root_logger()

				assert _default_handler is not None
				_get_library_root_logger().addHandler(_default_handler       )





def 				lowercase__		(					lowerCamelCase       ):
				_configure_library_root_logger()

				assert handler is not None
				_get_library_root_logger().addHandler(lowerCamelCase       )





def 				lowercase__		(					lowerCamelCase       ):
				_configure_library_root_logger()

				assert handler is not None and handler not in _get_library_root_logger().handlers
				_get_library_root_logger().removeHandler(lowerCamelCase       )





def 				lowercase__		(					):
				_configure_library_root_logger()
				_SCREAMING_SNAKE_CASE   :		List[str]       		=    False





def 				lowercase__		(					):
				_configure_library_root_logger()
				_SCREAMING_SNAKE_CASE   :		int       		=    True





def 				lowercase__		(					):
				_SCREAMING_SNAKE_CASE   :		Any       		=    _get_library_root_logger().handlers

				for handler in handlers:
								_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    logging.Formatter('[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s'       )
								handler.setFormatter(lowerCamelCase       )





def 				lowercase__		(					):
				_SCREAMING_SNAKE_CASE   :		List[str]       		=    _get_library_root_logger().handlers

				for handler in handlers:
								handler.setFormatter(lowerCamelCase       )





def 				lowercase__		(					self,						*lowerCamelCase,						**lowerCamelCase       ):
				_SCREAMING_SNAKE_CASE   :		str       		=    os.getenv('TRANSFORMERS_NO_ADVISORY_WARNINGS',						lowerCamelCase       )
				if no_advisory_warnings:
								return
				self.warning(*lowerCamelCase,						**lowerCamelCase       )


lowerCAmelCase__    						=			warning_advice





@functools.lru_cache(lowerCamelCase       )
def 				lowercase__		(					self,						*lowerCamelCase,						**lowerCamelCase       ):
				self.warning(*lowerCamelCase,						**lowerCamelCase       )


lowerCAmelCase__    						=			warning_once


class _lowerCAmelCase						:







						def __init__(    self   ,				*lowerCAmelCase_   ,				**lowerCAmelCase_							)		->							Optional[int]:  # pylint: disable=unused-argument
										_SCREAMING_SNAKE_CASE   :		Tuple       		=    args[0] if args else None







						def __iter__(    self							)		->							Dict:
										return iter(self._iterator							)







						def __getattr__(    self   ,				lowerCAmelCase_							)		->							Tuple:

										def empty_fn(*lowerCAmelCase_   ,				**lowerCAmelCase_							):  # pylint: disable=unused-argument
														return

										return empty_fn







						def __enter__(    self							)		->							List[Any]:
										return self







						def __exit__(    self   ,				lowerCAmelCase_   ,				lowerCAmelCase_   ,				lowerCAmelCase_							)		->							Optional[Any]:
										return







class _lowerCAmelCase						:







						def __call__(    self   ,				*lowerCAmelCase_   ,				**lowerCAmelCase_							)		->							Optional[Any]:
										if _tqdm_active:
														return tqdm_lib.tqdm(*lowerCAmelCase_   ,				**lowerCAmelCase_							)
										else:
														return EmptyTqdm(*lowerCAmelCase_   ,				**lowerCAmelCase_							)







						def 			A      (    self   ,				*lowerCAmelCase_   ,				**lowerCAmelCase_							)		->							List[str]:
										_SCREAMING_SNAKE_CASE   :		Optional[int]       		=    None
										if _tqdm_active:
														return tqdm_lib.tqdm.set_lock(*lowerCAmelCase_   ,				**lowerCAmelCase_							)







						def 			A      (    self							)		->							Any:
										if _tqdm_active:
														return tqdm_lib.tqdm.get_lock()







lowerCAmelCase__    						=			_tqdm_cls()





def 				lowercase__		(					):
				global _tqdm_active
				return bool(_tqdm_active       )





def 				lowercase__		(					):
				global _tqdm_active
				_SCREAMING_SNAKE_CASE   :		int       		=    True
				hf_hub_utils.enable_progress_bars()






def 				lowercase__		(					):
				global _tqdm_active
				_SCREAMING_SNAKE_CASE   :		Optional[Any]       		=    False
				hf_hub_utils.disable_progress_bars()

