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_          =			logging.get_logger(__name__)  # pylint: disable=invalid-name

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





def   __lowercase	(		lowerCamelCase   :	Optional[int]    ,    lowerCamelCase   :	int    ,    lowerCamelCase   :	Any=8      ):
 UpperCamelCase_   :		Tuple					=   height // scale_factor**2
 if height % scale_factor**2 != 0:
  new_height += 1
 UpperCamelCase_   :		Any					=   width // scale_factor**2
 if width % scale_factor**2 != 0:
  new_width += 1
 return new_height * scale_factor, new_width * scale_factor






class 						_lowercase		(   _SCREAMING_SNAKE_CASE			):

 def __init__(					self      :				Dict  , snake_case      :				UNetaDConditionModel  , snake_case      :				DDPMScheduler  , snake_case      :				VQModel  , )  ->		Optional[int]:

  """simple docstring"""





  super().__init__()

  self.register_modules(
      unet=_lowerCAmelCase  , scheduler=_lowerCAmelCase  , movq=_lowerCAmelCase  , )
  UpperCamelCase_   :		List[str]					=   2 ** (len(self.movq.config.block_out_channels     ) - 1)

 def 		SCREAMING_SNAKE_CASE__  (					self      :				Union[str, Any]  , snake_case      :				str  , snake_case      :				Optional[int]  , snake_case      :				Dict  , snake_case      :				Optional[int]  , snake_case      :				List[str]  , snake_case      :				List[Any]     )  ->		List[str]:

  """simple docstring"""





  if latents is None:
   UpperCamelCase_   :		Optional[Any]					=   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}"     )
   UpperCamelCase_   :		Union[str, Any]					=   latents.to(_lowerCAmelCase     )

  UpperCamelCase_   :		Dict					=   latents * scheduler.init_noise_sigma
  return latents

 def 		SCREAMING_SNAKE_CASE__  (					self      :				Optional[Any]  , snake_case      :				str=0     )  ->		List[Any]:

  """simple docstring"""





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

  UpperCamelCase_   :		Optional[int]					=   torch.device(f"cuda:{gpu_id}"     )

  UpperCamelCase_   :		Union[str, Any]					=   [
      self.unet,
      self.movq,
  ]
  for cpu_offloaded_model in models:
   if cpu_offloaded_model is not None:
    cpu_offload(_lowerCAmelCase  , _lowerCAmelCase     )

 def 		SCREAMING_SNAKE_CASE__  (					self      :				List[str]  , snake_case      :				Any=0     )  ->		List[str]:

  """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.'     )

  UpperCamelCase_   :		List[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)

  UpperCamelCase_   :		Dict					=   None
  for cpu_offloaded_model in [self.unet, self.movq]:
   UpperCamelCase_,					UpperCamelCase_   :		List[Any]					=   cpu_offload_with_hook(_lowerCAmelCase  , _lowerCAmelCase  , prev_module_hook=_lowerCAmelCase     )

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

 @property
 # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
 def 		SCREAMING_SNAKE_CASE__  (					self      :				int     )  ->		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      :				Any  , snake_case      :				Union[torch.FloatTensor, List[torch.FloatTensor]]  , snake_case      :				Union[torch.FloatTensor, List[torch.FloatTensor]]  , snake_case      :				int = 5_1_2  , snake_case      :				int = 5_1_2  , snake_case      :				int = 1_0_0  , snake_case      :				float = 4.0  , snake_case      :				int = 1  , snake_case      :				Optional[Union[torch.Generator, List[torch.Generator]]] = None  , snake_case      :				Optional[torch.FloatTensor] = None  , snake_case      :				Optional[str] = "pil"  , snake_case      :				bool = True  , )  ->		Optional[int]:

  """simple docstring"""





  UpperCamelCase_   :		List[str]					=   self._execution_device

  UpperCamelCase_   :		str					=   guidance_scale > 1.0

  if isinstance(_lowerCAmelCase  , _lowerCAmelCase     ):
   UpperCamelCase_   :		Dict					=   torch.cat(_lowerCAmelCase  , dim=0     )
  UpperCamelCase_   :		List[str]					=   image_embeds.shape[0] * num_images_per_prompt
  if isinstance(_lowerCAmelCase  , _lowerCAmelCase     ):
   UpperCamelCase_   :		str					=   torch.cat(_lowerCAmelCase  , dim=0     )

  if do_classifier_free_guidance:
   UpperCamelCase_   :		Dict					=   image_embeds.repeat_interleave(_lowerCAmelCase  , dim=0     )
   UpperCamelCase_   :		Tuple					=   negative_image_embeds.repeat_interleave(_lowerCAmelCase  , dim=0     )

  UpperCamelCase_   :		Union[str, Any]					=   torch.cat([negative_image_embeds, image_embeds]  , dim=0     ).to(dtype=self.unet.dtype  , device=_lowerCAmelCase     )

  self.scheduler.set_timesteps(_lowerCAmelCase  , device=_lowerCAmelCase     )
  UpperCamelCase_   :		Union[str, Any]					=   self.scheduler.timesteps

  UpperCamelCase_   :		Any					=   self.unet.config.in_channels

  UpperCamelCase_,					UpperCamelCase_   :		int					=   downscale_height_and_width(_lowerCAmelCase  , _lowerCAmelCase  , self.movq_scale_factor     )

  # create initial latent
  UpperCamelCase_   :		Optional[Any]					=   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
   UpperCamelCase_   :		Optional[Any]					=   torch.cat([latents] * 2     ) if do_classifier_free_guidance else latents

   UpperCamelCase_   :		Dict					=   {'image_embeds': image_embeds}
   UpperCamelCase_   :		Tuple					=   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_   :		Dict					=   noise_pred.split(latents.shape[1]  , dim=1     )
    UpperCamelCase_,					UpperCamelCase_   :		Dict					=   noise_pred.chunk(2     )
    UpperCamelCase_,					UpperCamelCase_   :		List[str]					=   variance_pred.chunk(2     )
    UpperCamelCase_   :		int					=   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
    UpperCamelCase_   :		Optional[int]					=   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_   :		Optional[int]					=   noise_pred.split(latents.shape[1]  , dim=1     )

   # compute the previous noisy sample x_t -> x_t-1
   UpperCamelCase_   :		List[str]					=   self.scheduler.step(
       _lowerCAmelCase  , _lowerCAmelCase  , _lowerCAmelCase  , generator=_lowerCAmelCase  , )[0]
  # post-processing
  UpperCamelCase_   :		Optional[int]					=   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_   :		Any					=   image * 0.5 + 0.5
   UpperCamelCase_   :		int					=   image.clamp(0  , 1     )
   UpperCamelCase_   :		Optional[Any]					=   image.cpu().permute(0  , 2  , 3  , 1     ).float().numpy()

  if output_type == "pil":
   UpperCamelCase_   :		Optional[int]					=   self.numpy_to_pil(_lowerCAmelCase     )

  if not return_dict:
   return (image,)

  return ImagePipelineOutput(images=_lowerCAmelCase     )









from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


lowerCamelCase__					:  Optional[int]      							=				{'configuration_mmbt': ['MMBTConfig']}

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowerCamelCase__					:  Any      							=				['MMBTForClassification', 'MMBTModel', 'ModalEmbeddings']


if TYPE_CHECKING:
   from .configuration_mmbt import MMBTConfig

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_mmbt import MMBTForClassification, MMBTModel, ModalEmbeddings

else:
   import sys

   lowerCamelCase__					:  str      							=				_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)