from typing import List, Optional, Union

import torch

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


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

SCREAMING_SNAKE_CASE :	Optional[int]							     =		'''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")
        >>> pipe_prior.to("cuda")
        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> zero_image_emb = out.negative_image_embeds
        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")
        >>> pipe.to("cuda")
        >>> image = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=50,
        ... ).images
        >>> image[0].save("cat.png")
        ```
'''



def   __A		(			_A							,				_A							,				_A=8					):



 """simple docstring"""


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






class    A_ (       a__			):

  def __init__(		self :      List[str]     ,							__SCREAMING_SNAKE_CASE :      List[str]     ,							__SCREAMING_SNAKE_CASE :      Optional[Any]     ,							__SCREAMING_SNAKE_CASE :      List[str]     ,							):
   super().__init__()

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

  def 		_UpperCAmelCase   (		self :      Optional[int]     ,							__SCREAMING_SNAKE_CASE :      Optional[int]     ,							__SCREAMING_SNAKE_CASE :      Tuple     ,							__SCREAMING_SNAKE_CASE :      List[Any]     ,							__SCREAMING_SNAKE_CASE :      List[Any]     ,							__SCREAMING_SNAKE_CASE :      Dict     ,							__SCREAMING_SNAKE_CASE :      Any       ):
   if latents is None:
    __a       =  randn_tensor(lowerCAmelCase__     ,							generator=lowerCAmelCase__     ,							device=lowerCAmelCase__     ,							dtype=lowerCAmelCase__       )
   else:
    if latents.shape != shape:
     raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {shape}"""       )
    __a       =  latents.to(lowerCAmelCase__       )

   __a       =  latents * scheduler.init_noise_sigma
   return latents

  def 		_UpperCAmelCase   (		self :      Dict     ,							__SCREAMING_SNAKE_CASE :      Tuple=0       ):
   if is_accelerate_available():
    from accelerate import cpu_offload
   else:
    raise ImportError("Please install accelerate via `pip install accelerate`"       )

   __a       =  torch.device(f"""cuda:{gpu_id}"""       )

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

  def 		_UpperCAmelCase   (		self :      Any     ,							__SCREAMING_SNAKE_CASE :      Any=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."       )

   __a       =  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       =  None
   for cpu_offloaded_model in [self.unet, self.movq]:
    __a       =  cpu_offload_with_hook(lowerCAmelCase__     ,							lowerCAmelCase__     ,							prev_module_hook=lowerCAmelCase__       )

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

  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def 		_UpperCAmelCase   (		self :      Optional[int]       ):
   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 :      Union[str, Any]     ,							__SCREAMING_SNAKE_CASE :      Tuple     ,							__SCREAMING_SNAKE_CASE :      List[str]     ,							__SCREAMING_SNAKE_CASE :      Union[str, Any] = 5_12     ,							__SCREAMING_SNAKE_CASE :      Optional[int] = 5_12     ,							__SCREAMING_SNAKE_CASE :      List[str] = 1_00     ,							__SCREAMING_SNAKE_CASE :      List[Any] = 4.0     ,							__SCREAMING_SNAKE_CASE :      str = 1     ,							__SCREAMING_SNAKE_CASE :      Union[str, Any] = None     ,							__SCREAMING_SNAKE_CASE :      Tuple = None     ,							__SCREAMING_SNAKE_CASE :      Optional[int] = "pil"     ,							__SCREAMING_SNAKE_CASE :      int = True     ,							):
   __a       =  self._execution_device

   __a       =  guidance_scale > 1.0

   if isinstance(lowerCAmelCase__     ,							lowerCAmelCase__       ):
    __a       =  torch.cat(lowerCAmelCase__     ,							dim=0       )
   __a       =  image_embeds.shape[0] * num_images_per_prompt
   if isinstance(lowerCAmelCase__     ,							lowerCAmelCase__       ):
    __a       =  torch.cat(lowerCAmelCase__     ,							dim=0       )

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

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

   self.scheduler.set_timesteps(lowerCAmelCase__     ,							device=lowerCAmelCase__       )
   __a       =  self.scheduler.timesteps

   __a       =  self.unet.config.in_channels

   __a       =  downscale_height_and_width(lowerCAmelCase__     ,							lowerCAmelCase__     ,							self.movq_scale_factor       )

   # create initial latent
   __a       =  self.prepare_latents(
       (batch_size, num_channels_latents, height, width)     ,							image_embeds.dtype     ,							lowerCAmelCase__     ,							lowerCAmelCase__     ,							lowerCAmelCase__     ,							self.scheduler     ,							)

   for i, t in enumerate(self.progress_bar(lowerCAmelCase__       )       ):
    # expand the latents if we are doing classifier free guidance
    __a       =  torch.cat([latents] * 2       ) if do_classifier_free_guidance else latents

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

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

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

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

   if not return_dict:
    return (image,)

   return ImagePipelineOutput(images=lowerCAmelCase__       )



import os
import sys
from contextlib import contextmanager


# Windows only
if os.name == "nt":
  import ctypes
  import msvcrt  # noqa
  class    A_ (       ctypes.Structure ):
    # _fields is a specific attr expected by ctypes
    _SCREAMING_SNAKE_CASE    				=			[("""size""", ctypes.c_int), ("""visible""", ctypes.c_byte)]



def   __A		(			):



 """simple docstring"""


 if os.name == "nt":
  __a       =  CursorInfo()
  __a       =  ctypes.windll.kernelaa.GetStdHandle(-11					)
  ctypes.windll.kernelaa.GetConsoleCursorInfo(_A							,				ctypes.byref(_A					)					)
  __a       =  False
  ctypes.windll.kernelaa.SetConsoleCursorInfo(_A							,				ctypes.byref(_A					)					)
 elif os.name == "posix":
  sys.stdout.write("\033[?25l"					)
  sys.stdout.flush()



def   __A		(			):



 """simple docstring"""


 if os.name == "nt":
  __a       =  CursorInfo()
  __a       =  ctypes.windll.kernelaa.GetStdHandle(-11					)
  ctypes.windll.kernelaa.GetConsoleCursorInfo(_A							,				ctypes.byref(_A					)					)
  __a       =  True
  ctypes.windll.kernelaa.SetConsoleCursorInfo(_A							,				ctypes.byref(_A					)					)
 elif os.name == "posix":
  sys.stdout.write("\033[?25h"					)
  sys.stdout.flush()






@contextmanager
def   __A		(			):



 """simple docstring"""


 try:
  hide_cursor()
  yield
 finally:
  show_cursor()


