
def   lowerCamelCase__							(					a      )      ->   int:

   if not isinstance(a		,		a      ):
      raise TypeError('''only integers accepted as input'''      )
   else:
      _A:  int					=       str(abs(a      )      )
      _A:  Optional[int]					=       [list(a      ) for char in range(len(a      )      )]
      for index in range(len(a      )      ):
         num_transpositions[index].pop(a      )
      return max(
          int(''''''.join(list(a      )      )      ) for transposition in num_transpositions      )


if __name__ == "__main__":
   __import__('doctest').testmod()





from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

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


UpperCAmelCase__		:     Optional[int]     			=      logging.get_logger(__name__)  # pylint: disable=invalid-name

UpperCAmelCase__		:     Dict     			=      '\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Img2ImgPipeline, KandinskyV22PriorPipeline\n        >>> from diffusers.utils import load_image\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior.to("cuda")\n\n        >>> prompt = "A red cartoon frog, 4k"\n        >>> image_emb, zero_image_emb = pipe_prior(prompt, return_dict=False)\n\n        >>> pipe = KandinskyV22Img2ImgPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-decoder", torch_dtype=torch.float16\n        ... )\n        >>> pipe.to("cuda")\n\n        >>> init_image = load_image(\n        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"\n        ...     "/kandinsky/frog.png"\n        ... )\n\n        >>> image = pipe(\n        ...     image=init_image,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ...     strength=0.2,\n        ... ).images\n\n        >>> image[0].save("red_frog.png")\n        ```\n'






def   lowerCamelCase__							(					a		,		a		,		a=8      )      ->   List[Any]:
   _A:  int					=       height // scale_factor**2
   if height % scale_factor**2 != 0:
      new_height += 1
   _A:  str					=       width // scale_factor**2
   if width % scale_factor**2 != 0:
      new_width += 1
   return new_height * scale_factor, new_width * scale_factor






def   lowerCamelCase__							(					a		,		a=5_12		,		a=5_12      )      ->   Dict:
   _A:  Union[str, Any]					=       pil_image.resize((w, h)		,		resample=Image.BICUBIC		,		reducing_gap=1      )
   _A:  Tuple					=       np.array(pil_image.convert('''RGB'''      )      )
   _A:  List[str]					=       arr.astype(np.floataa      ) / 127.5 - 1
   _A:  Tuple					=       np.transpose(a		,		[2, 0, 1]      )
   _A:  Any					=       torch.from_numpy(a      ).unsqueeze(0      )
   return image






