


from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

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


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

lowerCAmelCase	=							'\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Img2ImgPipeline, KandinskyV22PriorPipeline\n        >>> from diffusers.utils import load_image\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16\n        ... )\n        >>> pipe_prior.to("cuda")\n\n        >>> prompt = "A red cartoon frog, 4k"\n        >>> image_emb, zero_image_emb = pipe_prior(prompt, return_dict=False)\n\n        >>> pipe = KandinskyV22Img2ImgPipeline.from_pretrained(\n        ...     "kandinsky-community/kandinsky-2-2-decoder", torch_dtype=torch.float16\n        ... )\n        >>> pipe.to("cuda")\n\n        >>> init_image = load_image(\n        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"\n        ...     "/kandinsky/frog.png"\n        ... )\n\n        >>> image = pipe(\n        ...     image=init_image,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ...     strength=0.2,\n        ... ).images\n\n        >>> image[0].save("red_frog.png")\n        ```\n'
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE=8					):


		"""simple docstring"""




		lowercase__						=				height // scale_factor**2
		if height % scale_factor**2 != 0:
				new_height += 1
		lowercase__						=				width // scale_factor**2
		if width % scale_factor**2 != 0:
				new_width += 1
		return new_height * scale_factor, new_width * scale_factor
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE=5_12						,    SCREAMING_SNAKE_CASE=5_12					):


		"""simple docstring"""




		lowercase__						=				pil_image.resize((w, h)						,    resample=Image.BICUBIC						,    reducing_gap=1					)
		lowercase__						=				np.array(pil_image.convert('''RGB'''					)					)
		lowercase__						=				arr.astype(np.floataa					) / 127.5 - 1
		lowercase__						=				np.transpose(SCREAMING_SNAKE_CASE						,    [2, 0, 1]					)
		lowercase__						=				torch.from_numpy(SCREAMING_SNAKE_CASE					).unsqueeze(0					)
		return image


