


"""simple docstring"""







import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


lowerCAmelCase__				    =       logging.get_logger(__name__)

lowerCAmelCase__				    =       {'''vocab_file''': '''spiece.model'''}

lowerCAmelCase__				    =       {
    '''vocab_file''': {
        '''xlnet-base-cased''': '''https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model''',
        '''xlnet-large-cased''': '''https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model''',
    }
}

lowerCAmelCase__				    =       {
    '''xlnet-base-cased''': None,
    '''xlnet-large-cased''': None,
}

# Segments (not really needed)
lowerCAmelCase__				    =       0
lowerCAmelCase__				    =       1
lowerCAmelCase__				    =       2
lowerCAmelCase__				    =       3
lowerCAmelCase__				    =       4


class 				SCREAMING_SNAKE_CASE__    (		__UpperCamelCase    ):
       """simple docstring"""


       a  :    Optional[int]       =VOCAB_FILES_NAMES
       a  :    List[str]       =PRETRAINED_VOCAB_FILES_MAP
       a  :    str       =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       a  :    List[Any]       ="left"

       def __init__(						self		,					snake_case__		,					snake_case__=False		,					snake_case__=True		,					snake_case__=False		,					snake_case__="<s>"		,					snake_case__="</s>"		,					snake_case__="<unk>"		,					snake_case__="<sep>"		,					snake_case__="<pad>"		,					snake_case__="<cls>"		,					snake_case__="<mask>"		,					snake_case__=["<eop>", "<eod>"]		,					snake_case__ = None		,					**snake_case__		,					):






         """simple docstring"""





         lowerCAmelCase					:					Tuple								=							AddedToken(_lowerCAmelCase		,					lstrip=_lowerCAmelCase		,					rstrip=_lowerCAmelCase						) if isinstance(_lowerCAmelCase		,					_lowerCAmelCase						) else mask_token

         lowerCAmelCase					:					Optional[Any]								=							{} if sp_model_kwargs is None else sp_model_kwargs

         super().__init__(
             do_lower_case=_lowerCAmelCase		,					remove_space=_lowerCAmelCase		,					keep_accents=_lowerCAmelCase		,					bos_token=_lowerCAmelCase		,					eos_token=_lowerCAmelCase		,					unk_token=_lowerCAmelCase		,					sep_token=_lowerCAmelCase		,					pad_token=_lowerCAmelCase		,					cls_token=_lowerCAmelCase		,					mask_token=_lowerCAmelCase		,					additional_special_tokens=_lowerCAmelCase		,					sp_model_kwargs=self.sp_model_kwargs		,					**_lowerCAmelCase		,					)

         lowerCAmelCase					:					Union[str, Any]								=							3

         lowerCAmelCase					:					Optional[Any]								=							do_lower_case
         lowerCAmelCase					:					Optional[int]								=							remove_space
         lowerCAmelCase					:					Any								=							keep_accents
         lowerCAmelCase					:					List[Any]								=							vocab_file

         lowerCAmelCase					:					Optional[Any]								=							spm.SentencePieceProcessor(**self.sp_model_kwargs						)
         self.sp_model.Load(_lowerCAmelCase						)

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return len(self.sp_model						)

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					str								=							{self.convert_ids_to_tokens(_lowerCAmelCase						): i for i in range(self.vocab_size						)}
         vocab.update(self.added_tokens_encoder						)
         return vocab

       def __getstate__(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[Any]								=							self.__dict__.copy()
         lowerCAmelCase					:					Tuple								=							None
         return state

       def __setstate__(						self		,					snake_case__						):






         """simple docstring"""





         lowerCAmelCase					:					Dict								=							d

         # for backward compatibility
         if not hasattr(self		,					"sp_model_kwargs"						):
           lowerCAmelCase					:					List[Any]								=							{}

         lowerCAmelCase					:					Optional[int]								=							spm.SentencePieceProcessor(**self.sp_model_kwargs						)
         self.sp_model.Load(self.vocab_file						)

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         if self.remove_space:
           lowerCAmelCase					:					str								=							""" """.join(inputs.strip().split()						)
         else:
           lowerCAmelCase					:					int								=							inputs
         lowerCAmelCase					:					str								=							outputs.replace("``"		,					"\""						).replace("''"		,					"\""						)

         if not self.keep_accents:
           lowerCAmelCase					:					str								=							unicodedata.normalize("NFKD"		,					_lowerCAmelCase						)
           lowerCAmelCase					:					int								=							"""""".join([c for c in outputs if not unicodedata.combining(_lowerCAmelCase						)]						)
         if self.do_lower_case:
           lowerCAmelCase					:					List[str]								=							outputs.lower()

         return outputs

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[Any]								=							self.preprocess_text(_lowerCAmelCase						)
         lowerCAmelCase					:					Tuple								=							self.sp_model.encode(_lowerCAmelCase		,					out_type=_lowerCAmelCase						)
         lowerCAmelCase					:					List[str]								=							[]
         for piece in pieces:
           if len(_lowerCAmelCase						) > 1 and piece[-1] == str(","						) and piece[-2].isdigit():
             lowerCAmelCase					:					List[str]								=							self.sp_model.EncodeAsPieces(piece[:-1].replace(_lowerCAmelCase		,					""						)						)
             if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
               if len(cur_pieces[0]						) == 1:
                 lowerCAmelCase					:					List[Any]								=							cur_pieces[1:]
               else:
                 lowerCAmelCase					:					Optional[Any]								=							cur_pieces[0][1:]
             cur_pieces.append(piece[-1]						)
             new_pieces.extend(_lowerCAmelCase						)
           else:
             new_pieces.append(_lowerCAmelCase						)

         return new_pieces

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         return self.sp_model.PieceToId(_lowerCAmelCase						)

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         return self.sp_model.IdToPiece(_lowerCAmelCase						)

       def 	lowercase__							(						self		,					snake_case__						):






         """simple docstring"""





         lowerCAmelCase					:					int								=							"""""".join(_lowerCAmelCase						).replace(_lowerCAmelCase		,					" "						).strip()
         return out_string

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__ = False		,					snake_case__ = None		,					snake_case__ = True		,					**snake_case__		,					):






         """simple docstring"""





         lowerCAmelCase					:					Optional[int]								=							kwargs.pop("use_source_tokenizer"		,					_lowerCAmelCase						)

         lowerCAmelCase					:					str								=							self.convert_ids_to_tokens(_lowerCAmelCase		,					skip_special_tokens=_lowerCAmelCase						)

         # To avoid mixing byte-level and unicode for byte-level BPT
         # we need to build string separately for added tokens and byte-level tokens
         # cf. https://github.com/huggingface/transformers/issues/1133
         lowerCAmelCase					:					Optional[int]								=							[]
         lowerCAmelCase					:					Optional[int]								=							[]
         for token in filtered_tokens:
           if skip_special_tokens and token in self.all_special_ids:
             continue
           if token in self.added_tokens_encoder:
             if current_sub_text:
               sub_texts.append(self.convert_tokens_to_string(_lowerCAmelCase						)						)
               lowerCAmelCase					:					Dict								=							[]
             sub_texts.append(_lowerCAmelCase						)
           else:
             current_sub_text.append(_lowerCAmelCase						)
         if current_sub_text:
           sub_texts.append(self.convert_tokens_to_string(_lowerCAmelCase						)						)

         # Mimic the behavior of the Rust tokenizer:
         # By default, there are no spaces between special tokens
         lowerCAmelCase					:					Union[str, Any]								=							"""""".join(_lowerCAmelCase						)

         lowerCAmelCase					:					Tuple								=							(
             clean_up_tokenization_spaces
             if clean_up_tokenization_spaces is not None
             else self.clean_up_tokenization_spaces
         )
         if clean_up_tokenization_spaces:
           lowerCAmelCase					:					Union[str, Any]								=							self.clean_up_tokenization(_lowerCAmelCase						)
           return clean_text
         else:
           return text

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__ = None						):






         """simple docstring"""





         lowerCAmelCase					:					int								=							[self.sep_token_id]
         lowerCAmelCase					:					Dict								=							[self.cls_token_id]
         if token_ids_a is None:
           return token_ids_a + sep + cls
         return token_ids_a + sep + token_ids_a + sep + cls

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__ = None		,					snake_case__ = False						):






         """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 not None:
           return ([0] * len(_lowerCAmelCase						)) + [1] + ([0] * len(_lowerCAmelCase						)) + [1, 1]
         return ([0] * len(_lowerCAmelCase						)) + [1, 1]

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__ = None						):






         """simple docstring"""





         lowerCAmelCase					:					List[Any]								=							[self.sep_token_id]
         lowerCAmelCase					:					int								=							[2]

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







       def 	lowercase__							(						self		,					snake_case__		,					snake_case__ = None						):






         """simple docstring"""





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

         if os.path.abspath(self.vocab_file						) != os.path.abspath(_lowerCAmelCase						) and os.path.isfile(self.vocab_file						):
           copyfile(self.vocab_file		,					_lowerCAmelCase						)
         elif not os.path.isfile(self.vocab_file						):
           with open(_lowerCAmelCase		,					"wb"						) as fi:
             lowerCAmelCase					:					Optional[int]								=							self.sp_model.serialized_model_proto()
             fi.write(_lowerCAmelCase						)

         return (out_vocab_file,)







"""simple docstring"""







import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaImgaImgPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
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 				SCREAMING_SNAKE_CASE__    (		lowercase       ,   unittest.TestCase    ):
       """simple docstring"""


       a  :    Optional[Any]       =KandinskyVaaImgaImgPipeline
       a  :    Optional[int]       =["image_embeds", "negative_image_embeds", "image"]
       a  :    Optional[int]       =[
           "image_embeds",
           "negative_image_embeds",
           "image",
       ]
       a  :    str       =[
           "generator",
           "height",
           "width",
           "strength",
           "guidance_scale",
           "num_inference_steps",
           "return_dict",
           "guidance_scale",
           "num_images_per_prompt",
           "output_type",
           "return_dict",
       ]
       a  :    Dict       =False

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return 32

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return 32

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return self.time_input_dim

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return self.time_input_dim * 4

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         return 100

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         torch.manual_seed(0						)

         lowerCAmelCase					:					List[str]								=							{
             "in_channels": 4,
             # Out channels is double in channels because predicts mean and variance
             "out_channels": 8,
             "addition_embed_type": "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": "image_proj",
             "cross_attention_dim": self.cross_attention_dim,
             "attention_head_dim": 4,
             "resnet_time_scale_shift": "scale_shift",
             "class_embed_type": None,
         }

         lowerCAmelCase					:					int								=							UNetaDConditionModel(**snake_case__						)
         return model

       @property
       def 	lowercase__							(						self						):






         """simple docstring"""





         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 	lowercase__							(						self						):






         """simple docstring"""





         torch.manual_seed(0						)
         lowerCAmelCase					:					Optional[int]								=							VQModel(**self.dummy_movq_kwargs						)
         return model

       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Optional[Any]								=							self.dummy_unet
         lowerCAmelCase					:					Optional[int]								=							self.dummy_movq

         lowerCAmelCase					:					List[str]								=							{
             "num_train_timesteps": 1_000,
             "beta_schedule": "linear",
             "beta_start": 0.00085,
             "beta_end": 0.012,
             "clip_sample": False,
             "set_alpha_to_one": False,
             "steps_offset": 0,
             "prediction_type": "epsilon",
             "thresholding": False,
         }

         lowerCAmelCase					:					Tuple								=							DDIMScheduler(**snake_case__						)

         lowerCAmelCase					:					Optional[Any]								=							{
             "unet": unet,
             "scheduler": scheduler,
             "movq": movq,
         }

         return components

       def 	lowercase__							(						self		,					snake_case__		,					snake_case__=0						):






         """simple docstring"""





         lowerCAmelCase					:					Any								=							floats_tensor((1, self.text_embedder_hidden_size)		,					rng=random.Random(snake_case__						)						).to(snake_case__						)
         lowerCAmelCase					:					Any								=							floats_tensor((1, self.text_embedder_hidden_size)		,					rng=random.Random(seed + 1						)						).to(
             snake_case__						)
         # create init_image
         lowerCAmelCase					:					List[Any]								=							floats_tensor((1, 3, 64, 64)		,					rng=random.Random(snake_case__						)						).to(snake_case__						)
         lowerCAmelCase					:					List[str]								=							image.cpu().permute(0		,					2		,					3		,					1						)[0]
         lowerCAmelCase					:					List[str]								=							Image.fromarray(np.uinta(snake_case__						)						).convert("RGB"						).resize((256, 256)						)

         if str(snake_case__						).startswith("mps"						):
           lowerCAmelCase					:					Optional[int]								=							torch.manual_seed(snake_case__						)
         else:
           lowerCAmelCase					:					Optional[int]								=							torch.Generator(device=snake_case__						).manual_seed(snake_case__						)
         lowerCAmelCase					:					List[str]								=							{
             "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 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					Tuple								=							"cpu"

         lowerCAmelCase					:					Dict								=							self.get_dummy_components()

         lowerCAmelCase					:					Union[str, Any]								=							self.pipeline_class(**snake_case__						)
         lowerCAmelCase					:					Optional[int]								=							pipe.to(snake_case__						)

         pipe.set_progress_bar_config(disable=snake_case__						)

         lowerCAmelCase					:					int								=							pipe(**self.get_dummy_inputs(snake_case__						)						)
         lowerCAmelCase					:					Union[str, Any]								=							output.images

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

         lowerCAmelCase					:					Optional[Any]								=							image[0, -3:, -3:, -1]
         lowerCAmelCase					:					Optional[int]								=							image_from_tuple[0, -3:, -3:, -1]

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

         lowerCAmelCase					:					int								=							np.array(
             [0.6199778, 0.63984406, 0.46145785, 0.62944984, 0.5622215, 0.47306132, 0.47441456, 0.4607606, 0.48719263]						)
         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 				SCREAMING_SNAKE_CASE__    (		unittest.TestCase    ):
       """simple docstring"""


       def 	lowercase__							(						self						):






         """simple docstring"""





         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()







       def 	lowercase__							(						self						):






         """simple docstring"""





         lowerCAmelCase					:					List[Any]								=							load_numpy(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
             "/kandinskyv22/kandinskyv22_img2img_frog.npy"						)

         lowerCAmelCase					:					Tuple								=							load_image(
             "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"						)
         lowerCAmelCase					:					Optional[Any]								=							"A red cartoon frog, 4k"

         lowerCAmelCase					:					int								=							KandinskyVaaPriorPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-prior"		,					torch_dtype=torch.floataa						)
         pipe_prior.to(snake_case__						)

         lowerCAmelCase					:					List[Any]								=							KandinskyVaaImgaImgPipeline.from_pretrained(
             "kandinsky-community/kandinsky-2-2-decoder"		,					torch_dtype=torch.floataa						)
         lowerCAmelCase					:					Tuple								=							pipeline.to(snake_case__						)

         pipeline.set_progress_bar_config(disable=snake_case__						)

         lowerCAmelCase					:					str								=							torch.Generator(device="cpu"						).manual_seed(0						)
         lowerCAmelCase						,  lowerCAmelCase					:					Optional[Any]								=							pipe_prior(
             snake_case__		,					generator=snake_case__		,					num_inference_steps=5		,					negative_prompt=""		,					).to_tuple()

         lowerCAmelCase					:					Tuple								=							pipeline(
             image=snake_case__		,					image_embeds=snake_case__		,					negative_image_embeds=snake_case__		,					generator=snake_case__		,					num_inference_steps=100		,					height=768		,					width=768		,					strength=0.2		,					output_type="np"		,					)

         lowerCAmelCase					:					Optional[Any]								=							output.images[0]

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

         assert_mean_pixel_difference(snake_case__		,					snake_case__						)





