

import random
import unittest

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

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

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()







class 							a__							( UpperCAmelCase						,      unittest.TestCase							):







  """simple docstring"""




  UpperCAmelCase__   :			Optional[int]      							=KandinskyVaaPriorEmbaEmbPipeline
  UpperCAmelCase__   :			str      							=["""prompt""", """image"""]
  UpperCAmelCase__   :			Optional[Any]      							=["""prompt""", """image"""]
  UpperCAmelCase__   :			str      							=[
      """num_images_per_prompt""",
      """strength""",
      """generator""",
      """num_inference_steps""",
      """latents""",
      """negative_prompt""",
      """guidance_scale""",
      """output_type""",
      """return_dict""",
  ]
  UpperCAmelCase__   :			str      							=False


  @property
  def 			_lowercase   (      self     :			str		)       ->Optional[int]:
        """simple docstring"""







        return 3_2


  @property
  def 			_lowercase   (      self     :			List[str]		)       ->List[Any]:
        """simple docstring"""







        return 3_2


  @property
  def 			_lowercase   (      self     :			Tuple		)       ->List[str]:
        """simple docstring"""







        return self.time_input_dim


  @property
  def 			_lowercase   (      self     :			int		)       ->List[Any]:
        """simple docstring"""







        return self.time_input_dim * 4


  @property
  def 			_lowercase   (      self     :			int		)       ->List[str]:
        """simple docstring"""







        return 1_0_0


  @property
  def 			_lowercase   (      self     :			List[str]		)       ->List[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE			:   List[Any]	  = CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)
        return tokenizer


  @property
  def 			_lowercase   (      self     :			Dict		)       ->Any:
        """simple docstring"""







        torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE			:   List[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=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(UpperCAmelCase__		)


  @property
  def 			_lowercase   (      self     :			Dict		)       ->Any:
        """simple docstring"""







        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[Any]	  = PriorTransformer(**UpperCAmelCase__		)
        # 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			:   str	  = nn.Parameter(torch.ones(model.clip_std.shape		)		)
        return model


  @property
  def 			_lowercase   (      self     :			str		)       ->Union[str, Any]:
        """simple docstring"""







        torch.manual_seed(0		)
        SCREAMING_SNAKE_CASE			:   Optional[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			:   List[str]	  = CLIPVisionModelWithProjection(UpperCAmelCase__		)
        return model


  @property
  def 			_lowercase   (      self     :			List[str]		)       ->str:
        """simple docstring"""







        SCREAMING_SNAKE_CASE			:   Optional[Any]	  = CLIPImageProcessor(
            crop_size=2_2_4					,					do_center_crop=UpperCAmelCase__					,					do_normalize=UpperCAmelCase__					,					do_resize=UpperCAmelCase__					,					image_mean=[0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]					,					image_std=[0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]					,					resample=3					,					size=2_2_4					,					)

        return image_processor


  def 			_lowercase   (      self     :			Dict		)       ->Optional[int]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE			:   List[Any]	  = self.dummy_prior
        SCREAMING_SNAKE_CASE			:   Tuple	  = self.dummy_image_encoder
        SCREAMING_SNAKE_CASE			:   List[str]	  = self.dummy_text_encoder
        SCREAMING_SNAKE_CASE			:   int	  = self.dummy_tokenizer
        SCREAMING_SNAKE_CASE			:   int	  = self.dummy_image_processor

        SCREAMING_SNAKE_CASE			:   Optional[int]	  = UnCLIPScheduler(
            variance_type="""fixed_small_log"""					,					prediction_type="""sample"""					,					num_train_timesteps=1_0_0_0					,					clip_sample=UpperCAmelCase__					,					clip_sample_range=10.0					,					)

        SCREAMING_SNAKE_CASE			:   Union[str, Any]	  = {
            """prior""": prior,
            """image_encoder""": image_encoder,
            """text_encoder""": text_encoder,
            """tokenizer""": tokenizer,
            """scheduler""": scheduler,
            """image_processor""": image_processor,
        }

        return components


  def 			_lowercase   (      self     :			str					,					UpperCAmelCase__     :			Tuple					,					UpperCAmelCase__     :			Union[str, Any]=0		)       ->str:
        """simple docstring"""







        if str(UpperCAmelCase__		).startswith("""mps"""		):
              SCREAMING_SNAKE_CASE			:   Union[str, Any]	  = torch.manual_seed(UpperCAmelCase__		)
        else:
              SCREAMING_SNAKE_CASE			:   List[Any]	  = torch.Generator(device=UpperCAmelCase__		).manual_seed(UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE			:   int	  = floats_tensor((1, 3, 6_4, 6_4)					,					rng=random.Random(UpperCAmelCase__		)		).to(UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE			:   Optional[int]	  = image.cpu().permute(0					,					2					,					3					,					1		)[0]
        SCREAMING_SNAKE_CASE			:   List[str]	  = Image.fromarray(np.uinta(UpperCAmelCase__		)		).convert("""RGB"""		).resize((2_5_6, 2_5_6)		)

        SCREAMING_SNAKE_CASE			:   Optional[int]	  = {
            """prompt""": """horse""",
            """image""": init_image,
            """strength""": 0.5,
            """generator""": generator,
            """guidance_scale""": 4.0,
            """num_inference_steps""": 2,
            """output_type""": """np""",
        }
        return inputs


  def 			_lowercase   (      self     :			Tuple		)       ->Optional[Any]:
        """simple docstring"""







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

        SCREAMING_SNAKE_CASE			:   Tuple	  = self.get_dummy_components()

        SCREAMING_SNAKE_CASE			:   str	  = self.pipeline_class(**UpperCAmelCase__		)
        SCREAMING_SNAKE_CASE			:   List[Any]	  = pipe.to(UpperCAmelCase__		)

        pipe.set_progress_bar_config(disable=UpperCAmelCase__		)

        SCREAMING_SNAKE_CASE			:   List[Any]	  = pipe(**self.get_dummy_inputs(UpperCAmelCase__		)		)
        SCREAMING_SNAKE_CASE			:   Tuple	  = output.image_embeds

        SCREAMING_SNAKE_CASE			:   Tuple	  = pipe(
            **self.get_dummy_inputs(UpperCAmelCase__		)					,					return_dict=UpperCAmelCase__					,					)[0]

        SCREAMING_SNAKE_CASE			:   List[Any]	  = image[0, -1_0:]
        SCREAMING_SNAKE_CASE			:   int	  = image_from_tuple[0, -1_0:]

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

        SCREAMING_SNAKE_CASE			:   Tuple	  = np.array(
            [
                0.1_07_12_84,
                1.3_33_02_71,
                0.61_26_02_23,
                -0.6_69_10_65,
                -0.3_84_68_52,
                -1.0_30_36_61,
                0.22_71_61_11,
                0.03_34_89_01,
                0.30_04_06_75,
                -0.24_80_50_29,
            ]		)

        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 			_lowercase   (      self     :			str		)       ->Optional[Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE			:   int	  = torch_device == """cpu"""
        SCREAMING_SNAKE_CASE			:   Union[str, Any]	  = True
        SCREAMING_SNAKE_CASE			:   Optional[int]	  = False

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




  @skip_mps
  def 			_lowercase   (      self     :			Optional[Any]		)       ->Union[str, Any]:
        """simple docstring"""







        SCREAMING_SNAKE_CASE			:   Optional[int]	  = torch_device == """cpu"""
        SCREAMING_SNAKE_CASE			:   Optional[int]	  = False

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







from typing import List, Optional, Tuple, Union

from . import logging
from .import_utils import is_torch_available, is_torch_version


if is_torch_available():
  import torch

UpperCAmelCase__			:  int					     =     logging.get_logger(__name__)  # pylint: disable=invalid-name

try:
  from torch._dynamo import allow_in_graph as maybe_allow_in_graph
except (ImportError, ModuleNotFoundError):






  def        __lowercase   (  cls      )  ->    Any:
        return cls






def        __lowercase   (  _A    ,     _A = None    ,     _A = None    ,     _A = None    ,     _A = None    ,     )  ->    Tuple:
      SCREAMING_SNAKE_CASE			:   Optional[Any]	  = device
      SCREAMING_SNAKE_CASE			:   Tuple	  = shape[0]

      SCREAMING_SNAKE_CASE			:   Optional[int]	  = layout or torch.strided
      SCREAMING_SNAKE_CASE			:   Dict	  = device or torch.device("""cpu"""      )

      if generator is not None:
            SCREAMING_SNAKE_CASE			:   Optional[Any]	  = generator.device.type if not isinstance(_A    ,     _A      ) else generator[0].device.type
            if gen_device_type != device.type and gen_device_type == "cpu":
                  SCREAMING_SNAKE_CASE			:   Tuple	  = """cpu"""
                  if device != "mps":
                        logger.info(
                            F"The passed generator was created on 'cpu' even though a tensor on {device} was expected."
                            F" Tensors will be created on 'cpu' and then moved to {device}. Note that one can probably"
                            F" slighly speed up this function by passing a generator that was created on the {device} device."      )
            elif gen_device_type != device.type and gen_device_type == "cuda":
                  raise ValueError(F"Cannot generate a {device} tensor from a generator of type {gen_device_type}."      )

      if isinstance(_A    ,     _A      ):
            SCREAMING_SNAKE_CASE			:   int	  = (1,) + shape[1:]
            SCREAMING_SNAKE_CASE			:   Any	  = [
                torch.randn(_A    ,     generator=generator[i]    ,     device=_A    ,     dtype=_A    ,     layout=_A      )
                for i in range(_A      )
            ]
            SCREAMING_SNAKE_CASE			:   Any	  = torch.cat(_A    ,     dim=0      ).to(_A      )
      else:
            SCREAMING_SNAKE_CASE			:   int	  = torch.randn(_A    ,     generator=_A    ,     device=_A    ,     dtype=_A    ,     layout=_A      ).to(_A      )

      return latents







def        __lowercase   (  _A      )  ->    Dict:
      if is_torch_version("""<"""    ,     """2.0.0"""      ) or not hasattr(_A    ,     """_dynamo"""      ):
            return False
      return isinstance(_A    ,     torch._dynamo.eval_frame.OptimizedModule      )




