




import time
import warnings
from abc import ABC
from copy import deepcopy
from typing import Optional

import torch

from ..utils import add_start_docstrings, logging


SCREAMING_SNAKE_CASE__     :     Union[str, Any]		    =       logging.get_logger(__name__)


SCREAMING_SNAKE_CASE__     :     Dict		    =       r"\n    Args:\n        input_ids (`torch.LongTensor` of shape `(batch_size, sequence_length)`):\n            Indices of input sequence tokens in the vocabulary.\n\n            Indices can be obtained using [`AutoTokenizer`]. See [`PreTrainedTokenizer.encode`] and\n            [`PreTrainedTokenizer.__call__`] for details.\n\n            [What are input IDs?](../glossary#input-ids)\n        scores (`torch.FloatTensor` of shape `(batch_size, config.vocab_size)`):\n            Prediction scores of a language modeling head. These can be scores for each vocabulary token before SoftMax\n            or scores for each vocabulary token after SoftMax.\n        kwargs (`Dict[str, Any]`, *optional*):\n            Additional stopping criteria specific kwargs.\n\n    Return:\n        `bool`. `False` indicates we should continue, `True` indicates we should stop.\n\n"



class       a_       (     SCREAMING_SNAKE_CASE__     ):





	@add_start_docstrings(SCREAMING_SNAKE_CASE      )
	def __call__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ,     **SCREAMING_SNAKE_CASE      )  ->					bool:




				"""simple docstring"""



				raise NotImplementedError('StoppingCriteria needs to be subclassed'      )




class       a_       (     SCREAMING_SNAKE_CASE__     ):





	def __init__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE = None      )  ->					List[Any]:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						max_length
				SCREAMING_SNAKE_CASE_								=						max_position_embeddings





	@add_start_docstrings(SCREAMING_SNAKE_CASE      )
	def __call__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ,     **SCREAMING_SNAKE_CASE      )  ->					bool:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						input_ids.shape[-1]
				SCREAMING_SNAKE_CASE_								=						cur_len >= self.max_length
				if self.max_position_embeddings is not None and not is_done and cur_len >= self.max_position_embeddings:
							logger.warning_once(
							    'This is a friendly reminder - the current text generation call will exceed the model\'s predefined '
							    f'maximum length ({self.max_position_embeddings}). Depending on the model, you may observe '
							    'exceptions, performance degradation, or nothing at all.'      )
				return is_done




class       a_       (     SCREAMING_SNAKE_CASE__     ):





	def __init__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      )  ->					Any:




				"""simple docstring"""



				warnings.warn(
				    'The class `MaxNewTokensCriteria` is deprecated. '
				    f'Please use `MaxLengthCriteria(max_length={start_length + max_new_tokens})` '
				    'with `max_length = start_length + max_new_tokens` instead.'      ,     SCREAMING_SNAKE_CASE      ,     )
				SCREAMING_SNAKE_CASE_								=						start_length
				SCREAMING_SNAKE_CASE_								=						max_new_tokens
				SCREAMING_SNAKE_CASE_								=						start_length + max_new_tokens





	@add_start_docstrings(SCREAMING_SNAKE_CASE      )
	def __call__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ,     **SCREAMING_SNAKE_CASE      )  ->					bool:




				"""simple docstring"""



				return input_ids.shape[-1] >= self.max_length




class       a_       (     SCREAMING_SNAKE_CASE__     ):





	def __init__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE = None      )  ->					Optional[int]:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						max_time
				SCREAMING_SNAKE_CASE_								=						time.time() if initial_timestamp is None else initial_timestamp





	@add_start_docstrings(SCREAMING_SNAKE_CASE      )
	def __call__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ,     **SCREAMING_SNAKE_CASE      )  ->					bool:




				"""simple docstring"""



				return time.time() - self.initial_timestamp > self.max_time




class       a_       (     SCREAMING_SNAKE_CASE__     ):





	@add_start_docstrings(SCREAMING_SNAKE_CASE      )
	def __call__(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ,     **SCREAMING_SNAKE_CASE      )  ->					bool:




				"""simple docstring"""



				return any(criteria(SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ) for criteria in self      )





	@property
	def 						A_(							self      )  ->					Optional[int]:




				"""simple docstring"""



				for stopping_criterium in self:
							if isinstance(SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ):
										return stopping_criterium.max_length
							elif isinstance(SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      ):
										return stopping_criterium.max_length
				return None




