




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


UpperCAmelCase_									=	logging.get_logger(__name__)

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

UpperCAmelCase_									=	{
    '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',
    }
}

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

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





class     lowerCamelCase__(					__lowerCamelCase):
      UpperCAmelCase__      :	List[Any]    		=   VOCAB_FILES_NAMES
      UpperCAmelCase__      :	Optional[Any]    		=   PRETRAINED_VOCAB_FILES_MAP
      UpperCAmelCase__      :	Tuple    		=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      UpperCAmelCase__      :	Any    		=   'left'
      def __init__(				self: str    ,       UpperCamelCase_: List[str]    ,       UpperCamelCase_: Tuple=False    ,       UpperCamelCase_: str=True    ,       UpperCamelCase_: Dict=False    ,       UpperCamelCase_: Any="<s>"    ,       UpperCamelCase_: Dict="</s>"    ,       UpperCamelCase_: Optional[Any]="<unk>"    ,       UpperCamelCase_: List[Any]="<sep>"    ,       UpperCamelCase_: Any="<pad>"    ,       UpperCamelCase_: str="<cls>"    ,       UpperCamelCase_: str="<mask>"    ,       UpperCamelCase_: Optional[int]=["<eop>", "<eod>"]    ,       UpperCamelCase_: Optional[Dict[str, Any]] = None    ,       **UpperCamelCase_: List[Any]    ,       ):
             # Mask token behave like a normal word, i.e. include the space before it
             __lowerCamelCase   =						AddedToken(UpperCamelCase_    ,       lstrip=UpperCamelCase_    ,       rstrip=UpperCamelCase_   ) if isinstance(UpperCamelCase_    ,       UpperCamelCase_   ) else mask_token

             __lowerCamelCase   =						{} if sp_model_kwargs is None else sp_model_kwargs

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

             __lowerCamelCase   =						3

             __lowerCamelCase   =						do_lower_case
             __lowerCamelCase   =						remove_space
             __lowerCamelCase   =						keep_accents
             __lowerCamelCase   =						vocab_file

             __lowerCamelCase   =						spm.SentencePieceProcessor(**self.sp_model_kwargs   )
             self.sp_model.Load(UpperCamelCase_   )
      @property
      def 			lowerCAmelCase__						(				self: Union[str, Any]   ):
             return len(self.sp_model   )
      def 			lowerCAmelCase__						(				self: str   ):
             __lowerCamelCase   =						{self.convert_ids_to_tokens(UpperCamelCase_   ): i for i in range(self.vocab_size   )}
             vocab.update(self.added_tokens_encoder   )
             return vocab
      def __getstate__(				self: Dict   ):
             __lowerCamelCase   =						self.__dict__.copy()
             __lowerCamelCase   =						None
             return state
      def __setstate__(				self: Optional[Any]    ,       UpperCamelCase_: int   ):
             __lowerCamelCase   =						d

             # for backward compatibility
             if not hasattr(self    ,       """sp_model_kwargs"""   ):
                    __lowerCamelCase   =						{}

             __lowerCamelCase   =						spm.SentencePieceProcessor(**self.sp_model_kwargs   )
             self.sp_model.Load(self.vocab_file   )
      def 			lowerCAmelCase__						(				self: List[Any]    ,       UpperCamelCase_: Optional[Any]   ):
             if self.remove_space:
                    __lowerCamelCase   =						""" """.join(inputs.strip().split()   )
             else:
                    __lowerCamelCase   =						inputs
             __lowerCamelCase   =						outputs.replace("""``"""    ,       """\""""   ).replace("""''"""    ,       """\""""   )

             if not self.keep_accents:
                    __lowerCamelCase   =						unicodedata.normalize("""NFKD"""    ,       UpperCamelCase_   )
                    __lowerCamelCase   =						"""""".join([c for c in outputs if not unicodedata.combining(UpperCamelCase_   )]   )
             if self.do_lower_case:
                    __lowerCamelCase   =						outputs.lower()

             return outputs
      def 			lowerCAmelCase__						(				self: Any    ,       UpperCamelCase_: str   ):
             __lowerCamelCase   =						self.preprocess_text(UpperCamelCase_   )
             __lowerCamelCase   =						self.sp_model.encode(UpperCamelCase_    ,       out_type=UpperCamelCase_   )
             __lowerCamelCase   =						[]
             for piece in pieces:
                    if len(UpperCamelCase_   ) > 1 and piece[-1] == str(""","""   ) and piece[-2].isdigit():
                           __lowerCamelCase   =						self.sp_model.EncodeAsPieces(piece[:-1].replace(UpperCamelCase_    ,       """"""   )   )
                           if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
                                  if len(cur_pieces[0]   ) == 1:
                                         __lowerCamelCase   =						cur_pieces[1:]
                                  else:
                                         __lowerCamelCase   =						cur_pieces[0][1:]
                           cur_pieces.append(piece[-1]   )
                           new_pieces.extend(UpperCamelCase_   )
                    else:
                           new_pieces.append(UpperCamelCase_   )

             return new_pieces
      def 			lowerCAmelCase__						(				self: Optional[Any]    ,       UpperCamelCase_: Union[str, Any]   ):
             return self.sp_model.PieceToId(UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: List[Any]    ,       UpperCamelCase_: int   ):
             return self.sp_model.IdToPiece(UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: Dict    ,       UpperCamelCase_: Dict   ):
             __lowerCamelCase   =						"""""".join(UpperCamelCase_   ).replace(UpperCamelCase_    ,       """ """   ).strip()
             return out_string
      def 			lowerCAmelCase__						(				self: Optional[Any]    ,       UpperCamelCase_: List[int]    ,       UpperCamelCase_: bool = False    ,       UpperCamelCase_: bool = None    ,       UpperCamelCase_: bool = True    ,       **UpperCamelCase_: List[Any]    ,       ):
             __lowerCamelCase   =						kwargs.pop("""use_source_tokenizer"""    ,       UpperCamelCase_   )

             __lowerCamelCase   =						self.convert_ids_to_tokens(UpperCamelCase_    ,       skip_special_tokens=UpperCamelCase_   )

             # 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   =						[]
             __lowerCamelCase   =						[]
             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(UpperCamelCase_   )   )
                                  __lowerCamelCase   =						[]
                           sub_texts.append(UpperCamelCase_   )
                    else:
                           current_sub_text.append(UpperCamelCase_   )
             if current_sub_text:
                    sub_texts.append(self.convert_tokens_to_string(UpperCamelCase_   )   )

             # Mimic the behavior of the Rust tokenizer:
             # By default, there are no spaces between special tokens
             __lowerCamelCase   =						"""""".join(UpperCamelCase_   )

             __lowerCamelCase   =						(
                 clean_up_tokenization_spaces
                 if clean_up_tokenization_spaces is not None
                 else self.clean_up_tokenization_spaces
             )
             if clean_up_tokenization_spaces:
                    __lowerCamelCase   =						self.clean_up_tokenization(UpperCamelCase_   )
                    return clean_text
             else:
                    return text
      def 			lowerCAmelCase__						(				self: Dict    ,       UpperCamelCase_: List[int]    ,       UpperCamelCase_: Optional[List[int]] = None   ):
             __lowerCamelCase   =						[self.sep_token_id]
             __lowerCamelCase   =						[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 			lowerCAmelCase__						(				self: List[Any]    ,       UpperCamelCase_: List[int]    ,       UpperCamelCase_: Optional[List[int]] = None    ,       UpperCamelCase_: bool = False   ):

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

             if token_ids_a is not None:
                    return ([0] * len(UpperCamelCase_   )) + [1] + ([0] * len(UpperCamelCase_   )) + [1, 1]
             return ([0] * len(UpperCamelCase_   )) + [1, 1]
      def 			lowerCAmelCase__						(				self: List[str]    ,       UpperCamelCase_: List[int]    ,       UpperCamelCase_: Optional[List[int]] = None   ):
             __lowerCamelCase   =						[self.sep_token_id]
             __lowerCamelCase   =						[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 			lowerCAmelCase__						(				self: str    ,       UpperCamelCase_: str    ,       UpperCamelCase_: Optional[str] = None   ):
             if not os.path.isdir(UpperCamelCase_   ):
                    logger.error(F'Vocabulary path ({save_directory}) should be a directory'   )
                    return
             __lowerCamelCase   =						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_   ) and os.path.isfile(self.vocab_file   ):
                    copyfile(self.vocab_file    ,       UpperCamelCase_   )
             elif not os.path.isfile(self.vocab_file   ):
                    with open(UpperCamelCase_    ,       """wb"""   ) as fi:
                           __lowerCamelCase   =						self.sp_model.serialized_model_proto()
                           fi.write(UpperCamelCase_   )

             return (out_vocab_file,)







from typing import List, Optional, Union

import torch

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)


