

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

import numpy as np
import PIL
import torch
from transformers import CLIPTextModelWithProjection, CLIPTokenizer

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


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

lowerCAmelCase__							=			"\n    Examples:\n        ```py\n        >>> import torch\n        >>> from diffusers import DiffusionPipeline\n        >>> from diffusers.utils import export_to_gif\n\n        >>> device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n\n        >>> repo = \"openai/shap-e\"\n        >>> pipe = DiffusionPipeline.from_pretrained(repo, torch_dtype=torch.float16)\n        >>> pipe = pipe.to(device)\n\n        >>> guidance_scale = 15.0\n        >>> prompt = \"a shark\"\n\n        >>> images = pipe(\n        ...     prompt,\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], \"shark_3d.gif\")\n        ```\n"






@dataclass
class     __magic_name__     (							_snake_case     ):
 UpperCAmelCase				     =    42






class     __magic_name__     (							_snake_case     ):


 def __init__(    self		:			int      ,    lowerCAmelCase__		:			PriorTransformer      ,    lowerCAmelCase__		:			CLIPTextModelWithProjection      ,    lowerCAmelCase__		:			CLIPTokenizer      ,    lowerCAmelCase__		:			HeunDiscreteScheduler      ,    lowerCAmelCase__		:			ShapERenderer      ,    )     ->   Union[str, Any]:
   super().__init__()

   self.register_modules(
       prior=lowerCAmelCase__      ,    text_encoder=lowerCAmelCase__      ,    tokenizer=lowerCAmelCase__      ,    scheduler=lowerCAmelCase__      ,    renderer=lowerCAmelCase__      ,    )


 def _UpperCamelCase						(    self		:			Union[str, Any]      ,    lowerCAmelCase__		:			Any      ,    lowerCAmelCase__		:			str      ,    lowerCAmelCase__		:			Dict      ,    lowerCAmelCase__		:			Tuple      ,    lowerCAmelCase__		:			Union[str, Any]      ,    lowerCAmelCase__		:			Optional[int]     )     ->   List[Any]:
   if latents is None:
     UpperCAmelCase          =	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          =	latents.to(lowerCAmelCase__     )

   UpperCAmelCase          =	latents * scheduler.init_noise_sigma
   return latents


 def _UpperCamelCase						(    self		:			Union[str, Any]      ,    lowerCAmelCase__		:			Optional[int]=0     )     ->   Dict:
   if is_accelerate_available():
     from accelerate import cpu_offload
   else:
     raise ImportError("Please install accelerate via `pip install accelerate`"     )

   UpperCAmelCase          =	torch.device(f"cuda:{gpu_id}"     )

   UpperCAmelCase          =	[self.text_encoder, self.prior]
   for cpu_offloaded_model in models:
     if cpu_offloaded_model is not None:
       cpu_offload(lowerCAmelCase__      ,    lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			List[str]      ,    lowerCAmelCase__		:			List[str]=0     )     ->   List[str]:
   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."     )

   UpperCAmelCase          =	torch.device(f"cuda:{gpu_id}"     )

   if self.device.type != "cpu":
     self.to("cpu"      ,    silence_dtype_warnings=lowerCAmelCase__     )
     torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

   UpperCAmelCase          =	None
   for cpu_offloaded_model in [self.text_encoder, self.prior, self.renderer]:
     UpperCAmelCase   ,						UpperCAmelCase          =	cpu_offload_with_hook(lowerCAmelCase__      ,    lowerCAmelCase__      ,    prev_module_hook=lowerCAmelCase__     )

   if self.safety_checker is not None:
     UpperCAmelCase   ,						UpperCAmelCase          =	cpu_offload_with_hook(self.safety_checker      ,    lowerCAmelCase__      ,    prev_module_hook=lowerCAmelCase__     )

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


 @property
 def _UpperCamelCase						(    self		:			List[str]     )     ->   List[Any]:
   if self.device != torch.device("meta"     ) or not hasattr(self.text_encoder      ,    "_hf_hook"     ):
     return self.device
   for module in self.text_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 _UpperCamelCase						(    self		:			Any      ,    lowerCAmelCase__		:			int      ,    lowerCAmelCase__		:			Any      ,    lowerCAmelCase__		:			Dict      ,    lowerCAmelCase__		:			Any      ,    )     ->   Tuple:
   len(lowerCAmelCase__     ) if isinstance(lowerCAmelCase__      ,    lowerCAmelCase__     ) else 1

   # YiYi Notes: set pad_token_id to be 0, not sure why I can't set in the config file
   UpperCAmelCase          =	0
   # get prompt text embeddings
   UpperCAmelCase          =	self.tokenizer(
       lowerCAmelCase__      ,    padding="max_length"      ,    max_length=self.tokenizer.model_max_length      ,    truncation=lowerCAmelCase__      ,    return_tensors="pt"      ,    )
   UpperCAmelCase          =	text_inputs.input_ids
   UpperCAmelCase          =	self.tokenizer(lowerCAmelCase__      ,    padding="longest"      ,    return_tensors="pt"     ).input_ids

   if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(lowerCAmelCase__      ,    lowerCAmelCase__     ):
     UpperCAmelCase          =	self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]     )
     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}"     )

   UpperCAmelCase          =	self.text_encoder(text_input_ids.to(lowerCAmelCase__     )     )
   UpperCAmelCase          =	text_encoder_output.text_embeds

   UpperCAmelCase          =	prompt_embeds.repeat_interleave(lowerCAmelCase__      ,    dim=0     )
   # in Shap-E it normalize the prompt_embeds and then later rescale it
   UpperCAmelCase          =	prompt_embeds / torch.linalg.norm(lowerCAmelCase__      ,    dim=-1      ,    keepdim=lowerCAmelCase__     )

   if do_classifier_free_guidance:
     UpperCAmelCase          =	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          =	torch.cat([negative_prompt_embeds, prompt_embeds]     )

   # Rescale the features to have unit variance
   UpperCAmelCase          =	math.sqrt(prompt_embeds.shape[1]     ) * prompt_embeds

   return prompt_embeds


 @torch.no_grad()
 @replace_example_docstring(lowerCAmelCase__     )
 def __call__(    self		:			Optional[int]      ,    lowerCAmelCase__		:			str      ,    lowerCAmelCase__		:			int = 1      ,    lowerCAmelCase__		:			int = 2_5      ,    lowerCAmelCase__		:			Optional[Union[torch.Generator, List[torch.Generator]]] = None      ,    lowerCAmelCase__		:			Optional[torch.FloatTensor] = None      ,    lowerCAmelCase__		:			float = 4.0      ,    lowerCAmelCase__		:			int = 6_4      ,    lowerCAmelCase__		:			Optional[str] = "pil"      ,    lowerCAmelCase__		:			bool = True      ,    )     ->   Union[str, Any]:

   if isinstance(lowerCAmelCase__      ,    lowerCAmelCase__     ):
     UpperCAmelCase          =	1
   elif isinstance(lowerCAmelCase__      ,    lowerCAmelCase__     ):
     UpperCAmelCase          =	len(lowerCAmelCase__     )
   else:
     raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(lowerCAmelCase__     )}"     )

   UpperCAmelCase          =	self._execution_device

   UpperCAmelCase          =	batch_size * num_images_per_prompt

   UpperCAmelCase          =	guidance_scale > 1.0
   UpperCAmelCase          =	self._encode_prompt(lowerCAmelCase__      ,    lowerCAmelCase__      ,    lowerCAmelCase__      ,    lowerCAmelCase__     )

   # prior

   self.scheduler.set_timesteps(lowerCAmelCase__      ,    device=lowerCAmelCase__     )
   UpperCAmelCase          =	self.scheduler.timesteps

   UpperCAmelCase          =	self.prior.config.num_embeddings
   UpperCAmelCase          =	self.prior.config.embedding_dim

   UpperCAmelCase          =	self.prepare_latents(
       (batch_size, num_embeddings * embedding_dim)      ,    prompt_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          =	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          =	torch.cat([latents] * 2     ) if do_classifier_free_guidance else latents
     UpperCAmelCase          =	self.scheduler.scale_model_input(lowerCAmelCase__      ,    lowerCAmelCase__     )

     UpperCAmelCase          =	self.prior(
         lowerCAmelCase__      ,    timestep=lowerCAmelCase__      ,    proj_embedding=lowerCAmelCase__      ,    ).predicted_image_embedding

     # remove the variance
     UpperCAmelCase   ,						UpperCAmelCase          =	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          =	noise_pred.chunk(2     )
       UpperCAmelCase          =	noise_pred_uncond + guidance_scale * (noise_pred - noise_pred_uncond)

     UpperCAmelCase          =	self.scheduler.step(
         lowerCAmelCase__      ,    timestep=lowerCAmelCase__      ,    sample=lowerCAmelCase__      ,    ).prev_sample

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

   UpperCAmelCase          =	[]
   for i, latent in enumerate(lowerCAmelCase__     ):
     UpperCAmelCase          =	self.renderer.decode(
         latent[None, :]      ,    lowerCAmelCase__      ,    size=lowerCAmelCase__      ,    ray_batch_size=4_0_9_6      ,    n_coarse_samples=6_4      ,    n_fine_samples=1_2_8      ,    )
     images.append(lowerCAmelCase__     )

   UpperCAmelCase          =	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          =	images.cpu().numpy()

   if output_type == "pil":
     UpperCAmelCase          =	[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__     )






import sys
from collections.abc import Mapping

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter






class     __magic_name__     (							TensorFormatter[Mapping, np.ndarray, Mapping]     ):


 def __init__(    self		:			Optional[int]      ,    lowerCAmelCase__		:			str=None      ,    **lowerCAmelCase__		:			int     )     ->   Optional[Any]:
   super().__init__(features=lowerCAmelCase__     )
   UpperCAmelCase          =	np_array_kwargs


 def _UpperCamelCase						(    self		:			Any      ,    lowerCAmelCase__		:			Dict     )     ->   Union[str, Any]:
   if isinstance(lowerCAmelCase__      ,    lowerCAmelCase__     ):
     if column and all(
         isinstance(lowerCAmelCase__      ,    np.ndarray     ) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column     ):
       return np.stack(lowerCAmelCase__     )
     else:
       # don't use np.array(column, dtype=object)
       # since it fails in certain cases
       # see https://stackoverflow.com/q/51005699
       UpperCAmelCase          =	np.empty(len(lowerCAmelCase__     )      ,    dtype=lowerCAmelCase__     )
       UpperCAmelCase          =	column
       return out
   return column


 def _UpperCamelCase						(    self		:			Optional[Any]      ,    lowerCAmelCase__		:			Union[str, Any]     )     ->   List[Any]:
   if isinstance(lowerCAmelCase__      ,    (str, bytes, type(lowerCAmelCase__     ))     ):
     return value
   elif isinstance(lowerCAmelCase__      ,    (np.character, np.ndarray)     ) and np.issubdtype(value.dtype      ,    np.character     ):
     return value
   elif isinstance(lowerCAmelCase__      ,    np.number     ):
     return value

   UpperCAmelCase          =	{}

   if isinstance(lowerCAmelCase__      ,    np.ndarray     ) and np.issubdtype(value.dtype      ,    np.integer     ):
     UpperCAmelCase          =	{"dtype": np.intaa}
   elif isinstance(lowerCAmelCase__      ,    np.ndarray     ) and np.issubdtype(value.dtype      ,    np.floating     ):
     UpperCAmelCase          =	{"dtype": np.floataa}
   elif config.PIL_AVAILABLE and "PIL" in sys.modules:
     import PIL.Image

     if isinstance(lowerCAmelCase__      ,    PIL.Image.Image     ):
       return np.asarray(lowerCAmelCase__      ,    **self.np_array_kwargs     )

   return np.asarray(lowerCAmelCase__      ,    **{**default_dtype, **self.np_array_kwargs}     )


 def _UpperCamelCase						(    self		:			List[Any]      ,    lowerCAmelCase__		:			str     )     ->   Optional[int]:
   # support for torch, tf, jax etc.
   if config.TORCH_AVAILABLE and "torch" in sys.modules:
     import torch

     if isinstance(lowerCAmelCase__      ,    torch.Tensor     ):
       return self._tensorize(data_struct.detach().cpu().numpy()[()]     )
   if hasattr(lowerCAmelCase__      ,    "__array__"     ) and not isinstance(lowerCAmelCase__      ,    (np.ndarray, np.character, np.number)     ):
     UpperCAmelCase          =	data_struct.__array__()
   # support for nested types like struct of list of struct
   if isinstance(lowerCAmelCase__      ,    np.ndarray     ):
     if data_struct.dtype == object:
       return self._consolidate([self.recursive_tensorize(lowerCAmelCase__     ) for substruct in data_struct]     )
   if isinstance(lowerCAmelCase__      ,    (list, tuple)     ):
     return self._consolidate([self.recursive_tensorize(lowerCAmelCase__     ) for substruct in data_struct]     )
   return self._tensorize(lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			Union[str, Any]      ,    lowerCAmelCase__		:			dict     )     ->   Union[str, Any]:
   return map_nested(self._recursive_tensorize      ,    lowerCAmelCase__      ,    map_list=lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			int      ,    lowerCAmelCase__		:			pa.Table     )     ->   Mapping:
   UpperCAmelCase          =	self.numpy_arrow_extractor().extract_row(lowerCAmelCase__     )
   UpperCAmelCase          =	self.python_features_decoder.decode_row(lowerCAmelCase__     )
   return self.recursive_tensorize(lowerCAmelCase__     )


 def _UpperCamelCase						(    self		:			List[str]      ,    lowerCAmelCase__		:			pa.Table     )     ->   np.ndarray:
   UpperCAmelCase          =	self.numpy_arrow_extractor().extract_column(lowerCAmelCase__     )
   UpperCAmelCase          =	self.python_features_decoder.decode_column(lowerCAmelCase__      ,    pa_table.column_names[0]     )
   UpperCAmelCase          =	self.recursive_tensorize(lowerCAmelCase__     )
   UpperCAmelCase          =	self._consolidate(lowerCAmelCase__     )
   return column


 def _UpperCamelCase						(    self		:			List[Any]      ,    lowerCAmelCase__		:			pa.Table     )     ->   Mapping:
   UpperCAmelCase          =	self.numpy_arrow_extractor().extract_batch(lowerCAmelCase__     )
   UpperCAmelCase          =	self.python_features_decoder.decode_batch(lowerCAmelCase__     )
   UpperCAmelCase          =	self.recursive_tensorize(lowerCAmelCase__     )
   for column_name in batch:
     UpperCAmelCase          =	self._consolidate(batch[column_name]     )
   return batch



