






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


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

__snake_case      :      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				(      UpperCamelCase_    :   Optional[int]      ,    UpperCamelCase_    :   Optional[int]      ,    UpperCamelCase_    :   Union[str, Any]=8						)					->				Union[str, Any]:

						"""simple docstring"""



						lowerCAmelCase__        =  height // scale_factor**2
						if height % scale_factor**2 != 0:
												new_height += 1
						lowerCAmelCase__        =  width // scale_factor**2
						if width % scale_factor**2 != 0:
												new_width += 1
						return new_height * scale_factor, new_width * scale_factor


class       __SCREAMING_SNAKE_CASE   (				__lowercase):






						def __init__(		self       ,  _UpperCamelCase       ,  _UpperCamelCase       ,  _UpperCamelCase       ,  ):






												"""simple docstring"""
												super().__init__()

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






						def  UpperCamelCase__ (		self       ,  _UpperCamelCase       ,  _UpperCamelCase       ,  _UpperCamelCase       ,  _UpperCamelCase       ,  _UpperCamelCase       ,  _UpperCamelCase				):






												"""simple docstring"""
												if latents is None:
																		lowerCAmelCase__        =  randn_tensor(_UpperCamelCase       ,  generator=_UpperCamelCase       ,  device=_UpperCamelCase       ,  dtype=_UpperCamelCase				)
												else:
																		if latents.shape != shape:
																								raise ValueError(F"Unexpected latents shape, got {latents.shape}, expected {shape}"				)
																		lowerCAmelCase__        =  latents.to(_UpperCamelCase				)

												lowerCAmelCase__        =  latents * scheduler.init_noise_sigma
												return latents






						def  UpperCamelCase__ (		self       ,  _UpperCamelCase=0				):






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

												lowerCAmelCase__        =  torch.device(F"cuda:{gpu_id}"				)

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






						def  UpperCamelCase__ (		self       ,  _UpperCamelCase=0				):






												"""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.'				)

												lowerCAmelCase__        =  torch.device(F"cuda:{gpu_id}"				)

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

												lowerCAmelCase__        =  None
												for cpu_offloaded_model in [self.unet, self.movq]:
																		lowerCAmelCase__  ,    lowerCAmelCase__        =  cpu_offload_with_hook(_UpperCamelCase       ,  _UpperCamelCase       ,  prev_module_hook=_UpperCamelCase				)

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






						@property
						# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
						def  UpperCamelCase__ (		self				):






												"""simple docstring"""
												if not hasattr(self.unet       ,  '_hf_hook'				):
																		return self.device
												for module in self.unet.modules():
																		if (
																		    hasattr(_UpperCamelCase       ,  '_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(_UpperCamelCase				)
						def __call__(		self       ,  _UpperCamelCase       ,  _UpperCamelCase       ,  _UpperCamelCase = 5_12       ,  _UpperCamelCase = 5_12       ,  _UpperCamelCase = 1_00       ,  _UpperCamelCase = 4.0       ,  _UpperCamelCase = 1       ,  _UpperCamelCase = None       ,  _UpperCamelCase = None       ,  _UpperCamelCase = "pil"       ,  _UpperCamelCase = True       ,  ):






												"""simple docstring"""
												lowerCAmelCase__        =  self._execution_device

												lowerCAmelCase__        =  guidance_scale > 1.0

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

												if do_classifier_free_guidance:
																		lowerCAmelCase__        =  image_embeds.repeat_interleave(_UpperCamelCase       ,  dim=0				)
																		lowerCAmelCase__        =  negative_image_embeds.repeat_interleave(_UpperCamelCase       ,  dim=0				)

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

												self.scheduler.set_timesteps(_UpperCamelCase       ,  device=_UpperCamelCase				)
												lowerCAmelCase__        =  self.scheduler.timesteps

												lowerCAmelCase__        =  self.unet.config.in_channels

												lowerCAmelCase__  ,    lowerCAmelCase__        =  downscale_height_and_width(_UpperCamelCase       ,  _UpperCamelCase       ,  self.movq_scale_factor				)

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

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

																		lowerCAmelCase__        =  {'image_embeds': image_embeds}
																		lowerCAmelCase__        =  self.unet(
																		    sample=_UpperCamelCase       ,  timestep=_UpperCamelCase       ,  encoder_hidden_states=_UpperCamelCase       ,  added_cond_kwargs=_UpperCamelCase       ,  return_dict=_UpperCamelCase       ,  )[0]

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

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

												if output_type == "pil":
																		lowerCAmelCase__        =  self.numpy_to_pil(_UpperCamelCase				)

												if not return_dict:
																		return (image,)

												return ImagePipelineOutput(images=_UpperCamelCase				)













import colorsys

from PIL import Image  # type: ignore







def 		_UpperCamelCase				(      UpperCamelCase_    :   float      ,    UpperCamelCase_    :   float      ,    UpperCamelCase_    :   int						)					->				float:

						"""simple docstring"""



						lowerCAmelCase__        =  x
						lowerCAmelCase__        =  y
						for step in range(UpperCamelCase_						):  # noqa: B007
												lowerCAmelCase__        =  a * a - b * b + x
												lowerCAmelCase__        =  2 * a * b + y
												lowerCAmelCase__        =  a_new

												# divergence happens for all complex number with an absolute value
												# greater than 4
												if a * a + b * b > 4:
																		break
						return step / (max_step - 1)







def 		_UpperCamelCase				(      UpperCamelCase_    :   float						)					->				tuple:

						"""simple docstring"""



						if distance == 1:
												return (0, 0, 0)
						else:
												return (255, 255, 255)







def 		_UpperCamelCase				(      UpperCamelCase_    :   float						)					->				tuple:

						"""simple docstring"""



						if distance == 1:
												return (0, 0, 0)
						else:
												return tuple(round(i * 255						) for i in colorsys.hsv_to_rgb(UpperCamelCase_      ,    1      ,    1						)						)







def 		_UpperCamelCase				(      UpperCamelCase_    :   int = 800      ,    UpperCamelCase_    :   int = 600      ,    UpperCamelCase_    :   float = -0.6      ,    UpperCamelCase_    :   float = 0      ,    UpperCamelCase_    :   float = 3.2      ,    UpperCamelCase_    :   int = 50      ,    UpperCamelCase_    :   bool = True      ,    )					->				Image.Image:

						"""simple docstring"""



						lowerCAmelCase__        =  Image.new('RGB'      ,    (image_width, image_height)						)
						lowerCAmelCase__        =  img.load()

						# loop through the image-coordinates
						for image_x in range(UpperCamelCase_						):
												for image_y in range(UpperCamelCase_						):
																		# determine the figure-coordinates based on the image-coordinates
																		lowerCAmelCase__        =  figure_width / image_width * image_height
																		lowerCAmelCase__        =  figure_center_x + (image_x / image_width - 0.5) * figure_width
																		lowerCAmelCase__        =  figure_center_y + (image_y / image_height - 0.5) * figure_height

																		lowerCAmelCase__        =  get_distance(UpperCamelCase_      ,    UpperCamelCase_      ,    UpperCamelCase_						)

																		# color the corresponding pixel based on the selected coloring-function
																		if use_distance_color_coding:
																								lowerCAmelCase__        =  get_color_coded_rgb(UpperCamelCase_						)
																		else:
																								lowerCAmelCase__        =  get_black_and_white_rgb(UpperCamelCase_						)

						return img


if __name__ == "__main__":
					import doctest

					doctest.testmod()

					# colored version, full figure
					__snake_case      :      str             =       get_image()

					# uncomment for colored version, different section, zoomed in
					# img = get_image(figure_center_x = -0.6, figure_center_y = -0.4,
					# figure_width = 0.8)

					# uncomment for black and white version, full figure
					# img = get_image(use_distance_color_coding = False)

					# uncomment to save the image
					# img.save("mandelbrot.png")

					img.show()





