





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        snake_case_	(		__lowercase					,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 				UpperCAmelCase__						(							self			:					Dict      )->List[Any]:






   '''simple docstring'''

   return 32






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






   '''simple docstring'''

   return 32






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






   '''simple docstring'''

   return self.time_input_dim






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






   '''simple docstring'''

   return self.time_input_dim * 4






  @property
  def 				UpperCAmelCase__						(							self			:					int      )->List[Any]:






   '''simple docstring'''

   return 100






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






   '''simple docstring'''

   __lowerCAmelCase					:	Union[str, Any]      =       CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""      )
   return tokenizer






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






   '''simple docstring'''

   torch.manual_seed(0      )
   __lowerCAmelCase					:	Any      =       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(_snake_case      )






  @property
  def 				UpperCAmelCase__						(							self			:					Dict      )->Tuple:






   '''simple docstring'''

   torch.manual_seed(0      )

   __lowerCAmelCase					:	Any      =       {
       """num_attention_heads""": 2,
       """attention_head_dim""": 12,
       """embedding_dim""": self.text_embedder_hidden_size,
       """num_layers""": 1,
   }

   __lowerCAmelCase					:	Tuple      =       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
   __lowerCAmelCase					:	List[str]      =       nn.Parameter(torch.ones(model.clip_std.shape      )      )
   return model






  @property
  def 				UpperCAmelCase__						(							self			:					int      )->Dict:






   '''simple docstring'''

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

   __lowerCAmelCase					:	Optional[Any]      =       CLIPVisionModelWithProjection(_snake_case      )
   return model






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






   '''simple docstring'''

   __lowerCAmelCase					:	Dict      =       CLIPImageProcessor(
       crop_size=224       ,  do_center_crop=_snake_case       ,  do_normalize=_snake_case       ,  do_resize=_snake_case       ,  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=224       ,  )

   return image_processor






  def 				UpperCAmelCase__						(							self			:					Optional[Any]      )->Dict:






   '''simple docstring'''

   __lowerCAmelCase					:	Optional[Any]      =       self.dummy_prior
   __lowerCAmelCase					:	int      =       self.dummy_image_encoder
   __lowerCAmelCase					:	Dict      =       self.dummy_text_encoder
   __lowerCAmelCase					:	Tuple      =       self.dummy_tokenizer
   __lowerCAmelCase					:	Tuple      =       self.dummy_image_processor

   __lowerCAmelCase					:	List[str]      =       UnCLIPScheduler(
       variance_type="""fixed_small_log"""       ,  prediction_type="""sample"""       ,  num_train_timesteps=1000       ,  clip_sample=_snake_case       ,  clip_sample_range=10.0       ,  )

   __lowerCAmelCase					:	int      =       {
       """prior""": prior,
       """image_encoder""": image_encoder,
       """text_encoder""": text_encoder,
       """tokenizer""": tokenizer,
       """scheduler""": scheduler,
       """image_processor""": image_processor,
   }

   return components






  def 				UpperCAmelCase__						(							self			:					List[str]       ,  _snake_case			:					Optional[int]       ,  _snake_case			:					List[Any]=0      )->int:






   '''simple docstring'''

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






  def 				UpperCAmelCase__						(							self			:					str      )->Optional[Any]:






   '''simple docstring'''

   __lowerCAmelCase					:	Optional[int]      =       """cpu"""

   __lowerCAmelCase					:	List[Any]      =       self.get_dummy_components()

   __lowerCAmelCase					:	Dict      =       self.pipeline_class(**_snake_case      )
   __lowerCAmelCase					:	Tuple      =       pipe.to(_snake_case      )

   pipe.set_progress_bar_config(disable=_snake_case      )

   __lowerCAmelCase					:	List[Any]      =       pipe(**self.get_dummy_inputs(_snake_case      )      )
   __lowerCAmelCase					:	Optional[Any]      =       output.image_embeds

   __lowerCAmelCase					:	List[Any]      =       pipe(
       **self.get_dummy_inputs(_snake_case      )       ,  return_dict=_snake_case       ,  )[0]

   __lowerCAmelCase					:	Any      =       image[0, -10:]
   __lowerCAmelCase					:	List[Any]      =       image_from_tuple[0, -10:]

   assert image.shape == (1, 32)

   __lowerCAmelCase					:	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 				UpperCAmelCase__						(							self			:					int      )->List[Any]:






   '''simple docstring'''

   __lowerCAmelCase					:	Optional[Any]      =       torch_device == """cpu"""
   __lowerCAmelCase					:	int      =       True
   __lowerCAmelCase					:	Dict      =       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 				UpperCAmelCase__						(							self			:					Dict      )->str:






   '''simple docstring'''

   __lowerCAmelCase					:	Tuple      =       torch_device == """cpu"""
   __lowerCAmelCase					:	str      =       False

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






_UpperCAmelCase             =							{0: [2, 3], 1: [0], 2: [1], 3: [4], 4: []}

_UpperCAmelCase             =							{0: [1, 2, 3], 1: [2], 2: [0], 3: [4], 4: [5], 5: [3]}






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:dict[int, list[int]]   ,   SCREAMING_SNAKE_CASE		:int   ,   SCREAMING_SNAKE_CASE		:list[bool]   )      ->   list[int]:
 __lowerCAmelCase					:	str      =       True
 __lowerCAmelCase					:	str      =       []

 for neighbour in graph[vert]:
  if not visited[neighbour]:
   order += topology_sort(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )

 order.append(SCREAMING_SNAKE_CASE   )

 return order






def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:dict[int, list[int]]   ,   SCREAMING_SNAKE_CASE		:int   ,   SCREAMING_SNAKE_CASE		:list[bool]   )      ->   list[int]:
 __lowerCAmelCase					:	Optional[Any]      =       True
 __lowerCAmelCase					:	Union[str, Any]      =       [vert]

 for neighbour in reversed_graph[vert]:
  if not visited[neighbour]:
   component += find_components(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )

 return component







def 	_SCREAMING_SNAKE_CASE		(      SCREAMING_SNAKE_CASE		:dict[int, list[int]]   )      ->   list[list[int]]:
 __lowerCAmelCase					:	Optional[Any]      =       len(SCREAMING_SNAKE_CASE   ) * [False]
 __lowerCAmelCase					:	dict[int, list[int]]      =       {vert: [] for vert in range(len(SCREAMING_SNAKE_CASE   )   )}

 for vert, neighbours in graph.items():
  for neighbour in neighbours:
   reversed_graph[neighbour].append(SCREAMING_SNAKE_CASE   )

 __lowerCAmelCase					:	Tuple      =       []
 for i, was_visited in enumerate(SCREAMING_SNAKE_CASE   ):
  if not was_visited:
   order += topology_sort(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )

 __lowerCAmelCase					:	Optional[Any]      =       []
 __lowerCAmelCase					:	Any      =       len(SCREAMING_SNAKE_CASE   ) * [False]

 for i in range(len(SCREAMING_SNAKE_CASE   )   ):
  __lowerCAmelCase					:	Optional[int]      =       order[len(SCREAMING_SNAKE_CASE   ) - i - 1]
  if not visited[vert]:
   __lowerCAmelCase					:	Any      =       find_components(SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   ,   SCREAMING_SNAKE_CASE   )
   components_list.append(SCREAMING_SNAKE_CASE   )

 return components_list