


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							:	Dict					      =				logging.get_logger(__name__)  # pylint: disable=invalid-name

snake_case							:	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 							__lowercase       (			__lowerCAmelCase							:    Optional[int] ,   __lowerCAmelCase							:    Union[str, Any] ,   __lowerCAmelCase							:    List[Any]=8       ):
							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 	snake_case_       (lowerCamelCase_				):
					def __init__(				self       :List[str]      ,__snake_case       :UNetaDConditionModel      ,__snake_case       :DDPMScheduler      ,__snake_case       :VQModel      ,) ->   Any:
												super().__init__()

												self.register_modules(
												    unet=__snake_case      ,scheduler=__snake_case      ,movq=__snake_case      ,)
												a__					=					2 ** (len(self.movq.config.block_out_channels				) - 1)
					def 			lowerCamelCase__(				self       :Tuple      ,__snake_case       :Any      ,__snake_case       :Optional[Any]      ,__snake_case       :Any      ,__snake_case       :str      ,__snake_case       :int      ,__snake_case       :Any				) ->   Optional[Any]:
												if latents is None:
																			a__					=					randn_tensor(__snake_case      ,generator=__snake_case      ,device=__snake_case      ,dtype=__snake_case				)
												else:
																			if latents.shape != shape:
																										raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {shape}'				)
																			a__					=					latents.to(__snake_case				)

												a__					=					latents * scheduler.init_noise_sigma
												return latents
					def 			lowerCamelCase__(				self       :Tuple      ,__snake_case       :int=0				) ->   int:
												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(__snake_case      ,__snake_case				)
					def 			lowerCamelCase__(				self       :Any      ,__snake_case       :Optional[Any]=0				) ->   int:
												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=__snake_case				)
																			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(__snake_case      ,__snake_case      ,prev_module_hook=__snake_case				)

												# We'll offload the last model manually.
												a__					=					hook
					@property
					# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
					def 			lowerCamelCase__(				self       :Union[str, Any]				) ->   int:
												if not hasattr(self.unet      ,'_hf_hook'				):
																			return self.device
												for module in self.unet.modules():
																			if (
																			    hasattr(__snake_case      ,'_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(__snake_case				)
					def __call__(				self       :Union[str, Any]      ,__snake_case       :Union[torch.FloatTensor, List[torch.FloatTensor]]      ,__snake_case       :Union[torch.FloatTensor, List[torch.FloatTensor]]      ,__snake_case       :int = 5_12      ,__snake_case       :int = 5_12      ,__snake_case       :int = 1_00      ,__snake_case       :float = 4.0      ,__snake_case       :int = 1      ,__snake_case       :Optional[Union[torch.Generator, List[torch.Generator]]] = None      ,__snake_case       :Optional[torch.FloatTensor] = None      ,__snake_case       :Optional[str] = "pil"      ,__snake_case       :bool = True      ,) ->   Optional[int]:
												a__					=					self._execution_device

												a__					=					guidance_scale > 1.0

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

												if do_classifier_free_guidance:
																			a__					=					image_embeds.repeat_interleave(__snake_case      ,dim=0				)
																			a__					=					negative_image_embeds.repeat_interleave(__snake_case      ,dim=0				)

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

												self.scheduler.set_timesteps(__snake_case      ,device=__snake_case				)
												a__					=					self.scheduler.timesteps

												a__					=					self.unet.config.in_channels

												a__				, a__					=					downscale_height_and_width(__snake_case      ,__snake_case      ,self.movq_scale_factor				)

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

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

												if not return_dict:
																			return (image,)

												return ImagePipelineOutput(images=__snake_case				)








from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import rescale, resize, to_channel_dimension_format
from ...image_utils import (
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
						import PIL


snake_case							:	str					      =				logging.get_logger(__name__)






def 							__lowercase       (			__lowerCAmelCase							:    Union[str, Any] ,   __lowerCAmelCase							:    Optional[Any]       ):
							a__					=					b.T
							a__					=					np.sum(np.square(__lowerCAmelCase       ) ,   axis=1       )
							a__					=					np.sum(np.square(__lowerCAmelCase       ) ,   axis=0       )
							a__					=					np.matmul(__lowerCAmelCase ,   __lowerCAmelCase       )
							a__					=					aa[:, None] - 2 * ab + ba[None, :]
							return d






def 							__lowercase       (			__lowerCAmelCase							:    List[Any] ,   __lowerCAmelCase							:    str       ):
							a__					=					x.reshape(-1 ,   3       )
							a__					=					squared_euclidean_distance(__lowerCAmelCase ,   __lowerCAmelCase       )
							return np.argmin(__lowerCAmelCase ,   axis=1       )







class 	snake_case_       (lowerCamelCase_				):
					UpperCAmelCase__ :   Optional[Any]   				=      ['''pixel_values''']
					def __init__(				self       :Dict      ,__snake_case       :Optional[Union[List[List[int]], np.ndarray]] = None      ,__snake_case       :bool = True      ,__snake_case       :Dict[str, int] = None      ,__snake_case       :PILImageResampling = PILImageResampling.BILINEAR      ,__snake_case       :bool = True      ,__snake_case       :bool = True      ,**__snake_case       :Optional[int]      ,) ->   None:
												super().__init__(**__snake_case				)
												a__					=					size if size is not None else {'height': 2_56, 'width': 2_56}
												a__					=					get_size_dict(__snake_case				)
												a__					=					np.array(__snake_case				) if clusters is not None else None
												a__					=					do_resize
												a__					=					size
												a__					=					resample
												a__					=					do_normalize
												a__					=					do_color_quantize
					def 			lowerCamelCase__(				self       :Union[str, Any]      ,__snake_case       :np.ndarray      ,__snake_case       :Dict[str, int]      ,__snake_case       :PILImageResampling = PILImageResampling.BILINEAR      ,__snake_case       :Optional[Union[str, ChannelDimension]] = None      ,**__snake_case       :Any      ,) ->   np.ndarray:
												a__					=					get_size_dict(__snake_case				)
												if "height" not in size or "width" not in size:
																			raise ValueError(F'Size dictionary must contain both height and width keys. Got {size.keys()}'				)
												return resize(
												    __snake_case      ,size=(size['height'], size['width'])      ,resample=__snake_case      ,data_format=__snake_case      ,**__snake_case				)
					def 			lowerCamelCase__(				self       :List[Any]      ,__snake_case       :np.ndarray      ,__snake_case       :Optional[Union[str, ChannelDimension]] = None      ,) ->   np.ndarray:
												a__					=					rescale(image=__snake_case      ,scale=1 / 1_27.5      ,data_format=__snake_case				)
												a__					=					image - 1
												return image



					def 			lowerCamelCase__(				self       :Optional[Any]      ,__snake_case       :ImageInput      ,__snake_case       :bool = None      ,__snake_case       :Dict[str, int] = None      ,__snake_case       :PILImageResampling = None      ,__snake_case       :bool = None      ,__snake_case       :Optional[bool] = None      ,__snake_case       :Optional[Union[List[List[int]], np.ndarray]] = None      ,__snake_case       :Optional[Union[str, TensorType]] = None      ,__snake_case       :Optional[Union[str, ChannelDimension]] = ChannelDimension.FIRST      ,**__snake_case       :List[str]      ,) ->   PIL.Image.Image:
												a__					=					do_resize if do_resize is not None else self.do_resize
												a__					=					size if size is not None else self.size
												a__					=					get_size_dict(__snake_case				)
												a__					=					resample if resample is not None else self.resample
												a__					=					do_normalize if do_normalize is not None else self.do_normalize
												a__					=					do_color_quantize if do_color_quantize is not None else self.do_color_quantize
												a__					=					clusters if clusters is not None else self.clusters
												a__					=					np.array(__snake_case				)

												a__					=					make_list_of_images(__snake_case				)

												if not valid_images(__snake_case				):
																			raise ValueError(
																			    'Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '
																			    'torch.Tensor, tf.Tensor or jax.ndarray.'				)

												if do_resize and size is None or resample is None:
																			raise ValueError('Size and resample must be specified if do_resize is True.'				)

												if do_color_quantize and clusters is None:
																			raise ValueError('Clusters must be specified if do_color_quantize is True.'				)

												# All transformations expect numpy arrays.
												a__					=					[to_numpy_array(__snake_case				) for image in images]

												if do_resize:
																			a__					=					[self.resize(image=__snake_case      ,size=__snake_case      ,resample=__snake_case				) for image in images]

												if do_normalize:
																			a__					=					[self.normalize(image=__snake_case				) for image in images]

												if do_color_quantize:
																			a__					=					[to_channel_dimension_format(__snake_case      ,ChannelDimension.LAST				) for image in images]
																			# color quantize from (batch_size, height, width, 3) to (batch_size, height, width)
																			a__					=					np.array(__snake_case				)
																			a__					=					color_quantize(__snake_case      ,__snake_case				).reshape(images.shape[:-1]				)

																			# flatten to (batch_size, height*width)
																			a__					=					images.shape[0]
																			a__					=					images.reshape(__snake_case      ,-1				)

																			# We need to convert back to a list of images to keep consistent behaviour across processors.
																			a__					=					list(__snake_case				)
												else:
																			a__					=					[to_channel_dimension_format(__snake_case      ,__snake_case				) for image in images]

												a__					=					{'input_ids': images}
												return BatchFeature(data=__snake_case      ,tensor_type=__snake_case				)




