




'''simple docstring'''







from __future__ import annotations

from typing import Any



def        lowerCamelCase__				(      __lowercase				):
     if not postfix_notation:
          return 0

     snake_case			:      Optional[Any]        =			{"""+""", """-""", """*""", """/"""}
     snake_case			:      list[Any]        =			[]

     for token in postfix_notation:
          if token in operations:
               snake_case ,		snake_case			:      str        =			stack.pop(), stack.pop()
               if token == "+":
                    stack.append(a + b				)
               elif token == "-":
                    stack.append(a - b				)
               elif token == "*":
                    stack.append(a * b				)
               else:
                    if a * b < 0 and a % b != 0:
                         stack.append(a // b + 1				)
                    else:
                         stack.append(a // b				)
          else:
               stack.append(int(__lowercase				)				)

     return stack.pop()


if __name__ == "__main__":
      import doctest

      doctest.testmod()









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


lowercase      :   Union[str, Any]      =       logging.get_logger(__name__)  # pylint: disable=invalid-name

lowercase      :   List[str]      =       """
    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        lowerCamelCase__				(      __lowercase     ,   __lowercase     ,   __lowercase=8				):
     snake_case			:      int        =			h // scale_factor**2
     if h % scale_factor**2 != 0:
          new_h += 1
     snake_case			:      Tuple        =			w // scale_factor**2
     if w % scale_factor**2 != 0:
          new_w += 1
     return new_h * scale_factor, new_w * scale_factor







class 					_a (a__	):





     '''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							,)
          snake_case			:      int        =			2 ** (len(self.movq.config.block_out_channels							) - 1)







     def 			snake_case_      (				self							,__a							,__a							,__a							,__a							,__a							,__a							)      ->				Dict:
          if latents is None:
               snake_case			:      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}'''							)
               snake_case			:      Any        =			latents.to(__a							)

          snake_case			:      Tuple        =			latents * scheduler.init_noise_sigma
          return latents







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

          snake_case			:      str        =			text_inputs.input_ids
          snake_case			:      List[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							):
               snake_case			:      str        =			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			:      str        =			text_input_ids.to(__a							)
          snake_case			:      Tuple        =			text_inputs.attention_mask.to(__a							)

          snake_case ,		snake_case			:      Optional[Any]        =			self.text_encoder(
              input_ids=__a							,attention_mask=__a							)

          snake_case			:      Optional[Any]        =			prompt_embeds.repeat_interleave(__a							,dim=0							)
          snake_case			:      str        =			text_encoder_hidden_states.repeat_interleave(__a							,dim=0							)
          snake_case			:      str        =			text_mask.repeat_interleave(__a							,dim=0							)

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

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

               snake_case ,		snake_case			:      str        =			self.text_encoder(
                   input_ids=__a							,attention_mask=__a							)

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

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

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

               snake_case			:      List[str]        =			torch.cat([uncond_text_mask, text_mask]							)

          return prompt_embeds, text_encoder_hidden_states, text_mask







     def 			snake_case_      (				self							,__a=0							)      ->				Tuple:
          if is_accelerate_available():
               from accelerate import cpu_offload
          else:
               raise ImportError("""Please install accelerate via `pip install accelerate`"""							)

          snake_case			:      List[Any]        =			torch.device(F'''cuda:{gpu_id}'''							)

          snake_case			:      Tuple        =			[
              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[Any]:
          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			:      Union[str, Any]        =			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)

          snake_case			:      Any        =			None
          for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
               snake_case ,		snake_case			:      Optional[int]        =			cpu_offload_with_hook(__a							,__a							,prev_module_hook=__a							)

          if self.safety_checker is not None:
               snake_case ,		snake_case			:      List[str]        =			cpu_offload_with_hook(self.safety_checker							,__a							,prev_module_hook=__a							)

          # We'll offload the last model manually.
          snake_case			:      Any        =			hook







     @property
     # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
     def 			snake_case_      (				self							)      ->				List[Any]:
          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							,)      ->				List[str]:

          if isinstance(__a							,__a							):
               snake_case			:      Optional[Any]        =			1
          elif isinstance(__a							,__a							):
               snake_case			:      List[Any]        =			len(__a							)
          else:
               raise ValueError(F'''`prompt` has to be of type `str` or `list` but is {type(__a							)}'''							)

          snake_case			:      Optional[Any]        =			self._execution_device

          snake_case			:      Optional[Any]        =			batch_size * num_images_per_prompt
          snake_case			:      str        =			guidance_scale > 1.0

          snake_case ,		snake_case ,		snake_case			:      List[str]        =			self._encode_prompt(
              __a							,__a							,__a							,__a							,__a							)

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

          if do_classifier_free_guidance:
               snake_case			:      Dict        =			image_embeds.repeat_interleave(__a							,dim=0							)
               snake_case			:      Dict        =			negative_image_embeds.repeat_interleave(__a							,dim=0							)

          snake_case			:      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							)
          snake_case			:      Union[str, Any]        =			self.scheduler.timesteps

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

          snake_case ,		snake_case			:      str        =			get_new_h_w(__a							,__a							,self.movq_scale_factor							)

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

               snake_case			:      Any        =			{"""text_embeds""": prompt_embeds, """image_embeds""": image_embeds}
               snake_case			:      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:
                    snake_case ,		snake_case			:      Any        =			noise_pred.split(latents.shape[1]							,dim=1							)
                    snake_case ,		snake_case			:      Any        =			noise_pred.chunk(2							)
                    snake_case ,		snake_case			:      Union[str, Any]        =			variance_pred.chunk(2							)
                    snake_case			:      List[Any]        =			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
                    snake_case			:      Any        =			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			:      Tuple        =			noise_pred.split(latents.shape[1]							,dim=1							)

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

          if output_type == "pil":
               snake_case			:      Dict        =			self.numpy_to_pil(__a							)

          if not return_dict:
               return (image,)

          return ImagePipelineOutput(images=__a							)



