





'''simple docstring'''







from __future__ import annotations

import copy
import os
import tempfile
from importlib import import_module
from math import isnan

from transformers import is_tf_available
from transformers.models.auto import get_values
from transformers.testing_utils import _tf_gpu_memory_limit, require_tf, slow

from ..test_modeling_tf_common import ids_tensor


if is_tf_available():
					import numpy as np
					import tensorflow as tf

					from transformers import (
					    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
					    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
					    TF_MODEL_FOR_MASKED_LM_MAPPING,
					    TF_MODEL_FOR_MULTIPLE_CHOICE_MAPPING,
					    TF_MODEL_FOR_NEXT_SENTENCE_PREDICTION_MAPPING,
					    TF_MODEL_FOR_PRETRAINING_MAPPING,
					    TF_MODEL_FOR_QUESTION_ANSWERING_MAPPING,
					    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
					    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
					    TF_MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING,
					    TFSharedEmbeddings,
					)

					if _tf_gpu_memory_limit is not None:
										snake_case_  :							Tuple            =	tf.config.list_physical_devices("GPU")
										for gpu in gpus:
															# Restrict TensorFlow to only allocate x GB of memory on the GPUs
															try:
																				tf.config.set_logical_device_configuration(
																				    gpu, [tf.config.LogicalDeviceConfiguration(memory_limit=_tf_gpu_memory_limit)]
																				)
																				snake_case_  :							int            =	tf.config.list_logical_devices("GPU")
																				print("Logical GPUs", logical_gpus)
															except RuntimeError as e:
																				# Virtual devices must be set before GPUs have been initialized
																				print(e)


