






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import XLMRobertaTokenizerFast

from diffusers import DDIMScheduler, KandinskyImgaImgPipeline, KandinskyPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.pipelines.kandinsky.text_encoder import MCLIPConfig, MultilingualCLIP
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()







class      UpperCAmelCase_		(   lowercase,      unittest.TestCase					):



      """simple docstring"""
      UpperCamelCase_							:   Optional[Any] 	=KandinskyImgaImgPipeline
      UpperCamelCase_							:   str 	=['prompt', 'image_embeds', 'negative_image_embeds', 'image']
      UpperCamelCase_							:   Any 	=[
          'prompt',
          'negative_prompt',
          'image_embeds',
          'negative_image_embeds',
          'image',
      ]
      UpperCamelCase_							:   str 	=[
          'generator',
          'height',
          'width',
          'strength',
          'guidance_scale',
          'negative_prompt',
          'num_inference_steps',
          'return_dict',
          'guidance_scale',
          'num_images_per_prompt',
          'output_type',
          'return_dict',
      ]
      UpperCamelCase_							:   Union[str, Any] 	=False
      @property
      def        UpperCAmelCase		(							self					)							-> Optional[Any]:
          return 32
      @property
      def        UpperCAmelCase		(							self					)							-> Optional[int]:
          return 32
      @property
      def        UpperCAmelCase		(							self					)							-> Tuple:
          return self.time_input_dim
      @property
      def        UpperCAmelCase		(							self					)							-> Tuple:
          return self.time_input_dim * 4
      @property
      def        UpperCAmelCase		(							self					)							-> Optional[int]:
          return 100
      @property
      def        UpperCAmelCase		(							self					)							-> Optional[Any]:
          UpperCamelCase			:Union[str, Any]      				=  XLMRobertaTokenizerFast.from_pretrained('''YiYiXu/tiny-random-mclip-base'''					)
          return tokenizer
      @property
      def        UpperCAmelCase		(							self					)							-> Optional[Any]:
          torch.manual_seed(0					)
          UpperCamelCase			:Any      				=  MCLIPConfig(
              numDims=self.cross_attention_dim       ,							transformerDimensions=self.text_embedder_hidden_size       ,							hidden_size=self.text_embedder_hidden_size       ,							intermediate_size=37       ,							num_attention_heads=4       ,							num_hidden_layers=5       ,							vocab_size=1005       ,							)

          UpperCamelCase			:Tuple      				=  MultilingualCLIP(__UpperCAmelCase					)
          UpperCamelCase			:int      				=  text_encoder.eval()

          return text_encoder
      @property
      def        UpperCAmelCase		(							self					)							-> Optional[Any]:
          torch.manual_seed(0					)

          UpperCamelCase			:Any      				=  {
              '''in_channels''': 4,
              # Out channels is double in channels because predicts mean and variance
              '''out_channels''': 8,
              '''addition_embed_type''': '''text_image''',
              '''down_block_types''': ('''ResnetDownsampleBlock2D''', '''SimpleCrossAttnDownBlock2D'''),
              '''up_block_types''': ('''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D'''),
              '''mid_block_type''': '''UNetMidBlock2DSimpleCrossAttn''',
              '''block_out_channels''': (self.block_out_channels_a, self.block_out_channels_a * 2),
              '''layers_per_block''': 1,
              '''encoder_hid_dim''': self.text_embedder_hidden_size,
              '''encoder_hid_dim_type''': '''text_image_proj''',
              '''cross_attention_dim''': self.cross_attention_dim,
              '''attention_head_dim''': 4,
              '''resnet_time_scale_shift''': '''scale_shift''',
              '''class_embed_type''': None,
          }

          UpperCamelCase			:Dict      				=  UNetaDConditionModel(**__UpperCAmelCase					)
          return model
      @property
      def        UpperCAmelCase		(							self					)							-> int:
          return {
              "block_out_channels": [32, 64],
              "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
              "in_channels": 3,
              "latent_channels": 4,
              "layers_per_block": 1,
              "norm_num_groups": 8,
              "norm_type": "spatial",
              "num_vq_embeddings": 12,
              "out_channels": 3,
              "up_block_types": [
                  "AttnUpDecoderBlock2D",
                  "UpDecoderBlock2D",
              ],
              "vq_embed_dim": 4,
          }
      @property
      def        UpperCAmelCase		(							self					)							-> int:
          torch.manual_seed(0					)
          UpperCamelCase			:Any      				=  VQModel(**self.dummy_movq_kwargs					)
          return model
      def        UpperCAmelCase		(							self					)							-> int:
          UpperCamelCase			:Optional[Any]      				=  self.dummy_text_encoder
          UpperCamelCase			:str      				=  self.dummy_tokenizer
          UpperCamelCase			:str      				=  self.dummy_unet
          UpperCamelCase			:List[Any]      				=  self.dummy_movq

          UpperCamelCase			:List[str]      				=  {
              '''num_train_timesteps''': 1000,
              '''beta_schedule''': '''linear''',
              '''beta_start''': 0.0_0085,
              '''beta_end''': 0.012,
              '''clip_sample''': False,
              '''set_alpha_to_one''': False,
              '''steps_offset''': 0,
              '''prediction_type''': '''epsilon''',
              '''thresholding''': False,
          }

          UpperCamelCase			:Union[str, Any]      				=  DDIMScheduler(**__UpperCAmelCase					)

          UpperCamelCase			:Optional[int]      				=  {
              '''text_encoder''': text_encoder,
              '''tokenizer''': tokenizer,
              '''unet''': unet,
              '''scheduler''': scheduler,
              '''movq''': movq,
          }

          return components
      def        UpperCAmelCase		(							self       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_=0					)							-> Any:
          UpperCamelCase			:Union[str, Any]      				=  floats_tensor((1, self.cross_attention_dim)       ,							rng=random.Random(__UpperCAmelCase					)					).to(__UpperCAmelCase					)
          UpperCamelCase			:str      				=  floats_tensor((1, self.cross_attention_dim)       ,							rng=random.Random(seed + 1					)					).to(__UpperCAmelCase					)
          # create init_image
          UpperCamelCase			:List[str]      				=  floats_tensor((1, 3, 64, 64)       ,							rng=random.Random(__UpperCAmelCase					)					).to(__UpperCAmelCase					)
          UpperCamelCase			:Optional[Any]      				=  image.cpu().permute(0       ,							2       ,							3       ,							1					)[0]
          UpperCamelCase			:List[str]      				=  Image.fromarray(np.uinta(__UpperCAmelCase					)					).convert('''RGB'''					).resize((256, 256)					)

          if str(__UpperCAmelCase					).startswith('''mps'''					):
              UpperCamelCase			:Dict      				=  torch.manual_seed(__UpperCAmelCase					)
          else:
              UpperCamelCase			:List[str]      				=  torch.Generator(device=__UpperCAmelCase					).manual_seed(__UpperCAmelCase					)
          UpperCamelCase			:str      				=  {
              '''prompt''': '''horse''',
              '''image''': init_image,
              '''image_embeds''': image_embeds,
              '''negative_image_embeds''': negative_image_embeds,
              '''generator''': generator,
              '''height''': 64,
              '''width''': 64,
              '''num_inference_steps''': 10,
              '''guidance_scale''': 7.0,
              '''strength''': 0.2,
              '''output_type''': '''np''',
          }
          return inputs




      def        UpperCAmelCase		(							self					)							-> int:
          UpperCamelCase			:Dict      				=  '''cpu'''

          UpperCamelCase			:List[str]      				=  self.get_dummy_components()

          UpperCamelCase			:str      				=  self.pipeline_class(**__UpperCAmelCase					)
          UpperCamelCase			:int      				=  pipe.to(__UpperCAmelCase					)

          pipe.set_progress_bar_config(disable=__UpperCAmelCase					)

          UpperCamelCase			:Any      				=  pipe(**self.get_dummy_inputs(__UpperCAmelCase					)					)
          UpperCamelCase			:Optional[int]      				=  output.images

          UpperCamelCase			:Union[str, Any]      				=  pipe(
              **self.get_dummy_inputs(__UpperCAmelCase					)       ,							return_dict=__UpperCAmelCase       ,							)[0]

          UpperCamelCase			:Any      				=  image[0, -3:, -3:, -1]
          UpperCamelCase			:str      				=  image_from_tuple[0, -3:, -3:, -1]

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

          UpperCamelCase			:int      				=  np.array(
              [0.6147_4943, 0.607_3539, 0.4330_8544, 0.592_8269, 0.4749_3595, 0.4675_5973, 0.461_3838, 0.4536_8797, 0.5011_9233]					)
          assert (
              np.abs(image_slice.flatten() - expected_slice					).max() < 1e-2
          ), F''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''
          assert (
              np.abs(image_from_tuple_slice.flatten() - expected_slice					).max() < 1e-2
          ), F''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''







