



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


__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 						__snake_case(				_lowerCAmelCase     ,   _lowerCAmelCase     ,   _lowerCAmelCase=8		)		-> Any:
      snake_case__  :    Union[str, Any]    						=      height // scale_factor**2
      if height % scale_factor**2 != 0:
            new_height += 1
      snake_case__  :    List[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_       (       A__	):
    """simple docstring"""







    def __init__(    self					:     List[str] ,     snake_case_					:     UNetaDConditionModel ,     snake_case_					:     DDPMScheduler ,     snake_case_					:     VQModel ,     ):
          super().__init__()

          self.register_modules(
              unet=__snake_case ,     scheduler=__snake_case ,     movq=__snake_case ,     )
          snake_case__  :    Any    						=      2 ** (len(self.movq.config.block_out_channels						) - 1)




    def lowerCamelCase       (    self					:     Tuple ,     snake_case_					:     Tuple ,     snake_case_					:     List[str] ,     snake_case_					:     Any ,     snake_case_					:     int ,     snake_case_					:     Optional[Any] ,     snake_case_					:     int						):
          if latents is None:
                snake_case__  :    Optional[int]    						=      randn_tensor(__snake_case ,     generator=__snake_case ,     device=__snake_case ,     dtype=__snake_case						)
          else:
                if latents.shape != shape:
                      raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {shape}"						)
                snake_case__  :    List[Any]    						=      latents.to(__snake_case						)

          snake_case__  :    Optional[int]    						=      latents * scheduler.init_noise_sigma
          return latents




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

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

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




    def lowerCamelCase       (    self					:     List[str] ,     snake_case_					:     Tuple=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."""						)

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

          if self.device.type != "cpu":
                self.to("""cpu""" ,     silence_dtype_warnings=__snake_case						)
                torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

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

          # We'll offload the last model manually.
          snake_case__  :    Tuple    						=      hook




    @property
    # Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
    def lowerCamelCase       (    self					:     List[str]						):
          if not hasattr(self.unet ,     """_hf_hook"""						):
                return self.device
          for module in self.unet.modules():
                if (
                    hasattr(__snake_case ,     """_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(__snake_case						)
    def __call__(    self					:     Tuple ,     snake_case_					:     Union[torch.FloatTensor, List[torch.FloatTensor]] ,     snake_case_					:     Union[torch.FloatTensor, List[torch.FloatTensor]] ,     snake_case_					:     int = 512 ,     snake_case_					:     int = 512 ,     snake_case_					:     int = 100 ,     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 ,     ):
          snake_case__  :    int    						=      self._execution_device

          snake_case__  :    Optional[int]    						=      guidance_scale > 1.0

          if isinstance(__snake_case ,     __snake_case						):
                snake_case__  :    Union[str, Any]    						=      torch.cat(__snake_case ,     dim=0						)
          snake_case__  :    str    						=      image_embeds.shape[0] * num_images_per_prompt
          if isinstance(__snake_case ,     __snake_case						):
                snake_case__  :    Tuple    						=      torch.cat(__snake_case ,     dim=0						)

          if do_classifier_free_guidance:
                snake_case__  :    Optional[Any]    						=      image_embeds.repeat_interleave(__snake_case ,     dim=0						)
                snake_case__  :    int    						=      negative_image_embeds.repeat_interleave(__snake_case ,     dim=0						)

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

          self.scheduler.set_timesteps(__snake_case ,     device=__snake_case						)
          snake_case__  :    Optional[int]    						=      self.scheduler.timesteps

          snake_case__  :    Dict    						=      self.unet.config.in_channels

          snake_case__  :    str    						=      downscale_height_and_width(__snake_case ,     __snake_case ,     self.movq_scale_factor						)

          # create initial latent
          snake_case__  :    Optional[Any]    						=      self.prepare_latents(
              (batch_size, num_channels_latents, height, width) ,     image_embeds.dtype ,     __snake_case ,     __snake_case ,     __snake_case ,     self.scheduler ,     )

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

                snake_case__  :    str    						=      {'''image_embeds''': image_embeds}
                snake_case__  :    Tuple    						=      self.unet(
                    sample=__snake_case ,     timestep=__snake_case ,     encoder_hidden_states=__snake_case ,     added_cond_kwargs=__snake_case ,     return_dict=__snake_case ,     )[0]

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

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

          if output_type == "pil":
                snake_case__  :    Optional[Any]    						=      self.numpy_to_pil(__snake_case						)

          if not return_dict:
                return (image,)

          return ImagePipelineOutput(images=__snake_case						)





'''simple docstring'''






from __future__ import annotations


def  snake_case_     (					_lowerCAmelCase  :						list[int | float]   ,      _lowerCAmelCase  :						int   ,      _lowerCAmelCase  :						int  )      ->   int | float:
							if len(_lowerCAmelCase  ) == 0:
														raise ValueError('''find_max() arg is an empty sequence'''  )
							if (
							    left >= len(_lowerCAmelCase  )
							    or left < -len(_lowerCAmelCase  )
							    or right >= len(_lowerCAmelCase  )
							    or right < -len(_lowerCAmelCase  )
							):
														raise IndexError('''list index out of range'''  )
							if left == right:
														return nums[left]
							UpperCAmelCase :       List[Any] 							=		(left + right) >> 1  # the middle
							UpperCAmelCase :       Optional[Any] 							=		find_max(_lowerCAmelCase   ,      _lowerCAmelCase   ,      _lowerCAmelCase  )  # find max in range[left, mid]
							UpperCAmelCase :       Dict 							=		find_max(_lowerCAmelCase   ,      mid + 1   ,      _lowerCAmelCase  )  # find max in range[mid + 1, right]

							return left_max if left_max >= right_max else right_max


if __name__ == "__main__":
						import doctest

						doctest.testmod(verbose=True)





