






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 					_lowerCamelCase(			lowerCAmelCase__				:		Dict    ,    lowerCAmelCase__				:		Tuple    ,    lowerCAmelCase__				:		List[str]=8      ):





    '''simple docstring'''







    SCREAMING_SNAKE_CASE_   :	int													=   height // scale_factor**2
    if height % scale_factor**2 != 0:
        new_height += 1
    SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   width // scale_factor**2
    if width % scale_factor**2 != 0:
        new_width += 1
    return new_height * scale_factor, new_width * scale_factor



def 					_lowerCamelCase(			lowerCAmelCase__				:		str    ,    lowerCAmelCase__				:		Optional[Any]=512    ,    lowerCAmelCase__				:		Any=512      ):





    '''simple docstring'''







    SCREAMING_SNAKE_CASE_   :	Any													=   pil_image.resize((w, h)    ,    resample=Image.BICUBIC    ,    reducing_gap=1      )
    SCREAMING_SNAKE_CASE_   :	Tuple													=   np.array(pil_image.convert('RGB'      )      )
    SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   arr.astype(np.floataa      ) / 127.5 - 1
    SCREAMING_SNAKE_CASE_   :	List[str]													=   np.transpose(UpperCAmelCase__    ,    [2, 0, 1]      )
    SCREAMING_SNAKE_CASE_   :	Dict													=   torch.from_numpy(UpperCAmelCase__      ).unsqueeze(0      )
    return image


class 							__a							(  _a						):





  '''simple docstring'''



  def __init__(						self ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__ ,    ):
      super().__init__()

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


  def   __snake_case			(						self ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__       ):
      # get the original timestep using init_timestep
      SCREAMING_SNAKE_CASE_   :	str													=   min(int(num_inference_steps * strength       ) ,    _A       )

      SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   max(num_inference_steps - init_timestep ,    0       )
      SCREAMING_SNAKE_CASE_   :	List[str]													=   self.scheduler.timesteps[t_start:]

      return timesteps, num_inference_steps - t_start


  def   __snake_case			(						self ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__=None       ):
      if not isinstance(_A ,    (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(_A       )}'''       )

      SCREAMING_SNAKE_CASE_   :	Any													=   image.to(device=_A ,    dtype=_A       )

      SCREAMING_SNAKE_CASE_   :	Any													=   batch_size * num_images_per_prompt

      if image.shape[1] == 4:
          SCREAMING_SNAKE_CASE_   :	Tuple													=   image

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

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

          SCREAMING_SNAKE_CASE_   :	str													=   self.movq.config.scaling_factor * init_latents

      SCREAMING_SNAKE_CASE_   :	Tuple													=   torch.cat([init_latents] ,    dim=0       )

      SCREAMING_SNAKE_CASE_   :	List[Any]													=   init_latents.shape
      SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   randn_tensor(_A ,    generator=_A ,    device=_A ,    dtype=_A       )

      # get latents
      SCREAMING_SNAKE_CASE_   :	List[Any]													=   self.scheduler.add_noise(_A ,    _A ,    _A       )

      SCREAMING_SNAKE_CASE_   :	List[str]													=   init_latents

      return latents


  def   __snake_case			(						self ,    UpperCamelCase__=0       ):
      if is_accelerate_available():
          from accelerate import cpu_offload
      else:
          raise ImportError('Please install accelerate via `pip install accelerate`'       )

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

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


  def   __snake_case			(						self ,    UpperCamelCase__=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.'       )

      SCREAMING_SNAKE_CASE_   :	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)

      SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   None
      for cpu_offloaded_model in [self.unet, self.movq]:
          SCREAMING_SNAKE_CASE_							,       SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   cpu_offload_with_hook(_A ,    _A ,    prev_module_hook=_A       )

      # We'll offload the last model manually.
      SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   hook


  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def   __snake_case			(						self       ):
      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 ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__ ,    UpperCamelCase__ = 512 ,    UpperCamelCase__ = 512 ,    UpperCamelCase__ = 100 ,    UpperCamelCase__ = 4.0 ,    UpperCamelCase__ = 0.3 ,    UpperCamelCase__ = 1 ,    UpperCamelCase__ = None ,    UpperCamelCase__ = "pil" ,    UpperCamelCase__ = True ,    ):
      SCREAMING_SNAKE_CASE_   :	int													=   self._execution_device

      SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   guidance_scale > 1.0

      if isinstance(_A ,    _A       ):
          SCREAMING_SNAKE_CASE_   :	Optional[int]													=   torch.cat(_A ,    dim=0       )
      SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   image_embeds.shape[0]
      if isinstance(_A ,    _A       ):
          SCREAMING_SNAKE_CASE_   :	Tuple													=   torch.cat(_A ,    dim=0       )

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

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

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

      SCREAMING_SNAKE_CASE_   :	Any													=   torch.cat([prepare_image(_A ,    _A ,    _A       ) for i in image] ,    dim=0       )
      SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   image.to(dtype=image_embeds.dtype ,    device=_A       )

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

          SCREAMING_SNAKE_CASE_   :	Union[str, Any]													=   {'image_embeds': image_embeds}
          SCREAMING_SNAKE_CASE_   :	List[Any]													=   self.unet(
              sample=_A ,    timestep=_A ,    encoder_hidden_states=_A ,    added_cond_kwargs=_A ,    return_dict=_A ,    )[0]

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

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

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

      if output_type == "pil":
          SCREAMING_SNAKE_CASE_   :	str													=   self.numpy_to_pil(_A       )

      if not return_dict:
          return (image,)

      return ImagePipelineOutput(images=_A       )







from timeit import timeit



def 					_lowerCamelCase(			lowerCAmelCase__				:		int      ):





				'''simple docstring'''







				if number < 0:
								raise ValueError('the value of input must not be negative'      )
				SCREAMING_SNAKE_CASE_   :	Tuple													=   0
				while number:
								number &= number - 1
								result += 1
				return result



def 					_lowerCamelCase(			lowerCAmelCase__				:		int      ):





				'''simple docstring'''







				if number < 0:
								raise ValueError('the value of input must not be negative'      )
				SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   0
				while number:
								if number % 2 == 1:
												result += 1
								number >>= 1
				return result



def 					_lowerCamelCase(			):





				'''simple docstring'''







				def do_benchmark(lowerCAmelCase__				:		int      ) -> None:
								SCREAMING_SNAKE_CASE_   :	int													=   'import __main__ as z'
								print(F'''Benchmark when {number = }:'''      )
								print(F'''{get_set_bits_count_using_modulo_operator(lowerCAmelCase__      ) = }'''      )
								SCREAMING_SNAKE_CASE_   :	int													=   timeit('z.get_set_bits_count_using_modulo_operator(25)'    ,    setup=lowerCAmelCase__      )
								print(F'''timeit() runs in {timing} seconds'''      )
								print(F'''{get_set_bits_count_using_brian_kernighans_algorithm(lowerCAmelCase__      ) = }'''      )
								SCREAMING_SNAKE_CASE_   :	Optional[Any]													=   timeit(
								    'z.get_set_bits_count_using_brian_kernighans_algorithm(25)'    ,    setup=lowerCAmelCase__    ,    )
								print(F'''timeit() runs in {timing} seconds'''      )

				for number in (25, 37, 58, 0):
								do_benchmark(lowerCAmelCase__      )
								print()


if __name__ == "__main__":
				import doctest

				doctest.testmod()
				benchmark()