



'''simple docstring'''




from typing import List, Optional, Union

import torch
from transformers import (
    XLMRobertaTokenizer,
)

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .text_encoder import MultilingualCLIP


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

a   :			Optional[Any]					  =    """
    Examples:
        ```py
        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline
        >>> import torch

        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained(\"kandinsky-community/Kandinsky-2-1-prior\")
        >>> pipe_prior.to(\"cuda\")

        >>> prompt = \"red cat, 4k photo\"
        >>> out = pipe_prior(prompt)
        >>> image_emb = out.image_embeds
        >>> negative_image_emb = out.negative_image_embeds

        >>> pipe = KandinskyPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-1\")
        >>> pipe.to(\"cuda\")

        >>> image = pipe(
        ...     prompt,
        ...     image_embeds=image_emb,
        ...     negative_image_embeds=negative_image_emb,
        ...     height=768,
        ...     width=768,
        ...     num_inference_steps=100,
        ... ).images

        >>> image[0].save(\"cat.png\")
        ```
"""



def       __lowerCamelCase  (					_lowercase	,				_lowercase	,				_lowercase=8		)      ->			List[str]:
							UpperCAmelCase     :		List[Any]							     =		h // scale_factor**2
							if h % scale_factor**2 != 0:
														new_h += 1
							UpperCAmelCase     :		int							     =		w // scale_factor**2
							if w % scale_factor**2 != 0:
														new_w += 1
							return new_h * scale_factor, new_w * scale_factor