class UpperCAmelCase   (						SCREAMING_SNAKE_CASE__   ):





   '''simple docstring'''


   def __init__(					self     :							int			,			lowerCAmelCase_     :							UNetaDConditionModel			,			lowerCAmelCase_     :							DDPMScheduler			,			lowerCAmelCase_     :							VQModel			,			):

      """simple docstring"""
      super().__init__()

      self.register_modules(
          unet=lowerCAmelCase_			,			scheduler=lowerCAmelCase_			,			movq=lowerCAmelCase_			,			)
      _A:  List[Any]					=       2 ** (len(self.movq.config.block_out_channels     ) - 1)


   def  __magic_name__				(					self     :							List[str]			,			lowerCAmelCase_     :							Union[str, Any]			,			lowerCAmelCase_     :							List[Any]			,			lowerCAmelCase_     :							Union[str, Any]     ):

      """simple docstring"""
      # get the original timestep using init_timestep
      _A:  Union[str, Any]					=       min(int(num_inference_steps * strength     )			,			lowerCAmelCase_     )

      _A:  str					=       max(num_inference_steps - init_timestep			,			0     )
      _A:  str					=       self.scheduler.timesteps[t_start:]

      return timesteps, num_inference_steps - t_start


   def  __magic_name__				(					self     :							List[str]			,			lowerCAmelCase_     :							List[str]			,			lowerCAmelCase_     :							str			,			lowerCAmelCase_     :							List[Any]			,			lowerCAmelCase_     :							List[Any]			,			lowerCAmelCase_     :							Tuple			,			lowerCAmelCase_     :							Optional[int]			,			lowerCAmelCase_     :							Optional[int]=None     ):

      """simple docstring"""
      if not isinstance(lowerCAmelCase_			,			(torch.Tensor, PIL.Image.Image, list)     ):
         raise ValueError(
             F"""`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(lowerCAmelCase_     )}"""     )

      _A:  Optional[int]					=       image.to(device=lowerCAmelCase_			,			dtype=lowerCAmelCase_     )

      _A:  Union[str, Any]					=       batch_size * num_images_per_prompt

      if image.shape[1] == 4:
         _A:  Optional[int]					=       image

      else:
         if isinstance(lowerCAmelCase_			,			lowerCAmelCase_     ) and len(lowerCAmelCase_     ) != batch_size:
            raise ValueError(
                F"""You have passed a list of generators of length {len(lowerCAmelCase_     )}, but requested an effective batch"""
                F""" size of {batch_size}. Make sure the batch size matches the length of the generators."""     )

         elif isinstance(lowerCAmelCase_			,			lowerCAmelCase_     ):
            _A:  List[Any]					=       [
                self.movq.encode(image[i : i + 1]     ).latent_dist.sample(generator[i]     ) for i in range(lowerCAmelCase_     )
            ]
            _A:  Optional[Any]					=       torch.cat(lowerCAmelCase_			,			dim=0     )
         else:
            _A:  Optional[int]					=       self.movq.encode(lowerCAmelCase_     ).latent_dist.sample(lowerCAmelCase_     )

         _A:  int					=       self.movq.config.scaling_factor * init_latents

      _A:  Optional[Any]					=       torch.cat([init_latents]			,			dim=0     )

      _A:  Any					=       init_latents.shape
      _A:  Optional[Any]					=       randn_tensor(lowerCAmelCase_			,			generator=lowerCAmelCase_			,			device=lowerCAmelCase_			,			dtype=lowerCAmelCase_     )

      # get latents
      _A:  Union[str, Any]					=       self.scheduler.add_noise(lowerCAmelCase_			,			lowerCAmelCase_			,			lowerCAmelCase_     )

      _A:  List[str]					=       init_latents

      return latents


   def  __magic_name__				(					self     :							Optional[int]			,			lowerCAmelCase_     :							Optional[int]=0     ):

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

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

      _A:  int					=       [
          self.unet,
          self.movq,
      ]
      for cpu_offloaded_model in models:
         if cpu_offloaded_model is not None:
            cpu_offload(lowerCAmelCase_			,			lowerCAmelCase_     )


   def  __magic_name__				(					self     :							Any			,			lowerCAmelCase_     :							Any=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.'''     )

      _A:  Any					=       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)

      _A:  int					=       None
      for cpu_offloaded_model in [self.unet, self.movq]:
         _A			,  _A:  List[Any]					=       cpu_offload_with_hook(lowerCAmelCase_			,			lowerCAmelCase_			,			prev_module_hook=lowerCAmelCase_     )

      # We'll offload the last model manually.
      _A:  Tuple					=       hook


   @property
   # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
   def  __magic_name__				(					self     :							List[Any]     ):

      """simple docstring"""
      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     :							Optional[Any]			,			lowerCAmelCase_     :							Union[torch.FloatTensor, List[torch.FloatTensor]]			,			lowerCAmelCase_     :							Union[torch.FloatTensor, PIL.Image.Image, List[torch.FloatTensor], List[PIL.Image.Image]]			,			lowerCAmelCase_     :							Union[torch.FloatTensor, List[torch.FloatTensor]]			,			lowerCAmelCase_     :							int = 5_1_2			,			lowerCAmelCase_     :							int = 5_1_2			,			lowerCAmelCase_     :							int = 1_0_0			,			lowerCAmelCase_     :							float = 4.0			,			lowerCAmelCase_     :							float = 0.3			,			lowerCAmelCase_     :							int = 1			,			lowerCAmelCase_     :							Optional[Union[torch.Generator, List[torch.Generator]]] = None			,			lowerCAmelCase_     :							Optional[str] = "pil"			,			lowerCAmelCase_     :							bool = True			,			):

      """simple docstring"""
      _A:  Any					=       self._execution_device

      _A:  Any					=       guidance_scale > 1.0

      if isinstance(lowerCAmelCase_			,			lowerCAmelCase_     ):
         _A:  Any					=       torch.cat(lowerCAmelCase_			,			dim=0     )
      _A:  int					=       image_embeds.shape[0]
      if isinstance(lowerCAmelCase_			,			lowerCAmelCase_     ):
         _A:  Dict					=       torch.cat(lowerCAmelCase_			,			dim=0     )

      if do_classifier_free_guidance:
         _A:  Any					=       image_embeds.repeat_interleave(lowerCAmelCase_			,			dim=0     )
         _A:  str					=       negative_image_embeds.repeat_interleave(lowerCAmelCase_			,			dim=0     )

      _A:  Dict					=       torch.cat([negative_image_embeds, image_embeds]			,			dim=0     ).to(dtype=self.unet.dtype			,			device=lowerCAmelCase_     )

      if not isinstance(lowerCAmelCase_			,			lowerCAmelCase_     ):
         _A:  List[str]					=       [image]
      if not all(isinstance(lowerCAmelCase_			,			(PIL.Image.Image, torch.Tensor)     ) for i in image     ):
         raise ValueError(
             F"""Input is in incorrect format: {[type(lowerCAmelCase_     ) for i in image]}. Currently, we only support  PIL image and pytorch tensor"""     )

      _A:  List[str]					=       torch.cat([prepare_image(lowerCAmelCase_			,			lowerCAmelCase_			,			lowerCAmelCase_     ) for i in image]			,			dim=0     )
      _A:  Tuple					=       image.to(dtype=image_embeds.dtype			,			device=lowerCAmelCase_     )

      _A:  Optional[Any]					=       self.movq.encode(lowerCAmelCase_     )['''latents''']
      _A:  Optional[int]					=       latents.repeat_interleave(lowerCAmelCase_			,			dim=0     )
      self.scheduler.set_timesteps(lowerCAmelCase_			,			device=lowerCAmelCase_     )
      _A			,  _A:  List[Any]					=       self.get_timesteps(lowerCAmelCase_			,			lowerCAmelCase_			,			lowerCAmelCase_     )
      _A:  Dict					=       timesteps[:1].repeat(batch_size * num_images_per_prompt     )
      _A			,  _A:  Optional[int]					=       downscale_height_and_width(lowerCAmelCase_			,			lowerCAmelCase_			,			self.movq_scale_factor     )
      _A:  Any					=       self.prepare_latents(
          lowerCAmelCase_			,			lowerCAmelCase_			,			lowerCAmelCase_			,			lowerCAmelCase_			,			image_embeds.dtype			,			lowerCAmelCase_			,			lowerCAmelCase_     )
      for i, t in enumerate(self.progress_bar(lowerCAmelCase_     )     ):
         # expand the latents if we are doing classifier free guidance
         _A:  Dict					=       torch.cat([latents] * 2     ) if do_classifier_free_guidance else latents

         _A:  str					=       {'''image_embeds''': image_embeds}
         _A:  Optional[int]					=       self.unet(
             sample=lowerCAmelCase_			,			timestep=lowerCAmelCase_			,			encoder_hidden_states=lowerCAmelCase_			,			added_cond_kwargs=lowerCAmelCase_			,			return_dict=lowerCAmelCase_			,			)[0]

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

         # compute the previous noisy sample x_t -> x_t-1
         _A:  Any					=       self.scheduler.step(
             lowerCAmelCase_			,			lowerCAmelCase_			,			lowerCAmelCase_			,			generator=lowerCAmelCase_			,			)[0]

      # post-processing
      _A:  Tuple					=       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"]:
         _A:  int					=       image * 0.5 + 0.5
         _A:  Any					=       image.clamp(0			,			1     )
         _A:  Any					=       image.cpu().permute(0			,			2			,			3			,			1     ).float().numpy()

      if output_type == "pil":
         _A:  Union[str, Any]					=       self.numpy_to_pil(lowerCAmelCase_     )

      if not return_dict:
         return (image,)

      return ImagePipelineOutput(images=lowerCAmelCase_     )