UpperCAmelCase_									=	logging.get_logger(__name__)  # pylint: disable=invalid-name

UpperCAmelCase_									=	'\n    Examples:\n        ```py\n        >>> import torch\n        >>> import numpy as np\n\n        >>> from diffusers import KandinskyV22PriorPipeline, KandinskyV22ControlnetPipeline\n        >>> from transformers import pipeline\n        >>> from diffusers.utils import load_image\n\n\n        >>> def make_hint(image, depth_estimator):\n        ...     image = depth_estimator(image)["depth"]\n        ...     image = np.array(image)\n        ...     image = image[:, :, None]\n        ...     image = np.concatenate([image, image, image], axis=2)\n        ...     detected_map = torch.from_numpy(image).float() / 255.0\n        ...     hint = detected_map.permute(2, 0, 1)\n        ...     return hint\n\n\n        >>> depth_estimator = pipeline("depth-estimation")\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior = pipe_prior.to("cuda")\n\n        >>> pipe = KandinskyV22ControlnetPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-controlnet-depth", torch_dtype=torch.float16\n        ... )\n        >>> pipe = pipe.to("cuda")\n\n\n        >>> img = load_image(\n        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"\n        ...     "/kandinsky/cat.png"\n        ... ).resize((768, 768))\n\n        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to("cuda")\n\n        >>> prompt = "A robot, 4k photo"\n        >>> negative_prior_prompt = "lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature"\n\n        >>> generator = torch.Generator(device="cuda").manual_seed(43)\n\n        >>> image_emb, zero_image_emb = pipe_prior(\n        ...     prompt=prompt, negative_prompt=negative_prior_prompt, generator=generator\n        ... ).to_tuple()\n\n        >>> images = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     hint=hint,\n        ...     num_inference_steps=50,\n        ...     generator=generator,\n        ...     height=768,\n        ...     width=768,\n        ... ).images\n\n        >>> images[0].save("robot_cat.png")\n        ```\n'