class 	_a							(      UpperCamelCase__      ):

							def __init__(							self:     List[Any]   ,						UpperCamelCase_:     UNetaDConditionModel   ,						UpperCamelCase_:     DDPMScheduler   ,						UpperCamelCase_:     VQModel   ,						)      ->       int:





									"""simple docstring"""







									super().__init__()

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

							def 				lowerCamelCase_			(							self:     Optional[Any]   ,						UpperCamelCase_:     str   ,						UpperCamelCase_:     Union[str, Any]   ,						UpperCamelCase_:     int							)      ->       int:





									"""simple docstring"""







									lowercase__						=				min(int(num_inference_steps * strength							)   ,						UpperCamelCase_							)

									lowercase__						=				max(num_inference_steps - init_timestep   ,						0							)
									lowercase__						=				self.scheduler.timesteps[t_start:]

									return timesteps, num_inference_steps - t_start

							def 				lowerCamelCase_			(							self:     Any   ,						UpperCamelCase_:     Union[str, Any]   ,						UpperCamelCase_:     Union[str, Any]   ,						UpperCamelCase_:     Optional[Any]   ,						UpperCamelCase_:     Optional[int]   ,						UpperCamelCase_:     Dict   ,						UpperCamelCase_:     int   ,						UpperCamelCase_:     Dict=None							)      ->       Optional[Any]:





									"""simple docstring"""







									if not isinstance(UpperCamelCase_   ,						(torch.Tensor, PIL.Image.Image, list)							):
											raise ValueError(
											    f'`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(UpperCamelCase_							)}'							)

									lowercase__						=				image.to(device=UpperCamelCase_   ,						dtype=UpperCamelCase_							)

									lowercase__						=				batch_size * num_images_per_prompt

									if image.shape[1] == 4:
											lowercase__						=				image

									else:
											if isinstance(UpperCamelCase_   ,						UpperCamelCase_							) and len(UpperCamelCase_							) != batch_size:
													raise ValueError(
													    f'You have passed a list of generators of length {len(UpperCamelCase_							)}, but requested an effective batch'
													    f' size of {batch_size}. Make sure the batch size matches the length of the generators.'							)

											elif isinstance(UpperCamelCase_   ,						UpperCamelCase_							):
													lowercase__						=				[
													    self.movq.encode(image[i : i + 1]							).latent_dist.sample(generator[i]							) for i in range(UpperCamelCase_							)
													]
													lowercase__						=				torch.cat(UpperCamelCase_   ,						dim=0							)
											else:
													lowercase__						=				self.movq.encode(UpperCamelCase_							).latent_dist.sample(UpperCamelCase_							)

											lowercase__						=				self.movq.config.scaling_factor * init_latents

									lowercase__						=				torch.cat([init_latents]   ,						dim=0							)

									lowercase__						=				init_latents.shape
									lowercase__						=				randn_tensor(UpperCamelCase_   ,						generator=UpperCamelCase_   ,						device=UpperCamelCase_   ,						dtype=UpperCamelCase_							)

									# get latents
									lowercase__						=				self.scheduler.add_noise(UpperCamelCase_   ,						UpperCamelCase_   ,						UpperCamelCase_							)

									lowercase__						=				init_latents

									return latents

							def 				lowerCamelCase_			(							self:     str   ,						UpperCamelCase_:     List[Any]=0							)      ->       Optional[Any]:





									"""simple docstring"""







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

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

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

							def 				lowerCamelCase_			(							self:     int   ,						UpperCamelCase_:     Optional[Any]=0							)      ->       int:





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

									lowercase__						=				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__						=				None
									for cpu_offloaded_model in [self.unet, self.movq]:
											lowercase__					,		lowercase__						=				cpu_offload_with_hook(UpperCamelCase_   ,						UpperCamelCase_   ,						prev_module_hook=UpperCamelCase_							)

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

							@property
							# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
							def 				lowerCamelCase_			(							self:     List[str]							)      ->       str:





									"""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:     List[Any]   ,						UpperCamelCase_:     Union[torch.FloatTensor, List[torch.FloatTensor]]   ,						UpperCamelCase_:     Union[torch.FloatTensor, PIL.Image.Image, List[torch.FloatTensor], List[PIL.Image.Image]]   ,						UpperCamelCase_:     Union[torch.FloatTensor, List[torch.FloatTensor]]   ,						UpperCamelCase_:     int = 512   ,						UpperCamelCase_:     int = 512   ,						UpperCamelCase_:     int = 100   ,						UpperCamelCase_:     float = 4.0   ,						UpperCamelCase_:     float = 0.3   ,						UpperCamelCase_:     int = 1   ,						UpperCamelCase_:     Optional[Union[torch.Generator, List[torch.Generator]]] = None   ,						UpperCamelCase_:     Optional[str] = "pil"   ,						UpperCamelCase_:     bool = True   ,						)      ->       Optional[int]:





									"""simple docstring"""







									lowercase__						=				self._execution_device

									lowercase__						=				guidance_scale > 1.0

									if isinstance(UpperCamelCase_   ,						UpperCamelCase_							):
											lowercase__						=				torch.cat(UpperCamelCase_   ,						dim=0							)
									lowercase__						=				image_embeds.shape[0]
									if isinstance(UpperCamelCase_   ,						UpperCamelCase_							):
											lowercase__						=				torch.cat(UpperCamelCase_   ,						dim=0							)

									if do_classifier_free_guidance:
											lowercase__						=				image_embeds.repeat_interleave(UpperCamelCase_   ,						dim=0							)
											lowercase__						=				negative_image_embeds.repeat_interleave(UpperCamelCase_   ,						dim=0							)

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

									if not isinstance(UpperCamelCase_   ,						UpperCamelCase_							):
											lowercase__						=				[image]
									if not all(isinstance(UpperCamelCase_   ,						(PIL.Image.Image, torch.Tensor)							) for i in image							):
											raise ValueError(
											    f'Input is in incorrect format: {[type(UpperCamelCase_							) for i in image]}. Currently, we only support  PIL image and pytorch tensor'							)

									lowercase__						=				torch.cat([prepare_image(UpperCamelCase_   ,						UpperCamelCase_   ,						UpperCamelCase_							) for i in image]   ,						dim=0							)
									lowercase__						=				image.to(dtype=image_embeds.dtype   ,						device=UpperCamelCase_							)

									lowercase__						=				self.movq.encode(UpperCamelCase_							)['''latents''']
									lowercase__						=				latents.repeat_interleave(UpperCamelCase_   ,						dim=0							)
									self.scheduler.set_timesteps(UpperCamelCase_   ,						device=UpperCamelCase_							)
									lowercase__					,		lowercase__						=				self.get_timesteps(UpperCamelCase_   ,						UpperCamelCase_   ,						UpperCamelCase_							)
									lowercase__						=				timesteps[:1].repeat(batch_size * num_images_per_prompt							)
									lowercase__					,		lowercase__						=				downscale_height_and_width(UpperCamelCase_   ,						UpperCamelCase_   ,						self.movq_scale_factor							)
									lowercase__						=				self.prepare_latents(
									    UpperCamelCase_   ,						UpperCamelCase_   ,						UpperCamelCase_   ,						UpperCamelCase_   ,						image_embeds.dtype   ,						UpperCamelCase_   ,						UpperCamelCase_							)
									for i, t in enumerate(self.progress_bar(UpperCamelCase_							)							):
											# expand the latents if we are doing classifier free guidance
											lowercase__						=				torch.cat([latents] * 2							) if do_classifier_free_guidance else latents

											lowercase__						=				{'''image_embeds''': image_embeds}
											lowercase__						=				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__						=				noise_pred.split(latents.shape[1]   ,						dim=1							)
													lowercase__					,		lowercase__						=				noise_pred.chunk(2							)
													lowercase__					,		lowercase__						=				variance_pred.chunk(2							)
													lowercase__						=				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
													lowercase__						=				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__						=				noise_pred.split(latents.shape[1]   ,						dim=1							)

											# compute the previous noisy sample x_t -> x_t-1
											lowercase__						=				self.scheduler.step(
											    UpperCamelCase_   ,						UpperCamelCase_   ,						UpperCamelCase_   ,						generator=UpperCamelCase_   ,						)[0]

									# post-processing
									lowercase__						=				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__						=				image * 0.5 + 0.5
											lowercase__						=				image.clamp(0   ,						1							)
											lowercase__						=				image.cpu().permute(0   ,						2   ,						3   ,						1							).float().numpy()

									if output_type == "pil":
											lowercase__						=				self.numpy_to_pil(UpperCamelCase_							)

									if not return_dict:
											return (image,)

									return ImagePipelineOutput(images=UpperCamelCase_							)






from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCAmelCase	=							logging.get_logger(__name__)

lowerCAmelCase	=							{
    'unc-nlp/lxmert-base-uncased': 'https://huggingface.co/unc-nlp/lxmert-base-uncased/resolve/main/config.json',
}


class 	_a							(      UpperCamelCase__      ):
							_lowercase     :      List[Any]  				=			'''lxmert'''
							_lowercase     :      Any  				=			{}

							def __init__(							self:     Any   ,						UpperCamelCase_:     List[Any]=30_522   ,						UpperCamelCase_:     int=768   ,						UpperCamelCase_:     Optional[Any]=12   ,						UpperCamelCase_:     Dict=9_500   ,						UpperCamelCase_:     List[Any]=1_600   ,						UpperCamelCase_:     List[Any]=400   ,						UpperCamelCase_:     List[str]=3_072   ,						UpperCamelCase_:     int="gelu"   ,						UpperCamelCase_:     Any=0.1   ,						UpperCamelCase_:     int=0.1   ,						UpperCamelCase_:     List[str]=512   ,						UpperCamelCase_:     Optional[int]=2   ,						UpperCamelCase_:     Union[str, Any]=0.02   ,						UpperCamelCase_:     Dict=1E-1_2   ,						UpperCamelCase_:     List[Any]=9   ,						UpperCamelCase_:     List[Any]=5   ,						UpperCamelCase_:     Optional[Any]=5   ,						UpperCamelCase_:     str=2_048   ,						UpperCamelCase_:     Dict=4   ,						UpperCamelCase_:     Any=6.67   ,						UpperCamelCase_:     Dict=True   ,						UpperCamelCase_:     List[str]=True   ,						UpperCamelCase_:     Dict=True   ,						UpperCamelCase_:     Any=True   ,						UpperCamelCase_:     Tuple=True   ,						UpperCamelCase_:     Optional[Any]=True   ,						UpperCamelCase_:     Any=True   ,						**UpperCamelCase_:     Union[str, Any]   ,						)      ->       Union[str, Any]:





									"""simple docstring"""







									lowercase__						=				vocab_size
									lowercase__						=				hidden_size
									lowercase__						=				num_attention_heads
									lowercase__						=				hidden_act
									lowercase__						=				intermediate_size
									lowercase__						=				hidden_dropout_prob
									lowercase__						=				attention_probs_dropout_prob
									lowercase__						=				max_position_embeddings
									lowercase__						=				type_vocab_size
									lowercase__						=				initializer_range
									lowercase__						=				layer_norm_eps
									lowercase__						=				num_qa_labels
									lowercase__						=				num_object_labels
									lowercase__						=				num_attr_labels
									lowercase__						=				l_layers
									lowercase__						=				x_layers
									lowercase__						=				r_layers
									lowercase__						=				visual_feat_dim
									lowercase__						=				visual_pos_dim
									lowercase__						=				visual_loss_normalizer
									lowercase__						=				task_matched
									lowercase__						=				task_mask_lm
									lowercase__						=				task_obj_predict
									lowercase__						=				task_qa
									lowercase__						=				visual_obj_loss
									lowercase__						=				visual_attr_loss
									lowercase__						=				visual_feat_loss
									lowercase__						=				{'''vision''': r_layers, '''cross_encoder''': x_layers, '''language''': l_layers}
									super().__init__(**UpperCamelCase_							)


