



"""simple docstring"""

from typing import Callable, List, Optional, Tuple, Union

import torch
from transformers import CLIPTextModel, CLIPTokenizer

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin, TransformeraDModel, VQModel
from ...schedulers import VQDiffusionScheduler
from ...utils import logging
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


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







class    __lowercase     (					_UpperCAmelCase					,    _UpperCAmelCase):







						"""simple docstring"""







						@register_to_config
						def __init__(self   ,		lowercase__   ,		lowercase__ = None   ,		lowercase__ = None       ):
											super().__init__()

											snake_case_					:		Tuple 					=					learnable

											if self.learnable:
																assert hidden_size is not None, "learnable=True requires `hidden_size` to be set"
																assert length is not None, "learnable=True requires `length` to be set"

																snake_case_					:		Any 					=					torch.zeros(lowercase__   ,		lowercase__       )
											else:
																snake_case_					:		Optional[int] 					=					None

											snake_case_					:		Dict 					=					torch.nn.Parameter(lowercase__       )







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						_A		:							VQModel
						_A		:							CLIPTextModel
						_A		:							CLIPTokenizer
						_A		:							TransformeraDModel
						_A		:							LearnedClassifierFreeSamplingEmbeddings
						_A		:							VQDiffusionScheduler







						def __init__(self   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		):
											super().__init__()

											self.register_modules(
											    vqvae=lowercase__   ,		transformer=lowercase__   ,		text_encoder=lowercase__   ,		tokenizer=lowercase__   ,		scheduler=lowercase__   ,		learned_classifier_free_sampling_embeddings=lowercase__   ,		)







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__   ,		lowercase__       ):
											snake_case_					:		Dict 					=					len(lowercase__       ) if isinstance(lowercase__   ,		lowercase__       ) else 1

											# get prompt text embeddings
											snake_case_					:		Union[str, Any] 					=					self.tokenizer(
											    lowercase__   ,		padding="""max_length"""   ,		max_length=self.tokenizer.model_max_length   ,		return_tensors="""pt"""   ,		)
											snake_case_					:		Any 					=					text_inputs.input_ids

											if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
																snake_case_					:		Optional[int] 					=					self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]       )
																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_					:		Union[str, Any] 					=					text_input_ids[:, : self.tokenizer.model_max_length]
											snake_case_					:		int 					=					self.text_encoder(text_input_ids.to(self.device       )       )[0]

											# NOTE: This additional step of normalizing the text embeddings is from VQ-Diffusion.
											# While CLIP does normalize the pooled output of the text transformer when combining
											# the image and text embeddings, CLIP does not directly normalize the last hidden state.
											#
											# CLIP normalizing the pooled output.
											# https://github.com/huggingface/transformers/blob/d92e22d1f28324f513f3080e5c47c071a3916721/src/transformers/models/clip/modeling_clip.py#L1052-L1053
											snake_case_					:		List[str] 					=					prompt_embeds / prompt_embeds.norm(dim=-1   ,		keepdim=lowercase__       )

											# duplicate text embeddings for each generation per prompt
											snake_case_					:		Tuple 					=					prompt_embeds.repeat_interleave(lowercase__   ,		dim=0       )

											if do_classifier_free_guidance:
																if self.learned_classifier_free_sampling_embeddings.learnable:
																					snake_case_					:		Any 					=					self.learned_classifier_free_sampling_embeddings.embeddings
																					snake_case_					:		Union[str, Any] 					=					negative_prompt_embeds.unsqueeze(0       ).repeat(lowercase__   ,		1   ,		1       )
																else:
																					snake_case_					:		Tuple 					=					[""""""] * batch_size

																					snake_case_					:		Dict 					=					text_input_ids.shape[-1]
																					snake_case_					:		Tuple 					=					self.tokenizer(
																					    lowercase__   ,		padding="""max_length"""   ,		max_length=lowercase__   ,		truncation=lowercase__   ,		return_tensors="""pt"""   ,		)
																					snake_case_					:		Optional[int] 					=					self.text_encoder(uncond_input.input_ids.to(self.device       )       )[0]
																					# See comment for normalizing text embeddings
																					snake_case_					:		int 					=					negative_prompt_embeds / negative_prompt_embeds.norm(dim=-1   ,		keepdim=lowercase__       )

																# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
																snake_case_					:		Optional[Any] 					=					negative_prompt_embeds.shape[1]
																snake_case_					:		Optional[int] 					=					negative_prompt_embeds.repeat(1   ,		lowercase__   ,		1       )
																snake_case_					:		int 					=					negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,		lowercase__   ,		-1       )

																# 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_					:		Optional[Any] 					=					torch.cat([negative_prompt_embeds, prompt_embeds]       )

											return prompt_embeds







						@torch.no_grad()
						def __call__(self   ,		lowercase__   ,		lowercase__ = 1_00   ,		lowercase__ = 5.0   ,		lowercase__ = 1.0   ,		lowercase__ = 1   ,		lowercase__ = None   ,		lowercase__ = None   ,		lowercase__ = "pil"   ,		lowercase__ = True   ,		lowercase__ = None   ,		lowercase__ = 1   ,		):
											if isinstance(lowercase__   ,		lowercase__       ):
																snake_case_					:		List[Any] 					=					1
											elif isinstance(lowercase__   ,		lowercase__       ):
																snake_case_					:		Optional[int] 					=					len(lowercase__       )
											else:
																raise ValueError(f'`prompt` has to be of type `str` or `list` but is {type(lowercase__       )}'       )

											snake_case_					:		int 					=					batch_size * num_images_per_prompt

											snake_case_					:		Tuple 					=					guidance_scale > 1.0

											snake_case_					:		List[Any] 					=					self._encode_prompt(lowercase__   ,		lowercase__   ,		lowercase__       )

											if (callback_steps is None) or (
											    callback_steps is not None and (not isinstance(lowercase__   ,		lowercase__       ) or callback_steps <= 0)
											):
																raise ValueError(
																    f'`callback_steps` has to be a positive integer but is {callback_steps} of type'
																    f' {type(lowercase__       )}.'       )

											# get the initial completely masked latents unless the user supplied it

											snake_case_					:		int 					=					(batch_size, self.transformer.num_latent_pixels)
											if latents is None:
																snake_case_					:		Dict 					=					self.transformer.num_vector_embeds - 1
																snake_case_					:		Optional[int] 					=					torch.full(lowercase__   ,		lowercase__       ).to(self.device       )
											else:
																if latents.shape != latents_shape:
																					raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {latents_shape}'       )
																if (latents < 0).any() or (latents >= self.transformer.num_vector_embeds).any():
																					raise ValueError(
																					    """Unexpected latents value(s). All latents be valid embedding indices i.e. in the range 0,"""
																					    f' {self.transformer.num_vector_embeds - 1} (inclusive).'       )
																snake_case_					:		Tuple 					=					latents.to(self.device       )

											# set timesteps
											self.scheduler.set_timesteps(lowercase__   ,		device=self.device       )

											snake_case_					:		Optional[Any] 					=					self.scheduler.timesteps.to(self.device       )

											snake_case_					:		List[str] 					=					latents

											for i, t in enumerate(self.progress_bar(lowercase__       )       ):
																# expand the sample if we are doing classifier free guidance
																snake_case_					:		Any 					=					torch.cat([sample] * 2       ) if do_classifier_free_guidance else sample

																# predict the un-noised image
																# model_output == `log_p_x_0`
																snake_case_					:		int 					=					self.transformer(lowercase__   ,		encoder_hidden_states=lowercase__   ,		timestep=lowercase__       ).sample

																if do_classifier_free_guidance:
																					snake_case_					,			snake_case_					:		Union[str, Any] 					=					model_output.chunk(2       )
																					snake_case_					:		Optional[int] 					=					model_output_uncond + guidance_scale * (model_output_text - model_output_uncond)
																					model_output -= torch.logsumexp(lowercase__   ,		dim=1   ,		keepdim=lowercase__       )

																snake_case_					:		str 					=					self.truncate(lowercase__   ,		lowercase__       )

																# remove `log(0)`'s (`-inf`s)
																snake_case_					:		Union[str, Any] 					=					model_output.clamp(-70       )

																# compute the previous noisy sample x_t -> x_t-1
																snake_case_					:		Dict 					=					self.scheduler.step(lowercase__   ,		timestep=lowercase__   ,		sample=lowercase__   ,		generator=lowercase__       ).prev_sample

																# call the callback, if provided
																if callback is not None and i % callback_steps == 0:
																					callback(lowercase__   ,		lowercase__   ,		lowercase__       )

											snake_case_					:		Dict 					=					self.vqvae.config.vq_embed_dim
											snake_case_					:		Optional[int] 					=					(batch_size, self.transformer.height, self.transformer.width, embedding_channels)
											snake_case_					:		List[Any] 					=					self.vqvae.quantize.get_codebook_entry(lowercase__   ,		shape=lowercase__       )
											snake_case_					:		Dict 					=					self.vqvae.decode(lowercase__   ,		force_not_quantize=lowercase__       ).sample

											snake_case_					:		Union[str, Any] 					=					(image / 2 + 0.5).clamp(0   ,		1       )
											snake_case_					:		List[Any] 					=					image.cpu().permute(0   ,		2   ,		3   ,		1       ).numpy()

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

											if not return_dict:
																return (image,)

											return ImagePipelineOutput(images=lowercase__       )







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__       ):
											snake_case_					,			snake_case_					:		List[Any] 					=					torch.sort(lowercase__   ,		1   ,		descending=lowercase__       )
											snake_case_					:		Optional[Any] 					=					torch.exp(lowercase__       )
											snake_case_					:		int 					=					sorted_p_x_0.cumsum(dim=1       ) < truncation_rate

											# Ensure that at least the largest probability is not zeroed out
											snake_case_					:		Optional[int] 					=					torch.full_like(keep_mask[:, 0:1, :]   ,		lowercase__       )
											snake_case_					:		int 					=					torch.cat((all_true, keep_mask)   ,		dim=1       )
											snake_case_					:		List[Any] 					=					keep_mask[:, :-1, :]

											snake_case_					:		List[Any] 					=					keep_mask.gather(1   ,		indices.argsort(1       )       )

											snake_case_					:		Optional[int] 					=					log_p_x_0.clone()

											snake_case_					:		Dict 					=					-torch.inf  # -inf = log(0)

											return rv









"""simple docstring"""

def 							SCREAMING_SNAKE_CASE__      (		SCREAMING_SNAKE_CASE__ :  int				,    SCREAMING_SNAKE_CASE__ :  int			):



					"""simple docstring"""





					return numa ^ numa < 0


if __name__ == "__main__":
			import doctest

			doctest.testmod()




