


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


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

lowercase_			     =  "\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__ :     Union[str, Any]    ,					lowerCAmelCase__ :     Optional[int]    ,					lowerCAmelCase__ :     List[str]=8			) ->      str:
		__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 					__lowerCAmelCase					(   __SCREAMING_SNAKE_CASE       ):


		'''simple docstring'''



		def __init__(  self							, _a							, _a							, _a							, ):
				super().__init__()

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

		def 			__UpperCAmelCase       (  self							, _a							, _a							, _a							, _a							, _a							, _a      ):
				if latents is None:
						__a           =			randn_tensor(_a							, generator=_a							, device=_a							, dtype=_a      )
				else:
						if latents.shape != shape:
								raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {shape}'''      )
						__a           =			latents.to(_a      )

				__a           =			latents * scheduler.init_noise_sigma
				return latents

		def 			__UpperCAmelCase       (  self							, _a=0      ):
				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(_a							, _a      )

		def 			__UpperCAmelCase       (  self							, _a=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.'''      )

				__a           =			torch.device(f'''cuda:{gpu_id}'''      )

				if self.device.type != "cpu":
						self.to('''cpu'''							, silence_dtype_warnings=_a      )
						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(_a							, _a							, prev_module_hook=_a      )

				# We'll offload the last model manually.
				__a           =			hook

		@property
		# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
		def 			__UpperCAmelCase       (  self      ):
				if not hasattr(self.unet							, '''_hf_hook'''      ):
						return self.device
				for module in self.unet.modules():
						if (
						    hasattr(_a							, '''_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(_a      )
		def __call__(  self							, _a							, _a							, _a = 512							, _a = 512							, _a = 100							, _a = 4.0							, _a = 1							, _a = None							, _a = None							, _a = "pil"							, _a = True							, ):
				__a           =			self._execution_device

				__a           =			guidance_scale > 1.0

				if isinstance(_a							, _a      ):
						__a           =			torch.cat(_a							, dim=0      )
				__a           =			image_embeds.shape[0] * num_images_per_prompt
				if isinstance(_a							, _a      ):
						__a           =			torch.cat(_a							, dim=0      )

				if do_classifier_free_guidance:
						__a           =			image_embeds.repeat_interleave(_a							, dim=0      )
						__a           =			negative_image_embeds.repeat_interleave(_a							, dim=0      )

				__a           =			torch.cat([negative_image_embeds, image_embeds]							, dim=0      ).to(dtype=self.unet.dtype							, device=_a      )

				self.scheduler.set_timesteps(_a							, device=_a      )
				__a           =			self.scheduler.timesteps

				__a           =			self.unet.config.in_channels

				__a      ,     __a           =			downscale_height_and_width(_a							, _a							, self.movq_scale_factor      )

				# create initial latent
				__a           =			self.prepare_latents(
				    (batch_size, num_channels_latents, height, width)							, image_embeds.dtype							, _a							, _a							, _a							, self.scheduler							, )

				for i, t in enumerate(self.progress_bar(_a      )      ):
						# 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=_a							, timestep=_a							, encoder_hidden_states=_a							, added_cond_kwargs=_a							, return_dict=_a							, )[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(
						    _a							, _a							, _a							, generator=_a							, )[0]
				# post-processing
				__a           =			self.movq.decode(_a							, force_not_quantize=_a      )['''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(_a      )

				if not return_dict:
						return (image,)

				return ImagePipelineOutput(images=_a      )











def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							if density <= 0:
														raise ValueError("Impossible fluid density"  )
							if bulk_modulus <= 0:
														raise ValueError("Impossible bulk modulus"  )

							return (bulk_modulus / density) ** 0.5


if __name__ == "__main__":
						import doctest

						doctest.testmod()
