




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 		__UpperCAmelCase(		lowercase__							,						unittest.TestCase			):





     """simple docstring"""



     __UpperCAmelCase						:			Dict							=     KandinskyVaaPriorEmbaEmbPipeline
     __UpperCAmelCase						:			str							=     ['prompt', 'image']
     __UpperCAmelCase						:			List[str]							=     ['prompt', 'image']
     __UpperCAmelCase						:			int							=     [
         'num_images_per_prompt',
         'strength',
         'generator',
         'num_inference_steps',
         'latents',
         'negative_prompt',
         'guidance_scale',
         'output_type',
         'return_dict',
     ]
     __UpperCAmelCase						:			int							=     False

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






         """simple docstring"""
         return 32

     @property
     def 					__SCREAMING_SNAKE_CASE		(   self			:							Dict     )      ->      List[Any]:






         """simple docstring"""
         return 32

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






         """simple docstring"""
         return self.time_input_dim

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






         """simple docstring"""
         return self.time_input_dim * 4

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






         """simple docstring"""
         return 1_00

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






         """simple docstring"""
         a__  			=							CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""     )
         return tokenizer

     @property
     def 					__SCREAMING_SNAKE_CASE		(   self			:							str     )      ->      Optional[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=10_00    ,					)
         return CLIPTextModelWithProjection(_UpperCAmelCase     )

     @property
     def 					__SCREAMING_SNAKE_CASE		(   self			:							Any     )      ->      Optional[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(**_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
         a__  			=							nn.Parameter(torch.ones(model.clip_std.shape     )     )
         return model

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






         """simple docstring"""
         torch.manual_seed(0     )
         a__  			=							CLIPVisionConfig(
             hidden_size=self.text_embedder_hidden_size    ,					image_size=2_24    ,					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(_UpperCAmelCase     )
         return model

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






         """simple docstring"""
         a__  			=							CLIPImageProcessor(
             crop_size=2_24    ,					do_center_crop=_UpperCAmelCase    ,					do_normalize=_UpperCAmelCase    ,					do_resize=_UpperCAmelCase    ,					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=2_24    ,					)

         return image_processor

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Any     )      ->      Dict:






         """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=10_00    ,					clip_sample=_UpperCAmelCase    ,					clip_sample_range=1_0.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			:							List[str]    ,					A			:							List[str]    ,					A			:							List[str]=0     )      ->      str:






         """simple docstring"""
         if str(_UpperCAmelCase     ).startswith("""mps"""     ):
             a__  			=							torch.manual_seed(_UpperCAmelCase     )
         else:
             a__  			=							torch.Generator(device=_UpperCAmelCase     ).manual_seed(_UpperCAmelCase     )

         a__  			=							floats_tensor((1, 3, 64, 64)    ,					rng=random.Random(_UpperCAmelCase     )     ).to(_UpperCAmelCase     )
         a__  			=							image.cpu().permute(0    ,					2    ,					3    ,					1     )[0]
         a__  			=							Image.fromarray(np.uinta(_UpperCAmelCase     )     ).convert("""RGB"""     ).resize((2_56, 2_56)     )

         a__  			=							{
             """prompt""": """horse""",
             """image""": init_image,
             """strength""": 0.5,
             """generator""": generator,
             """guidance_scale""": 4.0,
             """num_inference_steps""": 2,
             """output_type""": """np""",
         }
         return inputs

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Optional[Any]     )      ->      Optional[int]:






         """simple docstring"""
         a__  			=							"""cpu"""

         a__  			=							self.get_dummy_components()

         a__  			=							self.pipeline_class(**_UpperCAmelCase     )
         a__  			=							pipe.to(_UpperCAmelCase     )

         pipe.set_progress_bar_config(disable=_UpperCAmelCase     )

         a__  			=							pipe(**self.get_dummy_inputs(_UpperCAmelCase     )     )
         a__  			=							output.image_embeds

         a__  			=							pipe(
             **self.get_dummy_inputs(_UpperCAmelCase     )    ,					return_dict=_UpperCAmelCase    ,					)[0]

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

         assert image.shape == (1, 32)

         a__  			=							np.array(
             [
                 0.1_0_7_1_2_8_4,
                 1.3_3_3_0_2_7_1,
                 0.6_1_2_6_0_2_2_3,
                 -0.6_6_9_1_0_6_5,
                 -0.3_8_4_6_8_5_2,
                 -1.0_3_0_3_6_6_1,
                 0.2_2_7_1_6_1_1_1,
                 0.0_3_3_4_8_9_0_1,
                 0.3_0_0_4_0_6_7_5,
                 -0.2_4_8_0_5_0_2_9,
             ]     )

         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			:							Optional[Any]     )      ->      str:






         """simple docstring"""
         a__  			=							torch_device == """cpu"""
         a__  			=							True
         a__  			=							False

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






     @skip_mps
     def 					__SCREAMING_SNAKE_CASE		(   self			:							Union[str, Any]     )      ->      str:






         """simple docstring"""
         a__  			=							torch_device == """cpu"""
         a__  			=							False

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








from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...file_utils import TensorType, is_torch_available
from ...onnx import OnnxConfig, OnnxConfigWithPast, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import logging


_snake_case		  =   logging.get_logger(__name__)

_snake_case		  =   {
    "facebook/blenderbot-3B": "https://huggingface.co/facebook/blenderbot-3B/resolve/main/config.json",
    # See all Blenderbot models at https://huggingface.co/models?filter=blenderbot
}


class 			UpperCamelCase_				(  A     ):




      '''simple docstring'''







      a		:Union[str, Any]						 =      'blenderbot'
      a		:List[str]						 =      ['past_key_values']
      a		:Any						 =      {'num_attention_heads': 'encoder_attention_heads', 'hidden_size': 'd_model'}







      def __init__( self					,					_UpperCAmelCase=8_008					,					_UpperCAmelCase=128					,					_UpperCAmelCase=2					,					_UpperCAmelCase=10_240					,					_UpperCAmelCase=32					,					_UpperCAmelCase=24					,					_UpperCAmelCase=10_240					,					_UpperCAmelCase=32					,					_UpperCAmelCase=0.0					,					_UpperCAmelCase=0.0					,					_UpperCAmelCase=True					,					_UpperCAmelCase=True					,					_UpperCAmelCase="gelu"					,					_UpperCAmelCase=2_560					,					_UpperCAmelCase=0.1					,					_UpperCAmelCase=0.0					,					_UpperCAmelCase=0.0					,					_UpperCAmelCase=0.02					,					_UpperCAmelCase=1					,					_UpperCAmelCase=False					,					_UpperCAmelCase=0					,					_UpperCAmelCase=1					,					_UpperCAmelCase=2					,					_UpperCAmelCase=3					,					_UpperCAmelCase=2					,					**_UpperCAmelCase					,					):
            lowerCAmelCase_						=      vocab_size
            lowerCAmelCase_						=      max_position_embeddings
            lowerCAmelCase_						=      d_model
            lowerCAmelCase_						=      encoder_ffn_dim
            lowerCAmelCase_						=      encoder_layers
            lowerCAmelCase_						=      encoder_attention_heads
            lowerCAmelCase_						=      decoder_ffn_dim
            lowerCAmelCase_						=      decoder_layers
            lowerCAmelCase_						=      decoder_attention_heads
            lowerCAmelCase_						=      dropout
            lowerCAmelCase_						=      attention_dropout
            lowerCAmelCase_						=      activation_dropout
            lowerCAmelCase_						=      activation_function
            lowerCAmelCase_						=      init_std
            lowerCAmelCase_						=      encoder_layerdrop
            lowerCAmelCase_						=      decoder_layerdrop
            lowerCAmelCase_						=      use_cache
            lowerCAmelCase_						=      encoder_layers
            lowerCAmelCase_						=      scale_embedding  # scale factor will be sqrt(d_model) if True

            super().__init__(
                pad_token_id=_UpperCAmelCase					,					bos_token_id=_UpperCAmelCase					,					eos_token_id=_UpperCAmelCase					,					is_encoder_decoder=_UpperCAmelCase					,					decoder_start_token_id=_UpperCAmelCase					,					encoder_no_repeat_ngram_size=_UpperCAmelCase					,					forced_eos_token_id=_UpperCAmelCase					,					**_UpperCAmelCase					,					)


class 			UpperCamelCase_				(  A     ):




      '''simple docstring'''







      @property
      def lowercase__		( self):
            if self.task in ["default", "seq2seq-lm"]:
                  lowerCAmelCase_						=      OrderedDict(
                      [
                          ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                          ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                      ])
                  if self.use_past:
                        lowerCAmelCase_						=      {0: '''batch'''}
                        lowerCAmelCase_						=      {0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
                  else:
                        lowerCAmelCase_						=      {0: '''batch''', 1: '''decoder_sequence'''}
                        lowerCAmelCase_						=      {0: '''batch''', 1: '''decoder_sequence'''}
                  if self.use_past:
                        self.fill_with_past_key_values_(_UpperCAmelCase					,					direction='''inputs''')
            elif self.task == "causal-lm":
                  lowerCAmelCase_						=      OrderedDict(
                      [
                          ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                          ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                      ])
                  if self.use_past:
                        lowerCAmelCase_			,		lowerCAmelCase_						=      self.num_layers
                        for i in range(_UpperCAmelCase):
                              lowerCAmelCase_						=      {0: '''batch''', 2: '''past_sequence + sequence'''}
                              lowerCAmelCase_						=      {0: '''batch''', 2: '''past_sequence + sequence'''}
            else:
                  lowerCAmelCase_						=      OrderedDict(
                      [
                          ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                          ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                          ('''decoder_input_ids''', {0: '''batch''', 1: '''decoder_sequence'''}),
                          ('''decoder_attention_mask''', {0: '''batch''', 1: '''decoder_sequence'''}),
                      ])

            return common_inputs







      @property
      # Copied from transformers.models.bart.configuration_bart.BartOnnxConfig.outputs
      def lowercase__		( self):
            if self.task in ["default", "seq2seq-lm"]:
                  lowerCAmelCase_						=      super().outputs
            else:
                  lowerCAmelCase_						=      super(_UpperCAmelCase					,					self).outputs
                  if self.use_past:
                        lowerCAmelCase_			,		lowerCAmelCase_						=      self.num_layers
                        for i in range(_UpperCAmelCase):
                              lowerCAmelCase_						=      {0: '''batch''', 2: '''past_sequence + sequence'''}
                              lowerCAmelCase_						=      {0: '''batch''', 2: '''past_sequence + sequence'''}
            return common_outputs







      def lowercase__		( self					,					_UpperCAmelCase					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = False					,					_UpperCAmelCase = None					,					):
            lowerCAmelCase_						=      self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                _UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase)
            # Generate decoder inputs
            lowerCAmelCase_						=      seq_length if not self.use_past else 1
            lowerCAmelCase_						=      self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                _UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase)
            lowerCAmelCase_						=      {f'decoder_{name}': tensor for name, tensor in decoder_inputs.items()}
            lowerCAmelCase_						=      dict(**_UpperCAmelCase					,					**_UpperCAmelCase)

            if self.use_past:
                  if not is_torch_available():
                        raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.''')
                  else:
                        import torch
                  lowerCAmelCase_			,		lowerCAmelCase_						=      common_inputs['''input_ids'''].shape
                  lowerCAmelCase_						=      common_inputs['''decoder_input_ids'''].shape[1]
                  lowerCAmelCase_			,		lowerCAmelCase_						=      self.num_attention_heads
                  lowerCAmelCase_						=      (
                      batch,
                      num_encoder_attention_heads,
                      encoder_seq_length,
                      self._config.hidden_size // num_encoder_attention_heads,
                  )
                  lowerCAmelCase_						=      decoder_seq_length
                  lowerCAmelCase_						=      (
                      batch,
                      num_decoder_attention_heads,
                      decoder_past_length,
                      self._config.hidden_size // num_decoder_attention_heads,
                  )
                  lowerCAmelCase_						=      torch.cat(
                      [common_inputs['''decoder_attention_mask'''], torch.ones(_UpperCAmelCase					,					_UpperCAmelCase)]					,					dim=1)
                  lowerCAmelCase_						=      []
                  lowerCAmelCase_			,		lowerCAmelCase_						=      self.num_layers

                  for _ in range(_UpperCAmelCase):
                        common_inputs["past_key_values"].append(
                            (
                                torch.zeros(_UpperCAmelCase),
                                torch.zeros(_UpperCAmelCase),
                                torch.zeros(_UpperCAmelCase),
                                torch.zeros(_UpperCAmelCase),
                            ))
            return common_inputs







      def lowercase__		( self					,					_UpperCAmelCase					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = False					,					_UpperCAmelCase = None					,					):
            lowerCAmelCase_						=      self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                _UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase)

            if self.use_past:
                  if not is_torch_available():
                        raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.''')
                  else:
                        import torch
                  lowerCAmelCase_			,		lowerCAmelCase_						=      common_inputs['''input_ids'''].shape
                  lowerCAmelCase_						=      seqlen
                  lowerCAmelCase_			,		lowerCAmelCase_						=      self.num_layers
                  lowerCAmelCase_			,		lowerCAmelCase_						=      self.num_attention_heads
                  lowerCAmelCase_						=      (
                      batch,
                      num_encoder_attention_heads,
                      past_key_values_length,
                      self._config.hidden_size // num_encoder_attention_heads,
                  )
                  lowerCAmelCase_						=      common_inputs['''attention_mask'''].dtype
                  lowerCAmelCase_						=      torch.cat(
                      [common_inputs['''attention_mask'''], torch.ones(_UpperCAmelCase					,					_UpperCAmelCase					,					dtype=_UpperCAmelCase)]					,					dim=1)
                  lowerCAmelCase_						=      [
                      (torch.zeros(_UpperCAmelCase), torch.zeros(_UpperCAmelCase)) for _ in range(_UpperCAmelCase)
                  ]
            return common_inputs







      def lowercase__		( self					,					_UpperCAmelCase					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = False					,					_UpperCAmelCase = None					,					):
            # Copied from OnnxConfig.generate_dummy_inputs
            # Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
            # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
            lowerCAmelCase_						=      compute_effective_axis_dimension(
                _UpperCAmelCase					,					fixed_dimension=OnnxConfig.default_fixed_batch					,					num_token_to_add=0)

            # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
            lowerCAmelCase_						=      tokenizer.num_special_tokens_to_add(_UpperCAmelCase)
            lowerCAmelCase_						=      compute_effective_axis_dimension(
                _UpperCAmelCase					,					fixed_dimension=OnnxConfig.default_fixed_sequence					,					num_token_to_add=_UpperCAmelCase)

            # Generate dummy inputs according to compute batch and sequence
            lowerCAmelCase_						=      [''' '''.join([tokenizer.unk_token]) * seq_length] * batch_size
            lowerCAmelCase_						=      dict(tokenizer(_UpperCAmelCase					,					return_tensors=_UpperCAmelCase))
            return common_inputs







      def lowercase__		( self					,					_UpperCAmelCase					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = -1					,					_UpperCAmelCase = False					,					_UpperCAmelCase = None					,					):
            if self.task in ["default", "seq2seq-lm"]:
                  lowerCAmelCase_						=      self._generate_dummy_inputs_for_default_and_seqaseq_lm(
                      _UpperCAmelCase					,					batch_size=_UpperCAmelCase					,					seq_length=_UpperCAmelCase					,					is_pair=_UpperCAmelCase					,					framework=_UpperCAmelCase)

            elif self.task == "causal-lm":
                  lowerCAmelCase_						=      self._generate_dummy_inputs_for_causal_lm(
                      _UpperCAmelCase					,					batch_size=_UpperCAmelCase					,					seq_length=_UpperCAmelCase					,					is_pair=_UpperCAmelCase					,					framework=_UpperCAmelCase)
            else:
                  lowerCAmelCase_						=      self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                      _UpperCAmelCase					,					batch_size=_UpperCAmelCase					,					seq_length=_UpperCAmelCase					,					is_pair=_UpperCAmelCase					,					framework=_UpperCAmelCase)

            return common_inputs







      def lowercase__		( self					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase):
            if self.task in ["default", "seq2seq-lm"]:
                  lowerCAmelCase_						=      super()._flatten_past_key_values_(_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase)
            else:
                  lowerCAmelCase_						=      super(_UpperCAmelCase					,					self)._flatten_past_key_values_(
                      _UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase					,					_UpperCAmelCase)







      def lowercase__		( self					,					_UpperCAmelCase					,					_UpperCAmelCase):
            if direction not in ["inputs", "outputs"]:
                  raise ValueError(f'direction must either be "inputs" or "outputs", but {direction} was given')

            lowerCAmelCase_						=      '''past_key_values''' if direction == '''inputs''' else '''present'''
            lowerCAmelCase_			,		lowerCAmelCase_						=      self.num_layers

            lowerCAmelCase_						=      '''past_encoder_sequence'''
            lowerCAmelCase_						=      '''past_decoder_sequence''' if direction == '''inputs''' else '''past_decoder_sequence + sequence'''

            for i in range(_UpperCAmelCase):
                  lowerCAmelCase_						=      {0: '''batch''', 2: decoder_sequence}
                  lowerCAmelCase_						=      {0: '''batch''', 2: decoder_sequence}
                  lowerCAmelCase_						=      {0: '''batch''', 2: encoder_sequence}
                  lowerCAmelCase_						=      {0: '''batch''', 2: encoder_sequence}