def lowercase    (  SCREAMING_SNAKE_CASE       ,    SCREAMING_SNAKE_CASE )			->							StoppingCriteriaList:


			'''simple docstring'''







			SCREAMING_SNAKE_CASE_								=						stopping_criteria.max_length
			SCREAMING_SNAKE_CASE_								=						deepcopy(SCREAMING_SNAKE_CASE )
			if stopping_max_length is not None and stopping_max_length != max_length:
						warnings.warn('You set different `max_length` for stopping criteria and `max_length` parameter'       ,    SCREAMING_SNAKE_CASE )
			elif stopping_max_length is None:
						new_stopping_criteria.append(MaxLengthCriteria(max_length=SCREAMING_SNAKE_CASE ) )
			return new_stopping_criteria






import gc
import unittest

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

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference



class       a_       (     SCREAMING_SNAKE_CASE__  ,    unittest.TestCase     ):
	A    							= ShapEPipeline
	A    							= ['''prompt''']
	A    							= ['''prompt''']
	A    							= [
	    '''num_images_per_prompt''',
	    '''num_inference_steps''',
	    '''generator''',
	    '''latents''',
	    '''guidance_scale''',
	    '''frame_size''',
	    '''output_type''',
	    '''return_dict''',
	]
	A    							= False





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




				"""simple docstring"""



				return 32





	@property
	def 						A_(							self      )  ->					Dict:




				"""simple docstring"""



				return 32





	@property
	def 						A_(							self      )  ->					str:




				"""simple docstring"""



				return self.time_input_dim * 4





	@property
	def 						A_(							self      )  ->					str:




				"""simple docstring"""



				return 8





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




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )
				return tokenizer





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




				"""simple docstring"""



				torch.manual_seed(0      )
				SCREAMING_SNAKE_CASE_								=						CLIPTextConfig(
				    bos_token_id=0      ,     eos_token_id=2      ,     hidden_size=self.text_embedder_hidden_size      ,     projection_dim=self.text_embedder_hidden_size      ,     intermediate_size=37      ,     layer_norm_eps=1e-05      ,     num_attention_heads=4      ,     num_hidden_layers=5      ,     pad_token_id=1      ,     vocab_size=1000      ,     )
				return CLIPTextModelWithProjection(SCREAMING_SNAKE_CASE      )





	@property
	def 						A_(							self      )  ->					List[Any]:




				"""simple docstring"""



				torch.manual_seed(0      )

				SCREAMING_SNAKE_CASE_								=						{
				    'num_attention_heads': 2,
				    'attention_head_dim': 16,
				    'embedding_dim': self.time_input_dim,
				    'num_embeddings': 32,
				    'embedding_proj_dim': self.text_embedder_hidden_size,
				    'time_embed_dim': self.time_embed_dim,
				    'num_layers': 1,
				    'clip_embed_dim': self.time_input_dim * 2,
				    'additional_embeddings': 0,
				    'time_embed_act_fn': 'gelu',
				    'norm_in_type': 'layer',
				    'encoder_hid_proj_type': None,
				    'added_emb_type': None,
				}

				SCREAMING_SNAKE_CASE_								=						PriorTransformer(**SCREAMING_SNAKE_CASE      )
				return model





	@property
	def 						A_(							self      )  ->					Union[str, Any]:




				"""simple docstring"""



				torch.manual_seed(0      )

				SCREAMING_SNAKE_CASE_								=						{
				    'param_shapes': (
				        (self.renderer_dim, 93),
				        (self.renderer_dim, 8),
				        (self.renderer_dim, 8),
				        (self.renderer_dim, 8),
				    ),
				    'd_latent': self.time_input_dim,
				    'd_hidden': self.renderer_dim,
				    'n_output': 12,
				    'background': (
				        0.1,
				        0.1,
				        0.1,
				    ),
				}
				SCREAMING_SNAKE_CASE_								=						ShapERenderer(**SCREAMING_SNAKE_CASE      )
				return model





	def 						A_(							self      )  ->					int:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						self.dummy_prior
				SCREAMING_SNAKE_CASE_								=						self.dummy_text_encoder
				SCREAMING_SNAKE_CASE_								=						self.dummy_tokenizer
				SCREAMING_SNAKE_CASE_								=						self.dummy_renderer

				SCREAMING_SNAKE_CASE_								=						HeunDiscreteScheduler(
				    beta_schedule='exp'      ,     num_train_timesteps=1024      ,     prediction_type='sample'      ,     use_karras_sigmas=SCREAMING_SNAKE_CASE      ,     clip_sample=SCREAMING_SNAKE_CASE      ,     clip_sample_range=1.0      ,     )
				SCREAMING_SNAKE_CASE_								=						{
				    'prior': prior,
				    'text_encoder': text_encoder,
				    'tokenizer': tokenizer,
				    'renderer': renderer,
				    'scheduler': scheduler,
				}

				return components





	def 						A_(							self      ,     SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE=0      )  ->					str:




				"""simple docstring"""



				if str(SCREAMING_SNAKE_CASE      ).startswith('mps'      ):
							SCREAMING_SNAKE_CASE_								=						torch.manual_seed(SCREAMING_SNAKE_CASE      )
				else:
							SCREAMING_SNAKE_CASE_								=						torch.Generator(device=SCREAMING_SNAKE_CASE      ).manual_seed(SCREAMING_SNAKE_CASE      )
				SCREAMING_SNAKE_CASE_								=						{
				    'prompt': 'horse',
				    'generator': generator,
				    'num_inference_steps': 1,
				    'frame_size': 32,
				    'output_type': 'np',
				}
				return inputs





	def 						A_(							self      )  ->					List[Any]:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						'cpu'

				SCREAMING_SNAKE_CASE_								=						self.get_dummy_components()

				SCREAMING_SNAKE_CASE_								=						self.pipeline_class(**SCREAMING_SNAKE_CASE      )
				SCREAMING_SNAKE_CASE_								=						pipe.to(SCREAMING_SNAKE_CASE      )

				pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )

				SCREAMING_SNAKE_CASE_								=						pipe(**self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )      )
				SCREAMING_SNAKE_CASE_								=						output.images[0]
				SCREAMING_SNAKE_CASE_								=						image[0, -3:, -3:, -1]

				assert image.shape == (20, 32, 32, 3)

				SCREAMING_SNAKE_CASE_								=						np.array(
				    [
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				        0.0_0_0_3_9_2_1_6,
				    ]      )

				assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2





	def 						A_(							self      )  ->					List[str]:




				"""simple docstring"""



				self._test_inference_batch_consistent(batch_sizes=[1, 2]      )





	def 						A_(							self      )  ->					str:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						torch_device == 'cpu'
				SCREAMING_SNAKE_CASE_								=						True

				self._test_inference_batch_single_identical(
				    batch_size=2      ,     test_max_difference=SCREAMING_SNAKE_CASE      ,     relax_max_difference=SCREAMING_SNAKE_CASE      ,     )





	def 						A_(							self      )  ->					Optional[Any]:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						self.get_dummy_components()
				SCREAMING_SNAKE_CASE_								=						self.pipeline_class(**SCREAMING_SNAKE_CASE      )
				SCREAMING_SNAKE_CASE_								=						pipe.to(SCREAMING_SNAKE_CASE      )
				pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )

				SCREAMING_SNAKE_CASE_								=						1
				SCREAMING_SNAKE_CASE_								=						2

				SCREAMING_SNAKE_CASE_								=						self.get_dummy_inputs(SCREAMING_SNAKE_CASE      )

				for key in inputs.keys():
							if key in self.batch_params:
										SCREAMING_SNAKE_CASE_								=						batch_size * [inputs[key]]

				SCREAMING_SNAKE_CASE_								=						pipe(**SCREAMING_SNAKE_CASE      ,     num_images_per_prompt=SCREAMING_SNAKE_CASE      )[0]

				assert images.shape[0] == batch_size * num_images_per_prompt




@slow
@require_torch_gpu
class       a_       (     unittest.TestCase     ):





	def 						A_(							self      )  ->					List[str]:




				"""simple docstring"""



				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()





	def 						A_(							self      )  ->					List[str]:




				"""simple docstring"""



				SCREAMING_SNAKE_CASE_								=						load_numpy(
				    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
				    '/shap_e/test_shap_e_np_out.npy'      )
				SCREAMING_SNAKE_CASE_								=						ShapEPipeline.from_pretrained('openai/shap-e'      )
				SCREAMING_SNAKE_CASE_								=						pipe.to(SCREAMING_SNAKE_CASE      )
				pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE      )

				SCREAMING_SNAKE_CASE_								=						torch.Generator(device=SCREAMING_SNAKE_CASE      ).manual_seed(0      )

				SCREAMING_SNAKE_CASE_								=						pipe(
				    'a shark'      ,     generator=SCREAMING_SNAKE_CASE      ,     guidance_scale=1_5.0      ,     num_inference_steps=64      ,     frame_size=64      ,     output_type='np'      ,     ).images[0]

				assert images.shape == (20, 64, 64, 3)

				assert_mean_pixel_difference(SCREAMING_SNAKE_CASE      ,     SCREAMING_SNAKE_CASE      )
