


import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__UpperCamelCase							:							Union[str, Any]    							=					logging.get_logger(__name__)


__UpperCamelCase							:							Union[str, Any]    							=					{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

# See all MVP models at https://huggingface.co/models?filter=mvp
__UpperCamelCase							:							str    							=					{
    '''vocab_file''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/vocab.json''',
    },
    '''added_tokens.json''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/added_tokens.json''',
    },
    '''merges_file''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/merges.txt''',
    },
}

__UpperCamelCase							:							str    							=					{
    '''RUCAIBox/mvp''': 1024,
}

@lru_cache()
def       _a	(					):







 """simple docstring"""



 UpperCamelCase__       :     Any 							= (
     list(range(ord('''!'''  )     ,			ord('''~'''  ) + 1  )  ) + list(range(ord('''¡'''  )     ,			ord('''¬'''  ) + 1  )  ) + list(range(ord('''®'''  )     ,			ord('''ÿ'''  ) + 1  )  )
 )
 UpperCamelCase__       :     Union[str, Any] 							= bs[:]
 UpperCamelCase__       :     Dict 							= 0
 for b in range(2**8  ):
  if b not in bs:
   bs.append(UpperCAmelCase_  )
   cs.append(2**8 + n  )
   n += 1
 UpperCamelCase__       :     List[str] 							= [chr(UpperCAmelCase_  ) for n in cs]
 return dict(zip(UpperCAmelCase_     ,			UpperCAmelCase_  )  )

def       _a	(					SCREAMING_SNAKE_CASE						:					str  ):







 """simple docstring"""



 UpperCamelCase__       :     Optional[Any] 							= set()
 UpperCamelCase__       :     str 							= word[0]
 for char in word[1:]:
  pairs.add((prev_char, char)  )
  UpperCamelCase__       :     Union[str, Any] 							= char
 return pairs