def 					lowerCamelCase__       (						A__       :					Optional[int]  ,			A__       :					Dict  ,			A__       :					Optional[int]=8	):





       '''simple docstring'''



       __lowerCamelCase   =						height // scale_factor**2
       if height % scale_factor**2 != 0:
              new_height += 1
       __lowerCamelCase   =						width // scale_factor**2
       if width % scale_factor**2 != 0:
              new_width += 1
       return new_height * scale_factor, new_width * scale_factor





class     lowerCamelCase__(					__lowerCamelCase):
      def __init__(				self: List[Any]    ,       UpperCamelCase_: UNetaDConditionModel    ,       UpperCamelCase_: DDPMScheduler    ,       UpperCamelCase_: VQModel    ,       ):
             super().__init__()

             self.register_modules(
                 unet=UpperCamelCase_    ,       scheduler=UpperCamelCase_    ,       movq=UpperCamelCase_    ,       )
             __lowerCamelCase   =						2 ** (len(self.movq.config.block_out_channels   ) - 1)
      def 			lowerCAmelCase__						(				self: int    ,       UpperCamelCase_: Any    ,       UpperCamelCase_: Tuple    ,       UpperCamelCase_: int    ,       UpperCamelCase_: Dict    ,       UpperCamelCase_: Dict    ,       UpperCamelCase_: int   ):
             if latents is None:
                    __lowerCamelCase   =						randn_tensor(UpperCamelCase_    ,       generator=UpperCamelCase_    ,       device=UpperCamelCase_    ,       dtype=UpperCamelCase_   )
             else:
                    if latents.shape != shape:
                           raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {shape}'   )
                    __lowerCamelCase   =						latents.to(UpperCamelCase_   )

             __lowerCamelCase   =						latents * scheduler.init_noise_sigma
             return latents
      def 			lowerCAmelCase__						(				self: Dict    ,       UpperCamelCase_: str=0   ):
             if is_accelerate_available():
                    from accelerate import cpu_offload
             else:
                    raise ImportError("""Please install accelerate via `pip install accelerate`"""   )

             __lowerCamelCase   =						torch.device(F'cuda:{gpu_id}'   )

             __lowerCamelCase   =						[
                 self.unet,
                 self.movq,
             ]
             for cpu_offloaded_model in models:
                    if cpu_offloaded_model is not None:
                           cpu_offload(UpperCamelCase_    ,       UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: List[str]    ,       UpperCamelCase_: Optional[int]=0   ):
             if is_accelerate_available() and is_accelerate_version(""">="""    ,       """0.17.0.dev0"""   ):
                    from accelerate import cpu_offload_with_hook
             else:
                    raise ImportError("""`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher."""   )

             __lowerCamelCase   =						torch.device(F'cuda:{gpu_id}'   )

             if self.device.type != "cpu":
                    self.to("""cpu"""    ,       silence_dtype_warnings=UpperCamelCase_   )
                    torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

             __lowerCamelCase   =						None
             for cpu_offloaded_model in [self.unet, self.movq]:
                    __lowerCamelCase,   __lowerCamelCase   =						cpu_offload_with_hook(UpperCamelCase_    ,       UpperCamelCase_    ,       prev_module_hook=UpperCamelCase_   )

             # We'll offload the last model manually.
             __lowerCamelCase   =						hook
      @property
      # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
      def 			lowerCAmelCase__						(				self: int   ):
             if not hasattr(self.unet    ,       """_hf_hook"""   ):
                    return self.device
             for module in self.unet.modules():
                    if (
                        hasattr(UpperCamelCase_    ,       """_hf_hook"""   )
                        and hasattr(module._hf_hook    ,       """execution_device"""   )
                        and module._hf_hook.execution_device is not None
                    ):
                           return torch.device(module._hf_hook.execution_device   )
             return self.device




      @torch.no_grad()
      @replace_example_docstring(UpperCamelCase_   )
      def __call__(				self: Tuple    ,       UpperCamelCase_: Union[torch.FloatTensor, List[torch.FloatTensor]]    ,       UpperCamelCase_: Union[torch.FloatTensor, List[torch.FloatTensor]]    ,       UpperCamelCase_: torch.FloatTensor    ,       UpperCamelCase_: int = 5_12    ,       UpperCamelCase_: int = 5_12    ,       UpperCamelCase_: int = 1_00    ,       UpperCamelCase_: float = 4.0    ,       UpperCamelCase_: int = 1    ,       UpperCamelCase_: Optional[Union[torch.Generator, List[torch.Generator]]] = None    ,       UpperCamelCase_: Optional[torch.FloatTensor] = None    ,       UpperCamelCase_: Optional[str] = "pil"    ,       UpperCamelCase_: bool = True    ,       ):
             __lowerCamelCase   =						self._execution_device

             __lowerCamelCase   =						guidance_scale > 1.0

             if isinstance(UpperCamelCase_    ,       UpperCamelCase_   ):
                    __lowerCamelCase   =						torch.cat(UpperCamelCase_    ,       dim=0   )
             if isinstance(UpperCamelCase_    ,       UpperCamelCase_   ):
                    __lowerCamelCase   =						torch.cat(UpperCamelCase_    ,       dim=0   )
             if isinstance(UpperCamelCase_    ,       UpperCamelCase_   ):
                    __lowerCamelCase   =						torch.cat(UpperCamelCase_    ,       dim=0   )

             __lowerCamelCase   =						image_embeds.shape[0] * num_images_per_prompt

             if do_classifier_free_guidance:
                    __lowerCamelCase   =						image_embeds.repeat_interleave(UpperCamelCase_    ,       dim=0   )
                    __lowerCamelCase   =						negative_image_embeds.repeat_interleave(UpperCamelCase_    ,       dim=0   )
                    __lowerCamelCase   =						hint.repeat_interleave(UpperCamelCase_    ,       dim=0   )

             __lowerCamelCase   =						torch.cat([negative_image_embeds, image_embeds]    ,       dim=0   ).to(dtype=self.unet.dtype    ,       device=UpperCamelCase_   )
             __lowerCamelCase   =						torch.cat([hint, hint]    ,       dim=0   ).to(dtype=self.unet.dtype    ,       device=UpperCamelCase_   )
             self.scheduler.set_timesteps(UpperCamelCase_    ,       device=UpperCamelCase_   )
             __lowerCamelCase   =						self.scheduler.timesteps

             __lowerCamelCase   =						self.movq.config.latent_channels

             __lowerCamelCase,   __lowerCamelCase   =						downscale_height_and_width(UpperCamelCase_    ,       UpperCamelCase_    ,       self.movq_scale_factor   )

             # create initial latent
             __lowerCamelCase   =						self.prepare_latents(
                 (batch_size, num_channels_latents, height, width)    ,       image_embeds.dtype    ,       UpperCamelCase_    ,       UpperCamelCase_    ,       UpperCamelCase_    ,       self.scheduler    ,       )

             for i, t in enumerate(self.progress_bar(UpperCamelCase_   )   ):
                    # expand the latents if we are doing classifier free guidance
                    __lowerCamelCase   =						torch.cat([latents] * 2   ) if do_classifier_free_guidance else latents

                    __lowerCamelCase   =						{"""image_embeds""": image_embeds, """hint""": hint}
                    __lowerCamelCase   =						self.unet(
                        sample=UpperCamelCase_    ,       timestep=UpperCamelCase_    ,       encoder_hidden_states=UpperCamelCase_    ,       added_cond_kwargs=UpperCamelCase_    ,       return_dict=UpperCamelCase_    ,       )[0]

                    if do_classifier_free_guidance:
                           __lowerCamelCase,   __lowerCamelCase   =						noise_pred.split(latents.shape[1]    ,       dim=1   )
                           __lowerCamelCase,   __lowerCamelCase   =						noise_pred.chunk(2   )
                           __lowerCamelCase,   __lowerCamelCase   =						variance_pred.chunk(2   )
                           __lowerCamelCase   =						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                           __lowerCamelCase   =						torch.cat([noise_pred, variance_pred_text]    ,       dim=1   )

                    if not (
                        hasattr(self.scheduler.config    ,       """variance_type"""   )
                        and self.scheduler.config.variance_type in ["learned", "learned_range"]
                    ):
                           __lowerCamelCase,   __lowerCamelCase   =						noise_pred.split(latents.shape[1]    ,       dim=1   )

                    # compute the previous noisy sample x_t -> x_t-1
                    __lowerCamelCase   =						self.scheduler.step(
                        UpperCamelCase_    ,       UpperCamelCase_    ,       UpperCamelCase_    ,       generator=UpperCamelCase_    ,       )[0]
             # post-processing
             __lowerCamelCase   =						self.movq.decode(UpperCamelCase_    ,       force_not_quantize=UpperCamelCase_   )["""sample"""]

             if output_type not in ["pt", "np", "pil"]:
                    raise ValueError(F'Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'   )

             if output_type in ["np", "pil"]:
                    __lowerCamelCase   =						image * 0.5 + 0.5
                    __lowerCamelCase   =						image.clamp(0    ,       1   )
                    __lowerCamelCase   =						image.cpu().permute(0    ,       2    ,       3    ,       1   ).float().numpy()

             if output_type == "pil":
                    __lowerCamelCase   =						self.numpy_to_pil(UpperCamelCase_   )

             if not return_dict:
                    return (image,)

             return ImagePipelineOutput(images=UpperCamelCase_   )

