



'''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


A_							: List[Any]						 =	logging.get_logger(__name__)  # pylint: disable=invalid-name

A_							: Any						 =	"""
    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     snake_case_					(     lowerCAmelCase_       ,   lowerCAmelCase_       ,   lowerCAmelCase_=8		)->		Tuple:





       '''simple docstring'''


       _UpperCAmelCase  :						int			     = h // scale_factor**2
       if h % scale_factor**2 != 0:
              new_h += 1
       _UpperCAmelCase  :						List[Any]			     = w // scale_factor**2
       if w % scale_factor**2 != 0:
              new_w += 1
       return new_h * scale_factor, new_w * scale_factor



class 							lowercase       (  _lowerCamelCase				):


 """simple docstring"""


 def __init__(							self  ,a_  ,a_  ,a_  ,a_  ,a_  ,)		->   Optional[Any]:
        super().__init__()

        self.register_modules(
            text_encoder=a_  ,tokenizer=a_  ,unet=a_  ,scheduler=a_  ,movq=a_  ,)
        _UpperCAmelCase  :						List[str]			     = 2 ** (len(self.movq.config.block_out_channels							) - 1)







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

        _UpperCAmelCase  :						int			     = latents * scheduler.init_noise_sigma
        return latents







 def _snake_case (							self  ,a_  ,a_  ,a_  ,a_  ,a_=None  ,)		->   List[Any]:
        _UpperCAmelCase  :						Union[str, Any]			     = len(a_							) if isinstance(a_  ,a_							) else 1
        # get prompt text embeddings
        _UpperCAmelCase  :						Optional[int]			     = self.tokenizer(
            a_  ,padding="""max_length"""  ,truncation=a_  ,max_length=77  ,return_attention_mask=a_  ,add_special_tokens=a_  ,return_tensors="""pt"""  ,)

        _UpperCAmelCase  :						Union[str, Any]			     = text_inputs.input_ids
        _UpperCAmelCase  :						Union[str, Any]			     = self.tokenizer(a_  ,padding="""longest"""  ,return_tensors="""pt"""							).input_ids

        if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(a_  ,a_							):
               _UpperCAmelCase  :						Union[str, Any]			     = 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}'''							)

        _UpperCAmelCase  :						Optional[Any]			     = text_input_ids.to(a_							)
        _UpperCAmelCase  :						List[str]			     = text_inputs.attention_mask.to(a_							)

        _UpperCAmelCase     ,_UpperCAmelCase  :						Tuple			     = self.text_encoder(
            input_ids=a_  ,attention_mask=a_							)

        _UpperCAmelCase  :						Any			     = prompt_embeds.repeat_interleave(a_  ,dim=0							)
        _UpperCAmelCase  :						List[str]			     = text_encoder_hidden_states.repeat_interleave(a_  ,dim=0							)
        _UpperCAmelCase  :						Tuple			     = text_mask.repeat_interleave(a_  ,dim=0							)

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

               _UpperCAmelCase  :						List[Any]			     = self.tokenizer(
                   a_  ,padding="""max_length"""  ,max_length=77  ,truncation=a_  ,return_attention_mask=a_  ,add_special_tokens=a_  ,return_tensors="""pt"""  ,)
               _UpperCAmelCase  :						Union[str, Any]			     = uncond_input.input_ids.to(a_							)
               _UpperCAmelCase  :						Any			     = uncond_input.attention_mask.to(a_							)

               _UpperCAmelCase     ,_UpperCAmelCase  :						Union[str, Any]			     = self.text_encoder(
                   input_ids=a_  ,attention_mask=a_							)

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

               _UpperCAmelCase  :						List[Any]			     = negative_prompt_embeds.shape[1]
               _UpperCAmelCase  :						Dict			     = negative_prompt_embeds.repeat(1  ,a_							)
               _UpperCAmelCase  :						Optional[int]			     = negative_prompt_embeds.view(batch_size * num_images_per_prompt  ,a_							)

               _UpperCAmelCase  :						int			     = uncond_text_encoder_hidden_states.shape[1]
               _UpperCAmelCase  :						Dict			     = uncond_text_encoder_hidden_states.repeat(1  ,a_  ,1							)
               _UpperCAmelCase  :						Tuple			     = uncond_text_encoder_hidden_states.view(
                   batch_size * num_images_per_prompt  ,a_  ,-1							)
               _UpperCAmelCase  :						str			     = uncond_text_mask.repeat_interleave(a_  ,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
               _UpperCAmelCase  :						List[Any]			     = torch.cat([negative_prompt_embeds, prompt_embeds]							)
               _UpperCAmelCase  :						Optional[int]			     = torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]							)

               _UpperCAmelCase  :						Optional[int]			     = torch.cat([uncond_text_mask, text_mask]							)

        return prompt_embeds, text_encoder_hidden_states, text_mask







 def _snake_case (							self  ,a_=0							)		->   Optional[Any]:
        if is_accelerate_available():
               from accelerate import cpu_offload
        else:
               raise ImportError("""Please install accelerate via `pip install accelerate`"""							)

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

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







 def _snake_case (							self  ,a_=0							)		->   List[str]:
        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."""							)

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

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

        _UpperCAmelCase  :						Union[str, Any]			     = None
        for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
               _UpperCAmelCase     ,_UpperCAmelCase  :						Tuple			     = cpu_offload_with_hook(a_  ,a_  ,prev_module_hook=a_							)

        if self.safety_checker is not None:
               _UpperCAmelCase     ,_UpperCAmelCase  :						Any			     = cpu_offload_with_hook(self.safety_checker  ,a_  ,prev_module_hook=a_							)

        # We'll offload the last model manually.
        _UpperCAmelCase  :						Optional[int]			     = hook







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

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

        _UpperCAmelCase  :						List[Any]			     = self._execution_device

        _UpperCAmelCase  :						List[Any]			     = batch_size * num_images_per_prompt
        _UpperCAmelCase  :						Optional[Any]			     = guidance_scale > 1.0

        _UpperCAmelCase     ,_UpperCAmelCase     ,_UpperCAmelCase  :						Optional[int]			     = self._encode_prompt(
            a_  ,a_  ,a_  ,a_  ,a_							)

        if isinstance(a_  ,a_							):
               _UpperCAmelCase  :						Union[str, Any]			     = torch.cat(a_  ,dim=0							)
        if isinstance(a_  ,a_							):
               _UpperCAmelCase  :						Union[str, Any]			     = torch.cat(a_  ,dim=0							)

        if do_classifier_free_guidance:
               _UpperCAmelCase  :						List[str]			     = image_embeds.repeat_interleave(a_  ,dim=0							)
               _UpperCAmelCase  :						Any			     = negative_image_embeds.repeat_interleave(a_  ,dim=0							)

        _UpperCAmelCase  :						Union[str, Any]			     = torch.cat([negative_image_embeds, image_embeds]  ,dim=0							).to(
            dtype=prompt_embeds.dtype  ,device=a_							)

        self.scheduler.set_timesteps(a_  ,device=a_							)
        _UpperCAmelCase  :						int			     = self.scheduler.timesteps

        _UpperCAmelCase  :						Optional[int]			     = self.unet.config.in_channels

        _UpperCAmelCase     ,_UpperCAmelCase  :						Union[str, Any]			     = get_new_h_w(a_  ,a_  ,self.movq_scale_factor							)

        # create initial latent
        _UpperCAmelCase  :						List[Any]			     = self.prepare_latents(
            (batch_size, num_channels_latents, height, width)  ,text_encoder_hidden_states.dtype  ,a_  ,a_  ,a_  ,self.scheduler  ,)

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

               _UpperCAmelCase  :						List[Any]			     = {"""text_embeds""": prompt_embeds, """image_embeds""": image_embeds}
               _UpperCAmelCase  :						List[str]			     = self.unet(
                   sample=a_  ,timestep=a_  ,encoder_hidden_states=a_  ,added_cond_kwargs=a_  ,return_dict=a_  ,)[0]

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

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

        if output_type == "pil":
               _UpperCAmelCase  :						List[str]			     = self.numpy_to_pil(a_							)

        if not return_dict:
               return (image,)

        return ImagePipelineOutput(images=a_							)








'''simple docstring'''






class 							lowercase       :


 """simple docstring"""


 def __init__(							self							)		->   List[str]:
        _UpperCAmelCase  :						int			     = 0
        _UpperCAmelCase  :						Union[str, Any]			     = 0
        _UpperCAmelCase  :						Optional[int]			     = {}







 def _snake_case (							self  ,a_							)		->   Optional[Any]:
        if vertex not in self.adjacency:
               _UpperCAmelCase  :						int			     = {}
               self.num_vertices += 1







 def _snake_case (							self  ,a_  ,a_  ,a_							)		->   int:
        self.add_vertex(a_							)
        self.add_vertex(a_							)

        if head == tail:
               return

        _UpperCAmelCase  :						List[Any]			     = weight
        _UpperCAmelCase  :						Dict			     = weight







 def _snake_case (							self							)		->   Dict:
        _UpperCAmelCase  :						Optional[int]			     = self.get_edges()
        for edge in edges:
               _UpperCAmelCase     ,_UpperCAmelCase     ,_UpperCAmelCase  :						Dict			     = edge
               edges.remove((tail, head, weight)							)
        for i in range(len(a_							)							):
               _UpperCAmelCase  :						str			     = list(edges[i]							)

        edges.sort(key=lambda a_							: e[2]							)
        for i in range(len(a_							) - 1							):
               if edges[i][2] >= edges[i + 1][2]:
                      _UpperCAmelCase  :						Optional[Any]			     = edges[i][2] + 1
        for edge in edges:
               _UpperCAmelCase     ,_UpperCAmelCase     ,_UpperCAmelCase  :						Union[str, Any]			     = edge
               _UpperCAmelCase  :						str			     = weight
               _UpperCAmelCase  :						List[str]			     = weight







 def __str__(							self							)		->   Any:
        _UpperCAmelCase  :						List[Any]			     = """"""
        for tail in self.adjacency:
               for head in self.adjacency[tail]:
                      _UpperCAmelCase  :						List[str]			     = self.adjacency[head][tail]
                      string += f'''{head} -> {tail} == {weight}\n'''
        return string.rstrip("""\n"""							)







 def _snake_case (							self							)		->   str:
        _UpperCAmelCase  :						int			     = []
        for tail in self.adjacency:
               for head in self.adjacency[tail]:
                      output.append((tail, head, self.adjacency[head][tail])							)
        return output







 def _snake_case (							self							)		->   Optional[int]:
        return self.adjacency.keys()







 @staticmethod
 def _snake_case (							a_=None  ,a_=None							)		->   Tuple:
        _UpperCAmelCase  :						List[Any]			     = Graph()
        if vertices is None:
               _UpperCAmelCase  :						List[str]			     = []
        if edges is None:
               _UpperCAmelCase  :						Optional[Any]			     = []
        for vertex in vertices:
               g.add_vertex(a_							)
        for edge in edges:
               g.add_edge(*a_							)
        return g



 class 							lowercase       :


  """simple docstring"""


  def __init__(							self							)		->   int:
         _UpperCAmelCase  :						List[str]			     = {}
         _UpperCAmelCase  :						int			     = {}







  def __len__(							self							)		->   Tuple:
         return len(self.parent							)







  def _snake_case (							self  ,a_							)		->   str:
         if item in self.parent:
                return self.find(a_							)

         _UpperCAmelCase  :						Optional[Any]			     = item
         _UpperCAmelCase  :						List[Any]			     = 0
         return item







  def _snake_case (							self  ,a_							)		->   List[str]:
         if item not in self.parent:
                return self.make_set(a_							)
         if item != self.parent[item]:
                _UpperCAmelCase  :						List[Any]			     = self.find(self.parent[item]							)
         return self.parent[item]







  def _snake_case (							self  ,a_  ,a_							)		->   Union[str, Any]:
         _UpperCAmelCase  :						Any			     = self.find(a_							)
         _UpperCAmelCase  :						List[str]			     = self.find(a_							)

         if roota == roota:
                return roota

         if self.rank[roota] > self.rank[roota]:
                _UpperCAmelCase  :						Any			     = roota
                return roota

         if self.rank[roota] < self.rank[roota]:
                _UpperCAmelCase  :						Any			     = roota
                return roota

         if self.rank[roota] == self.rank[roota]:
                self.rank[roota] += 1
                _UpperCAmelCase  :						List[str]			     = roota
                return roota
         return None







 @staticmethod
 def _snake_case (							a_							)		->   List[Any]:
        _UpperCAmelCase  :						int			     = graph.num_vertices

        _UpperCAmelCase  :						int			     = Graph.UnionFind()
        _UpperCAmelCase  :						Optional[int]			     = []
        while num_components > 1:
               _UpperCAmelCase  :						int			     = {}
               for vertex in graph.get_vertices():
                      _UpperCAmelCase  :						Union[str, Any]			     = -1

               _UpperCAmelCase  :						Tuple			     = graph.get_edges()
               for edge in edges:
                      _UpperCAmelCase     ,_UpperCAmelCase     ,_UpperCAmelCase  :						str			     = edge
                      edges.remove((tail, head, weight)							)
               for edge in edges:
                      _UpperCAmelCase     ,_UpperCAmelCase     ,_UpperCAmelCase  :						Optional[Any]			     = edge
                      _UpperCAmelCase  :						Any			     = union_find.find(a_							)
                      _UpperCAmelCase  :						Any			     = union_find.find(a_							)
                      if seta != seta:
                             if cheap_edge[seta] == -1 or cheap_edge[seta][2] > weight:
                                    _UpperCAmelCase  :						Tuple			     = [head, tail, weight]

                             if cheap_edge[seta] == -1 or cheap_edge[seta][2] > weight:
                                    _UpperCAmelCase  :						List[str]			     = [head, tail, weight]
               for vertex in cheap_edge:
                      if cheap_edge[vertex] != -1:
                             _UpperCAmelCase     ,_UpperCAmelCase     ,_UpperCAmelCase  :						str			     = cheap_edge[vertex]
                             if union_find.find(a_							) != union_find.find(a_							):
                                    union_find.union(a_  ,a_							)
                                    mst_edges.append(cheap_edge[vertex]							)
                                    _UpperCAmelCase  :						Tuple			     = num_components - 1
        _UpperCAmelCase  :						Optional[int]			     = Graph.build(edges=a_							)
        return mst



