


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


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





class     lowercase__  (						_snake_case     ,     _snake_case					):





							'''simple docstring'''


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

											_SCREAMING_SNAKE_CASE :       str     		=      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"

															_SCREAMING_SNAKE_CASE :       int     		=      torch.zeros(__snake_case    ,					__snake_case  )
											else:
															_SCREAMING_SNAKE_CASE :       int     		=      None

											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      torch.nn.Parameter(__snake_case  )





class     lowercase__  (						_snake_case					):





							'''simple docstring'''


							A_							:      VQModel
							A_							:      CLIPTextModel
							A_							:      CLIPTokenizer
							A_							:      TransformeraDModel
							A_							:      LearnedClassifierFreeSamplingEmbeddings
							A_							:      VQDiffusionScheduler






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

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






							def UpperCAmelCase_	(    self    ,					__snake_case    ,					__snake_case    ,					__snake_case  ):
											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      len(__snake_case  ) if isinstance(__snake_case    ,					__snake_case  ) else 1

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

											if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
															_SCREAMING_SNAKE_CASE :       Union[str, 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}"""  )
															_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      text_input_ids[:, : self.tokenizer.model_max_length]
											_SCREAMING_SNAKE_CASE :       Dict     		=      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
											_SCREAMING_SNAKE_CASE :       List[str]     		=      prompt_embeds / prompt_embeds.norm(dim=-1    ,					keepdim=__snake_case  )

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

											if do_classifier_free_guidance:
															if self.learned_classifier_free_sampling_embeddings.learnable:
																			_SCREAMING_SNAKE_CASE :       int     		=      self.learned_classifier_free_sampling_embeddings.embeddings
																			_SCREAMING_SNAKE_CASE :       Optional[int]     		=      negative_prompt_embeds.unsqueeze(0  ).repeat(__snake_case    ,					1    ,					1  )
															else:
																			_SCREAMING_SNAKE_CASE :       Tuple     		=      [""""""] * batch_size

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

															# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
															_SCREAMING_SNAKE_CASE :       Dict     		=      negative_prompt_embeds.shape[1]
															_SCREAMING_SNAKE_CASE :       int     		=      negative_prompt_embeds.repeat(1    ,					__snake_case    ,					1  )
															_SCREAMING_SNAKE_CASE :       Optional[int]     		=      negative_prompt_embeds.view(batch_size * num_images_per_prompt    ,					__snake_case    ,					-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
															_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      torch.cat([negative_prompt_embeds, prompt_embeds]  )

											return prompt_embeds






							@torch.no_grad()
							def __call__(    self    ,					__snake_case    ,					__snake_case = 100    ,					__snake_case = 5.0    ,					__snake_case = 1.0    ,					__snake_case = 1    ,					__snake_case = None    ,					__snake_case = None    ,					__snake_case = "pil"    ,					__snake_case = True    ,					__snake_case = None    ,					__snake_case = 1    ,					):
											if isinstance(__snake_case    ,					__snake_case  ):
															_SCREAMING_SNAKE_CASE :       int     		=      1
											elif isinstance(__snake_case    ,					__snake_case  ):
															_SCREAMING_SNAKE_CASE :       int     		=      len(__snake_case  )
											else:
															raise ValueError(f"""`prompt` has to be of type `str` or `list` but is {type(__snake_case  )}"""  )

											_SCREAMING_SNAKE_CASE :       Dict     		=      batch_size * num_images_per_prompt

											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      guidance_scale > 1.0

											_SCREAMING_SNAKE_CASE :       Any     		=      self._encode_prompt(__snake_case    ,					__snake_case    ,					__snake_case  )

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

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

											_SCREAMING_SNAKE_CASE :       Tuple     		=      (batch_size, self.transformer.num_latent_pixels)
											if latents is None:
															_SCREAMING_SNAKE_CASE :       Any     		=      self.transformer.num_vector_embeds - 1
															_SCREAMING_SNAKE_CASE :       Any     		=      torch.full(__snake_case    ,					__snake_case  ).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)."""  )
															_SCREAMING_SNAKE_CASE :       str     		=      latents.to(self.device  )

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

											_SCREAMING_SNAKE_CASE :       Dict     		=      self.scheduler.timesteps.to(self.device  )

											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      latents

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

															# predict the un-noised image
															# model_output == `log_p_x_0`
															_SCREAMING_SNAKE_CASE :       List[Any]     		=      self.transformer(__snake_case    ,					encoder_hidden_states=__snake_case    ,					timestep=__snake_case  ).sample

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

															_SCREAMING_SNAKE_CASE :       List[Any]     		=      self.truncate(__snake_case    ,					__snake_case  )

															# remove `log(0)`'s (`-inf`s)
															_SCREAMING_SNAKE_CASE :       Any     		=      model_output.clamp(-70  )

															# compute the previous noisy sample x_t -> x_t-1
															_SCREAMING_SNAKE_CASE :       List[str]     		=      self.scheduler.step(__snake_case    ,					timestep=__snake_case    ,					sample=__snake_case    ,					generator=__snake_case  ).prev_sample

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

											_SCREAMING_SNAKE_CASE :       Dict     		=      self.vqvae.config.vq_embed_dim
											_SCREAMING_SNAKE_CASE :       Dict     		=      (batch_size, self.transformer.height, self.transformer.width, embedding_channels)
											_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      self.vqvae.quantize.get_codebook_entry(__snake_case    ,					shape=__snake_case  )
											_SCREAMING_SNAKE_CASE :       List[str]     		=      self.vqvae.decode(__snake_case    ,					force_not_quantize=__snake_case  ).sample

											_SCREAMING_SNAKE_CASE :       Any     		=      (image / 2 + 0.5).clamp(0    ,					1  )
											_SCREAMING_SNAKE_CASE :       Optional[int]     		=      image.cpu().permute(0    ,					2    ,					3    ,					1  ).numpy()

											if output_type == "pil":
															_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      self.numpy_to_pil(__snake_case  )

											if not return_dict:
															return (image,)

											return ImagePipelineOutput(images=__snake_case  )






							def UpperCAmelCase_	(    self    ,					__snake_case    ,					__snake_case  ):
											_SCREAMING_SNAKE_CASE		,       _SCREAMING_SNAKE_CASE :       str     		=      torch.sort(__snake_case    ,					1    ,					descending=__snake_case  )
											_SCREAMING_SNAKE_CASE :       str     		=      torch.exp(__snake_case  )
											_SCREAMING_SNAKE_CASE :       Any     		=      sorted_p_x_0.cumsum(dim=1  ) < truncation_rate

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

											_SCREAMING_SNAKE_CASE :       str     		=      keep_mask.gather(1    ,					indices.argsort(1  )  )

											_SCREAMING_SNAKE_CASE :       int     		=      log_p_x_0.clone()

											_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      -torch.inf  # -inf = log(0)

											return rv








