

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

lowerCamelCase__					:	Optional[Any]				    =	'''
    Examples:
        ```py
        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")
        >>> pipe_prior.to("cuda")
        >>> prompt = "red cat, 4k photo"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> zero_image_emb = out.negative_image_embeds
        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")
        >>> pipe.to("cuda")
        >>> image = pipe(
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=zero_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=50,
        ... ).images
        >>> image[0].save("cat.png")
        ```
'''
def    UpperCamelCase       (      _lowerCAmelCase  :    Any,	_lowerCAmelCase  :    Dict,	_lowerCAmelCase  :    List[str]=8			)					->      Tuple:
	_UpperCAmelCase     :				Union[str, Any]          =       height // scale_factor**2
	if height % scale_factor**2 != 0:
		new_height += 1
	_UpperCAmelCase     :				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      (		__a):
			def __init__(		self   ,		_A   ,		_A   ,		_A   ,		)     ->     List[Any]:






				'''simple docstring'''







				super().__init__()

				self.register_modules(
				    unet=__lowerCAmelCase   ,		scheduler=__lowerCAmelCase   ,		movq=__lowerCAmelCase   ,		)
				_UpperCAmelCase     :				Tuple          =       2 ** (len(self.movq.config.block_out_channels  ) - 1)
			def 				__snake_case			(		self   ,		_A   ,		_A   ,		_A   ,		_A   ,		_A   ,		_A  )     ->     str:






				'''simple docstring'''







				if latents is None:
					_UpperCAmelCase     :				int          =       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     :				int          =       latents.to(__lowerCAmelCase  )

				_UpperCAmelCase     :				str          =       latents * scheduler.init_noise_sigma
				return latents
			def 				__snake_case			(		self   ,		_A=0  )     ->     Optional[Any]:






				'''simple docstring'''







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

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

				_UpperCAmelCase     :				List[str]          =       [
				    self.unet,
				    self.movq,
				]
				for cpu_offloaded_model in models:
					if cpu_offloaded_model is not None:
						cpu_offload(__lowerCAmelCase   ,		__lowerCAmelCase  )
			def 				__snake_case			(		self   ,		_A=0  )     ->     Optional[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."""  )

				_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     :				Tuple          =       hook
			@property
			# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
			def 				__snake_case			(		self  )     ->     List[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   ,		_A   ,		_A   ,		_A = 5_12   ,		_A = 5_12   ,		_A = 1_00   ,		_A = 4.0   ,		_A = 1   ,		_A = None   ,		_A = None   ,		_A = "pil"   ,		_A = True   ,		)     ->     List[str]:






				'''simple docstring'''







				_UpperCAmelCase     :				Union[str, Any]          =       self._execution_device

				_UpperCAmelCase     :				Union[str, Any]          =       guidance_scale > 1.0

				if isinstance(__lowerCAmelCase   ,		__lowerCAmelCase  ):
					_UpperCAmelCase     :				Any          =       torch.cat(__lowerCAmelCase   ,		dim=0  )
				_UpperCAmelCase     :				Tuple          =       image_embeds.shape[0] * num_images_per_prompt
				if isinstance(__lowerCAmelCase   ,		__lowerCAmelCase  ):
					_UpperCAmelCase     :				Tuple          =       torch.cat(__lowerCAmelCase   ,		dim=0  )

				if do_classifier_free_guidance:
					_UpperCAmelCase     :				Optional[int]          =       image_embeds.repeat_interleave(__lowerCAmelCase   ,		dim=0  )
					_UpperCAmelCase     :				Union[str, Any]          =       negative_image_embeds.repeat_interleave(__lowerCAmelCase   ,		dim=0  )

				_UpperCAmelCase     :				Optional[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     :				Dict          =       self.unet.config.in_channels

				_UpperCAmelCase    , _UpperCAmelCase     :				Optional[Any]          =       downscale_height_and_width(__lowerCAmelCase   ,		__lowerCAmelCase   ,		self.movq_scale_factor  )

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

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

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

				if output_type == "pil":
					_UpperCAmelCase     :				Optional[Any]          =       self.numpy_to_pil(__lowerCAmelCase  )

				if not return_dict:
					return (image,)

				return ImagePipelineOutput(images=__lowerCAmelCase  )












import numpy as np







def   __lowerCamelCase   (				__a  :np.array       )						->							np.array:






		"""simple docstring"""

		return 1 / (1 + np.exp(-vector       ))







def   __lowerCamelCase   (				__a  :np.array       )						->							np.array:






		"""simple docstring"""

		return vector * sigmoid(1.702 * vector       )


if __name__ == "__main__":
					import doctest

					doctest.testmod()



