






'''simple docstring'''
import argparse
from collections import defaultdict

import yaml


UpperCamelCase__ =			'''docs/source/en/_toctree.yml'''




def 		__SCREAMING_SNAKE_CASE					(  _UpperCamelCase							):


   """simple docstring"""




   lowercase_	:    List[str]           =     defaultdict(lowerCAmelCase__							)
   lowercase_	:    Tuple           =     []
   lowercase_	:    Optional[int]           =     []
   for doc in doc_list:
      if "local" in doc:
         counts[doc["local"]] += 1

      if doc["title"].lower() == "overview":
         overview_doc.append({"local": doc["local"], "title": doc["title"]}							)
      else:
         new_doc_list.append(lowerCAmelCase__							)

   lowercase_	:    Optional[Any]           =     new_doc_list
   lowercase_	:    Optional[Any]           =     [key for key, value in counts.items() if value > 1]

   lowercase_	:    Tuple           =     []
   for duplicate_key in duplicates:
      lowercase_	:    Any           =     list({doc["title"] for doc in doc_list if doc["local"] == duplicate_key}							)
      if len(lowerCAmelCase__							) > 1:
         raise ValueError(
             F"""{duplicate_key} is present several times in the documentation table of content at """
             "`docs/source/en/_toctree.yml` with different *Title* values. Choose one of those and remove the "
             "others."							)
      # Only add this once
      new_doc.append({"local": duplicate_key, "title": titles[0]}							)

   # Add none duplicate-keys
   new_doc.extend([doc for doc in doc_list if "local" not in counts or counts[doc["local"]] == 1]							)
   lowercase_	:    Tuple           =     sorted(lowerCAmelCase__						,		key=lambda _UpperCamelCase							: s["title"].lower()							)

   # "overview" gets special treatment and is always first
   if len(lowerCAmelCase__							) > 1:
      raise ValueError("{doc_list} has two 'overview' docs which is not allowed."							)

   overview_doc.extend(lowerCAmelCase__							)

   # Sort
   return overview_doc




def 		__SCREAMING_SNAKE_CASE					(  _UpperCamelCase=False							):


   """simple docstring"""




   with open(lowerCAmelCase__						,		encoding="utf-8"							) as f:
      lowercase_	:    Union[str, Any]           =     yaml.safe_load(f.read()							)

   # Get to the API doc
   lowercase_	:    Optional[int]           =     0
   while content[api_idx]["title"] != "API":
      api_idx += 1
   lowercase_	:    List[str]           =     content[api_idx]["sections"]

   # Then to the model doc
   lowercase_	:    int           =     0
   while api_doc[scheduler_idx]["title"] != "Schedulers":
      scheduler_idx += 1

   lowercase_	:    str           =     api_doc[scheduler_idx]["sections"]
   lowercase_	:    Any           =     clean_doc_toc(lowerCAmelCase__							)

   lowercase_	:    List[str]           =     False
   if new_scheduler_doc != scheduler_doc:
      lowercase_	:    Optional[int]           =     True
      if overwrite:
         lowercase_	:    List[Any]           =     new_scheduler_doc

   if diff:
      if overwrite:
         lowercase_	:    List[Any]           =     api_doc
         with open(lowerCAmelCase__						,		"w"						,		encoding="utf-8"							) as f:
            f.write(yaml.dump(lowerCAmelCase__						,		allow_unicode=lowerCAmelCase__							)							)
      else:
         raise ValueError(
             "The model doc part of the table of content is not properly sorted, run `make style` to fix this."							)




def 		__SCREAMING_SNAKE_CASE					(  _UpperCamelCase=False							):


   """simple docstring"""




   with open(lowerCAmelCase__						,		encoding="utf-8"							) as f:
      lowercase_	:    Union[str, Any]           =     yaml.safe_load(f.read()							)

   # Get to the API doc
   lowercase_	:    int           =     0
   while content[api_idx]["title"] != "API":
      api_idx += 1
   lowercase_	:    str           =     content[api_idx]["sections"]

   # Then to the model doc
   lowercase_	:    Union[str, Any]           =     0
   while api_doc[pipeline_idx]["title"] != "Pipelines":
      pipeline_idx += 1

   lowercase_	:    Tuple           =     False
   lowercase_	:    int           =     api_doc[pipeline_idx]["sections"]
   lowercase_	:    Any           =     []

   # sort sub pipeline docs
   for pipeline_doc in pipeline_docs:
      if "section" in pipeline_doc:
         lowercase_	:    Dict           =     pipeline_doc["section"]
         lowercase_	:    Optional[Any]           =     clean_doc_toc(lowerCAmelCase__							)
         if overwrite:
            lowercase_	:    Optional[Any]           =     new_sub_pipeline_doc
      new_pipeline_docs.append(lowerCAmelCase__							)

   # sort overall pipeline doc
   lowercase_	:    Dict           =     clean_doc_toc(lowerCAmelCase__							)

   if new_pipeline_docs != pipeline_docs:
      lowercase_	:    Optional[int]           =     True
      if overwrite:
         lowercase_	:    Tuple           =     new_pipeline_docs

   if diff:
      if overwrite:
         lowercase_	:    Any           =     api_doc
         with open(lowerCAmelCase__						,		"w"						,		encoding="utf-8"							) as f:
            f.write(yaml.dump(lowerCAmelCase__						,		allow_unicode=lowerCAmelCase__							)							)
      else:
         raise ValueError(
             "The model doc part of the table of content is not properly sorted, run `make style` to fix this."							)


