

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


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

lowerCAmelCase__   		=      '''
    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 						_A	(       A__	,							A__	,							A__=8 ):






 """simple docstring"""



 __lowercase				    =					h // scale_factor**2
 if h % scale_factor**2 != 0:
  new_h += 1
 __lowercase				    =					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  :	List[Any]   ,lowercase__  :	MultilingualCLIP   ,lowercase__  :	XLMRobertaTokenizer   ,lowercase__  :	UNetaDConditionModel   ,lowercase__  :	Union[DDIMScheduler, DDPMScheduler]   ,lowercase__  :	VQModel   ,):
   super().__init__()

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






  def    SCREAMING_SNAKE_CASE ( self  :	Tuple   ,lowercase__  :	Dict   ,lowercase__  :	List[Any]   ,lowercase__  :	Dict   ,lowercase__  :	Union[str, Any]   ,lowercase__  :	int   ,lowercase__  :	Dict     ):
   if latents is None:
    __lowercase				    =					randn_tensor(lowercase__   ,generator=lowercase__   ,device=lowercase__   ,dtype=lowercase__     )
   else:
    if latents.shape != shape:
     raise ValueError(F"Unexpected latents shape, got {latents.shape}, expected {shape}"     )
    __lowercase				    =					latents.to(lowercase__     )

   __lowercase				    =					latents * scheduler.init_noise_sigma
   return latents






  def    SCREAMING_SNAKE_CASE ( self  :	Optional[Any]   ,lowercase__  :	Union[str, Any]   ,lowercase__  :	Optional[Any]   ,lowercase__  :	Any   ,lowercase__  :	List[Any]   ,lowercase__  :	Tuple=None   ,):
   __lowercase				    =					len(lowercase__     ) if isinstance(lowercase__   ,lowercase__     ) else 1
   # get prompt text embeddings
   __lowercase				    =					self.tokenizer(
       lowercase__   ,padding='''max_length'''   ,truncation=lowercase__   ,max_length=7_7   ,return_attention_mask=lowercase__   ,add_special_tokens=lowercase__   ,return_tensors='''pt'''   ,)

   __lowercase				    =					text_inputs.input_ids
   __lowercase				    =					self.tokenizer(lowercase__   ,padding='''longest'''   ,return_tensors='''pt'''     ).input_ids

   if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(lowercase__   ,lowercase__     ):
    __lowercase				    =					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}"     )

   __lowercase				    =					text_input_ids.to(lowercase__     )
   __lowercase				    =					text_inputs.attention_mask.to(lowercase__     )

   __lowercase    ,      __lowercase				    =					self.text_encoder(
       input_ids=lowercase__   ,attention_mask=lowercase__     )

   __lowercase				    =					prompt_embeds.repeat_interleave(lowercase__   ,dim=0     )
   __lowercase				    =					text_encoder_hidden_states.repeat_interleave(lowercase__   ,dim=0     )
   __lowercase				    =					text_mask.repeat_interleave(lowercase__   ,dim=0     )

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

    __lowercase				    =					self.tokenizer(
        lowercase__   ,padding='''max_length'''   ,max_length=7_7   ,truncation=lowercase__   ,return_attention_mask=lowercase__   ,add_special_tokens=lowercase__   ,return_tensors='''pt'''   ,)
    __lowercase				    =					uncond_input.input_ids.to(lowercase__     )
    __lowercase				    =					uncond_input.attention_mask.to(lowercase__     )

    __lowercase    ,      __lowercase				    =					self.text_encoder(
        input_ids=lowercase__   ,attention_mask=lowercase__     )

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

    __lowercase				    =					negative_prompt_embeds.shape[1]
    __lowercase				    =					negative_prompt_embeds.repeat(1   ,lowercase__     )
    __lowercase				    =					negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,lowercase__     )

    __lowercase				    =					uncond_text_encoder_hidden_states.shape[1]
    __lowercase				    =					uncond_text_encoder_hidden_states.repeat(1   ,lowercase__   ,1     )
    __lowercase				    =					uncond_text_encoder_hidden_states.view(
        batch_size * num_images_per_prompt   ,lowercase__   ,-1     )
    __lowercase				    =					uncond_text_mask.repeat_interleave(lowercase__   ,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
    __lowercase				    =					torch.cat([negative_prompt_embeds, prompt_embeds]     )
    __lowercase				    =					torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]     )

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

   return prompt_embeds, text_encoder_hidden_states, text_mask






  def    SCREAMING_SNAKE_CASE ( self  :	List[str]   ,lowercase__  :	List[Any]=0     ):
   if is_accelerate_available():
    from accelerate import cpu_offload
   else:
    raise ImportError('''Please install accelerate via `pip install accelerate`'''     )

   __lowercase				    =					torch.device(F"cuda:{gpu_id}"     )

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






  def    SCREAMING_SNAKE_CASE ( self  :	Dict   ,lowercase__  :	Union[str, Any]=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.'''     )

   __lowercase				    =					torch.device(F"cuda:{gpu_id}"     )

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

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

   if self.safety_checker is not None:
    __lowercase    ,      __lowercase				    =					cpu_offload_with_hook(self.safety_checker   ,lowercase__   ,prev_module_hook=lowercase__     )

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






  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def    SCREAMING_SNAKE_CASE ( self  :	Dict     ):
   if not hasattr(self.unet   ,'''_hf_hook'''     ):
    return self.device
   for module in self.unet.modules():
    if (
        hasattr(lowercase__   ,'''_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(lowercase__     )
  def __call__( self  :	Union[str, Any]   ,lowercase__  :	Union[str, List[str]]   ,lowercase__  :	Union[torch.FloatTensor, List[torch.FloatTensor]]   ,lowercase__  :	Union[torch.FloatTensor, List[torch.FloatTensor]]   ,lowercase__  :	Optional[Union[str, List[str]]] = None   ,lowercase__  :	int = 5_1_2   ,lowercase__  :	int = 5_1_2   ,lowercase__  :	int = 1_0_0   ,lowercase__  :	float = 4.0   ,lowercase__  :	int = 1   ,lowercase__  :	Optional[Union[torch.Generator, List[torch.Generator]]] = None   ,lowercase__  :	Optional[torch.FloatTensor] = None   ,lowercase__  :	Optional[str] = "pil"   ,lowercase__  :	bool = True   ,):

   if isinstance(lowercase__   ,lowercase__     ):
    __lowercase				    =					1
   elif isinstance(lowercase__   ,lowercase__     ):
    __lowercase				    =					len(lowercase__     )
   else:
    raise ValueError(F"`prompt` has to be of type `str` or `list` but is {type(lowercase__     )}"     )

   __lowercase				    =					self._execution_device

   __lowercase				    =					batch_size * num_images_per_prompt
   __lowercase				    =					guidance_scale > 1.0

   __lowercase    ,      __lowercase    ,      __lowercase				    =					self._encode_prompt(
       lowercase__   ,lowercase__   ,lowercase__   ,lowercase__   ,lowercase__     )

   if isinstance(lowercase__   ,lowercase__     ):
    __lowercase				    =					torch.cat(lowercase__   ,dim=0     )
   if isinstance(lowercase__   ,lowercase__     ):
    __lowercase				    =					torch.cat(lowercase__   ,dim=0     )

   if do_classifier_free_guidance:
    __lowercase				    =					image_embeds.repeat_interleave(lowercase__   ,dim=0     )
    __lowercase				    =					negative_image_embeds.repeat_interleave(lowercase__   ,dim=0     )

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

   self.scheduler.set_timesteps(lowercase__   ,device=lowercase__     )
   __lowercase				    =					self.scheduler.timesteps

   __lowercase				    =					self.unet.config.in_channels

   __lowercase    ,      __lowercase				    =					get_new_h_w(lowercase__   ,lowercase__   ,self.movq_scale_factor     )

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

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

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

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

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

   if output_type == "pil":
    __lowercase				    =					self.numpy_to_pil(lowercase__     )

   if not return_dict:
    return (image,)

   return ImagePipelineOutput(images=lowercase__     )









from __future__ import annotations

from typing import Generic, TypeVar

__a   :							str						= TypeVar("T")


class    __lowercase   (						Generic[T]	):







     '''simple docstring'''




     def __init__(					self							:      Any			,						UpperCamelCase_							:      T					):


         """simple docstring"""



         __A       =						data
         __A       =						self
         __A       =						0


class    __lowercase   (						Generic[T]	):







     '''simple docstring'''




     def __init__(					self							:      Optional[Any]					):


         """simple docstring"""



         __A       =						{}




     def 		lowerCAmelCase_  (					self							:      Union[str, Any]			,						UpperCamelCase_							:      T					):


         """simple docstring"""



         __A       =						DisjointSetTreeNode(UpperCamelCase_					)




     def 		lowerCAmelCase_  (					self							:      List[Any]			,						UpperCamelCase_							:      T					):


         """simple docstring"""



         __A       =						self.map[data]
         if elem_ref != elem_ref.parent:
             __A       =						self.find_set(elem_ref.parent.data					)
         return elem_ref.parent




     def 		lowerCAmelCase_  (					self							:      int			,						UpperCamelCase_							:      DisjointSetTreeNode[T]			,						UpperCamelCase_							:      DisjointSetTreeNode[T]					):


         """simple docstring"""



         if nodea.rank > nodea.rank:
             __A       =						nodea
         else:
             __A       =						nodea
             if nodea.rank == nodea.rank:
                 nodea.rank += 1





     def 		lowerCAmelCase_  (					self							:      Tuple			,						UpperCamelCase_							:      T			,						UpperCamelCase_							:      T					):


         """simple docstring"""



         self.link(self.find_set(UpperCamelCase_					)			,						self.find_set(UpperCamelCase_					)					)


class    __lowercase   (						Generic[T]	):







     '''simple docstring'''




     def __init__(					self							:      Optional[int]					):


         """simple docstring"""



         __A       =						{}




     def 		lowerCAmelCase_  (					self							:      int			,						UpperCamelCase_							:      T					):


         """simple docstring"""



         if node not in self.connections:
             __A       =						{}




     def 		lowerCAmelCase_  (					self							:      Dict			,						UpperCamelCase_							:      T			,						UpperCamelCase_							:      T			,						UpperCamelCase_							:      int					):


         """simple docstring"""



         self.add_node(UpperCamelCase_					)
         self.add_node(UpperCamelCase_					)
         __A       =						weight
         __A       =						weight





     def 		lowerCAmelCase_  (					self							:      Dict					):


         """simple docstring"""



         __A       =						[]
         __A       =						set()
         for start in self.connections:
             for end in self.connections[start]:
                 if (start, end) not in seen:
                     seen.add((end, start)					)
                     edges.append((start, end, self.connections[start][end])					)
         edges.sort(key=lambda UpperCamelCase_					: x[2]					)

         # creating the disjoint set
         __A       =						DisjointSetTree[T]()
         for node in self.connections:
             disjoint_set.make_set(UpperCamelCase_					)

         # MST generation
         __A       =						0
         __A       =						0
         __A       =						GraphUndirectedWeighted[T]()
         while num_edges < len(self.connections					) - 1:
             __A     ,						__A     ,						__A       =						edges[index]
             index += 1
             __A       =						disjoint_set.find_set(UpperCamelCase_					)
             __A       =						disjoint_set.find_set(UpperCamelCase_					)
             if parent_u != parent_v:
                 num_edges += 1
                 graph.add_edge(UpperCamelCase_			,						UpperCamelCase_			,						UpperCamelCase_					)
                 disjoint_set.union(UpperCamelCase_			,						UpperCamelCase_					)
         return graph