@require_tf
class        __a   :
							__a       : Optional[Any]          =							None
							__a       : str          =							()
							__a       : Union[str, Any]          =							()
							__a       : Tuple          =							True
							__a       : Tuple          =							True
							__a       : int          =							True
							__a       : Optional[Any]          =							False

							def      UpperCAmelCase__						(				self				:  Tuple							,  __magic_name__				:  List[str]							,  __magic_name__				:  Union[str, Any]							,  __magic_name__				:  Dict=False )       ->							dict:







													"""simple docstring"""
													UpperCAmelCase_      :   int      =	copy.deepcopy(__magic_name__ )

													if model_class in get_values(__magic_name__ ):
																			UpperCAmelCase_      :   List[str]      =	{
																			    k: tf.tile(tf.expand_dims(__magic_name__							,  1 )							,  (1, self.model_tester.num_choices) + (1,) * (v.ndim - 1) )
																			    if isinstance(__magic_name__							,  tf.Tensor ) and v.ndim > 0
																			    else v
																			    for k, v in inputs_dict.items()
																			}

													if return_labels:
																			if model_class in get_values(__magic_name__ ):
																									UpperCAmelCase_      :   Any      =	tf.ones(self.model_tester.batch_size							,  dtype=tf.intaa )
																			elif model_class in get_values(__magic_name__ ):
																									UpperCAmelCase_      :   Dict      =	tf.zeros(self.model_tester.batch_size							,  dtype=tf.intaa )
																									UpperCAmelCase_      :   Dict      =	tf.zeros(self.model_tester.batch_size							,  dtype=tf.intaa )
																			elif model_class in [
																			    *get_values(__magic_name__ ),
																			    *get_values(__magic_name__ ),
																			]:
																									UpperCAmelCase_      :   Optional[int]      =	tf.zeros(self.model_tester.batch_size							,  dtype=tf.intaa )
																			elif model_class in get_values(__magic_name__ ):
																									UpperCAmelCase_      :   List[str]      =	tf.zeros(self.model_tester.batch_size							,  dtype=tf.intaa )
																			elif model_class in [
																			    *get_values(__magic_name__ ),
																			    *get_values(__magic_name__ ),
																			    *get_values(__magic_name__ ),
																			    *get_values(__magic_name__ ),
																			    *get_values(__magic_name__ ),
																			]:
																									UpperCAmelCase_      :   Optional[Any]      =	tf.zeros(
																									    (self.model_tester.batch_size, self.model_tester.seq_length)							,  dtype=tf.intaa )
													return inputs_dict

							@slow
							def      UpperCAmelCase__						(				self				:  Tuple )       ->							str:







													"""simple docstring"""
													UpperCAmelCase_    ,							UpperCAmelCase_      :   str      =	self.model_tester.prepare_config_and_inputs_for_common()
													for model_class in self.all_model_classes[:2]:
																			UpperCAmelCase_      :   Union[str, Any]      =	self._prepare_for_class(__magic_name__							,  __magic_name__ )
																			UpperCAmelCase_      :   Dict      =	model_class(__magic_name__ )

																			@tf.function
																			def run_in_graph_mode():
																									return model(__magic_name__ )

																			UpperCAmelCase_      :   Any      =	run_in_graph_mode()
																			self.assertIsNotNone(__magic_name__ )

							@slow
							def      UpperCAmelCase__						(				self				:  int )       ->							List[Any]:







													"""simple docstring"""
													UpperCAmelCase_    ,							UpperCAmelCase_      :   int      =	self.model_tester.prepare_config_and_inputs_for_common()
													for model_class in self.all_model_classes[:2]:
																			UpperCAmelCase_      :   Union[str, Any]      =	self._prepare_for_class(__magic_name__							,  __magic_name__ )
																			UpperCAmelCase_      :   List[str]      =	model_class(__magic_name__ )

																			@tf.function(experimental_compile=__magic_name__ )
																			def run_in_graph_mode():
																									return model(__magic_name__ )

																			UpperCAmelCase_      :   int      =	run_in_graph_mode()
																			self.assertIsNotNone(__magic_name__ )

							@slow
							def      UpperCAmelCase__						(				self				:  Union[str, Any] )       ->							Optional[int]:







													"""simple docstring"""
													# This is a copy of the test_keras_fit method, but we use XLA compilation instead of eager
													UpperCAmelCase_    ,							UpperCAmelCase_      :   Optional[Any]      =	self.model_tester.prepare_config_and_inputs_for_common()
													for model_class in self.all_model_classes[:2]:
																			UpperCAmelCase_      :   str      =	model_class(__magic_name__ )
																			if getattr(__magic_name__							,  '''hf_compute_loss'''							,  __magic_name__ ):
																									# Test that model correctly compute the loss with kwargs
																									UpperCAmelCase_      :   Dict      =	self._prepare_for_class(inputs_dict.copy()							,  __magic_name__							,  return_labels=__magic_name__ )
																									# Is there a better way to remove these decoder inputs?
																									UpperCAmelCase_      :   Optional[Any]      =	{
																									    key: val
																									    for key, val in prepared_for_class.items()
																									    if key not in ('''head_mask''', '''decoder_head_mask''', '''cross_attn_head_mask''', '''decoder_input_ids''')
																									}

																									UpperCAmelCase_      :   Any      =	{
																									    '''labels''',
																									    '''label''',
																									    '''label_ids''',
																									    '''start_positions''',
																									    '''start_position''',
																									    '''end_positions''',
																									    '''end_position''',
																									    '''next_sentence_label''',
																									}
																									UpperCAmelCase_      :   int      =	possible_label_cols.intersection(set(__magic_name__ ) )
																									self.assertGreater(len(__magic_name__ )							,  0							,  msg='''No matching label names found!''' )
																									UpperCAmelCase_      :   Optional[Any]      =	{key: val for key, val in prepared_for_class.items() if key in label_names}
																									UpperCAmelCase_      :   Dict      =	{key: val for key, val in prepared_for_class.items() if key not in label_names}
																									self.assertGreater(len(__magic_name__ )							,  0 )

																									# Make sure it works with XLA!
																									model.compile(optimizer=tf.keras.optimizers.SGD(0.0 )							,  jit_compile=__magic_name__ )
																									# Make sure the model fits without crashing regardless of where we pass the labels
																									UpperCAmelCase_      :   Union[str, Any]      =	model.fit(
																									    __magic_name__							,  validation_data=__magic_name__							,  steps_per_epoch=1							,  validation_steps=1							,  shuffle=__magic_name__							,  verbose=0							,  )
																									UpperCAmelCase_      :   Union[str, Any]      =	history.history['''loss'''][0]
																									self.assertTrue(not isnan(__magic_name__ ) )
																									UpperCAmelCase_      :   Union[str, Any]      =	history.history['''val_loss'''][0]
																									self.assertTrue(not isnan(__magic_name__ ) )

																									# Now test it with separate labels, to make sure that path works in XLA too.
																									UpperCAmelCase_      :   List[str]      =	model_class(__magic_name__ )
																									model.compile(optimizer=tf.keras.optimizers.SGD(0.0 )							,  jit_compile=__magic_name__ )
																									UpperCAmelCase_      :   int      =	model.fit(
																									    __magic_name__							,  __magic_name__							,  validation_data=(inputs_minus_labels, labels)							,  steps_per_epoch=1							,  validation_steps=1							,  shuffle=__magic_name__							,  verbose=0							,  )

																									UpperCAmelCase_      :   List[Any]      =	history.history['''loss'''][0]
																									self.assertTrue(not isnan(__magic_name__ ) )
																									UpperCAmelCase_      :   Union[str, Any]      =	history.history['''val_loss'''][0]
																									self.assertTrue(not isnan(__magic_name__ ) )

							@slow
							def      UpperCAmelCase__						(				self				:  Tuple )       ->							Optional[int]:







													"""simple docstring"""
													UpperCAmelCase_    ,							UpperCAmelCase_      :   List[Any]      =	self.model_tester.prepare_config_and_inputs_for_common()
													UpperCAmelCase_      :   str      =	True
													UpperCAmelCase_      :   Dict      =	True

													if hasattr(__magic_name__							,  '''use_cache''' ):
																			UpperCAmelCase_      :   int      =	True

													UpperCAmelCase_      :   Optional[Any]      =	getattr(self.model_tester							,  '''encoder_seq_length'''							,  self.model_tester.seq_length )
													UpperCAmelCase_      :   Any      =	getattr(self.model_tester							,  '''key_length'''							,  __magic_name__ )

													for model_class in self.all_model_classes[:2]:
																			UpperCAmelCase_      :   int      =	self._prepare_for_class(__magic_name__							,  __magic_name__ )
																			UpperCAmelCase_      :   Any      =	model_class(__magic_name__ )
																			model.build()
																			UpperCAmelCase_      :   Dict      =	len(model(__magic_name__ ) )

																			for key in list(class_inputs_dict.keys() ):
																									# Remove keys not in the serving signature, as the SavedModel will not be compiled to deal with them
																									if key not in model.input_signature:
																															del class_inputs_dict[key]
																									# Check it's a tensor, in case the inputs dict has some bools in it too
																									elif isinstance(class_inputs_dict[key]							,  tf.Tensor ) and class_inputs_dict[key].dtype.is_integer:
																															UpperCAmelCase_      :   Optional[Any]      =	tf.cast(class_inputs_dict[key]							,  tf.intaa )

																			if set(class_inputs_dict.keys() ) != set(model.input_signature.keys() ):
																									continue  # Some models have inputs that the preparation functions don't create, we skip those

																			with tempfile.TemporaryDirectory() as tmpdirname:
																									model.save_pretrained(__magic_name__							,  saved_model=__magic_name__ )
																									UpperCAmelCase_      :   List[str]      =	os.path.join(__magic_name__							,  '''saved_model'''							,  '''1''' )
																									UpperCAmelCase_      :   Union[str, Any]      =	tf.keras.models.load_model(__magic_name__ )
																									UpperCAmelCase_      :   Union[str, Any]      =	model(__magic_name__ )

																									if self.is_encoder_decoder:
																															UpperCAmelCase_      :   Optional[Any]      =	outputs['''encoder_hidden_states''']
																															UpperCAmelCase_      :   Optional[int]      =	outputs['''encoder_attentions''']
																									else:
																															UpperCAmelCase_      :   Optional[int]      =	outputs['''hidden_states''']
																															UpperCAmelCase_      :   int      =	outputs['''attentions''']

																									self.assertEqual(len(__magic_name__ )							,  __magic_name__ )

																									UpperCAmelCase_      :   List[Any]      =	getattr(
																									    self.model_tester							,  '''expected_num_hidden_layers'''							,  self.model_tester.num_hidden_layers + 1 )

																									self.assertEqual(len(__magic_name__ )							,  __magic_name__ )
																									self.assertListEqual(
																									    list(output_hidden_states[0].shape[-2:] )							,  [self.model_tester.seq_length, self.model_tester.hidden_size]							,  )

																									self.assertEqual(len(__magic_name__ )							,  self.model_tester.num_hidden_layers )
																									self.assertListEqual(
																									    list(output_attentions[0].shape[-3:] )							,  [self.model_tester.num_attention_heads, encoder_seq_length, encoder_key_length]							,  )

							@slow
							def      UpperCAmelCase__						(				self				:  Dict )       ->							Tuple:







													"""simple docstring"""
													tf.keras.mixed_precision.set_global_policy('''mixed_float16''' )

													# try/finally block to ensure subsequent tests run in float32
													try:
																			UpperCAmelCase_    ,							UpperCAmelCase_      :   str      =	self.model_tester.prepare_config_and_inputs_for_common()
																			for model_class in self.all_model_classes[:2]:
																									UpperCAmelCase_      :   Optional[int]      =	self._prepare_for_class(__magic_name__							,  __magic_name__ )
																									UpperCAmelCase_      :   Any      =	model_class(__magic_name__ )
																									UpperCAmelCase_      :   Union[str, Any]      =	model(__magic_name__ )

																									self.assertIsNotNone(__magic_name__ )
													finally:
																			tf.keras.mixed_precision.set_global_policy('''float32''' )

							@slow
							def      UpperCAmelCase__						(				self				:  Tuple )       ->							Any:







													"""simple docstring"""
													UpperCAmelCase_    ,							UpperCAmelCase_      :   Union[str, Any]      =	self.model_tester.prepare_config_and_inputs_for_common()
													# head_mask and decoder_head_mask has different shapes than other input args
													if "head_mask" in inputs_dict:
																			del inputs_dict["head_mask"]
													if "decoder_head_mask" in inputs_dict:
																			del inputs_dict["decoder_head_mask"]
													if "cross_attn_head_mask" in inputs_dict:
																			del inputs_dict["cross_attn_head_mask"]
													UpperCAmelCase_      :   Optional[int]      =	{
													    module_member
													    for model_class in self.all_model_classes
													    for module in (import_module(model_class.__module__ ),)
													    for module_member_name in dir(__magic_name__ )
													    if module_member_name.endswith('''MainLayer''' )
													    for module_member in (getattr(__magic_name__							,  __magic_name__ ),)
													    if isinstance(__magic_name__							,  __magic_name__ )
													    and tf.keras.layers.Layer in module_member.__bases__
													    and getattr(__magic_name__							,  '''_keras_serializable'''							,  __magic_name__ )
													}

													for main_layer_class in tf_main_layer_classes:
																			# T5MainLayer needs an embed_tokens parameter when called without the inputs_embeds parameter
																			if "T5" in main_layer_class.__name__:
																									# Take the same values than in TFT5ModelTester for this shared layer
																									UpperCAmelCase_      :   Tuple      =	TFSharedEmbeddings(self.model_tester.vocab_size							,  self.model_tester.hidden_size							,  name='''shared''' )
																									UpperCAmelCase_      :   Dict      =	False
																									UpperCAmelCase_      :   Optional[int]      =	main_layer_class(__magic_name__							,  embed_tokens=__magic_name__ )
																			else:
																									UpperCAmelCase_      :   List[str]      =	main_layer_class(__magic_name__ )

																			UpperCAmelCase_      :   Union[str, Any]      =	{
																			    name: tf.keras.Input(tensor.shape[1:]							,  dtype=tensor.dtype ) for name, tensor in inputs_dict.items()
																			}

																			if hasattr(self.model_tester							,  '''num_labels''' ):
																									UpperCAmelCase_      :   List[str]      =	self.model_tester.num_labels
																			else:
																									UpperCAmelCase_      :   Optional[int]      =	2

																			UpperCAmelCase_      :   Any      =	tf.data.Dataset.from_tensor_slices(
																			    (inputs_dict, np.ones((self.model_tester.batch_size, self.model_tester.seq_length, num_labels, 1) )) ).batch(1 )

																			UpperCAmelCase_      :   Optional[Any]      =	main_layer(__magic_name__ )[0]
																			UpperCAmelCase_      :   List[str]      =	tf.keras.layers.Dense(__magic_name__							,  activation='''softmax'''							,  name='''outputs''' )(__magic_name__ )
																			UpperCAmelCase_      :   int      =	tf.keras.models.Model(inputs=__magic_name__							,  outputs=[outputs] )

																			model.compile(loss='''binary_crossentropy'''							,  optimizer='''adam'''							,  metrics=['''binary_accuracy'''] )
																			model.fit(__magic_name__							,  epochs=1 )

																			with tempfile.TemporaryDirectory() as tmpdirname:
																									UpperCAmelCase_      :   List[str]      =	os.path.join(__magic_name__							,  '''keras_model.h5''' )
																									model.save(__magic_name__ )
																									if "T5" in main_layer_class.__name__:
																															UpperCAmelCase_      :   Any      =	tf.keras.models.load_model(
																															    __magic_name__							,  custom_objects={
																															        main_layer_class.__name__: main_layer_class,
																															        '''TFSharedEmbeddings''': TFSharedEmbeddings,
																															    }							,  )
																									else:
																															UpperCAmelCase_      :   Dict      =	tf.keras.models.load_model(
																															    __magic_name__							,  custom_objects={main_layer_class.__name__: main_layer_class} )
																									assert isinstance(__magic_name__							,  tf.keras.Model )
																									model(__magic_name__ )

							@slow
							def      UpperCAmelCase__						(				self				:  List[Any] )       ->							Any:







													"""simple docstring"""
													UpperCAmelCase_    ,							UpperCAmelCase_      :   Any      =	self.model_tester.prepare_config_and_inputs_for_common()

													for model_class in self.all_model_classes[:2]:
																			UpperCAmelCase_      :   Optional[Any]      =	model_class(__magic_name__ )

																			UpperCAmelCase_      :   Optional[Any]      =	copy.deepcopy(__magic_name__ )

																			if not self.is_encoder_decoder:
																									UpperCAmelCase_      :   Tuple      =	inputs['''input_ids''']
																									del inputs["input_ids"]
																			else:
																									UpperCAmelCase_      :   Dict      =	inputs['''input_ids''']
																									UpperCAmelCase_      :   str      =	inputs.get('''decoder_input_ids'''							,  __magic_name__ )
																									del inputs["input_ids"]
																									inputs.pop('''decoder_input_ids'''							,  __magic_name__ )

																			if not self.is_encoder_decoder:
																									UpperCAmelCase_      :   Optional[int]      =	model.get_input_embeddings()(__magic_name__ )
																			else:
																									UpperCAmelCase_      :   Tuple      =	model.get_input_embeddings()(__magic_name__ )
																									UpperCAmelCase_      :   List[str]      =	model.get_input_embeddings()(__magic_name__ )

																			UpperCAmelCase_      :   List[Any]      =	self._prepare_for_class(__magic_name__							,  __magic_name__ )

																			@tf.function
																			def run_in_graph_mode():
																									return model(__magic_name__ )

																			UpperCAmelCase_      :   List[str]      =	run_in_graph_mode()
																			self.assertIsNotNone(__magic_name__ )

							def      UpperCAmelCase__						(				self				:  int							,  __magic_name__				:  Any							,  __magic_name__				:  List[Any] )       ->							List[Any]:







													"""simple docstring"""
													# special tokens cannot be bad tokens
													UpperCAmelCase_      :   Any      =	[]
													if model.config.bos_token_id is not None:
																			special_tokens.append(model.config.bos_token_id )
													if model.config.pad_token_id is not None:
																			special_tokens.append(model.config.pad_token_id )
													if model.config.eos_token_id is not None:
																			special_tokens.append(model.config.eos_token_id )

													# create random bad tokens that are not special tokens
													UpperCAmelCase_      :   Optional[int]      =	[]
													while len(__magic_name__ ) < num_bad_tokens:
																			UpperCAmelCase_      :   Optional[Any]      =	tf.squeeze(ids_tensor((1, 1)							,  self.model_tester.vocab_size )							,  0 ).numpy()[0]
																			if token not in special_tokens:
																									bad_tokens.append(__magic_name__ )
													return bad_tokens

							def      UpperCAmelCase__						(				self				:  Optional[int]							,  __magic_name__				:  str )       ->							int:







													"""simple docstring"""
													for token_id in output_ids[0].numpy().tolist():
																			self.assertGreaterEqual(__magic_name__							,  0 )
																			self.assertLess(__magic_name__							,  self.model_tester.vocab_size )



							def      UpperCAmelCase__						(				self				:  Any							,  __magic_name__				:  Dict							,  __magic_name__				:  int )       ->							str:







													"""simple docstring"""
													# for all bad word tokens
													for bad_word_ids in bad_words_ids:
																			# for all slices in batch
																			for generated_ids_slice in generated_ids:
																									# for all word idx
																									for i in range(len(__magic_name__ )							,  len(__magic_name__ ) ):
																															# if tokens match
																															if generated_ids_slice[i - len(__magic_name__ ) : i] == bad_word_ids:
																																					return True
													return False