if __name__ == "__main__":
       UpperCamelCase__ =			argparse.ArgumentParser()
       parser.add_argument('--fix_and_overwrite', action='store_true', help='Whether to fix inconsistencies.')
       UpperCamelCase__ =			parser.parse_args()

       check_scheduler_doc(args.fix_and_overwrite)
       check_pipeline_doc(args.fix_and_overwrite)












'''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__ =			'\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained("kandinsky-community/Kandinsky-2-1-prior")\n        >>> pipe_prior.to("cuda")\n\n        >>> prompt = "red cat, 4k photo"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> negative_image_emb = out.negative_image_embeds\n\n        >>> pipe = KandinskyPipeline.from_pretrained("kandinsky-community/kandinsky-2-1")\n        >>> pipe.to("cuda")\n\n        >>> image = pipe(\n        ...     prompt,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=negative_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ... ).images\n\n        >>> image[0].save("cat.png")\n        ```\n'




def 		__SCREAMING_SNAKE_CASE					(  _UpperCamelCase						,		_UpperCamelCase						,		_UpperCamelCase=8							):


   """simple docstring"""




   lowercase_	:    int           =     h // scale_factor**2
   if h % scale_factor**2 != 0:
      new_h += 1
   lowercase_	:    str           =     w // scale_factor**2
   if w % scale_factor**2 != 0:
      new_w += 1
   return new_h * scale_factor, new_w * scale_factor





class        _UpperCAmelCase						(		snake_case			):




    def __init__(			self  :       int  ,     a  :       MultilingualCLIP  ,     a  :       XLMRobertaTokenizer  ,     a  :       UNetaDConditionModel  ,     a  :       Union[DDIMScheduler, DDPMScheduler]  ,     a  :       VQModel  ,     ):






       '''simple docstring'''






       super().__init__()

       self.register_modules(
           text_encoder=a  ,     tokenizer=a  ,     unet=a  ,     scheduler=a  ,     movq=a  ,     )
       lowercase_	:    Dict           =     2 ** (len(self.movq.config.block_out_channels			) - 1)




    def 				lowerCAmelCase__				(			self  :       List[Any]  ,     a  :       Tuple  ,     a  :       List[str]  ,     a  :       Optional[Any]  ,     a  :       str  ,     a  :       Tuple  ,     a  :       List[str]			):






       '''simple docstring'''






       if latents is None:
          lowercase_	:    List[str]           =     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}"""			)
          lowercase_	:    Optional[int]           =     latents.to(a			)

       lowercase_	:    str           =     latents * scheduler.init_noise_sigma
       return latents




    def 				lowerCAmelCase__				(			self  :       Optional[Any]  ,     a  :       List[str]  ,     a  :       List[Any]  ,     a  :       Union[str, Any]  ,     a  :       str  ,     a  :       Tuple=None  ,     ):






       '''simple docstring'''






       lowercase_	:    Tuple           =     len(a			) if isinstance(a  ,     a			) else 1
       # get prompt text embeddings
       lowercase_	:    Any           =     self.tokenizer(
           a  ,     padding="max_length"  ,     truncation=a  ,     max_length=7_7  ,     return_attention_mask=a  ,     add_special_tokens=a  ,     return_tensors="pt"  ,     )

       lowercase_	:    Union[str, Any]           =     text_inputs.input_ids
       lowercase_	:    Tuple           =     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			):
          lowercase_	:    Optional[int]           =     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_	:    List[str]           =     text_input_ids.to(a			)
       lowercase_	:    int           =     text_inputs.attention_mask.to(a			)

       lowercase_    ,		lowercase_	:    Optional[int]           =     self.text_encoder(
           input_ids=a  ,     attention_mask=a			)

       lowercase_	:    str           =     prompt_embeds.repeat_interleave(a  ,     dim=0			)
       lowercase_	:    int           =     text_encoder_hidden_states.repeat_interleave(a  ,     dim=0			)
       lowercase_	:    int           =     text_mask.repeat_interleave(a  ,     dim=0			)

       if do_classifier_free_guidance:
          lowercase_	:    List[str]
          if negative_prompt is None:
             lowercase_	:    int           =     [""] * 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			):
             lowercase_	:    Tuple           =     [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:
             lowercase_	:    Dict           =     negative_prompt

          lowercase_	:    str           =     self.tokenizer(
              a  ,     padding="max_length"  ,     max_length=7_7  ,     truncation=a  ,     return_attention_mask=a  ,     add_special_tokens=a  ,     return_tensors="pt"  ,     )
          lowercase_	:    List[Any]           =     uncond_input.input_ids.to(a			)
          lowercase_	:    Optional[int]           =     uncond_input.attention_mask.to(a			)

          lowercase_    ,		lowercase_	:    int           =     self.text_encoder(
              input_ids=a  ,     attention_mask=a			)

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

          lowercase_	:    List[str]           =     negative_prompt_embeds.shape[1]
          lowercase_	:    Dict           =     negative_prompt_embeds.repeat(1  ,     a			)
          lowercase_	:    Optional[Any]           =     negative_prompt_embeds.view(batch_size * num_images_per_prompt  ,     a			)

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

          lowercase_	:    Any           =     torch.cat([uncond_text_mask, text_mask]			)

       return prompt_embeds, text_encoder_hidden_states, text_mask




    def 				lowerCAmelCase__				(			self  :       Tuple  ,     a  :       Optional[Any]=0			):






       '''simple docstring'''






       if is_accelerate_available():
          from accelerate import cpu_offload
       else:
          raise ImportError("Please install accelerate via `pip install accelerate`"			)

       lowercase_	:    List[str]           =     torch.device(f"""cuda:{gpu_id}"""			)

       lowercase_	:    str           =     [
           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 				lowerCAmelCase__				(			self  :       Union[str, Any]  ,     a  :       List[str]=0			):






       '''simple docstring'''






       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_	:    List[str]           =     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)

       lowercase_	:    List[str]           =     None
       for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
          lowercase_    ,		lowercase_	:    Optional[int]           =     cpu_offload_with_hook(a  ,     a  ,     prev_module_hook=a			)

       if self.safety_checker is not None:
          lowercase_    ,		lowercase_	:    Optional[int]           =     cpu_offload_with_hook(self.safety_checker  ,     a  ,     prev_module_hook=a			)

       # We'll offload the last model manually.
       lowercase_	:    Dict           =     hook




    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def 				lowerCAmelCase__				(			self  :       Tuple			):






       '''simple docstring'''






       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  :       Tuple  ,     a  :       Union[str, List[str]]  ,     a  :       Union[torch.FloatTensor, List[torch.FloatTensor]]  ,     a  :       Union[torch.FloatTensor, List[torch.FloatTensor]]  ,     a  :       Optional[Union[str, List[str]]] = None  ,     a  :       int = 5_1_2  ,     a  :       int = 5_1_2  ,     a  :       int = 1_0_0  ,     a  :       float = 4.0  ,     a  :       int = 1  ,     a  :       Optional[Union[torch.Generator, List[torch.Generator]]] = None  ,     a  :       Optional[torch.FloatTensor] = None  ,     a  :       Optional[str] = "pil"  ,     a  :       bool = True  ,     ):






       '''simple docstring'''






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

       lowercase_	:    Tuple           =     self._execution_device

       lowercase_	:    Dict           =     batch_size * num_images_per_prompt
       lowercase_	:    Dict           =     guidance_scale > 1.0

       lowercase_    ,		lowercase_    ,		lowercase_	:    List[str]           =     self._encode_prompt(
           a  ,     a  ,     a  ,     a  ,     a			)

       if isinstance(a  ,     a			):
          lowercase_	:    Optional[int]           =     torch.cat(a  ,     dim=0			)
       if isinstance(a  ,     a			):
          lowercase_	:    int           =     torch.cat(a  ,     dim=0			)

       if do_classifier_free_guidance:
          lowercase_	:    Optional[int]           =     image_embeds.repeat_interleave(a  ,     dim=0			)
          lowercase_	:    int           =     negative_image_embeds.repeat_interleave(a  ,     dim=0			)

       lowercase_	:    str           =     torch.cat([negative_image_embeds, image_embeds]  ,     dim=0			).to(
           dtype=prompt_embeds.dtype  ,     device=a			)

       self.scheduler.set_timesteps(a  ,     device=a			)
       lowercase_	:    List[str]           =     self.scheduler.timesteps

       lowercase_	:    str           =     self.unet.config.in_channels

       lowercase_    ,		lowercase_	:    int           =     get_new_h_w(a  ,     a  ,     self.movq_scale_factor			)

       # create initial latent
       lowercase_	:    str           =     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
          lowercase_	:    Optional[int]           =     torch.cat([latents] * 2			) if do_classifier_free_guidance else latents

          lowercase_	:    Optional[int]           =     {"text_embeds": prompt_embeds, "image_embeds": image_embeds}
          lowercase_	:    Optional[int]           =     self.unet(
              sample=a  ,     timestep=a  ,     encoder_hidden_states=a  ,     added_cond_kwargs=a  ,     return_dict=a  ,     )[0]

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

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

       if output_type == "pil":
          lowercase_	:    List[str]           =     self.numpy_to_pil(a			)

       if not return_dict:
          return (image,)

       return ImagePipelineOutput(images=a			)




