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


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

UpperCAmelCase__ :       Optional[int]       				=					'''
    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 lowercase_    (  _snake_case  ,_snake_case  ,_snake_case=8		):
							SCREAMING_SNAKE_CASE__		:			Tuple	 =  height // scale_factor**2
							if height % scale_factor**2 != 0:
														new_height += 1
							SCREAMING_SNAKE_CASE__		:			Union[str, Any]	 =  width // scale_factor**2
							if width % scale_factor**2 != 0:
														new_width += 1
							return new_height * scale_factor, new_width * scale_factor



class       lowerCAmelCase_		(_snake_case    ):


				"""simple docstring"""



				def __init__(self  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__  ,			)   ->						Tuple:



											"""simple docstring"""




											super().__init__()

											self.register_modules(
											    unet=_lowerCamelCase  ,			scheduler=_lowerCamelCase  ,			movq=_lowerCamelCase  ,			)
											SCREAMING_SNAKE_CASE__		:			int	 =  2 ** (len(self.movq.config.block_out_channels       ) - 1)



				def       __magic_name__       (self  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__       )   ->						Union[str, Any]:



											"""simple docstring"""




											if latents is None:
																		SCREAMING_SNAKE_CASE__		:			Optional[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__		:			Dict	 =  latents.to(_lowerCamelCase       )

											SCREAMING_SNAKE_CASE__		:			Dict	 =  latents * scheduler.init_noise_sigma
											return latents



				def       __magic_name__       (self  ,			SCREAMING_SNAKE_CASE__=0       )   ->						Optional[int]:



											"""simple docstring"""




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

											SCREAMING_SNAKE_CASE__		:			List[str]	 =  torch.device(F'''cuda:{gpu_id}'''       )

											SCREAMING_SNAKE_CASE__		:			Union[str, Any]	 =  [
											    self.unet,
											    self.movq,
											]
											for cpu_offloaded_model in models:
																		if cpu_offloaded_model is not None:
																									cpu_offload(_lowerCamelCase  ,			_lowerCamelCase       )



				def       __magic_name__       (self  ,			SCREAMING_SNAKE_CASE__=0       )   ->						Dict:



											"""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__		:			str	 =  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__		:			Union[str, Any]	 =  None
											for cpu_offloaded_model in [self.unet, self.movq]:
																		SCREAMING_SNAKE_CASE__		:			Union[str, Any]	 =  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       __magic_name__       (self       )   ->						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  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__  ,			SCREAMING_SNAKE_CASE__ = 5_12  ,			SCREAMING_SNAKE_CASE__ = 5_12  ,			SCREAMING_SNAKE_CASE__ = 1_00  ,			SCREAMING_SNAKE_CASE__ = 4.0  ,			SCREAMING_SNAKE_CASE__ = 1  ,			SCREAMING_SNAKE_CASE__ = None  ,			SCREAMING_SNAKE_CASE__ = None  ,			SCREAMING_SNAKE_CASE__ = "pil"  ,			SCREAMING_SNAKE_CASE__ = True  ,			)   ->						Any:



											"""simple docstring"""




											SCREAMING_SNAKE_CASE__		:			Dict	 =  self._execution_device

											SCREAMING_SNAKE_CASE__		:			Any	 =  guidance_scale > 1.0

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

											if do_classifier_free_guidance:
																		SCREAMING_SNAKE_CASE__		:			Union[str, Any]	 =  image_embeds.repeat_interleave(_lowerCamelCase  ,			dim=0       )
																		SCREAMING_SNAKE_CASE__		:			Any	 =  negative_image_embeds.repeat_interleave(_lowerCamelCase  ,			dim=0       )

											SCREAMING_SNAKE_CASE__		:			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       )
											SCREAMING_SNAKE_CASE__		:			Union[str, Any]	 =  self.scheduler.timesteps

											SCREAMING_SNAKE_CASE__		:			Optional[int]	 =  self.unet.config.in_channels

											SCREAMING_SNAKE_CASE__		:			Dict	 =  downscale_height_and_width(_lowerCamelCase  ,			_lowerCamelCase  ,			self.movq_scale_factor       )

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

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

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

											if output_type == "pil":
																		SCREAMING_SNAKE_CASE__		:			Union[str, Any]	 =  self.numpy_to_pil(_lowerCamelCase       )

											if not return_dict:
																		return (image,)

											return ImagePipelineOutput(images=_lowerCamelCase       )








def 							__lowerCamelCase		(					UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						int		):




      """simple docstring"""





      while b:
            a				,							a		:int									=			b, a % b
      return a
def 							__lowerCamelCase		(					UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						int		):




      """simple docstring"""





      return a if b == 0 else euclidean_gcd_recursive(UpperCAmelCase_					,    a % b		)
def 							__lowerCamelCase		(					):




      """simple docstring"""





      print(F'''euclidean_gcd(3, 5) = {euclidean_gcd(3					,    5		)}'''		)
      print(F'''euclidean_gcd(5, 3) = {euclidean_gcd(5					,    3		)}'''		)
      print(F'''euclidean_gcd(1, 3) = {euclidean_gcd(1					,    3		)}'''		)
      print(F'''euclidean_gcd(3, 6) = {euclidean_gcd(3					,    6		)}'''		)
      print(F'''euclidean_gcd(6, 3) = {euclidean_gcd(6					,    3		)}'''		)

      print(F'''euclidean_gcd_recursive(3, 5) = {euclidean_gcd_recursive(3					,    5		)}'''		)
      print(F'''euclidean_gcd_recursive(5, 3) = {euclidean_gcd_recursive(5					,    3		)}'''		)
      print(F'''euclidean_gcd_recursive(1, 3) = {euclidean_gcd_recursive(1					,    3		)}'''		)
      print(F'''euclidean_gcd_recursive(3, 6) = {euclidean_gcd_recursive(3					,    6		)}'''		)
      print(F'''euclidean_gcd_recursive(6, 3) = {euclidean_gcd_recursive(6					,    3		)}'''		)


if __name__ == "__main__":
     main()