"""simple docstring"""




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


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

UpperCAmelCase       :    List[str]									=			"\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        __SCREAMING_SNAKE_CASE     (   _lowerCAmelCase							):
				UpperCAmelCase	=      42





class        __SCREAMING_SNAKE_CASE     (   _lowerCAmelCase							):







				def __init__(    self    :Optional[Any]      ,__UpperCAmelCase    :PriorTransformer      ,__UpperCAmelCase    :CLIPTextModelWithProjection      ,__UpperCAmelCase    :CLIPTokenizer      ,__UpperCAmelCase    :HeunDiscreteScheduler      ,__UpperCAmelCase    :ShapERenderer      ,)      ->					List[Any]:




									"""simple docstring"""



									super().__init__()

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







				def   lowercase_		(    self    :Any      ,__UpperCAmelCase    :Tuple      ,__UpperCAmelCase    :Dict      ,__UpperCAmelCase    :Optional[int]      ,__UpperCAmelCase    :Dict      ,__UpperCAmelCase    :Any      ,__UpperCAmelCase    :Dict  )      ->					Optional[Any]:




									"""simple docstring"""



									if latents is None:
														lowerCamelCase__		:   Optional[Any]  				=					randn_tensor(__UpperCAmelCase      ,generator=__UpperCAmelCase      ,device=__UpperCAmelCase      ,dtype=__UpperCAmelCase  )
									else:
														if latents.shape != shape:
																			raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {shape}"""  )
														lowerCamelCase__		:   Optional[Any]  				=					latents.to(__UpperCAmelCase  )

									lowerCamelCase__		:   Dict  				=					latents * scheduler.init_noise_sigma
									return latents







				def   lowercase_		(    self    :Dict      ,__UpperCAmelCase    :List[Any]=0  )      ->					str:




									"""simple docstring"""



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

									lowerCamelCase__		:   List[str]  				=					torch.device(F"""cuda:{gpu_id}"""  )

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







				def   lowercase_		(    self    :int      ,__UpperCAmelCase    :Dict=0  )      ->					Dict:




									"""simple docstring"""



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

									lowerCamelCase__		:   List[Any]  				=					torch.device(F"""cuda:{gpu_id}"""  )

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

									lowerCamelCase__		:   List[Any]  				=					None
									for cpu_offloaded_model in [self.text_encoder, self.prior, self.renderer]:
														lowerCamelCase__				,						lowerCamelCase__		:   Tuple  				=					cpu_offload_with_hook(__UpperCAmelCase      ,__UpperCAmelCase      ,prev_module_hook=__UpperCAmelCase  )

									if self.safety_checker is not None:
														lowerCamelCase__				,						lowerCamelCase__		:   Any  				=					cpu_offload_with_hook(self.safety_checker      ,__UpperCAmelCase      ,prev_module_hook=__UpperCAmelCase  )

									# We'll offload the last model manually.
									lowerCamelCase__		:   Dict  				=					hook







				@property
				def   lowercase_		(    self    :Dict  )      ->					Optional[Any]:




									"""simple docstring"""



									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(__UpperCAmelCase      ,'''_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    :Tuple      ,__UpperCAmelCase    :Optional[int]      ,__UpperCAmelCase    :Optional[Any]      ,__UpperCAmelCase    :Any      ,__UpperCAmelCase    :Optional[Any]      ,)      ->					str:




									"""simple docstring"""



									len(__UpperCAmelCase  ) if isinstance(__UpperCAmelCase      ,__UpperCAmelCase  ) else 1

									# YiYi Notes: set pad_token_id to be 0, not sure why I can't set in the config file
									lowerCamelCase__		:   Optional[int]  				=					0
									# get prompt text embeddings
									lowerCamelCase__		:   Union[str, Any]  				=					self.tokenizer(
									    __UpperCAmelCase      ,padding='''max_length'''      ,max_length=self.tokenizer.model_max_length      ,truncation=__UpperCAmelCase      ,return_tensors='''pt'''      ,)
									lowerCamelCase__		:   Union[str, Any]  				=					text_inputs.input_ids
									lowerCamelCase__		:   Any  				=					self.tokenizer(__UpperCAmelCase      ,padding='''longest'''      ,return_tensors='''pt'''  ).input_ids

									if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(__UpperCAmelCase      ,__UpperCAmelCase  ):
														lowerCamelCase__		:   List[str]  				=					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}"""  )

									lowerCamelCase__		:   Optional[int]  				=					self.text_encoder(text_input_ids.to(__UpperCAmelCase  )  )
									lowerCamelCase__		:   List[Any]  				=					text_encoder_output.text_embeds

									lowerCamelCase__		:   Optional[Any]  				=					prompt_embeds.repeat_interleave(__UpperCAmelCase      ,dim=0  )
									# in Shap-E it normalize the prompt_embeds and then later rescale it
									lowerCamelCase__		:   Union[str, Any]  				=					prompt_embeds / torch.linalg.norm(__UpperCAmelCase      ,dim=-1      ,keepdim=__UpperCAmelCase  )

									if do_classifier_free_guidance:
														lowerCamelCase__		:   Any  				=					torch.zeros_like(__UpperCAmelCase  )

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

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

									return prompt_embeds







				@torch.no_grad()
				@replace_example_docstring(__UpperCAmelCase  )
				def __call__(    self    :Any      ,__UpperCAmelCase    :str      ,__UpperCAmelCase    :int = 1      ,__UpperCAmelCase    :int = 25      ,__UpperCAmelCase    :Optional[Union[torch.Generator, List[torch.Generator]]] = None      ,__UpperCAmelCase    :Optional[torch.FloatTensor] = None      ,__UpperCAmelCase    :float = 4.0      ,__UpperCAmelCase    :int = 64      ,__UpperCAmelCase    :Optional[str] = "pil"      ,__UpperCAmelCase    :bool = True      ,)      ->					Tuple:




									"""simple docstring"""



									if isinstance(__UpperCAmelCase      ,__UpperCAmelCase  ):
														lowerCamelCase__		:   Union[str, Any]  				=					1
									elif isinstance(__UpperCAmelCase      ,__UpperCAmelCase  ):
														lowerCamelCase__		:   Any  				=					len(__UpperCAmelCase  )
									else:
														raise ValueError(F"""`prompt` has to be of type `str` or `list` but is {type(__UpperCAmelCase  )}"""  )

									lowerCamelCase__		:   Optional[Any]  				=					self._execution_device

									lowerCamelCase__		:   List[Any]  				=					batch_size * num_images_per_prompt

									lowerCamelCase__		:   Optional[int]  				=					guidance_scale > 1.0
									lowerCamelCase__		:   Tuple  				=					self._encode_prompt(__UpperCAmelCase      ,__UpperCAmelCase      ,__UpperCAmelCase      ,__UpperCAmelCase  )

									# prior

									self.scheduler.set_timesteps(__UpperCAmelCase      ,device=__UpperCAmelCase  )
									lowerCamelCase__		:   str  				=					self.scheduler.timesteps

									lowerCamelCase__		:   Dict  				=					self.prior.config.num_embeddings
									lowerCamelCase__		:   Optional[Any]  				=					self.prior.config.embedding_dim

									lowerCamelCase__		:   int  				=					self.prepare_latents(
									    (batch_size, num_embeddings * embedding_dim)      ,prompt_embeds.dtype      ,__UpperCAmelCase      ,__UpperCAmelCase      ,__UpperCAmelCase      ,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
									lowerCamelCase__		:   Tuple  				=					latents.reshape(latents.shape[0]      ,__UpperCAmelCase      ,__UpperCAmelCase  )

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

														lowerCamelCase__		:   Dict  				=					self.prior(
														    __UpperCAmelCase      ,timestep=__UpperCAmelCase      ,proj_embedding=__UpperCAmelCase      ,).predicted_image_embedding

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

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

														lowerCamelCase__		:   Union[str, Any]  				=					self.scheduler.step(
														    __UpperCAmelCase      ,timestep=__UpperCAmelCase      ,sample=__UpperCAmelCase      ,).prev_sample

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

									lowerCamelCase__		:   List[Any]  				=					[]
									for i, latent in enumerate(__UpperCAmelCase  ):
														lowerCamelCase__		:   Tuple  				=					self.renderer.decode(
														    latent[None, :]      ,__UpperCAmelCase      ,size=__UpperCAmelCase      ,ray_batch_size=40_96      ,n_coarse_samples=64      ,n_fine_samples=1_28      ,)
														images.append(__UpperCAmelCase  )

									lowerCamelCase__		:   Union[str, Any]  				=					torch.stack(__UpperCAmelCase  )

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

									lowerCamelCase__		:   Optional[int]  				=					images.cpu().numpy()

									if output_type == "pil":
														lowerCamelCase__		:   Optional[Any]  				=					[self.numpy_to_pil(__UpperCAmelCase  ) 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=__UpperCAmelCase  )






