






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


lowercase	:				Optional[int]      	=     logging.get_logger(__name__)  # pylint: disable=invalid-name

lowercase	:				Union[str, Any]      	=     "\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained(\"kandinsky-community/Kandinsky-2-1-prior\")\n        >>> pipe_prior.to(\"cuda\")\n\n        >>> prompt = \"red cat, 4k photo\"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> negative_image_emb = out.negative_image_embeds\n\n        >>> pipe = KandinskyPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-1\")\n        >>> pipe.to(\"cuda\")\n\n        >>> image = pipe(\n        ...     prompt,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=negative_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ... ).images\n\n        >>> image[0].save(\"cat.png\")\n        ```\n"

def   lowerCAmelCase__						(     _a							:      Union[str, Any]	,   _a							:      Optional[int]	,   _a							:      Union[str, Any]=8     ):
		snake_case_					:				Union[str, Any]    			=							h // scale_factor**2
		if h % scale_factor**2 != 0:
				new_h += 1
		snake_case_					:				List[Any]    			=							w // scale_factor**2
		if w % scale_factor**2 != 0:
				new_w += 1
		return new_h * scale_factor, new_w * scale_factor




class 		UpperCAmelCase_    (						lowerCAmelCase__       ):


						'''simple docstring'''




						def __init__(						self		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					)			->					int:
								super().__init__()

								self.register_modules(
								    text_encoder=__a		,					tokenizer=__a		,					unet=__a		,					scheduler=__a		,					movq=__a		,					)
								snake_case_					:				Optional[int]    			=							2 ** (len(self.movq.config.block_out_channels					) - 1)






						def    _lowerCAmelCase							(						self		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE					)			->					Tuple:
								if latents is None:
										snake_case_					:				str    			=							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}'''					)
										snake_case_					:				Any    			=							latents.to(__a					)

								snake_case_					:				Union[str, Any]    			=							latents * scheduler.init_noise_sigma
								return latents






						def    _lowerCAmelCase							(						self		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE=None		,					)			->					Optional[int]:
								snake_case_					:				Optional[int]    			=							len(__a					) if isinstance(__a		,					__a					) else 1
								# get prompt text embeddings
								snake_case_					:				Optional[int]    			=							self.tokenizer(
								    __a		,					padding="max_length"		,					truncation=__a		,					max_length=77		,					return_attention_mask=__a		,					add_special_tokens=__a		,					return_tensors="pt"		,					)

								snake_case_					:				Optional[int]    			=							text_inputs.input_ids
								snake_case_					:				Optional[int]    			=							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					):
										snake_case_					:				Optional[int]    			=							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}'''					)

								snake_case_					:				List[str]    			=							text_input_ids.to(__a					)
								snake_case_					:				str    			=							text_inputs.attention_mask.to(__a					)

								snake_case_			,							snake_case_					:				Optional[int]    			=							self.text_encoder(
								    input_ids=__a		,					attention_mask=__a					)

								snake_case_					:				List[Any]    			=							prompt_embeds.repeat_interleave(__a		,					dim=0					)
								snake_case_					:				int    			=							text_encoder_hidden_states.repeat_interleave(__a		,					dim=0					)
								snake_case_					:				List[str]    			=							text_mask.repeat_interleave(__a		,					dim=0					)

								if do_classifier_free_guidance:
										snake_case_					:				Optional[int]    			=							42
										if negative_prompt is None:
												snake_case_					:				Optional[int]    			=							[""] * 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					):
												snake_case_					:				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:
												snake_case_					:				Optional[Any]    			=							negative_prompt

										snake_case_					:				Tuple    			=							self.tokenizer(
										    __a		,					padding="max_length"		,					max_length=77		,					truncation=__a		,					return_attention_mask=__a		,					add_special_tokens=__a		,					return_tensors="pt"		,					)
										snake_case_					:				List[str]    			=							uncond_input.input_ids.to(__a					)
										snake_case_					:				Any    			=							uncond_input.attention_mask.to(__a					)

										snake_case_			,							snake_case_					:				Any    			=							self.text_encoder(
										    input_ids=__a		,					attention_mask=__a					)

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

										snake_case_					:				Dict    			=							negative_prompt_embeds.shape[1]
										snake_case_					:				str    			=							negative_prompt_embeds.repeat(1		,					__a					)
										snake_case_					:				List[str]    			=							negative_prompt_embeds.view(batch_size * num_images_per_prompt		,					__a					)

										snake_case_					:				Union[str, Any]    			=							uncond_text_encoder_hidden_states.shape[1]
										snake_case_					:				Tuple    			=							uncond_text_encoder_hidden_states.repeat(1		,					__a		,					1					)
										snake_case_					:				Any    			=							uncond_text_encoder_hidden_states.view(
										    batch_size * num_images_per_prompt		,					__a		,					-1					)
										snake_case_					:				Dict    			=							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
										snake_case_					:				Union[str, Any]    			=							torch.cat([negative_prompt_embeds, prompt_embeds]					)
										snake_case_					:				Dict    			=							torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]					)

										snake_case_					:				int    			=							torch.cat([uncond_text_mask, text_mask]					)

								return prompt_embeds, text_encoder_hidden_states, text_mask






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

								snake_case_					:				Any    			=							torch.device(f'''cuda:{gpu_id}'''					)

								snake_case_					:				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    _lowerCAmelCase							(						self		,					_SCREAMING_SNAKE_CASE=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."					)

								snake_case_					:				Optional[Any]    			=							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)

								snake_case_					:				str    			=							None
								for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
										snake_case_			,							snake_case_					:				Dict    			=							cpu_offload_with_hook(__a		,					__a		,					prev_module_hook=__a					)

								if self.safety_checker is not None:
										snake_case_			,							snake_case_					:				List[str]    			=							cpu_offload_with_hook(self.safety_checker		,					__a		,					prev_module_hook=__a					)

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






						@property
						# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
						def    _lowerCAmelCase							(						self					)			->					Dict:
								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		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE		,					_SCREAMING_SNAKE_CASE = None		,					_SCREAMING_SNAKE_CASE = 512		,					_SCREAMING_SNAKE_CASE = 512		,					_SCREAMING_SNAKE_CASE = 100		,					_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		,					)			->					Optional[int]:

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

								snake_case_					:				Tuple    			=							self._execution_device

								snake_case_					:				str    			=							batch_size * num_images_per_prompt
								snake_case_					:				int    			=							guidance_scale > 1.0

								snake_case_			,							snake_case_			,							snake_case_					:				Optional[Any]    			=							self._encode_prompt(
								    __a		,					__a		,					__a		,					__a		,					__a					)

								if isinstance(__a		,					__a					):
										snake_case_					:				Optional[Any]    			=							torch.cat(__a		,					dim=0					)
								if isinstance(__a		,					__a					):
										snake_case_					:				List[Any]    			=							torch.cat(__a		,					dim=0					)

								if do_classifier_free_guidance:
										snake_case_					:				List[str]    			=							image_embeds.repeat_interleave(__a		,					dim=0					)
										snake_case_					:				Union[str, Any]    			=							negative_image_embeds.repeat_interleave(__a		,					dim=0					)

								snake_case_					:				List[str]    			=							torch.cat([negative_image_embeds, image_embeds]		,					dim=0					).to(
								    dtype=prompt_embeds.dtype		,					device=__a					)

								self.scheduler.set_timesteps(__a		,					device=__a					)
								snake_case_					:				Dict    			=							self.scheduler.timesteps

								snake_case_					:				Tuple    			=							self.unet.config.in_channels

								snake_case_			,							snake_case_					:				Tuple    			=							get_new_h_w(__a		,					__a		,					self.movq_scale_factor					)

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

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

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

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

								if output_type == "pil":
										snake_case_					:				Tuple    			=							self.numpy_to_pil(__a					)

								if not return_dict:
										return (image,)

								return ImagePipelineOutput(images=__a					)




