

"""simple docstring"""






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


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

_lowerCAmelCase						=		"\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 		lowerCamelCase__		(					_lowerCamelCase ,   _lowerCamelCase ,   _lowerCamelCase=8   ):

    '''simple docstring'''



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







class 	__UpperCamelCase    (    __UpperCAmelCase						):

  def __init__(		self    ,_A    ,_A    ,_A    ,):
      '''simple docstring'''

      super().__init__()

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

  def __lowerCamelCase		(		self    ,_A    ,_A    ,_A    ,_A    ,_A    ,_A      ):
      '''simple docstring'''

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

      _lowerCAmelCase       :   List[str]  =    latents * scheduler.init_noise_sigma
      return latents

  def __lowerCamelCase		(		self    ,_A=0      ):
      '''simple docstring'''

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

      _lowerCAmelCase       :   int  =    torch.device(F"""cuda:{gpu_id}"""      )

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

  def __lowerCamelCase		(		self    ,_A=0      ):
      '''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.'      )

      _lowerCAmelCase       :   Tuple  =    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)

      _lowerCAmelCase       :   str  =    None
      for cpu_offloaded_model in [self.unet, self.movq]:
          _lowerCAmelCase,					_lowerCAmelCase       :   Tuple  =    cpu_offload_with_hook(lowerCAmelCase_    ,lowerCAmelCase_    ,prev_module_hook=lowerCAmelCase_      )

      # We'll offload the last model manually.
      _lowerCAmelCase       :   Union[str, Any]  =    hook

  @property
  # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
  def __lowerCamelCase		(		self      ):
      '''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    ,_A    ,_A    ,_A = 512    ,_A = 512    ,_A = 100    ,_A = 4.0    ,_A = 1    ,_A = None    ,_A = None    ,_A = "pil"    ,_A = True    ,):
      '''simple docstring'''

      _lowerCAmelCase       :   int  =    self._execution_device

      _lowerCAmelCase       :   str  =    guidance_scale > 1.0

      if isinstance(lowerCAmelCase_    ,lowerCAmelCase_      ):
          _lowerCAmelCase       :   Optional[Any]  =    torch.cat(lowerCAmelCase_    ,dim=0      )
      _lowerCAmelCase       :   Tuple  =    image_embeds.shape[0] * num_images_per_prompt
      if isinstance(lowerCAmelCase_    ,lowerCAmelCase_      ):
          _lowerCAmelCase       :   Dict  =    torch.cat(lowerCAmelCase_    ,dim=0      )

      if do_classifier_free_guidance:
          _lowerCAmelCase       :   Tuple  =    image_embeds.repeat_interleave(lowerCAmelCase_    ,dim=0      )
          _lowerCAmelCase       :   List[Any]  =    negative_image_embeds.repeat_interleave(lowerCAmelCase_    ,dim=0      )

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

      self.scheduler.set_timesteps(lowerCAmelCase_    ,device=lowerCAmelCase_      )
      _lowerCAmelCase       :   Dict  =    self.scheduler.timesteps

      _lowerCAmelCase       :   Union[str, Any]  =    self.unet.config.in_channels

      _lowerCAmelCase,					_lowerCAmelCase       :   Optional[int]  =    downscale_height_and_width(lowerCAmelCase_    ,lowerCAmelCase_    ,self.movq_scale_factor      )

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

          _lowerCAmelCase       :   List[str]  =    {'image_embeds': image_embeds}
          _lowerCAmelCase       :   Dict  =    self.unet(
              sample=lowerCAmelCase_    ,timestep=lowerCAmelCase_    ,encoder_hidden_states=lowerCAmelCase_    ,added_cond_kwargs=lowerCAmelCase_    ,return_dict=lowerCAmelCase_    ,)[0]

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

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

      if output_type == "pil":
          _lowerCAmelCase       :   List[Any]  =    self.numpy_to_pil(lowerCAmelCase_      )

      if not return_dict:
          return (image,)

      return ImagePipelineOutput(images=lowerCAmelCase_      )








"""simple docstring"""






import logging
import os

from .state import PartialState







class 	__UpperCamelCase    (    logging.LoggerAdapter						):

  @staticmethod
  def __lowerCamelCase		(		_A      ):
      '''simple docstring'''

      _lowerCAmelCase       :   Optional[Any]  =    PartialState()
      return not main_process_only or (main_process_only and state.is_main_process)







  def __lowerCamelCase		(		self    ,_A    ,_A    ,*_A    ,**_A      ):
      '''simple docstring'''

      if PartialState._shared_state == {}:
          raise RuntimeError(
              'You must initialize the accelerate state by calling either `PartialState()` or `Accelerator()` before using the logging utility.'      )
      _lowerCAmelCase       :   Tuple  =    kwargs.pop('main_process_only'    ,_A      )
      _lowerCAmelCase       :   Any  =    kwargs.pop('in_order'    ,_A      )

      if self.isEnabledFor(_A      ):
          if self._should_log(_A      ):
              _lowerCAmelCase,					_lowerCAmelCase       :   Optional[int]  =    self.process(_A    ,_A      )
              self.logger.log(_A    ,_A    ,*_A    ,**_A      )

          elif in_order:
              _lowerCAmelCase       :   str  =    PartialState()
              for i in range(state.num_processes      ):
                  if i == state.process_index:
                      _lowerCAmelCase,					_lowerCAmelCase       :   Optional[int]  =    self.process(_A    ,_A      )
                      self.logger.log(_A    ,_A    ,*_A    ,**_A      )
                  state.wait_for_everyone()





def 		lowerCamelCase__		(					_lowerCamelCase ,   _lowerCamelCase = None   ):

    '''simple docstring'''



    if log_level is None:
        _lowerCAmelCase       :   Union[str, Any]  =    os.environ.get('ACCELERATE_LOG_LEVEL' ,   _lowerCamelCase   )
    _lowerCAmelCase       :   int  =    logging.getLogger(_lowerCamelCase   )
    if log_level is not None:
        logger.setLevel(log_level.upper()   )
        logger.root.setLevel(log_level.upper()   )
    return MultiProcessAdapter(_lowerCamelCase ,   {}   )