@slow
@require_torch_gpu
class      UpperCAmelCase_		(   unittest.TestCase					):



      """simple docstring"""
      def        UpperCAmelCase		(							self					)							-> str:
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()




      def        UpperCAmelCase		(							self					)							-> Any:
          UpperCamelCase			:Tuple      				=  load_numpy(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
              '''/kandinsky/kandinsky_img2img_frog.npy'''					)

          UpperCamelCase			:Union[str, Any]      				=  load_image(
              '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main''' '''/kandinsky/cat.png'''					)
          UpperCamelCase			:List[str]      				=  '''A red cartoon frog, 4k'''

          UpperCamelCase			:List[str]      				=  KandinskyPriorPipeline.from_pretrained(
              '''kandinsky-community/kandinsky-2-1-prior'''       ,							torch_dtype=torch.floataa					)
          pipe_prior.to(__UpperCAmelCase					)

          UpperCamelCase			:Optional[Any]      				=  KandinskyImgaImgPipeline.from_pretrained(
              '''kandinsky-community/kandinsky-2-1'''       ,							torch_dtype=torch.floataa					)
          UpperCamelCase			:Optional[int]      				=  pipeline.to(__UpperCAmelCase					)

          pipeline.set_progress_bar_config(disable=__UpperCAmelCase					)

          UpperCamelCase			:Union[str, Any]      				=  torch.Generator(device='''cpu'''					).manual_seed(0					)
          UpperCamelCase				,  UpperCamelCase			:Tuple      				=  pipe_prior(
              __UpperCAmelCase       ,							generator=__UpperCAmelCase       ,							num_inference_steps=5       ,							negative_prompt=''''''       ,							).to_tuple()

          UpperCamelCase			:Union[str, Any]      				=  pipeline(
              __UpperCAmelCase       ,							image=__UpperCAmelCase       ,							image_embeds=__UpperCAmelCase       ,							negative_image_embeds=__UpperCAmelCase       ,							generator=__UpperCAmelCase       ,							num_inference_steps=100       ,							height=768       ,							width=768       ,							strength=0.2       ,							output_type='''np'''       ,							)

          UpperCamelCase			:int      				=  output.images[0]

          assert image.shape == (768, 768, 3)

          assert_mean_pixel_difference(__UpperCAmelCase       ,							__UpperCAmelCase					)