"""simple docstring"""


import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import ClassLabel, Features, Image
from .base import TaskTemplate
@dataclass(frozen=lowerCAmelCase__						)
class     _UpperCamelCase    (  lowerCAmelCase__						):
							'''simple docstring'''







							__UpperCAmelCase		:       str							 =field(default="""image-classification"""			,metadata={"""include_in_asdict_even_if_is_default""": True}						)
							__UpperCAmelCase		:       ClassVar[Features]							 =Features({"""image""": Image()}						)
							__UpperCAmelCase		:       ClassVar[Features]							 =Features({"""labels""": ClassLabel}						)
							__UpperCAmelCase		:       str							 ="image"
							__UpperCAmelCase		:       str							 ="labels"

							def      snake_case							(	self       ,  __a      ):
										if self.label_column not in features:
													raise ValueError(f"Column {self.label_column} is not present in features."      )
										if not isinstance(features[self.label_column]       ,  __a      ):
													raise ValueError(f"Column {self.label_column} is not a ClassLabel."      )
										__lowerCAmelCase							       =     copy.deepcopy(self      )
										__lowerCAmelCase							       =     self.label_schema.copy()
										__lowerCAmelCase							       =     features[self.label_column]
										__lowerCAmelCase							       =     label_schema
										return task_template



							@property
							def      snake_case							(	self      ):
										return {
										    self.image_column: "image",
										    self.label_column: "labels",
										}




