






from dataclasses import dataclass
from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from transformers import CLIPImageProcessor, CLIPVisionModel

from ...models import PriorTransformer
from ...pipelines import DiffusionPipeline
from ...schedulers import HeunDiscreteScheduler
from ...utils import (
    BaseOutput,
    is_accelerate_available,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .renderer import ShapERenderer


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

A_       :Dict       =						'\n    Examples:\n        ```py\n        >>> from PIL import Image\n        >>> import torch\n        >>> from diffusers import DiffusionPipeline\n        >>> from diffusers.utils import export_to_gif, load_image\n\n        >>> device = torch.device("cuda" if torch.cuda.is_available() else "cpu")\n\n        >>> repo = "openai/shap-e-img2img"\n        >>> pipe = DiffusionPipeline.from_pretrained(repo, torch_dtype=torch.float16)\n        >>> pipe = pipe.to(device)\n\n        >>> guidance_scale = 3.0\n        >>> image_url = "https://hf.co/datasets/diffusers/docs-images/resolve/main/shap-e/corgi.png"\n        >>> image = load_image(image_url).convert("RGB")\n\n        >>> images = pipe(\n        ...     image,\n        ...     guidance_scale=guidance_scale,\n        ...     num_inference_steps=64,\n        ...     frame_size=256,\n        ... ).images\n\n        >>> gif_path = export_to_gif(images[0], "corgi_3d.gif")\n        ```\n'




@dataclass
class 				__A     (			lowercase_ ):





				"""simple docstring"""







				UpperCamelCase__ :      Union[PIL.Image.Image, np.ndarray]




class 				__A     (			lowercase_ ):





				"""simple docstring"""







				def __init__( self							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  ):




									"""simple docstring"""

									super().__init__()

									self.register_modules(
									    prior=lowerCamelCase__							,  image_encoder=lowerCamelCase__							,  image_processor=lowerCamelCase__							,  scheduler=lowerCamelCase__							,  renderer=lowerCamelCase__							,  )


				def       __lowercase							( self							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__    ):




									"""simple docstring"""

									if latents is None:
														__UpperCamelCase      :					Tuple       		=randn_tensor(lowerCamelCase__							,  generator=lowerCamelCase__							,  device=lowerCamelCase__							,  dtype=lowerCamelCase__    )
									else:
														if latents.shape != shape:
																			raise ValueError(f'Unexpected latents shape, got {latents.shape}, expected {shape}'    )
														__UpperCamelCase      :					List[str]       		=latents.to(lowerCamelCase__    )

									__UpperCamelCase      :					Dict       		=latents * scheduler.init_noise_sigma
									return latents


				def       __lowercase							( self							,  lowerCamelCase__=0    ):




									"""simple docstring"""

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

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

									__UpperCamelCase      :					int       		=[self.image_encoder, self.prior]
									for cpu_offloaded_model in models:
														if cpu_offloaded_model is not None:
																			cpu_offload(lowerCamelCase__							,  lowerCamelCase__    )


				@property
				def       __lowercase							( self    ):




									"""simple docstring"""

									if self.device != torch.device('meta'    ) or not hasattr(self.image_encoder							,  '_hf_hook'    ):
														return self.device
									for module in self.image_encoder.modules():
														if (
														    hasattr(lowerCamelCase__							,  '_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


				def       __lowercase							( self							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  ):




									"""simple docstring"""

									if isinstance(lowerCamelCase__							,  lowerCamelCase__    ) and isinstance(image[0]							,  torch.Tensor    ):
														__UpperCamelCase      :					Optional[Any]       		=torch.cat(lowerCamelCase__							,  axis=0    ) if image[0].ndim == 4 else torch.stack(lowerCamelCase__							,  axis=0    )

									if not isinstance(lowerCamelCase__							,  torch.Tensor    ):
														__UpperCamelCase      :					List[Any]       		=self.image_processor(lowerCamelCase__							,  return_tensors='pt'    ).pixel_values[0].unsqueeze(0    )

									__UpperCamelCase      :					Any       		=image.to(dtype=self.image_encoder.dtype							,  device=lowerCamelCase__    )

									__UpperCamelCase      :					List[str]       		=self.image_encoder(lowerCamelCase__    )['last_hidden_state']
									__UpperCamelCase      :					Any       		=image_embeds[:, 1:, :].contiguous()  # batch_size, dim, 256

									__UpperCamelCase      :					Dict       		=image_embeds.repeat_interleave(lowerCamelCase__							,  dim=0    )

									if do_classifier_free_guidance:
														__UpperCamelCase      :					Union[str, Any]       		=torch.zeros_like(lowerCamelCase__    )

														# 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
														__UpperCamelCase      :					List[Any]       		=torch.cat([negative_image_embeds, image_embeds]    )

									return image_embeds







				@torch.no_grad()
				@replace_example_docstring(lowerCamelCase__    )
				def __call__( self							,  lowerCamelCase__							,  lowerCamelCase__ = 1							,  lowerCamelCase__ = 25							,  lowerCamelCase__ = None							,  lowerCamelCase__ = None							,  lowerCamelCase__ = 4.0							,  lowerCamelCase__ = 64							,  lowerCamelCase__ = "pil"							,  lowerCamelCase__ = True							,  ):




									"""simple docstring"""

									if isinstance(lowerCamelCase__							,  PIL.Image.Image    ):
														__UpperCamelCase      :					Any       		=1
									elif isinstance(lowerCamelCase__							,  torch.Tensor    ):
														__UpperCamelCase      :					str       		=image.shape[0]
									elif isinstance(lowerCamelCase__							,  lowerCamelCase__    ) and isinstance(image[0]							,  (torch.Tensor, PIL.Image.Image)    ):
														__UpperCamelCase      :					Union[str, Any]       		=len(lowerCamelCase__    )
									else:
														raise ValueError(
														    f'`image` has to be of type `PIL.Image.Image`, `torch.Tensor`, `List[PIL.Image.Image]` or `List[torch.Tensor]` but is {type(lowerCamelCase__    )}'    )

									__UpperCamelCase      :					Tuple       		=self._execution_device

									__UpperCamelCase      :					List[str]       		=batch_size * num_images_per_prompt

									__UpperCamelCase      :					Optional[Any]       		=guidance_scale > 1.0
									__UpperCamelCase      :					List[Any]       		=self._encode_image(lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__    )

									# prior

									self.scheduler.set_timesteps(lowerCamelCase__							,  device=lowerCamelCase__    )
									__UpperCamelCase      :					List[str]       		=self.scheduler.timesteps

									__UpperCamelCase      :					Optional[int]       		=self.prior.config.num_embeddings
									__UpperCamelCase      :					Tuple       		=self.prior.config.embedding_dim

									__UpperCamelCase      :					Optional[Any]       		=self.prepare_latents(
									    (batch_size, num_embeddings * embedding_dim)							,  image_embeds.dtype							,  lowerCamelCase__							,  lowerCamelCase__							,  lowerCamelCase__							,  self.scheduler							,  )

									# YiYi notes: for testing only to match ldm, we can directly create a latents with desired shape: batch_size, num_embeddings, embedding_dim
									__UpperCamelCase      :					Tuple       		=latents.reshape(latents.shape[0]							,  lowerCamelCase__							,  lowerCamelCase__    )

									for i, t in enumerate(self.progress_bar(lowerCamelCase__    )    ):
														# expand the latents if we are doing classifier free guidance
														__UpperCamelCase      :					List[str]       		=torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents
														__UpperCamelCase      :					List[str]       		=self.scheduler.scale_model_input(lowerCamelCase__							,  lowerCamelCase__    )

														__UpperCamelCase      :					int       		=self.prior(
														    lowerCamelCase__							,  timestep=lowerCamelCase__							,  proj_embedding=lowerCamelCase__							,  ).predicted_image_embedding

														# remove the variance
														__UpperCamelCase		,				__UpperCamelCase      :					Tuple       		=noise_pred.split(
														    scaled_model_input.shape[2]							,  dim=2    )  # batch_size, num_embeddings, embedding_dim

														if do_classifier_free_guidance is not None:
																			__UpperCamelCase		,				__UpperCamelCase      :					List[str]       		=noise_pred.chunk(2    )
																			__UpperCamelCase      :					Optional[Any]       		=noise_pred_uncond + guidance_scale * (noise_pred - noise_pred_uncond)

														__UpperCamelCase      :					Tuple       		=self.scheduler.step(
														    lowerCamelCase__							,  timestep=lowerCamelCase__							,  sample=lowerCamelCase__							,  ).prev_sample

									if output_type == "latent":
														return ShapEPipelineOutput(images=lowerCamelCase__    )

									__UpperCamelCase      :					List[Any]       		=[]
									for i, latent in enumerate(lowerCamelCase__    ):
														print()
														__UpperCamelCase      :					int       		=self.renderer.decode(
														    latent[None, :]							,  lowerCamelCase__							,  size=lowerCamelCase__							,  ray_batch_size=4096							,  n_coarse_samples=64							,  n_fine_samples=128							,  )

														images.append(lowerCamelCase__    )

									__UpperCamelCase      :					Tuple       		=torch.stack(lowerCamelCase__    )

									if output_type not in ["np", "pil"]:
														raise ValueError(f'Only the output types `pil` and `np` are supported not output_type={output_type}'    )

									__UpperCamelCase      :					str       		=images.cpu().numpy()

									if output_type == "pil":
														__UpperCamelCase      :					Optional[int]       		=[self.numpy_to_pil(lowerCamelCase__    ) for image in images]

									# Offload last model to CPU
									if hasattr(self							,  'final_offload_hook'    ) and self.final_offload_hook is not None:
														self.final_offload_hook.offload()

									if not return_dict:
														return (images,)

									return ShapEPipelineOutput(images=lowerCamelCase__    )





'''simple docstring'''

import dataclasses
import re
import string
from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple

import numpy as np

from . import residue_constants


lowerCamelCase							:   str  						=	Mapping[str, np.ndarray]
lowerCamelCase							:   List[Any]  						=	Mapping[str, Any]  # Is a nested dict.
lowerCamelCase							:   Any  						=	0.0_1
@dataclasses.dataclass(frozen=lowercase_     )
class   __lowerCAmelCase						:





      '''simple docstring'''






      lowerCAmelCase__					:     np.ndarray  # [num_res, num_atom_type, 3]

      # Amino-acid type for each residue represented as an integer between 0 and
      # 20, where 20 is 'X'.
      lowerCAmelCase__					:     np.ndarray  # [num_res]

      # Binary float mask to indicate presence of a particular atom. 1.0 if an atom
      # is present and 0.0 if not. This should be used for loss masking.
      lowerCAmelCase__					:     np.ndarray  # [num_res, num_atom_type]

      # Residue index as used in PDB. It is not necessarily continuous or 0-indexed.
      lowerCAmelCase__					:     np.ndarray  # [num_res]

      # B-factors, or temperature factors, of each residue (in sq. angstroms units),
      # representing the displacement of the residue from its ground truth mean
      # value.
      lowerCAmelCase__					:     np.ndarray  # [num_res, num_atom_type]

      # Chain indices for multi-chain predictions
      lowerCAmelCase__					:     Optional[np.ndarray]  					=							None

      # Optional remark about the protein. Included as a comment in output PDB
      # files
      lowerCAmelCase__					:     Optional[str]  					=							None

      # Templates used to generate this protein (prediction-only)
      lowerCAmelCase__					:     Optional[Sequence[str]]  					=							None

      # Chain corresponding to each parent
      lowerCAmelCase__					:     Optional[Sequence[int]]  					=							None






def _SCREAMING_SNAKE_CASE      (A       )       ->       Protein:






       """simple docstring"""







       lowercase__      			= R'''(\[[A-Z]+\]\n)'''
       lowercase__      			= [tag.strip() for tag in re.split(A  , A       ) if len(A       ) > 0]
       lowercase__      			= zip(tags[0::2]  , [l.split('''\n'''       ) for l in tags[1::2]]       )

       lowercase__      			= ["N", "CA", "C"]
       lowercase__      			= None
       lowercase__      			= None
       lowercase__      			= None
       for g in groups:
              if "[PRIMARY]" == g[0]:
                     lowercase__      			= g[1][0].strip()
                     for i in range(len(A       )       ):
                            if seq[i] not in residue_constants.restypes:
                                   lowercase__      			= '''X'''  # FIXME: strings are immutable
                     lowercase__      			= np.array(
                         [residue_constants.restype_order.get(A  , residue_constants.restype_num       ) for res_symbol in seq]       )
              elif "[TERTIARY]" == g[0]:
                     lowercase__      			= []
                     for axis in range(3       ):
                            tertiary.append(list(map(A  , g[1][axis].split()       )       )       )
                     lowercase__      			= np.array(A       )
                     lowercase__      			= np.zeros((len(tertiary[0]       ) // 3, residue_constants.atom_type_num, 3)       ).astype(np.floataa       )
                     for i, atom in enumerate(A       ):
                            lowercase__      			= np.transpose(tertiary_np[:, i::3]       )
                     atom_positions *= PICO_TO_ANGSTROM
              elif "[MASK]" == g[0]:
                     lowercase__      			= np.array(list(map({'''-''': 0, '''+''': 1}.get  , g[1][0].strip()       )       )       )
                     lowercase__      			= np.zeros(
                         (
                             len(A       ),
                             residue_constants.atom_type_num,
                         )       ).astype(np.floataa       )
                     for i, atom in enumerate(A       ):
                            lowercase__      			= 1
                     atom_mask *= mask[..., None]

       assert aatype is not None

       return Protein(
           atom_positions=A  , atom_mask=A  , aatype=A  , residue_index=np.arange(len(A       )       )  , b_factors=A  , )

def _SCREAMING_SNAKE_CASE      (A  , A = 0       )       ->       List[str]:






       """simple docstring"""







       lowercase__      			= []

       lowercase__      			= prot.remark
       if remark is not None:
              pdb_headers.append(f"REMARK {remark}"       )

       lowercase__      			= prot.parents
       lowercase__      			= prot.parents_chain_index
       if parents is not None and parents_chain_index is not None:
              lowercase__      			= [p for i, p in zip(A  , A       ) if i == chain_id]

       if parents is None or len(A       ) == 0:
              lowercase__      			= ['''N/A''']

       pdb_headers.append(f"PARENT {' '.join(A       )}"       )

       return pdb_headers

def _SCREAMING_SNAKE_CASE      (A  , A       )       ->       str:






       """simple docstring"""







       lowercase__      			= []
       lowercase__      			= pdb_str.split('''\n'''       )

       lowercase__      			= prot.remark
       if remark is not None:
              out_pdb_lines.append(f"REMARK {remark}"       )

       lowercase__      			= 42
       if prot.parents is not None and len(prot.parents       ) > 0:
              lowercase__      			= []
              if prot.parents_chain_index is not None:
                     lowercase__      			= {}
                     for p, i in zip(prot.parents  , prot.parents_chain_index       ):
                            parent_dict.setdefault(str(A       )  , []       )
                            parent_dict[str(A       )].append(A       )

                     lowercase__      			= max([int(A       ) for chain_idx in parent_dict]       )
                     for i in range(max_idx + 1       ):
                            lowercase__      			= parent_dict.get(str(A       )  , ['''N/A''']       )
                            parents_per_chain.append(A       )
              else:
                     parents_per_chain.append(list(prot.parents       )       )
       else:
              lowercase__      			= [['''N/A''']]

       def make_parent_line(A       ) -> str:
              return f"PARENT {' '.join(A       )}"

       out_pdb_lines.append(make_parent_line(parents_per_chain[0]       )       )

       lowercase__      			= 0
       for i, l in enumerate(A       ):
              if "PARENT" not in l and "REMARK" not in l:
                     out_pdb_lines.append(A       )
              if "TER" in l and "END" not in lines[i + 1]:
                     chain_counter += 1
                     if not chain_counter >= len(A       ):
                            lowercase__      			= parents_per_chain[chain_counter]
                     else:
                            lowercase__      			= ['''N/A''']

                     out_pdb_lines.append(make_parent_line(A       )       )

       return "\n".join(A       )

def _SCREAMING_SNAKE_CASE      (A       )       ->       str:






       """simple docstring"""







       lowercase__      			= residue_constants.restypes + ['''X''']

       def res_atoa(A       ) -> str:
              return residue_constants.restype_atoa.get(restypes[r]  , '''UNK'''       )

       lowercase__      			= residue_constants.atom_types

       lowercase__      			= []

       lowercase__      			= prot.atom_mask
       lowercase__      			= prot.aatype
       lowercase__      			= prot.atom_positions
       lowercase__      			= prot.residue_index.astype(np.intaa       )
       lowercase__      			= prot.b_factors
       lowercase__      			= prot.chain_index

       if np.any(aatype > residue_constants.restype_num       ):
              raise ValueError('''Invalid aatypes.'''       )

       lowercase__      			= get_pdb_headers(A       )
       if len(A       ) > 0:
              pdb_lines.extend(A       )

       lowercase__      			= aatype.shape[0]
       lowercase__      			= 1
       lowercase__      			= 0
       lowercase__      			= string.ascii_uppercase
       lowercase__      			= None
       # Add all atom sites.
       for i in range(A       ):
              lowercase__      			= res_atoa(aatype[i]       )
              for atom_name, pos, mask, b_factor in zip(A  , atom_positions[i]  , atom_mask[i]  , b_factors[i]       ):
                     if mask < 0.5:
                            continue

                     lowercase__      			= '''ATOM'''
                     lowercase__      			= atom_name if len(A       ) == 4 else f" {atom_name}"
                     lowercase__      			= ''''''
                     lowercase__      			= ''''''
                     lowercase__      			= 1.00
                     lowercase__      			= atom_name[0]  # Protein supports only C, N, O, S, this works.
                     lowercase__      			= ''''''

                     lowercase__      			= '''A'''
                     if chain_index is not None:
                            lowercase__      			= chain_tags[chain_index[i]]

                     # PDB is a columnar format, every space matters here!
                     lowercase__      			= (
                         f"{record_type:<6}{atom_index:>5} {name:<4}{alt_loc:>1}"
                         f"{res_name_a:>3} {chain_tag:>1}"
                         f"{residue_index[i]:>4}{insertion_code:>1}   "
                         f"{pos[0]:>8.3f}{pos[1]:>8.3f}{pos[2]:>8.3f}"
                         f"{occupancy:>6.2f}{b_factor:>6.2f}          "
                         f"{element:>2}{charge:>2}"
                     )
                     pdb_lines.append(A       )
                     atom_index += 1

              lowercase__      			= i == n - 1
              if chain_index is not None:
                     if i != n - 1 and chain_index[i + 1] != prev_chain_index:
                            lowercase__      			= True
                            lowercase__      			= chain_index[i + 1]

              if should_terminate:
                     # Close the chain.
                     lowercase__      			= '''TER'''
                     lowercase__      			= (
                         f"{chain_end:<6}{atom_index:>5}      {res_atoa(aatype[i]       ):>3} {chain_tag:>1}{residue_index[i]:>4}"
                     )
                     pdb_lines.append(A       )
                     atom_index += 1

                     if i != n - 1:
                            # "prev" is a misnomer here. This happens at the beginning of
                            # each new chain.
                            pdb_lines.extend(get_pdb_headers(A  , A       )       )

       pdb_lines.append('''END'''       )
       pdb_lines.append(''''''       )
       return "\n".join(A       )

def _SCREAMING_SNAKE_CASE      (A       )       ->       np.ndarray:






       """simple docstring"""







       return residue_constants.STANDARD_ATOM_MASK[prot.aatype]

def _SCREAMING_SNAKE_CASE      (A  , A  , A = None  , A = None  , A = None  , A = None  , A = None  , )       ->       Protein:






       """simple docstring"""







       return Protein(
           aatype=features['''aatype''']  , atom_positions=result['''final_atom_positions''']  , atom_mask=result['''final_atom_mask''']  , residue_index=features['''residue_index'''] + 1  , b_factors=b_factors if b_factors is not None else np.zeros_like(result['''final_atom_mask''']       )  , chain_index=A  , remark=A  , parents=A  , parents_chain_index=A  , )