'''simple docstring'''


from typing import Dict

import numpy as np
import torch

from . import residue_constants as rc
from .tensor_utils import tensor_tree_map, tree_map







def 			snake_case_   (  SCREAMING_SNAKE_CASE__     ):







				"""simple docstring"""





				_SCREAMING_SNAKE_CASE :       Tuple     		=      []
				_SCREAMING_SNAKE_CASE :       Dict     		=      []
				_SCREAMING_SNAKE_CASE :       str     		=      []

				for rt in rc.restypes:
								_SCREAMING_SNAKE_CASE :       Optional[int]     		=      rc.restype_name_to_atomaa_names[rc.restype_atoa[rt]]
								restype_atomaa_to_atomaa_list.append([(rc.atom_order[name] if name else 0) for name in atom_names]     )
								_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      {name: i for i, name in enumerate(SCREAMING_SNAKE_CASE__     )}
								restype_atomaa_to_atomaa_list.append(
								    [(atom_name_to_idxaa[name] if name in atom_name_to_idxaa else 0) for name in rc.atom_types]     )

								restype_atomaa_mask_list.append([(1.0 if name else 0.0) for name in atom_names]     )

				# Add dummy mapping for restype 'UNK'
				restype_atomaa_to_atomaa_list.append([0] * 14     )
				restype_atomaa_to_atomaa_list.append([0] * 37     )
				restype_atomaa_mask_list.append([0.0] * 14     )

				_SCREAMING_SNAKE_CASE :       Dict     		=      torch.tensor(
				    SCREAMING_SNAKE_CASE__							,      dtype=torch.intaa							,      device=protein["""aatype"""].device							,      )
				_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      torch.tensor(
				    SCREAMING_SNAKE_CASE__							,      dtype=torch.intaa							,      device=protein["""aatype"""].device							,      )
				_SCREAMING_SNAKE_CASE :       str     		=      torch.tensor(
				    SCREAMING_SNAKE_CASE__							,      dtype=torch.floataa							,      device=protein["""aatype"""].device							,      )
				_SCREAMING_SNAKE_CASE :       int     		=      protein["""aatype"""].to(torch.long     )

				# create the mapping for (residx, atom14) --> atom37, i.e. an array
				# with shape (num_res, 14) containing the atom37 indices for this protein
				_SCREAMING_SNAKE_CASE :       List[str]     		=      restype_atomaa_to_atomaa[protein_aatype]
				_SCREAMING_SNAKE_CASE :       str     		=      restype_atomaa_mask[protein_aatype]

				_SCREAMING_SNAKE_CASE :       List[Any]     		=      residx_atomaa_mask
				_SCREAMING_SNAKE_CASE :       List[str]     		=      residx_atomaa_to_atomaa.long()

				# create the gather indices for mapping back
				_SCREAMING_SNAKE_CASE :       int     		=      restype_atomaa_to_atomaa[protein_aatype]
				_SCREAMING_SNAKE_CASE :       Dict     		=      residx_atomaa_to_atomaa.long()

				# create the corresponding mask
				_SCREAMING_SNAKE_CASE :       str     		=      torch.zeros([21, 37]							,      dtype=torch.floataa							,      device=protein["""aatype"""].device     )
				for restype, restype_letter in enumerate(rc.restypes     ):
								_SCREAMING_SNAKE_CASE :       int     		=      rc.restype_atoa[restype_letter]
								_SCREAMING_SNAKE_CASE :       Dict     		=      rc.residue_atoms[restype_name]
								for atom_name in atom_names:
												_SCREAMING_SNAKE_CASE :       List[Any]     		=      rc.atom_order[atom_name]
												_SCREAMING_SNAKE_CASE :       Union[str, Any]     		=      1

				_SCREAMING_SNAKE_CASE :       int     		=      restype_atomaa_mask[protein_aatype]
				_SCREAMING_SNAKE_CASE :       int     		=      residx_atomaa_mask

				return protein







def 			snake_case_   (  SCREAMING_SNAKE_CASE__     ):







				"""simple docstring"""





				_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      tree_map(lambda SCREAMING_SNAKE_CASE__     : torch.tensor(SCREAMING_SNAKE_CASE__							,      device=batch["""aatype"""].device     )							,      SCREAMING_SNAKE_CASE__							,      np.ndarray     )
				_SCREAMING_SNAKE_CASE :       Optional[Any]     		=      tensor_tree_map(lambda SCREAMING_SNAKE_CASE__     : np.array(SCREAMING_SNAKE_CASE__     )							,      make_atomaa_masks(SCREAMING_SNAKE_CASE__     )     )
				return out




