


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__ ( __lowerCamelCase							,						unittest.TestCase			):
      a__   :     Union[str, Any]								=						KandinskyVaaPriorEmbaEmbPipeline
      a__   :     int								=						['prompt', 'image']
      a__   :     List[Any]								=						['prompt', 'image']
      a__   :     Any								=						[
          'num_images_per_prompt',
          'strength',
          'generator',
          'num_inference_steps',
          'latents',
          'negative_prompt',
          'guidance_scale',
          'output_type',
          'return_dict',
      ]
      a__   :     Tuple								=						False






      @property
      def 	__lowercase(     self					:   List[Any]   )  -> Dict:
           return 32






      @property
      def 	__lowercase(     self					:   str   )  -> Tuple:
           return 32






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






      @property
      def 	__lowercase(     self					:   Optional[int]   )  -> Tuple:
           return self.time_input_dim * 4






      @property
      def 	__lowercase(     self					:   Dict   )  -> Dict:
           return 1_00






      @property
      def 	__lowercase(     self					:   Dict   )  -> List[Any]:
           UpperCamelCase__							:  List[str]								=     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''   )
           return tokenizer






      @property
      def 	__lowercase(     self					:   Optional[Any]   )  -> Union[str, Any]:
           torch.manual_seed(0   )
           UpperCamelCase__							:  Tuple								=     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(__lowerCamelCase   )






      @property
      def 	__lowercase(     self					:   Tuple   )  -> Optional[Any]:
           torch.manual_seed(0   )

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

           UpperCamelCase__							:  Any								=     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
           UpperCamelCase__							:  int								=     nn.Parameter(torch.ones(model.clip_std.shape   )   )
           return model






      @property
      def 	__lowercase(     self					:   Tuple   )  -> str:
           torch.manual_seed(0   )
           UpperCamelCase__							:  Any								=     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,		)

           UpperCamelCase__							:  int								=     CLIPVisionModelWithProjection(__lowerCamelCase   )
           return model






      @property
      def 	__lowercase(     self					:   List[Any]   )  -> int:
           UpperCamelCase__							:  Any								=     CLIPImageProcessor(
               crop_size=2_24,		do_center_crop=__lowerCamelCase,		do_normalize=__lowerCamelCase,		do_resize=__lowerCamelCase,		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=2_24,		)

           return image_processor






      def 	__lowercase(     self					:   str   )  -> Tuple:
           UpperCamelCase__							:  str								=     self.dummy_prior
           UpperCamelCase__							:  str								=     self.dummy_image_encoder
           UpperCamelCase__							:  List[str]								=     self.dummy_text_encoder
           UpperCamelCase__							:  Optional[int]								=     self.dummy_tokenizer
           UpperCamelCase__							:  List[Any]								=     self.dummy_image_processor

           UpperCamelCase__							:  Any								=     UnCLIPScheduler(
               variance_type='''fixed_small_log''',		prediction_type='''sample''',		num_train_timesteps=10_00,		clip_sample=__lowerCamelCase,		clip_sample_range=10.0,		)

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

           return components






      def 	__lowercase(     self					:   Tuple,		__lowerCamelCase					:   int,		__lowerCamelCase					:   str=0   )  -> int:
           if str(__lowerCamelCase   ).startswith('''mps'''   ):
                UpperCamelCase__							:  str								=     torch.manual_seed(__lowerCamelCase   )
           else:
                UpperCamelCase__							:  Any								=     torch.Generator(device=__lowerCamelCase   ).manual_seed(__lowerCamelCase   )

           UpperCamelCase__							:  List[Any]								=     floats_tensor((1, 3, 64, 64),		rng=random.Random(__lowerCamelCase   )   ).to(__lowerCamelCase   )
           UpperCamelCase__							:  Optional[int]								=     image.cpu().permute(0,		2,		3,		1   )[0]
           UpperCamelCase__							:  List[Any]								=     Image.fromarray(np.uinta(__lowerCamelCase   )   ).convert('''RGB'''   ).resize((2_56, 2_56)   )

           UpperCamelCase__							:  List[Any]								=     {
               '''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					:   str   )  -> Optional[int]:
           UpperCamelCase__							:  Dict								=     '''cpu'''

           UpperCamelCase__							:  int								=     self.get_dummy_components()

           UpperCamelCase__							:  Tuple								=     self.pipeline_class(**__lowerCamelCase   )
           UpperCamelCase__							:  Optional[int]								=     pipe.to(__lowerCamelCase   )

           pipe.set_progress_bar_config(disable=__lowerCamelCase   )

           UpperCamelCase__							:  str								=     pipe(**self.get_dummy_inputs(__lowerCamelCase   )   )
           UpperCamelCase__							:  Optional[int]								=     output.image_embeds

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

           UpperCamelCase__							:  Optional[Any]								=     image[0, -10:]
           UpperCamelCase__							:  Optional[Any]								=     image_from_tuple[0, -10:]

           assert image.shape == (1, 32)

           UpperCamelCase__							:  List[str]								=     np.array(
               [
                   0.107_1284,
                   1.333_0271,
                   0.6126_0223,
                   -0.669_1065,
                   -0.384_6852,
                   -1.030_3661,
                   0.2271_6111,
                   0.0334_8901,
                   0.3004_0675,
                   -0.2480_5029,
               ]   )

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

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






      @skip_mps
      def 	__lowercase(     self					:   int   )  -> Dict:
           UpperCamelCase__							:  Tuple								=     torch_device == '''cpu'''
           UpperCamelCase__							:  List[str]								=     False

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








import json
from typing import TYPE_CHECKING, List, Optional, Tuple

from tokenizers import pre_tokenizers

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_deberta import DebertaTokenizer


if TYPE_CHECKING:
   from transformers.pipelines.conversational import Conversation


_SCREAMING_SNAKE_CASE :					Tuple			=	logging.get_logger(__name__)

_SCREAMING_SNAKE_CASE :					Tuple			=	{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt""", """tokenizer_file""": """tokenizer.json"""}

_SCREAMING_SNAKE_CASE :					str			=	{
    """vocab_file""": {
        """microsoft/deberta-base""": """https://huggingface.co/microsoft/deberta-base/resolve/main/vocab.json""",
        """microsoft/deberta-large""": """https://huggingface.co/microsoft/deberta-large/resolve/main/vocab.json""",
        """microsoft/deberta-xlarge""": """https://huggingface.co/microsoft/deberta-xlarge/resolve/main/vocab.json""",
        """microsoft/deberta-base-mnli""": """https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/vocab.json""",
        """microsoft/deberta-large-mnli""": """https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/vocab.json""",
        """microsoft/deberta-xlarge-mnli""": (
            """https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/vocab.json"""
        ),
    },
    """merges_file""": {
        """microsoft/deberta-base""": """https://huggingface.co/microsoft/deberta-base/resolve/main/merges.txt""",
        """microsoft/deberta-large""": """https://huggingface.co/microsoft/deberta-large/resolve/main/merges.txt""",
        """microsoft/deberta-xlarge""": """https://huggingface.co/microsoft/deberta-xlarge/resolve/main/merges.txt""",
        """microsoft/deberta-base-mnli""": """https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/merges.txt""",
        """microsoft/deberta-large-mnli""": """https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/merges.txt""",
        """microsoft/deberta-xlarge-mnli""": (
            """https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/merges.txt"""
        ),
    },
}

_SCREAMING_SNAKE_CASE :					Any			=	{
    """microsoft/deberta-base""": 512,
    """microsoft/deberta-large""": 512,
    """microsoft/deberta-xlarge""": 512,
    """microsoft/deberta-base-mnli""": 512,
    """microsoft/deberta-large-mnli""": 512,
    """microsoft/deberta-xlarge-mnli""": 512,
}

_SCREAMING_SNAKE_CASE :					Any			=	{
    """microsoft/deberta-base""": {"""do_lower_case""": False},
    """microsoft/deberta-large""": {"""do_lower_case""": False},
}

class  UpperCamelCase__ ( __lowerCamelCase			):
      a__   :     int								=						VOCAB_FILES_NAMES
      a__   :     Optional[Any]								=						PRETRAINED_VOCAB_FILES_MAP
      a__   :     int								=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      a__   :     str								=						['input_ids', 'attention_mask', 'token_type_ids']
      a__   :     int								=						DebertaTokenizer






      def __init__(     self					:   Optional[int],		__lowerCamelCase					:   List[str]=None,		__lowerCamelCase					:   Optional[Any]=None,		__lowerCamelCase					:   Union[str, Any]=None,		__lowerCamelCase					:   str="replace",		__lowerCamelCase					:   Any="[CLS]",		__lowerCamelCase					:   Optional[int]="[SEP]",		__lowerCamelCase					:   List[Any]="[SEP]",		__lowerCamelCase					:   Optional[Any]="[CLS]",		__lowerCamelCase					:   List[Any]="[UNK]",		__lowerCamelCase					:   Optional[Any]="[PAD]",		__lowerCamelCase					:   Dict="[MASK]",		__lowerCamelCase					:   List[Any]=False,		**__lowerCamelCase					:   Union[str, Any],		)  -> Optional[int]:
           super().__init__(
               __lowerCamelCase,		__lowerCamelCase,		tokenizer_file=__lowerCamelCase,		errors=__lowerCamelCase,		bos_token=__lowerCamelCase,		eos_token=__lowerCamelCase,		unk_token=__lowerCamelCase,		sep_token=__lowerCamelCase,		cls_token=__lowerCamelCase,		pad_token=__lowerCamelCase,		mask_token=__lowerCamelCase,		add_prefix_space=__lowerCamelCase,		**__lowerCamelCase,		)
           UpperCamelCase__							:  Any								=     kwargs.pop('''add_bos_token''',		__lowerCamelCase   )

           UpperCamelCase__							:  List[Any]								=     json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()   )
           if pre_tok_state.get('''add_prefix_space''',		__lowerCamelCase   ) != add_prefix_space:
                UpperCamelCase__							:  Tuple								=     getattr(__lowerCamelCase,		pre_tok_state.pop('''type'''   )   )
                UpperCamelCase__							:  Tuple								=     add_prefix_space
                UpperCamelCase__							:  Union[str, Any]								=     pre_tok_class(**__lowerCamelCase   )

           UpperCamelCase__							:  Optional[int]								=     add_prefix_space






      @property
      def 	__lowercase(     self					:   Dict   )  -> str:
           if self._mask_token is None:
                if self.verbose:
                     logger.error('''Using mask_token, but it is not set yet.'''   )
                return None
           return str(self._mask_token   )






      @mask_token.setter
      def 	__lowercase(     self					:   List[Any],		__lowerCamelCase					:   Union[str, Any]   )  -> List[str]:
           UpperCamelCase__							:  List[Any]								=     AddedToken(__lowerCamelCase,		lstrip=__lowerCamelCase,		rstrip=__lowerCamelCase   ) if isinstance(__lowerCamelCase,		__lowerCamelCase   ) else value
           UpperCamelCase__							:  List[str]								=     value






      def 	__lowercase(     self					:   List[str],		__lowerCamelCase					:   List[int],		__lowerCamelCase					:   Optional[List[int]] = None   )  -> List[int]:
           if token_ids_a is None:
                return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
           UpperCamelCase__							:  Union[str, Any]								=     [self.cls_token_id]
           UpperCamelCase__							:  Union[str, Any]								=     [self.sep_token_id]
           return cls + token_ids_a + sep + token_ids_a + sep






      def 	__lowercase(     self					:   str,		__lowerCamelCase					:   List[int],		__lowerCamelCase					:   Optional[List[int]] = None   )  -> List[int]:
           UpperCamelCase__							:  Dict								=     [self.sep_token_id]
           UpperCamelCase__							:  int								=     [self.cls_token_id]

           if token_ids_a is None:
                return len(cls + token_ids_a + sep   ) * [0]
           return len(cls + token_ids_a + sep   ) * [0] + len(token_ids_a + sep   ) * [1]






      def 	__lowercase(     self					:   Union[str, Any],		*__lowerCamelCase					:   Any,		**__lowerCamelCase					:   List[Any]   )  -> BatchEncoding:
           UpperCamelCase__							:  str								=     kwargs.get('''is_split_into_words''',		__lowerCamelCase   )
           assert self.add_prefix_space or not is_split_into_words, (
               f'You need to instantiate {self.__class__.__name__} with add_prefix_space=True '
               "to use it with pretokenized inputs."
           )

           return super()._batch_encode_plus(*__lowerCamelCase,		**__lowerCamelCase   )






      def 	__lowercase(     self					:   Optional[Any],		*__lowerCamelCase					:   str,		**__lowerCamelCase					:   int   )  -> BatchEncoding:
           UpperCamelCase__							:  str								=     kwargs.get('''is_split_into_words''',		__lowerCamelCase   )

           assert self.add_prefix_space or not is_split_into_words, (
               f'You need to instantiate {self.__class__.__name__} with add_prefix_space=True '
               "to use it with pretokenized inputs."
           )

           return super()._encode_plus(*__lowerCamelCase,		**__lowerCamelCase   )






      def 	__lowercase(     self					:   Union[str, Any],		__lowerCamelCase					:   str,		__lowerCamelCase					:   Optional[str] = None   )  -> Tuple[str]:
           UpperCamelCase__							:  Dict								=     self._tokenizer.model.save(__lowerCamelCase,		name=__lowerCamelCase   )
           return tuple(__lowerCamelCase   )






      def 	__lowercase(     self					:   int,		__lowerCamelCase					:   "Conversation"   )  -> List[int]:
           UpperCamelCase__							:  List[Any]								=     []
           for is_user, text in conversation.iter_texts():
                input_ids.extend(self.encode(__lowerCamelCase,		add_special_tokens=__lowerCamelCase   ) + [self.eos_token_id]   )

           if len(__lowerCamelCase   ) > self.model_max_length:
                UpperCamelCase__							:  List[Any]								=     input_ids[-self.model_max_length :]
           return input_ids




