



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


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



class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__  ,				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		@register_to_config
		def __init__(					self							,    _lowercase							,    _lowercase = None							,    _lowercase = None   ) ->						List[str]:







								'''simple docstring'''

								super().__init__()

								snake_case_   : int						=			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_   : Union[str, Any]						=			torch.zeros(_lowercase							,    _lowercase   )
								else:
														snake_case_   : Tuple						=			None

								snake_case_   : Optional[Any]						=			torch.nn.Parameter(_lowercase   )




class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       42
		_lowerCamelCase			  =       42
		_lowerCamelCase			  =       42
		_lowerCamelCase			  =       42
		_lowerCamelCase			  =       42
		_lowerCamelCase			  =       42





		def __init__(					self							,    _lowercase							,    _lowercase							,    _lowercase							,    _lowercase							,    _lowercase							,    _lowercase							,    ) ->						Optional[Any]:







								'''simple docstring'''

								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   ) ->						int:







								'''simple docstring'''

								snake_case_   : Union[str, Any]						=			len(_lowercase   ) if isinstance(_lowercase							,    _lowercase   ) else 1

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

								if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
														snake_case_   : Any						=			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_   : Optional[int]						=			text_input_ids[:, : self.tokenizer.model_max_length]
								snake_case_   : Any						=			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_   : Union[str, Any]						=			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_   : Optional[Any]						=			self.learned_classifier_free_sampling_embeddings.embeddings
																				snake_case_   : Any						=			negative_prompt_embeds.unsqueeze(0   ).repeat(_lowercase							,    1							,    1   )
														else:
																				snake_case_   : Union[str, Any]						=			[""""""] * batch_size

																				snake_case_   : Tuple						=			text_input_ids.shape[-1]
																				snake_case_   : Dict						=			self.tokenizer(
																				    _lowercase							,    padding="""max_length"""							,    max_length=_lowercase							,    truncation=_lowercase							,    return_tensors="""pt"""							,    )
																				snake_case_   : Dict						=			self.text_encoder(uncond_input.input_ids.to(self.device   )   )[0]
																				# See comment for normalizing text embeddings
																				snake_case_   : Optional[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_   : List[Any]						=			negative_prompt_embeds.shape[1]
														snake_case_   : 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_   : Tuple						=			torch.cat([negative_prompt_embeds, prompt_embeds]   )

								return prompt_embeds





		@torch.no_grad()
		def __call__(					self							,    _lowercase							,    _lowercase = 1_0_0							,    _lowercase = 5.0							,    _lowercase = 1.0							,    _lowercase = 1							,    _lowercase = None							,    _lowercase = None							,    _lowercase = "pil"							,    _lowercase = True							,    _lowercase = None							,    _lowercase = 1							,    ) ->						Union[ImagePipelineOutput, Tuple]:







								'''simple docstring'''

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

								snake_case_   : List[str]						=			batch_size * num_images_per_prompt

								snake_case_   : List[str]						=			guidance_scale > 1.0

								snake_case_   : Tuple						=			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_   : Optional[Any]						=			(batch_size, self.transformer.num_latent_pixels)
								if latents is None:
														snake_case_   : Optional[Any]						=			self.transformer.num_vector_embeds - 1
														snake_case_   : List[str]						=			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_   : Optional[int]						=			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_   : int						=			latents

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

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

														if do_classifier_free_guidance:
																				snake_case_			,		snake_case_   : Any						=			model_output.chunk(2   )
																				snake_case_   : Any						=			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(-7_0   )

														# compute the previous noisy sample x_t -> x_t-1
														snake_case_   : List[str]						=			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_   : str						=			self.vqvae.config.vq_embed_dim
								snake_case_   : Optional[Any]						=			(batch_size, self.transformer.height, self.transformer.width, embedding_channels)
								snake_case_   : Optional[int]						=			self.vqvae.quantize.get_codebook_entry(_lowercase							,    shape=_lowercase   )
								snake_case_   : int						=			self.vqvae.decode(_lowercase							,    force_not_quantize=_lowercase   ).sample

								snake_case_   : int						=			(image / 2 + 0.5).clamp(0							,    1   )
								snake_case_   : int						=			image.cpu().permute(0							,    2							,    3							,    1   ).numpy()

								if output_type == "pil":
														snake_case_   : Any						=			self.numpy_to_pil(_lowercase   )

								if not return_dict:
														return (image,)

								return ImagePipelineOutput(images=_lowercase   )





		def    UpperCAmelCase__							(					self							,    _lowercase							,    _lowercase   ) ->						torch.FloatTensor:







								'''simple docstring'''

								snake_case_			,		snake_case_   : List[Any]						=			torch.sort(_lowercase							,    1							,    descending=_lowercase   )
								snake_case_   : Optional[Any]						=			torch.exp(_lowercase   )
								snake_case_   : Optional[Any]						=			sorted_p_x_0.cumsum(dim=1   ) < truncation_rate

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

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

								snake_case_   : int						=			log_p_x_0.clone()

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

								return rv






"""simple docstring"""





from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ...utils.backbone_utils import BackboneConfigMixin, get_aligned_output_features_output_indices


__lowerCAmelCase       :					str      					=					logging.get_logger(__name__)

__lowerCAmelCase       :					Tuple      					=					{
    '''shi-labs/nat-mini-in1k-224''': '''https://huggingface.co/shi-labs/nat-mini-in1k-224/resolve/main/config.json''',
    # See all Nat models at https://huggingface.co/models?filter=nat
}



class    _lowerCAmelCase    (				SCREAMING_SNAKE_CASE__  ,				SCREAMING_SNAKE_CASE__		):





		"""simple docstring"""






		_lowerCamelCase			  =       '''nat'''

		_lowerCamelCase			  =       {
		    '''num_attention_heads''': '''num_heads''',
		    '''num_hidden_layers''': '''num_layers''',
		}





		def __init__(					self							,    _lowercase=4							,    _lowercase=3							,    _lowercase=6_4							,    _lowercase=[3, 4, 6, 5]							,    _lowercase=[2, 4, 8, 1_6]							,    _lowercase=7							,    _lowercase=3.0							,    _lowercase=True							,    _lowercase=0.0							,    _lowercase=0.0							,    _lowercase=0.1							,    _lowercase="gelu"							,    _lowercase=0.02							,    _lowercase=1E-5							,    _lowercase=0.0							,    _lowercase=None							,    _lowercase=None							,    **_lowercase							,    ) ->						Optional[Any]:







								'''simple docstring'''

								super().__init__(**_lowercase   )

								snake_case_   : Any						=			patch_size
								snake_case_   : Optional[Any]						=			num_channels
								snake_case_   : Optional[Any]						=			embed_dim
								snake_case_   : Tuple						=			depths
								snake_case_   : int						=			len(_lowercase   )
								snake_case_   : Optional[int]						=			num_heads
								snake_case_   : List[str]						=			kernel_size
								snake_case_   : str						=			mlp_ratio
								snake_case_   : str						=			qkv_bias
								snake_case_   : str						=			hidden_dropout_prob
								snake_case_   : Tuple						=			attention_probs_dropout_prob
								snake_case_   : Tuple						=			drop_path_rate
								snake_case_   : Dict						=			hidden_act
								snake_case_   : Union[str, Any]						=			layer_norm_eps
								snake_case_   : Tuple						=			initializer_range
								# we set the hidden_size attribute in order to make Nat work with VisionEncoderDecoderModel
								# this indicates the channel dimension after the last stage of the model
								snake_case_   : Union[str, Any]						=			int(embed_dim * 2 ** (len(_lowercase   ) - 1)   )
								snake_case_   : Union[str, Any]						=			layer_scale_init_value
								snake_case_   : Optional[Any]						=			["""stem"""] + [f'stage{idx}' for idx in range(1							,    len(_lowercase   ) + 1   )]
								snake_case_			,		snake_case_   : Any						=			get_aligned_output_features_output_indices(
								    out_features=_lowercase							,    out_indices=_lowercase							,    stage_names=self.stage_names   )

