



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


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

A_		:     Dict        =     '''
    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 	UpperCAmelCase__      (      UpperCAmelCase__ :Optional[Any] ,					UpperCAmelCase__ :Optional[int] ,					UpperCAmelCase__ :Optional[int]=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 		_lowercase		(       UpperCAmelCase__	):

   def __init__(      self			:       Optional[Any]							,   __lowerCAmelCase			:       UNetaDConditionModel							,   __lowerCAmelCase			:       DDPMScheduler							,   __lowerCAmelCase			:       VQModel							,   )       ->				Union[str, Any]:


    """simple docstring"""






    super().__init__()

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

   def 		A							(      self			:       str							,   __lowerCAmelCase			:       Optional[Any]							,   __lowerCAmelCase			:       Optional[Any]							,   __lowerCAmelCase			:       List[str]							,   __lowerCAmelCase			:       Optional[Any]							,   __lowerCAmelCase			:       Tuple							,   __lowerCAmelCase			:       Dict							)       ->				Optional[int]:


    """simple docstring"""






    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 		A							(      self			:       Any							,   __lowerCAmelCase			:       Any=0							)       ->				List[Any]:


    """simple docstring"""






    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 		A							(      self			:       Dict							,   __lowerCAmelCase			:       Any=0							)       ->				Any:


    """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						=		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		,  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 		A							(      self			:       List[Any]							)       ->				str:


    """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			:       List[str]							,   __lowerCAmelCase			:       Union[torch.FloatTensor, List[torch.FloatTensor]]							,   __lowerCAmelCase			:       Union[torch.FloatTensor, List[torch.FloatTensor]]							,   __lowerCAmelCase			:       int = 512							,   __lowerCAmelCase			:       int = 512							,   __lowerCAmelCase			:       int = 100							,   __lowerCAmelCase			:       float = 4.0							,   __lowerCAmelCase			:       int = 1							,   __lowerCAmelCase			:       Optional[Union[torch.Generator, List[torch.Generator]]] = None							,   __lowerCAmelCase			:       Optional[torch.FloatTensor] = None							,   __lowerCAmelCase			:       Optional[str] = "pil"							,   __lowerCAmelCase			:       bool = True							,   )       ->				List[str]:


    """simple docstring"""






    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		,  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		,  a						=		noise_pred.split(latents.shape[1]							,   dim=1							)
      a		,  a						=		noise_pred.chunk(2							)
      a		,  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		,  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							)






from typing import List, Optional, Tuple, Union

import torch

from ...models import UNetaDModel
from ...schedulers import ScoreSdeVeScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput



class 		_lowercase		(       UpperCAmelCase__	):
   _UpperCAmelCase     	=			42
   _UpperCAmelCase     	=			42

   def __init__(      self			:       Optional[Any]							,   __lowerCAmelCase			:       UNetaDModel							,   __lowerCAmelCase			:       ScoreSdeVeScheduler							)       ->				str:


    """simple docstring"""






    super().__init__()
    self.register_modules(unet=__lowerCAmelCase							,   scheduler=__lowerCAmelCase							)




   @torch.no_grad()
   def __call__(      self			:       int							,   __lowerCAmelCase			:       int = 1							,   __lowerCAmelCase			:       int = 2000							,   __lowerCAmelCase			:       Optional[Union[torch.Generator, List[torch.Generator]]] = None							,   __lowerCAmelCase			:       Optional[str] = "pil"							,   __lowerCAmelCase			:       bool = True							,   **__lowerCAmelCase			:       Any							,   )       ->				Union[ImagePipelineOutput, Tuple]:


    """simple docstring"""






    a						=		self.unet.config.sample_size
    a						=		(batch_size, 3, img_size, img_size)

    a						=		self.unet

    a						=		randn_tensor(__lowerCAmelCase							,   generator=__lowerCAmelCase							) * self.scheduler.init_noise_sigma
    a						=		sample.to(self.device							)

    self.scheduler.set_timesteps(__lowerCAmelCase							)
    self.scheduler.set_sigmas(__lowerCAmelCase							)

    for i, t in enumerate(self.progress_bar(self.scheduler.timesteps							)							):
     a						=		self.scheduler.sigmas[i] * torch.ones(shape[0]							,   device=self.device							)

     # correction step
     for _ in range(self.scheduler.config.correct_steps							):
      a						=		self.unet(__lowerCAmelCase							,   __lowerCAmelCase							).sample
      a						=		self.scheduler.step_correct(__lowerCAmelCase							,   __lowerCAmelCase							,   generator=__lowerCAmelCase							).prev_sample

     # prediction step
     a						=		model(__lowerCAmelCase							,   __lowerCAmelCase							).sample
     a						=		self.scheduler.step_pred(__lowerCAmelCase							,   __lowerCAmelCase							,   __lowerCAmelCase							,   generator=__lowerCAmelCase							)

     a		,  a						=		output.prev_sample, output.prev_sample_mean

    a						=		sample_mean.clamp(0							,   1							)
    a						=		sample.cpu().permute(0							,   2							,   3							,   1							).numpy()
    if output_type == "pil":
     a						=		self.numpy_to_pil(__lowerCAmelCase							)

    if not return_dict:
     return (sample,)

    return ImagePipelineOutput(images=__lowerCAmelCase							)