class    __magic_name__						(  lowercase__):
     A:		List[Any]								=   VOCAB_FILES_NAMES
     A:		List[str]								=   PRETRAINED_VOCAB_FILES_MAP
     A:		Tuple								=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     A:		Any								=   ["""input_ids""", """attention_mask"""]


     def __init__(   self				:  Any				,						lowerCamelCase__				:  Tuple				,						lowerCamelCase__				:  Optional[Any]				,						lowerCamelCase__				:  str="replace"				,						lowerCamelCase__				:  int="<s>"				,						lowerCamelCase__				:  Union[str, Any]="</s>"				,						lowerCamelCase__				:  Union[str, Any]="</s>"				,						lowerCamelCase__				:  Union[str, Any]="<s>"				,						lowerCamelCase__				:  List[str]="<unk>"				,						lowerCamelCase__				:  Optional[int]="<pad>"				,						lowerCamelCase__				:  Optional[int]="<mask>"				,						lowerCamelCase__				:  Union[str, Any]=False				,						**lowerCamelCase__				:  List[Any]				,						)	->   Optional[int]:





      '''simple docstring'''






      UpperCamelCase__       :     Union[str, Any] 							= AddedToken(lowerCamelCase__				,						lstrip=lowerCamelCase__				,						rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__				,						lowerCamelCase__      ) else bos_token
      UpperCamelCase__       :     str 							= AddedToken(lowerCamelCase__				,						lstrip=lowerCamelCase__				,						rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__				,						lowerCamelCase__      ) else eos_token
      UpperCamelCase__       :     Optional[Any] 							= AddedToken(lowerCamelCase__				,						lstrip=lowerCamelCase__				,						rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__				,						lowerCamelCase__      ) else sep_token
      UpperCamelCase__       :     Union[str, Any] 							= AddedToken(lowerCamelCase__				,						lstrip=lowerCamelCase__				,						rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__				,						lowerCamelCase__      ) else cls_token
      UpperCamelCase__       :     List[str] 							= AddedToken(lowerCamelCase__				,						lstrip=lowerCamelCase__				,						rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__				,						lowerCamelCase__      ) else unk_token
      UpperCamelCase__       :     Optional[Any] 							= AddedToken(lowerCamelCase__				,						lstrip=lowerCamelCase__				,						rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__				,						lowerCamelCase__      ) else pad_token

      # Mask token behave like a normal word, i.e. include the space before it
      UpperCamelCase__       :     Optional[Any] 							= AddedToken(lowerCamelCase__				,						lstrip=lowerCamelCase__				,						rstrip=lowerCamelCase__      ) if isinstance(lowerCamelCase__				,						lowerCamelCase__      ) else mask_token

      super().__init__(
          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__				,						)
      with open(lowerCamelCase__				,						encoding='''utf-8'''      ) as vocab_handle:
       UpperCamelCase__       :     Dict 							= json.load(lowerCamelCase__      )
      UpperCamelCase__       :     Any 							= {v: k for k, v in self.encoder.items()}
      UpperCamelCase__       :     List[str] 							= errors  # how to handle errors in decoding
      UpperCamelCase__       :     List[Any] 							= bytes_to_unicode()
      UpperCamelCase__       :     Optional[int] 							= {v: k for k, v in self.byte_encoder.items()}
      with open(lowerCamelCase__				,						encoding='''utf-8'''      ) as merges_handle:
       UpperCamelCase__       :     Any 							= merges_handle.read().split('''\n'''      )[1:-1]
      UpperCamelCase__       :     Any 							= [tuple(merge.split()      ) for merge in bpe_merges]
      UpperCamelCase__       :     Dict 							= dict(zip(lowerCamelCase__				,						range(len(lowerCamelCase__      )      )      )      )
      UpperCamelCase__       :     List[Any] 							= {}
      UpperCamelCase__       :     List[str] 							= add_prefix_space

      # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
      UpperCamelCase__       :     Any 							= re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''      )


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





      '''simple docstring'''






      return len(self.encoder      )


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





      '''simple docstring'''






      return dict(self.encoder				,						**self.added_tokens_encoder      )


     def   UpperCAmelCase__						(   self				:  Dict				,						lowerCamelCase__				:  str      )	->   str:





      '''simple docstring'''






      if token in self.cache:
       return self.cache[token]
      UpperCamelCase__       :     List[str] 							= tuple(lowerCamelCase__      )
      UpperCamelCase__       :     Any 							= get_pairs(lowerCamelCase__      )

      if not pairs:
       return token

      while True:
       UpperCamelCase__       :     Union[str, Any] 							= min(lowerCamelCase__				,						key=lambda lowerCamelCase__      : self.bpe_ranks.get(lowerCamelCase__				,						float('''inf'''      )      )      )
       if bigram not in self.bpe_ranks:
        break
       UpperCamelCase__							,					UpperCamelCase__       :     List[str] 							= bigram
       UpperCamelCase__       :     List[Any] 							= []
       UpperCamelCase__       :     Dict 							= 0
       while i < len(lowerCamelCase__      ):
        try:
         UpperCamelCase__       :     Union[str, Any] 							= word.index(lowerCamelCase__				,						lowerCamelCase__      )
        except ValueError:
         new_word.extend(word[i:]      )
         break
        else:
         new_word.extend(word[i:j]      )
         UpperCamelCase__       :     Optional[Any] 							= j

        if word[i] == first and i < len(lowerCamelCase__      ) - 1 and word[i + 1] == second:
         new_word.append(first + second      )
         i += 2
        else:
         new_word.append(word[i]      )
         i += 1
       UpperCamelCase__       :     Union[str, Any] 							= tuple(lowerCamelCase__      )
       UpperCamelCase__       :     Tuple 							= new_word
       if len(lowerCamelCase__      ) == 1:
        break
       else:
        UpperCamelCase__       :     Optional[Any] 							= get_pairs(lowerCamelCase__      )
      UpperCamelCase__       :     str 							= ''' '''.join(lowerCamelCase__      )
      UpperCamelCase__       :     str 							= word
      return word


     def   UpperCAmelCase__						(   self				:  List[Any]				,						lowerCamelCase__				:  int      )	->   List[str]:





      '''simple docstring'''






      UpperCamelCase__       :     Optional[Any] 							= []
      for token in re.findall(self.pat				,						lowerCamelCase__      ):
       UpperCamelCase__       :     List[Any] 							= ''''''.join(
           self.byte_encoder[b] for b in token.encode('''utf-8'''      )      )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
       bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCamelCase__      ).split(''' '''      )      )
      return bpe_tokens


     def   UpperCAmelCase__						(   self				:  Union[str, Any]				,						lowerCamelCase__				:  List[str]      )	->   Union[str, Any]:





      '''simple docstring'''






      return self.encoder.get(lowerCamelCase__				,						self.encoder.get(self.unk_token      )      )


     def   UpperCAmelCase__						(   self				:  Tuple				,						lowerCamelCase__				:  int      )	->   Union[str, Any]:





      '''simple docstring'''






      return self.decoder.get(lowerCamelCase__      )


     def   UpperCAmelCase__						(   self				:  int				,						lowerCamelCase__				:  Union[str, Any]      )	->   Any:





      '''simple docstring'''






      UpperCamelCase__       :     List[str] 							= ''''''.join(lowerCamelCase__      )
      UpperCamelCase__       :     int 							= bytearray([self.byte_decoder[c] for c in text]      ).decode('''utf-8'''				,						errors=self.errors      )
      return text


     def   UpperCAmelCase__						(   self				:  List[str]				,						lowerCamelCase__				:  str				,						lowerCamelCase__				:  Optional[str] = None      )	->   Tuple[str]:





      '''simple docstring'''






      if not os.path.isdir(lowerCamelCase__      ):
       logger.error(F"Vocabulary path ({save_directory}) should be a directory"      )
       return
      UpperCamelCase__       :     str 							= os.path.join(
          lowerCamelCase__				,						(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']      )
      UpperCamelCase__       :     str 							= os.path.join(
          lowerCamelCase__				,						(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']      )

      with open(lowerCamelCase__				,						'''w'''				,						encoding='''utf-8'''      ) as f:
       f.write(json.dumps(self.encoder				,						indent=2				,						sort_keys=lowerCamelCase__				,						ensure_ascii=lowerCamelCase__      ) + '''\n'''      )

      UpperCamelCase__       :     Dict 							= 0
      with open(lowerCamelCase__				,						'''w'''				,						encoding='''utf-8'''      ) as writer:
       writer.write('''#version: 0.2\n'''      )
       for bpe_tokens, token_index in sorted(self.bpe_ranks.items()				,						key=lambda lowerCamelCase__      : kv[1]      ):
        if index != token_index:
         logger.warning(
             F"Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."
             ''' Please check that the tokenizer is not corrupted!'''      )
         UpperCamelCase__       :     Optional[Any] 							= token_index
        writer.write(''' '''.join(lowerCamelCase__      ) + '''\n'''      )
        index += 1

      return vocab_file, merge_file


     def   UpperCAmelCase__						(   self				:  Optional[Any]				,						lowerCamelCase__				:  List[int]				,						lowerCamelCase__				:  Optional[List[int]] = None      )	->   List[int]:





      '''simple docstring'''






      if token_ids_a is None:
       return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
      UpperCamelCase__       :     Tuple 							= [self.cls_token_id]
      UpperCamelCase__       :     Optional[Any] 							= [self.sep_token_id]
      return cls + token_ids_a + sep + sep + token_ids_a + sep


     def   UpperCAmelCase__						(   self				:  Union[str, Any]				,						lowerCamelCase__				:  List[int]				,						lowerCamelCase__				:  Optional[List[int]] = None				,						lowerCamelCase__				:  bool = False      )	->   List[int]:





      '''simple docstring'''






      if already_has_special_tokens:
       return super().get_special_tokens_mask(
           token_ids_a=lowerCamelCase__				,						token_ids_a=lowerCamelCase__				,						already_has_special_tokens=lowerCamelCase__      )

      if token_ids_a is None:
       return [1] + ([0] * len(lowerCamelCase__      )) + [1]
      return [1] + ([0] * len(lowerCamelCase__      )) + [1, 1] + ([0] * len(lowerCamelCase__      )) + [1]


     def   UpperCAmelCase__						(   self				:  int				,						lowerCamelCase__				:  List[int]				,						lowerCamelCase__				:  Optional[List[int]] = None      )	->   List[int]:





      '''simple docstring'''






      UpperCamelCase__       :     str 							= [self.sep_token_id]
      UpperCamelCase__       :     str 							= [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 + sep + token_ids_a + sep      ) * [0]


     def   UpperCAmelCase__						(   self				:  Optional[int]				,						lowerCamelCase__				:  List[str]				,						lowerCamelCase__				:  Dict=False				,						**lowerCamelCase__				:  Tuple      )	->   Optional[int]:





      '''simple docstring'''






      UpperCamelCase__       :     Union[str, Any] 							= kwargs.pop('''add_prefix_space'''				,						self.add_prefix_space      )
      if (is_split_into_words or add_prefix_space) and (len(lowerCamelCase__      ) > 0 and not text[0].isspace()):
       UpperCamelCase__       :     Optional[int] 							= ''' ''' + text
      return (text, kwargs)








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						:			Optional[int]							=     KandinskyVaaPriorEmbaEmbPipeline
     __UpperCAmelCase						:			str							=     ["""prompt""", """image"""]
     __UpperCAmelCase						:			Dict							=     ["""prompt""", """image"""]
     __UpperCAmelCase						:			Optional[Any]							=     [
         """num_images_per_prompt""",
         """strength""",
         """generator""",
         """num_inference_steps""",
         """latents""",
         """negative_prompt""",
         """guidance_scale""",
         """output_type""",
         """return_dict""",
     ]
     __UpperCAmelCase						:			Dict							=     False

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






         """simple docstring"""
         return 32

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






         """simple docstring"""
         return 32

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






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

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






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

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






         """simple docstring"""
         return 1_00

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






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

     @property
     def 					__SCREAMING_SNAKE_CASE		(   self			:							int     )      ->      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(A     )

     @property
     def 					__SCREAMING_SNAKE_CASE		(   self			:							Dict     )      ->      List[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(**A     )
         # 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			:							Dict     )      ->      Optional[int]:






         """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(A     )
         return model

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






         """simple docstring"""
         a__  			=							CLIPImageProcessor(
             crop_size=2_24    ,					do_center_crop=A    ,					do_normalize=A    ,					do_resize=A    ,					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			:							List[Any]     )      ->      Tuple:






         """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=A    ,					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			:							Optional[int]    ,					A			:							Tuple    ,					A			:							Optional[Any]=0     )      ->      Any:






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

         a__  			=							floats_tensor((1, 3, 64, 64)    ,					rng=random.Random(A     )     ).to(A     )
         a__  			=							image.cpu().permute(0    ,					2    ,					3    ,					1     )[0]
         a__  			=							Image.fromarray(np.uinta(A     )     ).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			:							Dict     )      ->      Optional[int]:






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

         a__  			=							self.get_dummy_components()

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

         pipe.set_progress_bar_config(disable=A     )

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

         a__  			=							pipe(
             **self.get_dummy_inputs(A     )    ,					return_dict=A    ,					)[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			:							str     )      ->      Optional[Any]:






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

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






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






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

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






