




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


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

a_        =       '\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained("kandinsky-community/kandinsky-2-2-prior")\n        >>> pipe_prior.to("cuda")\n        >>> prompt = "red cat, 4k photo"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> zero_image_emb = out.negative_image_embeds\n        >>> pipe = KandinskyV22Pipeline.from_pretrained("kandinsky-community/kandinsky-2-2-decoder")\n        >>> pipe.to("cuda")\n        >>> image = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=50,\n        ... ).images\n        >>> image[0].save("cat.png")\n        ```\n'







def      __UpperCAmelCase     ( __UpperCamelCase	,				__UpperCamelCase	,				__UpperCamelCase=8	):
			__lowercase			:  Optional[int]								=       height // scale_factor**2
			if height % scale_factor**2 != 0:
						new_height += 1
			__lowercase			:  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   UpperCAmelCase_    (			snake_case		):
		def __init__(						self				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						)						->       Union[str, Any]:
					super().__init__()

					self.register_modules(
					    unet=UpperCamelCase_				,						scheduler=UpperCamelCase_				,						movq=UpperCamelCase_				,						)
					__lowercase			:  List[str]								=       2 ** (len(self.movq.config.block_out_channels       ) - 1)
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_				,						UpperCamelCase_       )						->       List[Any]:
					if latents is None:
								__lowercase			:  Optional[int]								=       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}"""       )
								__lowercase			:  Optional[Any]								=       latents.to(UpperCamelCase_       )

					__lowercase			:  Optional[Any]								=       latents * scheduler.init_noise_sigma
					return latents
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_=0       )						->       Optional[int]:
					if is_accelerate_available():
								from accelerate import cpu_offload
					else:
								raise ImportError('''Please install accelerate via `pip install accelerate`'''       )

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

					__lowercase			:  Any								=       [
					    self.unet,
					    self.movq,
					]
					for cpu_offloaded_model in models:
								if cpu_offloaded_model is not None:
											cpu_offload(UpperCamelCase_				,						UpperCamelCase_       )
		def 	_lowerCamelCase							(						self				,						UpperCamelCase_=0       )						->       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.'''       )

					__lowercase			:  str								=       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)

					__lowercase			:  Optional[int]								=       None
					for cpu_offloaded_model in [self.unet, self.movq]:
								__lowercase			,__lowercase			:  Any								=       cpu_offload_with_hook(UpperCamelCase_				,						UpperCamelCase_				,						prev_module_hook=UpperCamelCase_       )

					# We'll offload the last model manually.
					__lowercase			:  List[str]								=       hook
		@property
		# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
		def 	_lowerCamelCase							(						self       )						->       List[Any]:
					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				,						)						->       Union[str, Any]:
					__lowercase			:  Optional[int]								=       self._execution_device

					__lowercase			:  int								=       guidance_scale > 1.0

					if isinstance(UpperCamelCase_				,						UpperCamelCase_       ):
								__lowercase			:  Tuple								=       torch.cat(UpperCamelCase_				,						dim=0       )
					__lowercase			:  List[str]								=       image_embeds.shape[0] * num_images_per_prompt
					if isinstance(UpperCamelCase_				,						UpperCamelCase_       ):
								__lowercase			:  Any								=       torch.cat(UpperCamelCase_				,						dim=0       )

					if do_classifier_free_guidance:
								__lowercase			:  Any								=       image_embeds.repeat_interleave(UpperCamelCase_				,						dim=0       )
								__lowercase			:  List[str]								=       negative_image_embeds.repeat_interleave(UpperCamelCase_				,						dim=0       )

					__lowercase			:  List[str]								=       torch.cat([negative_image_embeds, image_embeds]				,						dim=0       ).to(dtype=self.unet.dtype				,						device=UpperCamelCase_       )

					self.scheduler.set_timesteps(UpperCamelCase_				,						device=UpperCamelCase_       )
					__lowercase			:  Dict								=       self.scheduler.timesteps

					__lowercase			:  List[Any]								=       self.unet.config.in_channels

					__lowercase			,__lowercase			:  Any								=       downscale_height_and_width(UpperCamelCase_				,						UpperCamelCase_				,						self.movq_scale_factor       )

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

								__lowercase			:  List[Any]								=       {'''image_embeds''': image_embeds}
								__lowercase			:  Dict								=       self.unet(
								    sample=UpperCamelCase_				,						timestep=UpperCamelCase_				,						encoder_hidden_states=UpperCamelCase_				,						added_cond_kwargs=UpperCamelCase_				,						return_dict=UpperCamelCase_				,						)[0]

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

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

					if output_type == "pil":
								__lowercase			:  Optional[Any]								=       self.numpy_to_pil(UpperCamelCase_       )

					if not return_dict:
								return (image,)

					return ImagePipelineOutput(images=UpperCamelCase_       )




'''simple docstring'''




from collections import OrderedDict
from typing import Mapping

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig


__lowercase        =    {
    '''albert-base-v1''': '''https://huggingface.co/albert-base-v1/resolve/main/config.json''',
    '''albert-large-v1''': '''https://huggingface.co/albert-large-v1/resolve/main/config.json''',
    '''albert-xlarge-v1''': '''https://huggingface.co/albert-xlarge-v1/resolve/main/config.json''',
    '''albert-xxlarge-v1''': '''https://huggingface.co/albert-xxlarge-v1/resolve/main/config.json''',
    '''albert-base-v2''': '''https://huggingface.co/albert-base-v2/resolve/main/config.json''',
    '''albert-large-v2''': '''https://huggingface.co/albert-large-v2/resolve/main/config.json''',
    '''albert-xlarge-v2''': '''https://huggingface.co/albert-xlarge-v2/resolve/main/config.json''',
    '''albert-xxlarge-v2''': '''https://huggingface.co/albert-xxlarge-v2/resolve/main/config.json''',
}






class 							a__(   lowerCAmelCase__      ):


			'''simple docstring'''






			UpperCAmelCase_     :	List[str]						=			'''albert'''



			def __init__(  self			,       __lowerCAmelCase=30000			,       __lowerCAmelCase=128			,       __lowerCAmelCase=4096			,       __lowerCAmelCase=12			,       __lowerCAmelCase=1			,       __lowerCAmelCase=64			,       __lowerCAmelCase=16384			,       __lowerCAmelCase=1			,       __lowerCAmelCase="gelu_new"			,       __lowerCAmelCase=0			,       __lowerCAmelCase=0			,       __lowerCAmelCase=512			,       __lowerCAmelCase=2			,       __lowerCAmelCase=0.02			,       __lowerCAmelCase=1E-1_2			,       __lowerCAmelCase=0.1			,       __lowerCAmelCase="absolute"			,       __lowerCAmelCase=0			,       __lowerCAmelCase=2			,       __lowerCAmelCase=3			,       **__lowerCAmelCase			,       ):




					"""simple docstring"""


					super().__init__(pad_token_id=__lowerCAmelCase			,       bos_token_id=__lowerCAmelCase			,       eos_token_id=__lowerCAmelCase			,       **__lowerCAmelCase)

					lowerCAmelCase     =    vocab_size
					lowerCAmelCase     =    embedding_size
					lowerCAmelCase     =    hidden_size
					lowerCAmelCase     =    num_hidden_layers
					lowerCAmelCase     =    num_hidden_groups
					lowerCAmelCase     =    num_attention_heads
					lowerCAmelCase     =    inner_group_num
					lowerCAmelCase     =    hidden_act
					lowerCAmelCase     =    intermediate_size
					lowerCAmelCase     =    hidden_dropout_prob
					lowerCAmelCase     =    attention_probs_dropout_prob
					lowerCAmelCase     =    max_position_embeddings
					lowerCAmelCase     =    type_vocab_size
					lowerCAmelCase     =    initializer_range
					lowerCAmelCase     =    layer_norm_eps
					lowerCAmelCase     =    classifier_dropout_prob
					lowerCAmelCase     =    position_embedding_type







class 							a__(   lowerCAmelCase__      ):


			'''simple docstring'''






			@property
			def 				a_			(  self):




					"""simple docstring"""


					if self.task == "multiple-choice":
							lowerCAmelCase     =    {0: """batch""", 1: """choice""", 2: """sequence"""}
					else:
							lowerCAmelCase     =    {0: """batch""", 1: """sequence"""}
					return OrderedDict(
					    [
					        ("""input_ids""", dynamic_axis),
					        ("""attention_mask""", dynamic_axis),
					        ("""token_type_ids""", dynamic_axis),
					    ])




