






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,
)


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

__A					=	"\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       lowercase__	(					A_:						Dict  ,      A_:						List[Any]  ,      A_:						str=8		)	->				str:







       """simple docstring"""






       __UpperCAmelCase     				=height // scale_factor**2
       if height % scale_factor**2 != 0:
              new_height += 1
       __UpperCAmelCase     				=width // scale_factor**2
       if width % scale_factor**2 != 0:
              new_width += 1
       return new_height * scale_factor, new_width * scale_factor







def       lowercase__	(					A_:						Dict  ,      A_:						Tuple=512  ,      A_:						Any=512		)	->				Optional[Any]:







       """simple docstring"""






       __UpperCAmelCase     				=pil_image.resize((w, h)  ,      resample=Image.BICUBIC  ,      reducing_gap=1		)
       __UpperCAmelCase     				=np.array(pil_image.convert("""RGB"""		)		)
       __UpperCAmelCase     				=arr.astype(np.floataa		) / 1_2_7.5 - 1
       __UpperCAmelCase     				=np.transpose(_snake_case  ,      [2, 0, 1]		)
       __UpperCAmelCase     				=torch.from_numpy(_snake_case		).unsqueeze(0		)
       return image





class 				_A			(     _A				):
    """simple docstring"""

    def __init__(				self      :     Optional[Any] ,     __SCREAMING_SNAKE_CASE      :     UNetaDConditionModel ,     __SCREAMING_SNAKE_CASE      :     DDPMScheduler ,     __SCREAMING_SNAKE_CASE      :     VQModel ,     )      ->							Tuple:
           super().__init__()

           self.register_modules(
               unet=__lowerCAmelCase ,     scheduler=__lowerCAmelCase ,     movq=__lowerCAmelCase ,     )
           __UpperCAmelCase     				=2 ** (len(self.movq.config.block_out_channels      ) - 1)



    def _a							(				self      :     Dict ,     __SCREAMING_SNAKE_CASE      :     List[str] ,     __SCREAMING_SNAKE_CASE      :     List[Any] ,     __SCREAMING_SNAKE_CASE      :     List[str]      )      ->							Dict:
           # get the original timestep using init_timestep
           __UpperCAmelCase     				=min(int(num_inference_steps * strength      ) ,     __lowerCAmelCase      )

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

           return timesteps, num_inference_steps - t_start



    def _a							(				self      :     int ,     __SCREAMING_SNAKE_CASE      :     List[Any] ,     __SCREAMING_SNAKE_CASE      :     Any ,     __SCREAMING_SNAKE_CASE      :     str ,     __SCREAMING_SNAKE_CASE      :     Tuple ,     __SCREAMING_SNAKE_CASE      :     Optional[Any] ,     __SCREAMING_SNAKE_CASE      :     Optional[int] ,     __SCREAMING_SNAKE_CASE      :     List[str]=None      )      ->							List[Any]:
           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      )}'''      )

           __UpperCAmelCase     				=image.to(device=__lowerCAmelCase ,     dtype=__lowerCAmelCase      )

           __UpperCAmelCase     				=batch_size * num_images_per_prompt

           if image.shape[1] == 4:
                  __UpperCAmelCase     				=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      ):
                         __UpperCAmelCase     				=[
                             self.movq.encode(image[i : i + 1]      ).latent_dist.sample(generator[i]      ) for i in range(__lowerCAmelCase      )
                         ]
                         __UpperCAmelCase     				=torch.cat(__lowerCAmelCase ,     dim=0      )
                  else:
                         __UpperCAmelCase     				=self.movq.encode(__lowerCAmelCase      ).latent_dist.sample(__lowerCAmelCase      )

                  __UpperCAmelCase     				=self.movq.config.scaling_factor * init_latents

           __UpperCAmelCase     				=torch.cat([init_latents] ,     dim=0      )

           __UpperCAmelCase     				=init_latents.shape
           __UpperCAmelCase     				=randn_tensor(__lowerCAmelCase ,     generator=__lowerCAmelCase ,     device=__lowerCAmelCase ,     dtype=__lowerCAmelCase      )

           # get latents
           __UpperCAmelCase     				=self.scheduler.add_noise(__lowerCAmelCase ,     __lowerCAmelCase ,     __lowerCAmelCase      )

           __UpperCAmelCase     				=init_latents

           return latents



    def _a							(				self      :     Optional[int] ,     __SCREAMING_SNAKE_CASE      :     str=0      )      ->							Tuple:
           if is_accelerate_available():
                  from accelerate import cpu_offload
           else:
                  raise ImportError("""Please install accelerate via `pip install accelerate`"""      )

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

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



    def _a							(				self      :     int ,     __SCREAMING_SNAKE_CASE      :     str=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     				=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)

           __UpperCAmelCase     				=None
           for cpu_offloaded_model in [self.unet, self.movq]:
                  __UpperCAmelCase				, __UpperCAmelCase     				=cpu_offload_with_hook(__lowerCAmelCase ,     __lowerCAmelCase ,     prev_module_hook=__lowerCAmelCase      )

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



    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def _a							(				self      :     Union[str, Any]      )      ->							Any:
           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      :     Any ,     __SCREAMING_SNAKE_CASE      :     Union[torch.FloatTensor, List[torch.FloatTensor]] ,     __SCREAMING_SNAKE_CASE      :     Union[torch.FloatTensor, PIL.Image.Image, List[torch.FloatTensor], List[PIL.Image.Image]] ,     __SCREAMING_SNAKE_CASE      :     Union[torch.FloatTensor, List[torch.FloatTensor]] ,     __SCREAMING_SNAKE_CASE      :     int = 512 ,     __SCREAMING_SNAKE_CASE      :     int = 512 ,     __SCREAMING_SNAKE_CASE      :     int = 100 ,     __SCREAMING_SNAKE_CASE      :     float = 4.0 ,     __SCREAMING_SNAKE_CASE      :     float = 0.3 ,     __SCREAMING_SNAKE_CASE      :     int = 1 ,     __SCREAMING_SNAKE_CASE      :     Optional[Union[torch.Generator, List[torch.Generator]]] = None ,     __SCREAMING_SNAKE_CASE      :     Optional[str] = "pil" ,     __SCREAMING_SNAKE_CASE      :     bool = True ,     )      ->							List[Any]:
           __UpperCAmelCase     				=self._execution_device

           __UpperCAmelCase     				=guidance_scale > 1.0

           if isinstance(__lowerCAmelCase ,     __lowerCAmelCase      ):
                  __UpperCAmelCase     				=torch.cat(__lowerCAmelCase ,     dim=0      )
           __UpperCAmelCase     				=image_embeds.shape[0]
           if isinstance(__lowerCAmelCase ,     __lowerCAmelCase      ):
                  __UpperCAmelCase     				=torch.cat(__lowerCAmelCase ,     dim=0      )

           if do_classifier_free_guidance:
                  __UpperCAmelCase     				=image_embeds.repeat_interleave(__lowerCAmelCase ,     dim=0      )
                  __UpperCAmelCase     				=negative_image_embeds.repeat_interleave(__lowerCAmelCase ,     dim=0      )

           __UpperCAmelCase     				=torch.cat([negative_image_embeds, image_embeds] ,     dim=0      ).to(dtype=self.unet.dtype ,     device=__lowerCAmelCase      )

           if not isinstance(__lowerCAmelCase ,     __lowerCAmelCase      ):
                  __UpperCAmelCase     				=[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'''      )

           __UpperCAmelCase     				=torch.cat([prepare_image(__lowerCAmelCase ,     __lowerCAmelCase ,     __lowerCAmelCase      ) for i in image] ,     dim=0      )
           __UpperCAmelCase     				=image.to(dtype=image_embeds.dtype ,     device=__lowerCAmelCase      )

           __UpperCAmelCase     				=self.movq.encode(__lowerCAmelCase      )["""latents"""]
           __UpperCAmelCase     				=latents.repeat_interleave(__lowerCAmelCase ,     dim=0      )
           self.scheduler.set_timesteps(__lowerCAmelCase ,     device=__lowerCAmelCase      )
           __UpperCAmelCase				, __UpperCAmelCase     				=self.get_timesteps(__lowerCAmelCase ,     __lowerCAmelCase ,     __lowerCAmelCase      )
           __UpperCAmelCase     				=timesteps[:1].repeat(batch_size * num_images_per_prompt      )
           __UpperCAmelCase				, __UpperCAmelCase     				=downscale_height_and_width(__lowerCAmelCase ,     __lowerCAmelCase ,     self.movq_scale_factor      )
           __UpperCAmelCase     				=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
                  __UpperCAmelCase     				=torch.cat([latents] * 2      ) if do_classifier_free_guidance else latents

                  __UpperCAmelCase     				={"""image_embeds""": image_embeds}
                  __UpperCAmelCase     				=self.unet(
                      sample=__lowerCAmelCase ,     timestep=__lowerCAmelCase ,     encoder_hidden_states=__lowerCAmelCase ,     added_cond_kwargs=__lowerCAmelCase ,     return_dict=__lowerCAmelCase ,     )[0]

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

                  # compute the previous noisy sample x_t -> x_t-1
                  __UpperCAmelCase     				=self.scheduler.step(
                      __lowerCAmelCase ,     __lowerCAmelCase ,     __lowerCAmelCase ,     generator=__lowerCAmelCase ,     )[0]

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

           if output_type == "pil":
                  __UpperCAmelCase     				=self.numpy_to_pil(__lowerCAmelCase      )

           if not return_dict:
                  return (image,)

           return ImagePipelineOutput(images=__lowerCAmelCase      )









def 	SCREAMING_SNAKE_CASE_							(	_snake_case	:int				)  ->     int:
   if not isinstance(_snake_case	,   _snake_case				):
      raise ValueError('''Input must be an integer'''				)
   if input_num <= 0:
      raise ValueError('''Input must be positive'''				)
   return sum(
       divisor for divisor in range(1	,   input_num // 2 + 1				) if input_num % divisor == 0				)


if __name__ == "__main__":
   import doctest

   doctest.testmod()