class 			UpperCamelCase_ (							__magic_name__			):





	def __init__(  self		,		A		,		A		,		A		,		A		,		A		,		)			->			List[str]:
								super().__init__()

								self.register_modules(
								    text_encoder=A		,		tokenizer=A		,		unet=A		,		scheduler=A		,		movq=A		,		)
								UpperCAmelCase     :		Dict							     =		2 ** (len(self.movq.config.block_out_channels    ) - 1)





	def  _lowercase(  self		,		A		,		A		,		A		,		A		,		A		,		A    )			->			List[Any]:
								if latents is None:
															UpperCAmelCase     :		List[Any]							     =		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}'''    )
															UpperCAmelCase     :		List[str]							     =		latents.to(A    )

								UpperCAmelCase     :		Dict							     =		latents * scheduler.init_noise_sigma
								return latents





	def  _lowercase(  self		,		A		,		A		,		A		,		A		,		A=None		,		)			->			Union[str, Any]:
								UpperCAmelCase     :		Any							     =		len(A    ) if isinstance(A		,		A    ) else 1
								# get prompt text embeddings
								UpperCAmelCase     :		str							     =		self.tokenizer(
								    A		,		padding="""max_length"""		,		truncation=A		,		max_length=77		,		return_attention_mask=A		,		add_special_tokens=A		,		return_tensors="""pt"""		,		)

								UpperCAmelCase     :		List[str]							     =		text_inputs.input_ids
								UpperCAmelCase     :		Tuple							     =		self.tokenizer(A		,		padding="""longest"""		,		return_tensors="""pt"""    ).input_ids

								if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(A		,		A    ):
															UpperCAmelCase     :		Tuple							     =		self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]    )
															logger.warning(
															    """The following part of your input was truncated because CLIP can only handle sequences up to"""
															    f''' {self.tokenizer.model_max_length} tokens: {removed_text}'''    )

								UpperCAmelCase     :		Tuple							     =		text_input_ids.to(A    )
								UpperCAmelCase     :		Union[str, Any]							     =		text_inputs.attention_mask.to(A    )

								UpperCAmelCase     ,     UpperCAmelCase     :		Optional[int]							     =		self.text_encoder(
								    input_ids=A		,		attention_mask=A    )

								UpperCAmelCase     :		List[Any]							     =		prompt_embeds.repeat_interleave(A		,		dim=0    )
								UpperCAmelCase     :		int							     =		text_encoder_hidden_states.repeat_interleave(A		,		dim=0    )
								UpperCAmelCase     :		Tuple							     =		text_mask.repeat_interleave(A		,		dim=0    )

								if do_classifier_free_guidance:
															UpperCAmelCase     :		List[str]
															if negative_prompt is None:
																						UpperCAmelCase     :		Optional[Any]							     =		[""""""] * batch_size
															elif type(A    ) is not type(A    ):
																						raise TypeError(
																						    f'''`negative_prompt` should be the same type to `prompt`, but got {type(A    )} !='''
																						    f''' {type(A    )}.'''    )
															elif isinstance(A		,		A    ):
																						UpperCAmelCase     :		List[str]							     =		[negative_prompt]
															elif batch_size != len(A    ):
																						raise ValueError(
																						    f'''`negative_prompt`: {negative_prompt} has batch size {len(A    )}, but `prompt`:'''
																						    f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
																						    """ the batch size of `prompt`."""    )
															else:
																						UpperCAmelCase     :		Union[str, Any]							     =		negative_prompt

															UpperCAmelCase     :		Union[str, Any]							     =		self.tokenizer(
															    A		,		padding="""max_length"""		,		max_length=77		,		truncation=A		,		return_attention_mask=A		,		add_special_tokens=A		,		return_tensors="""pt"""		,		)
															UpperCAmelCase     :		Dict							     =		uncond_input.input_ids.to(A    )
															UpperCAmelCase     :		Union[str, Any]							     =		uncond_input.attention_mask.to(A    )

															UpperCAmelCase     ,     UpperCAmelCase     :		Dict							     =		self.text_encoder(
															    input_ids=A		,		attention_mask=A    )

															# duplicate unconditional embeddings for each generation per prompt, using mps friendly method

															UpperCAmelCase     :		Any							     =		negative_prompt_embeds.shape[1]
															UpperCAmelCase     :		List[Any]							     =		negative_prompt_embeds.repeat(1		,		A    )
															UpperCAmelCase     :		str							     =		negative_prompt_embeds.view(batch_size * num_images_per_prompt		,		A    )

															UpperCAmelCase     :		Union[str, Any]							     =		uncond_text_encoder_hidden_states.shape[1]
															UpperCAmelCase     :		int							     =		uncond_text_encoder_hidden_states.repeat(1		,		A		,		1    )
															UpperCAmelCase     :		List[Any]							     =		uncond_text_encoder_hidden_states.view(
															    batch_size * num_images_per_prompt		,		A		,		-1    )
															UpperCAmelCase     :		int							     =		uncond_text_mask.repeat_interleave(A		,		dim=0    )

															# done duplicates

															# For classifier free guidance, we need to do two forward passes.
															# Here we concatenate the unconditional and text embeddings into a single batch
															# to avoid doing two forward passes
															UpperCAmelCase     :		str							     =		torch.cat([negative_prompt_embeds, prompt_embeds]    )
															UpperCAmelCase     :		int							     =		torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]    )

															UpperCAmelCase     :		Dict							     =		torch.cat([uncond_text_mask, text_mask]    )

								return prompt_embeds, text_encoder_hidden_states, text_mask





	def  _lowercase(  self		,		A=0    )			->			Tuple:
								if is_accelerate_available():
															from accelerate import cpu_offload
								else:
															raise ImportError("""Please install accelerate via `pip install accelerate`"""    )

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

								UpperCAmelCase     :		Optional[Any]							     =		[
								    self.unet,
								    self.text_encoder,
								    self.movq,
								]
								for cpu_offloaded_model in models:
															if cpu_offloaded_model is not None:
																						cpu_offload(A		,		A    )





	def  _lowercase(  self		,		A=0    )			->			Union[str, Any]:
								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[str]							     =		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)

								UpperCAmelCase     :		List[str]							     =		None
								for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
															UpperCAmelCase     ,     UpperCAmelCase     :		Any							     =		cpu_offload_with_hook(A		,		A		,		prev_module_hook=A    )

								if self.safety_checker is not None:
															UpperCAmelCase     ,     UpperCAmelCase     :		Tuple							     =		cpu_offload_with_hook(self.safety_checker		,		A		,		prev_module_hook=A    )

								# We'll offload the last model manually.
								UpperCAmelCase     :		Optional[Any]							     =		hook





	@property
	# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
	def  _lowercase(  self    )			->			Optional[Any]:
								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		,		A = None		,		A = 512		,		A = 512		,		A = 100		,		A = 4.0		,		A = 1		,		A = None		,		A = None		,		A = "pil"		,		A = True		,		)			->			Dict:

								if isinstance(A		,		A    ):
															UpperCAmelCase     :		Optional[int]							     =		1
								elif isinstance(A		,		A    ):
															UpperCAmelCase     :		List[Any]							     =		len(A    )
								else:
															raise ValueError(f'''`prompt` has to be of type `str` or `list` but is {type(A    )}'''    )

								UpperCAmelCase     :		int							     =		self._execution_device

								UpperCAmelCase     :		Optional[Any]							     =		batch_size * num_images_per_prompt
								UpperCAmelCase     :		List[Any]							     =		guidance_scale > 1.0

								UpperCAmelCase     ,     UpperCAmelCase     ,     UpperCAmelCase     :		List[str]							     =		self._encode_prompt(
								    A		,		A		,		A		,		A		,		A    )

								if isinstance(A		,		A    ):
															UpperCAmelCase     :		str							     =		torch.cat(A		,		dim=0    )
								if isinstance(A		,		A    ):
															UpperCAmelCase     :		str							     =		torch.cat(A		,		dim=0    )

								if do_classifier_free_guidance:
															UpperCAmelCase     :		Optional[Any]							     =		image_embeds.repeat_interleave(A		,		dim=0    )
															UpperCAmelCase     :		List[Any]							     =		negative_image_embeds.repeat_interleave(A		,		dim=0    )

								UpperCAmelCase     :		Union[str, Any]							     =		torch.cat([negative_image_embeds, image_embeds]		,		dim=0    ).to(
								    dtype=prompt_embeds.dtype		,		device=A    )

								self.scheduler.set_timesteps(A		,		device=A    )
								UpperCAmelCase     :		Tuple							     =		self.scheduler.timesteps

								UpperCAmelCase     :		Union[str, Any]							     =		self.unet.config.in_channels

								UpperCAmelCase     ,     UpperCAmelCase     :		Any							     =		get_new_h_w(A		,		A		,		self.movq_scale_factor    )

								# create initial latent
								UpperCAmelCase     :		Any							     =		self.prepare_latents(
								    (batch_size, num_channels_latents, height, width)		,		text_encoder_hidden_states.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
															UpperCAmelCase     :		Union[str, Any]							     =		torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents

															UpperCAmelCase     :		List[str]							     =		{"""text_embeds""": prompt_embeds, """image_embeds""": image_embeds}
															UpperCAmelCase     :		Any							     =		self.unet(
															    sample=A		,		timestep=A		,		encoder_hidden_states=A		,		added_cond_kwargs=A		,		return_dict=A		,		)[0]

															if do_classifier_free_guidance:
																						UpperCAmelCase     ,     UpperCAmelCase     :		Union[str, Any]							     =		noise_pred.split(latents.shape[1]		,		dim=1    )
																						UpperCAmelCase     ,     UpperCAmelCase     :		Optional[int]							     =		noise_pred.chunk(2    )
																						UpperCAmelCase     ,     UpperCAmelCase     :		Any							     =		variance_pred.chunk(2    )
																						UpperCAmelCase     :		int							     =		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     :		Tuple							     =		noise_pred.split(latents.shape[1]		,		dim=1    )

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

								if output_type == "pil":
															UpperCAmelCase     :		Optional[int]							     =		self.numpy_to_pil(A    )

								if not return_dict:
															return (image,)

								return ImagePipelineOutput(images=A    )






'''simple docstring'''




def       __lowerCamelCase  (					_lowercase	,				_lowercase		)      ->			int:
							if b == 0:
														return 1
							if (b % 2) == 0:
														return actual_power(_lowercase	,				int(b / 2		)		) * actual_power(_lowercase	,				int(b / 2		)		)
							else:
														return a * actual_power(_lowercase	,				int(b / 2		)		) * actual_power(_lowercase	,				int(b / 2		)		)



def       __lowerCamelCase  (					_lowercase	,				_lowercase		)      ->			float:
							if b < 0:
														return 1 / actual_power(_lowercase	,				_lowercase		)
							return actual_power(_lowercase	,				_lowercase		)


if __name__ == "__main__":
							print(power(-2, -3))

