
"""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__ :       Union[str, Any]						     =			logging.get_logger(__name__)  # pylint: disable=invalid-name

lowerCamelCase__ :       Dict						     =			"\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 __A						(   a_ :							Any		,		a_ :							Tuple		,		a_ :							Optional[int]=8     )->							List[str]:




		'''simple docstring'''






		SCREAMING_SNAKE_CASE       :     Union[str, Any]				= height // scale_factor**2
		if height % scale_factor**2 != 0:
				new_height += 1
		SCREAMING_SNAKE_CASE       :     str				= 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   ):
	'''simple docstring'''

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







			'''simple docstring'''




			super().__init__()

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





	def   __lowerCAmelCase (						self    :Optional[Any]		, lowerCamelCase_    :Union[str, Any]		, lowerCamelCase_    :Optional[int]		, lowerCamelCase_    :int		, lowerCamelCase_    :Union[str, Any]		, lowerCamelCase_    :str		, lowerCamelCase_    :int					)							->			Optional[int]:







			'''simple docstring'''




			if latents is None:
					SCREAMING_SNAKE_CASE       :     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}"					)
					SCREAMING_SNAKE_CASE       :     List[str]				= latents.to(lowerCamelCase_					)

			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= latents * scheduler.init_noise_sigma
			return latents





	def   __lowerCAmelCase (						self    :Optional[int]		, lowerCamelCase_    :Optional[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`'''					)

			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= torch.device(f"cuda:{gpu_id}"					)

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





	def   __lowerCAmelCase (						self    :Tuple		, lowerCamelCase_    :Union[str, Any]=0					)							->			Union[str, 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.'''					)

			SCREAMING_SNAKE_CASE       :     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)

			SCREAMING_SNAKE_CASE       :     str				= None
			for cpu_offloaded_model in [self.unet, self.movq]:
					SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE       :     Dict				= cpu_offload_with_hook(lowerCamelCase_		, lowerCamelCase_		, prev_module_hook=lowerCamelCase_					)

			# We'll offload the last model manually.
			SCREAMING_SNAKE_CASE       :     str				= hook





	@property
	# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
	def   __lowerCAmelCase (						self    :Optional[Any]					)							->			Dict:







			'''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    :Union[str, Any]		, lowerCamelCase_    :Union[torch.FloatTensor, List[torch.FloatTensor]]		, lowerCamelCase_    :Union[torch.FloatTensor, List[torch.FloatTensor]]		, lowerCamelCase_    :int = 5_12		, lowerCamelCase_    :int = 5_12		, lowerCamelCase_    :int = 1_00		, 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		, )							->			str:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Any				= self._execution_device

			SCREAMING_SNAKE_CASE       :     Optional[int]				= guidance_scale > 1.0

			if isinstance(lowerCamelCase_		, lowerCamelCase_					):
					SCREAMING_SNAKE_CASE       :     Optional[Any]				= torch.cat(lowerCamelCase_		, dim=0					)
			SCREAMING_SNAKE_CASE       :     Optional[int]				= image_embeds.shape[0] * num_images_per_prompt
			if isinstance(lowerCamelCase_		, lowerCamelCase_					):
					SCREAMING_SNAKE_CASE       :     List[str]				= torch.cat(lowerCamelCase_		, dim=0					)

			if do_classifier_free_guidance:
					SCREAMING_SNAKE_CASE       :     Optional[int]				= image_embeds.repeat_interleave(lowerCamelCase_		, dim=0					)
					SCREAMING_SNAKE_CASE       :     str				= negative_image_embeds.repeat_interleave(lowerCamelCase_		, dim=0					)

			SCREAMING_SNAKE_CASE       :     Any				= torch.cat([negative_image_embeds, image_embeds]		, dim=0					).to(dtype=self.unet.dtype		, device=lowerCamelCase_					)

			self.scheduler.set_timesteps(lowerCamelCase_		, device=lowerCamelCase_					)
			SCREAMING_SNAKE_CASE       :     str				= self.scheduler.timesteps

			SCREAMING_SNAKE_CASE       :     Optional[Any]				= self.unet.config.in_channels

			SCREAMING_SNAKE_CASE,					SCREAMING_SNAKE_CASE       :     List[str]				= downscale_height_and_width(lowerCamelCase_		, lowerCamelCase_		, self.movq_scale_factor					)

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

					SCREAMING_SNAKE_CASE       :     Union[str, Any]				= {'''image_embeds''': image_embeds}
					SCREAMING_SNAKE_CASE       :     Optional[int]				= self.unet(
					    sample=lowerCamelCase_		, timestep=lowerCamelCase_		, encoder_hidden_states=lowerCamelCase_		, added_cond_kwargs=lowerCamelCase_		, return_dict=lowerCamelCase_		, )[0]

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

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

			if output_type == "pil":
					SCREAMING_SNAKE_CASE       :     str				= self.numpy_to_pil(lowerCamelCase_					)

			if not return_dict:
					return (image,)

			return ImagePipelineOutput(images=lowerCamelCase_					)







"""simple docstring"""






def __A						(   a_ :							list		,		a_ :							int		,		a_ :							int = 0		,		a_ :							int = 0     )->							int:




		'''simple docstring'''






		SCREAMING_SNAKE_CASE       :     str				= right or len(a_     ) - 1
		if left > right:
				return -1
		elif list_data[left] == key:
				return left
		elif list_data[right] == key:
				return right
		else:
				return search(a_		,		a_		,		left + 1		,		right - 1     )


if __name__ == "__main__":
					import doctest

					doctest.testmod()





