"""simple docstring"""


from typing import List, Optional, Union

import torch
from transformers import (
    XLMRobertaTokenizer,
)

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


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

__UpperCamelCase									=			'''
    Examples:
        ```py
        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained("kandinsky-community/Kandinsky-2-1-prior")
        >>> pipe_prior.to("cuda")

        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> negative_image_emb = out.negative_image_embeds

        >>> pipe = KandinskyPipeline.from_pretrained("kandinsky-community/kandinsky-2-1")
        >>> pipe.to("cuda")

        >>> image = pipe(
        ...     prompt,
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=negative_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=100,
        ... ).images

        >>> image[0].save("cat.png")
        ```
'''







def 					UpperCAmelCase    (		UpperCAmelCase						,					UpperCAmelCase						,					UpperCAmelCase=8					)				->      Optional[Any]:
     snake_case_					       =					h // scale_factor**2
     if h % scale_factor**2 != 0:
          new_h += 1
     snake_case_					       =					w // scale_factor**2
     if w % scale_factor**2 != 0:
          new_w += 1
     return new_h * scale_factor, new_w * scale_factor




class  UpperCamelCase					(				lowerCAmelCase__   ):







       def __init__( self,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							)      ->      List[str]:
            super().__init__()

            self.register_modules(
                text_encoder=lowerCAmelCase__,							tokenizer=lowerCAmelCase__,							unet=lowerCAmelCase__,							scheduler=lowerCAmelCase__,							movq=lowerCAmelCase__,							)
            snake_case_					       =					2 ** (len(self.movq.config.block_out_channels) - 1)







       def 		a_						( self,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__)      ->      Optional[Any]:
            if latents is None:
                 snake_case_					       =					randn_tensor(lowerCAmelCase__,							generator=lowerCAmelCase__,							device=lowerCAmelCase__,							dtype=lowerCAmelCase__)
            else:
                 if latents.shape != shape:
                      raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {shape}')
                 snake_case_					       =					latents.to(lowerCAmelCase__)

            snake_case_					       =					latents * scheduler.init_noise_sigma
            return latents







       def 		a_						( self,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__=None,							)      ->      int:
            snake_case_					       =					len(lowerCAmelCase__) if isinstance(lowerCAmelCase__,							lowerCAmelCase__) else 1
            # get prompt text embeddings
            snake_case_					       =					self.tokenizer(
                lowerCAmelCase__,							padding='max_length',							truncation=lowerCAmelCase__,							max_length=77,							return_attention_mask=lowerCAmelCase__,							add_special_tokens=lowerCAmelCase__,							return_tensors='pt',							)

            snake_case_					       =					text_inputs.input_ids
            snake_case_					       =					self.tokenizer(lowerCAmelCase__,							padding='longest',							return_tensors='pt').input_ids

            if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_					       =					self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1])
                 logger.warning(
                     'The following part of your input was truncated because CLIP can only handle sequences up to'
                     f' {self.tokenizer.model_max_length} tokens: {removed_text}')

            snake_case_					       =					text_input_ids.to(lowerCAmelCase__)
            snake_case_					       =					text_inputs.attention_mask.to(lowerCAmelCase__)

            snake_case_       ,	snake_case_					       =					self.text_encoder(
                input_ids=lowerCAmelCase__,							attention_mask=lowerCAmelCase__)

            snake_case_					       =					prompt_embeds.repeat_interleave(lowerCAmelCase__,							dim=0)
            snake_case_					       =					text_encoder_hidden_states.repeat_interleave(lowerCAmelCase__,							dim=0)
            snake_case_					       =					text_mask.repeat_interleave(lowerCAmelCase__,							dim=0)

            if do_classifier_free_guidance:
                 snake_case_					       =					42
                 if negative_prompt is None:
                      snake_case_					       =					[''] * batch_size
                 elif type(lowerCAmelCase__) is not type(lowerCAmelCase__):
                      raise TypeError(
                          f'`negative_prompt` should be the same type to `prompt`, but got {type(lowerCAmelCase__)} !='
                          f' {type(lowerCAmelCase__)}.')
                 elif isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                      snake_case_					       =					[negative_prompt]
                 elif batch_size != len(lowerCAmelCase__):
                      raise ValueError(
                          f'`negative_prompt`: {negative_prompt} has batch size {len(lowerCAmelCase__)}, but `prompt`:'
                          f' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
                          ' the batch size of `prompt`.')
                 else:
                      snake_case_					       =					negative_prompt

                 snake_case_					       =					self.tokenizer(
                     lowerCAmelCase__,							padding='max_length',							max_length=77,							truncation=lowerCAmelCase__,							return_attention_mask=lowerCAmelCase__,							add_special_tokens=lowerCAmelCase__,							return_tensors='pt',							)
                 snake_case_					       =					uncond_input.input_ids.to(lowerCAmelCase__)
                 snake_case_					       =					uncond_input.attention_mask.to(lowerCAmelCase__)

                 snake_case_       ,	snake_case_					       =					self.text_encoder(
                     input_ids=lowerCAmelCase__,							attention_mask=lowerCAmelCase__)

                 # duplicate unconditional embeddings for each generation per prompt, using mps friendly method

                 snake_case_					       =					negative_prompt_embeds.shape[1]
                 snake_case_					       =					negative_prompt_embeds.repeat(1,							lowerCAmelCase__)
                 snake_case_					       =					negative_prompt_embeds.view(batch_size * num_images_per_prompt,							lowerCAmelCase__)

                 snake_case_					       =					uncond_text_encoder_hidden_states.shape[1]
                 snake_case_					       =					uncond_text_encoder_hidden_states.repeat(1,							lowerCAmelCase__,							1)
                 snake_case_					       =					uncond_text_encoder_hidden_states.view(
                     batch_size * num_images_per_prompt,							lowerCAmelCase__,							-1)
                 snake_case_					       =					uncond_text_mask.repeat_interleave(lowerCAmelCase__,							dim=0)

                 # done duplicates

                 # For classifier free guidance, we need to do two forward passes.
                 # Here we concatenate the unconditional and text embeddings into a single batch
                 # to avoid doing two forward passes
                 snake_case_					       =					torch.cat([negative_prompt_embeds, prompt_embeds])
                 snake_case_					       =					torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states])

                 snake_case_					       =					torch.cat([uncond_text_mask, text_mask])

            return prompt_embeds, text_encoder_hidden_states, text_mask







       def 		a_						( self,							lowerCAmelCase__=0)      ->      List[str]:
            if is_accelerate_available():
                 from accelerate import cpu_offload
            else:
                 raise ImportError('Please install accelerate via `pip install accelerate`')

            snake_case_					       =					torch.device(f'cuda:{gpu_id}')

            snake_case_					       =					[
                self.unet,
                self.text_encoder,
                self.movq,
            ]
            for cpu_offloaded_model in models:
                 if cpu_offloaded_model is not None:
                      cpu_offload(lowerCAmelCase__,							lowerCAmelCase__)







       def 		a_						( self,							lowerCAmelCase__=0)      ->      int:
            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.')

            snake_case_					       =					torch.device(f'cuda:{gpu_id}')

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

            snake_case_					       =					None
            for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
                 snake_case_       ,	snake_case_					       =					cpu_offload_with_hook(lowerCAmelCase__,							lowerCAmelCase__,							prev_module_hook=lowerCAmelCase__)

            if self.safety_checker is not None:
                 snake_case_       ,	snake_case_					       =					cpu_offload_with_hook(self.safety_checker,							lowerCAmelCase__,							prev_module_hook=lowerCAmelCase__)

            # We'll offload the last model manually.
            snake_case_					       =					hook







       @property
       # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
       def 		a_						( self)      ->      Tuple:
            if not hasattr(self.unet,							'_hf_hook'):
                 return self.device
            for module in self.unet.modules():
                 if (
                     hasattr(lowerCAmelCase__,							'_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(lowerCAmelCase__)
       def __call__( self,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__ = None,							lowerCAmelCase__ = 512,							lowerCAmelCase__ = 512,							lowerCAmelCase__ = 100,							lowerCAmelCase__ = 4.0,							lowerCAmelCase__ = 1,							lowerCAmelCase__ = None,							lowerCAmelCase__ = None,							lowerCAmelCase__ = "pil",							lowerCAmelCase__ = True,							)      ->      Union[str, Any]:

            if isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_					       =					1
            elif isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_					       =					len(lowerCAmelCase__)
            else:
                 raise ValueError(f'`prompt` has to be of type `str` or `list` but is {type(lowerCAmelCase__)}')

            snake_case_					       =					self._execution_device

            snake_case_					       =					batch_size * num_images_per_prompt
            snake_case_					       =					guidance_scale > 1.0

            snake_case_       ,	snake_case_       ,	snake_case_					       =					self._encode_prompt(
                lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__)

            if isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_					       =					torch.cat(lowerCAmelCase__,							dim=0)
            if isinstance(lowerCAmelCase__,							lowerCAmelCase__):
                 snake_case_					       =					torch.cat(lowerCAmelCase__,							dim=0)

            if do_classifier_free_guidance:
                 snake_case_					       =					image_embeds.repeat_interleave(lowerCAmelCase__,							dim=0)
                 snake_case_					       =					negative_image_embeds.repeat_interleave(lowerCAmelCase__,							dim=0)

            snake_case_					       =					torch.cat([negative_image_embeds, image_embeds],							dim=0).to(
                dtype=prompt_embeds.dtype,							device=lowerCAmelCase__)

            self.scheduler.set_timesteps(lowerCAmelCase__,							device=lowerCAmelCase__)
            snake_case_					       =					self.scheduler.timesteps

            snake_case_					       =					self.unet.config.in_channels

            snake_case_       ,	snake_case_					       =					get_new_h_w(lowerCAmelCase__,							lowerCAmelCase__,							self.movq_scale_factor)

            # create initial latent
            snake_case_					       =					self.prepare_latents(
                (batch_size, num_channels_latents, height, width),							text_encoder_hidden_states.dtype,							lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							self.scheduler,							)

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

                 snake_case_					       =					{'text_embeds': prompt_embeds, 'image_embeds': image_embeds}
                 snake_case_					       =					self.unet(
                     sample=lowerCAmelCase__,							timestep=lowerCAmelCase__,							encoder_hidden_states=lowerCAmelCase__,							added_cond_kwargs=lowerCAmelCase__,							return_dict=lowerCAmelCase__,							)[0]

                 if do_classifier_free_guidance:
                      snake_case_       ,	snake_case_					       =					noise_pred.split(latents.shape[1],							dim=1)
                      snake_case_       ,	snake_case_					       =					noise_pred.chunk(2)
                      snake_case_       ,	snake_case_					       =					variance_pred.chunk(2)
                      snake_case_					       =					noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                      snake_case_					       =					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"]
                 ):
                      snake_case_       ,	snake_case_					       =					noise_pred.split(latents.shape[1],							dim=1)

                 # compute the previous noisy sample x_t -> x_t-1
                 snake_case_					       =					self.scheduler.step(
                     lowerCAmelCase__,							lowerCAmelCase__,							lowerCAmelCase__,							generator=lowerCAmelCase__,							).prev_sample
            # post-processing
            snake_case_					       =					self.movq.decode(lowerCAmelCase__,							force_not_quantize=lowerCAmelCase__)['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"]:
                 snake_case_					       =					image * 0.5 + 0.5
                 snake_case_					       =					image.clamp(0,							1)
                 snake_case_					       =					image.cpu().permute(0,							2,							3,							1).float().numpy()

            if output_type == "pil":
                 snake_case_					       =					self.numpy_to_pil(lowerCAmelCase__)

            if not return_dict:
                 return (image,)

            return ImagePipelineOutput(images=lowerCAmelCase__)







import os
import shutil
from pathlib import Path
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ..utils import ONNX_EXTERNAL_WEIGHTS_NAME, ONNX_WEIGHTS_NAME, is_onnx_available, logging


if is_onnx_available():
       import onnxruntime as ort


UpperCAmelCase_									=	logging.get_logger(__name__)

UpperCAmelCase_									=	{
    'tensor(bool)': np.bool_,
    'tensor(int8)': np.inta,
    'tensor(uint8)': np.uinta,
    'tensor(int16)': np.intaa,
    'tensor(uint16)': np.uintaa,
    'tensor(int32)': np.intaa,
    'tensor(uint32)': np.uintaa,
    'tensor(int64)': np.intaa,
    'tensor(uint64)': np.uintaa,
    'tensor(float16)': np.floataa,
    'tensor(float)': np.floataa,
    'tensor(double)': np.floataa,
}





class     lowerCamelCase__:
      def __init__(				self: str    ,       UpperCamelCase_: List[Any]=None    ,       **UpperCamelCase_: str   ):
             logger.info("""`diffusers.OnnxRuntimeModel` is experimental and might change in the future."""   )
             __lowerCamelCase   =						model
             __lowerCamelCase   =						kwargs.get("""model_save_dir"""    ,       UpperCamelCase_   )
             __lowerCamelCase   =						kwargs.get("""latest_model_name"""    ,       UpperCamelCase_   )
      def __call__(				self: Dict    ,       **UpperCamelCase_: Any   ):
             __lowerCamelCase   =						{k: np.array(UpperCamelCase_   ) for k, v in kwargs.items()}
             return self.model.run(UpperCamelCase_    ,       UpperCamelCase_   )
      @staticmethod
      def 			lowerCAmelCase__						(				UpperCamelCase_: Union[str, Path]    ,       UpperCamelCase_: Tuple=None    ,       UpperCamelCase_: Tuple=None   ):
             if provider is None:
                    logger.info("""No onnxruntime provider specified, using CPUExecutionProvider"""   )
                    __lowerCamelCase   =						"""CPUExecutionProvider"""

             return ort.InferenceSession(UpperCamelCase_    ,       providers=[provider]    ,       sess_options=UpperCamelCase_   )
      def 			lowerCAmelCase__						(				self: int    ,       UpperCamelCase_: Union[str, Path]    ,       UpperCamelCase_: Optional[str] = None    ,       **UpperCamelCase_: Optional[int]   ):
             __lowerCamelCase   =						file_name if file_name is not None else ONNX_WEIGHTS_NAME

             __lowerCamelCase   =						self.model_save_dir.joinpath(self.latest_model_name   )
             __lowerCamelCase   =						Path(UpperCamelCase_   ).joinpath(UpperCamelCase_   )
             try:
                    shutil.copyfile(UpperCamelCase_    ,       UpperCamelCase_   )
             except shutil.SameFileError:
                    pass

             # copy external weights (for models >2GB)
             __lowerCamelCase   =						self.model_save_dir.joinpath(UpperCamelCase_   )
             if src_path.exists():
                    __lowerCamelCase   =						Path(UpperCamelCase_   ).joinpath(UpperCamelCase_   )
                    try:
                           shutil.copyfile(UpperCamelCase_    ,       UpperCamelCase_   )
                    except shutil.SameFileError:
                           pass
      def 			lowerCAmelCase__						(				self: List[Any]    ,       UpperCamelCase_: Union[str, os.PathLike]    ,       **UpperCamelCase_: Optional[Any]    ,       ):
             if os.path.isfile(UpperCamelCase_   ):
                    logger.error(F'Provided path ({save_directory}) should be a directory, not a file'   )
                    return

             os.makedirs(UpperCamelCase_    ,       exist_ok=UpperCamelCase_   )

             # saving model weights/files
             self._save_pretrained(UpperCamelCase_    ,       **UpperCamelCase_   )
      @classmethod
      def 			lowerCAmelCase__						(				cls: str    ,       UpperCamelCase_: Union[str, Path]    ,       UpperCamelCase_: Optional[Union[bool, str, None]] = None    ,       UpperCamelCase_: Optional[Union[str, None]] = None    ,       UpperCamelCase_: bool = False    ,       UpperCamelCase_: Optional[str] = None    ,       UpperCamelCase_: Optional[str] = None    ,       UpperCamelCase_: Optional[str] = None    ,       UpperCamelCase_: Optional["ort.SessionOptions"] = None    ,       **UpperCamelCase_: int    ,       ):
             __lowerCamelCase   =						file_name if file_name is not None else ONNX_WEIGHTS_NAME
             # load model from local directory
             if os.path.isdir(UpperCamelCase_   ):
                    __lowerCamelCase   =						OnnxRuntimeModel.load_model(
                        os.path.join(UpperCamelCase_    ,       UpperCamelCase_   )    ,       provider=UpperCamelCase_    ,       sess_options=UpperCamelCase_   )
                    __lowerCamelCase   =						Path(UpperCamelCase_   )
             # load model from hub
             else:
                    # download model
                    __lowerCamelCase   =						hf_hub_download(
                        repo_id=UpperCamelCase_    ,       filename=UpperCamelCase_    ,       use_auth_token=UpperCamelCase_    ,       revision=UpperCamelCase_    ,       cache_dir=UpperCamelCase_    ,       force_download=UpperCamelCase_    ,       )
                    __lowerCamelCase   =						Path(UpperCamelCase_   ).parent
                    __lowerCamelCase   =						Path(UpperCamelCase_   ).name
                    __lowerCamelCase   =						OnnxRuntimeModel.load_model(UpperCamelCase_    ,       provider=UpperCamelCase_    ,       sess_options=UpperCamelCase_   )
             return cls(model=UpperCamelCase_    ,       **UpperCamelCase_   )




      @classmethod
      def 			lowerCAmelCase__						(				cls: Optional[int]    ,       UpperCamelCase_: Union[str, Path]    ,       UpperCamelCase_: bool = True    ,       UpperCamelCase_: Optional[str] = None    ,       UpperCamelCase_: Optional[str] = None    ,       **UpperCamelCase_: int    ,       ):
             __lowerCamelCase   =						None
             if len(str(UpperCamelCase_   ).split("""@"""   )   ) == 2:
                    __lowerCamelCase,   __lowerCamelCase   =						model_id.split("""@"""   )

             return cls._from_pretrained(
                 model_id=UpperCamelCase_    ,       revision=UpperCamelCase_    ,       cache_dir=UpperCamelCase_    ,       force_download=UpperCamelCase_    ,       use_auth_token=UpperCamelCase_    ,       **UpperCamelCase_    ,       )