import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
     from .tokenization_pegasus import PegasusTokenizer
else:
     UpperCAmelCase__							  =						None


UpperCAmelCase__							  =						logging.get_logger(__name__)


UpperCAmelCase__							  =						"▁"

UpperCAmelCase__							  =						{"vocab_file": "spiece.model", "tokenizer_file": "tokenizer.json"}

UpperCAmelCase__							  =						{
    "vocab_file": {"google/pegasus-xsum": "https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model"},
    "tokenizer_file": {
        "google/pegasus-xsum": "https://huggingface.co/google/pegasus-xsum/resolve/main/tokenizer.json"
    },
}

UpperCAmelCase__							  =						{
    "google/pegasus-xsum": 512,
}






class 						lowercase_   (      lowercase  ):






      '''simple docstring'''




      __snake_case	    =    VOCAB_FILES_NAMES
      __snake_case	    =    PRETRAINED_VOCAB_FILES_MAP
      __snake_case	    =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      __snake_case	    =    PegasusTokenizer
      __snake_case	    =    ['''input_ids''', '''attention_mask''']







      def __init__( self :		List[Any]					,   __UpperCAmelCase :		str=None					,   __UpperCAmelCase :		List[Any]=None					,   __UpperCAmelCase :		Any="<pad>"					,   __UpperCAmelCase :		str="</s>"					,   __UpperCAmelCase :		List[str]="<unk>"					,   __UpperCAmelCase :		List[str]="<mask_2>"					,   __UpperCAmelCase :		Tuple="<mask_1>"					,   __UpperCAmelCase :		int=None					,   __UpperCAmelCase :		str=103					,   **__UpperCAmelCase :		Optional[int]					,   )       ->Dict:




       """simple docstring"""


       a						    = offset

       if additional_special_tokens is not None:
        if not isinstance(__UpperCAmelCase					,   __UpperCAmelCase    ):
         raise TypeError(
             F"""additional_special_tokens should be of type {type(__UpperCAmelCase    )}, but is"""
             F""" {type(__UpperCAmelCase    )}"""    )

        a						    = (
            ([mask_token_sent] + additional_special_tokens)
            if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
            else additional_special_tokens
        )
        # fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
        additional_special_tokens_extended += [
            F"""<unk_{i}>""" for i in range(len(__UpperCAmelCase    )					,   self.offset - 1    )
        ]

        if len(set(__UpperCAmelCase    )    ) != len(__UpperCAmelCase    ):
         raise ValueError(
             '''Please make sure that the provided additional_special_tokens do not contain an incorrectly'''
             F""" shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}."""    )
        a						    = additional_special_tokens_extended
       else:
        a						    = [mask_token_sent] if mask_token_sent is not None else []
        additional_special_tokens += [F"""<unk_{i}>""" for i in range(2					,   self.offset    )]

       super().__init__(
           __UpperCAmelCase					,   tokenizer_file=__UpperCAmelCase					,   pad_token=__UpperCAmelCase					,   eos_token=__UpperCAmelCase					,   unk_token=__UpperCAmelCase					,   mask_token=__UpperCAmelCase					,   mask_token_sent=__UpperCAmelCase					,   offset=__UpperCAmelCase					,   additional_special_tokens=__UpperCAmelCase					,   **__UpperCAmelCase					,   )
       a						    = vocab_file
       a						    = False if not self.vocab_file else True







      def  __lowerCAmelCase   ( self :		Optional[Any]					,   __UpperCAmelCase :		int    )       ->int:




       """simple docstring"""


       a						    = set(self.all_special_ids    )  # call it once instead of inside list comp
       all_special_ids.remove(self.unk_token_id    )  # <unk> is only sometimes special

       if all_special_ids != set(range(len(self.additional_special_tokens    ) + 3    )    ):
        raise ValueError(
            '''There should be 3 special tokens: mask_token, pad_token, and eos_token +'''
            F""" {len(self.additional_special_tokens    )} additional_special_tokens, but got {all_special_ids}"""    )

       return [1 if x in all_special_ids else 0 for x in seq]







      def  __lowerCAmelCase   ( self :		Dict					,   __UpperCAmelCase :		List					,   __UpperCAmelCase :		Optional[List] = None					,   __UpperCAmelCase :		bool = False    )       ->List[int]:




       """simple docstring"""


       if already_has_special_tokens:
        return self._special_token_mask(__UpperCAmelCase    )
       elif token_ids_a is None:
        return self._special_token_mask(__UpperCAmelCase    ) + [1]
       else:
        return self._special_token_mask(token_ids_a + token_ids_a    ) + [1]







      def  __lowerCAmelCase   ( self :		Union[str, Any]					,   __UpperCAmelCase :		int					,   __UpperCAmelCase :		Optional[int]=None    )       ->List[int]:




       """simple docstring"""


       if token_ids_a is None:
        return token_ids_a + [self.eos_token_id]
       # We don't expect to process pairs, but leave the pair logic for API consistency
       return token_ids_a + token_ids_a + [self.eos_token_id]







      def  __lowerCAmelCase   ( self :		Dict					,   __UpperCAmelCase :		str					,   __UpperCAmelCase :		Optional[str] = None    )       ->Tuple[str]:




       """simple docstring"""


       if not self.can_save_slow_tokenizer:
        raise ValueError(
            '''Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '''
            '''tokenizer.'''    )

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

       if os.path.abspath(self.vocab_file    ) != os.path.abspath(__UpperCAmelCase    ):
        copyfile(self.vocab_file					,   __UpperCAmelCase    )

       return (out_vocab_file,)


