


'''simple docstring'''




import unittest

from transformers import MobileBertConfig, is_torch_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
				import torch

				from transformers import (
				    MODEL_FOR_PRETRAINING_MAPPING,
				    MobileBertForMaskedLM,
				    MobileBertForMultipleChoice,
				    MobileBertForNextSentencePrediction,
				    MobileBertForPreTraining,
				    MobileBertForQuestionAnswering,
				    MobileBertForSequenceClassification,
				    MobileBertForTokenClassification,
				    MobileBertModel,
				)



class  lowerCAmelCase				:



	def __init__(					self ,					snake_case__ ,					snake_case__=13 ,					snake_case__=7 ,					snake_case__=True ,					snake_case__=True ,					snake_case__=True ,					snake_case__=True ,					snake_case__=99 ,					snake_case__=64 ,					snake_case__=32 ,					snake_case__=5 ,					snake_case__=4 ,					snake_case__=37 ,					snake_case__="gelu" ,					snake_case__=0.1 ,					snake_case__=0.1 ,					snake_case__=512 ,					snake_case__=16 ,					snake_case__=2 ,					snake_case__=0.0_2 ,					snake_case__=3 ,					snake_case__=4 ,					snake_case__=None ,					):
		lowerCAmelCase					:    str      							= parent
		lowerCAmelCase					:    Optional[int]      							= batch_size
		lowerCAmelCase					:    Optional[Any]      							= seq_length
		lowerCAmelCase					:    Optional[Any]      							= is_training
		lowerCAmelCase					:    Dict      							= use_input_mask
		lowerCAmelCase					:    Tuple      							= use_token_type_ids
		lowerCAmelCase					:    int      							= use_labels
		lowerCAmelCase					:    int      							= vocab_size
		lowerCAmelCase					:    Any      							= hidden_size
		lowerCAmelCase					:    Optional[Any]      							= embedding_size
		lowerCAmelCase					:    int      							= num_hidden_layers
		lowerCAmelCase					:    List[str]      							= num_attention_heads
		lowerCAmelCase					:    List[Any]      							= intermediate_size
		lowerCAmelCase					:    Dict      							= hidden_act
		lowerCAmelCase					:    Optional[int]      							= hidden_dropout_prob
		lowerCAmelCase					:    int      							= attention_probs_dropout_prob
		lowerCAmelCase					:    List[Any]      							= max_position_embeddings
		lowerCAmelCase					:    int      							= type_vocab_size
		lowerCAmelCase					:    List[str]      							= type_sequence_label_size
		lowerCAmelCase					:    Dict      							= initializer_range
		lowerCAmelCase					:    Any      							= num_labels
		lowerCAmelCase					:    str      							= num_choices
		lowerCAmelCase					:    int      							= scope



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    List[Any]      							= ids_tensor([self.batch_size, self.seq_length] ,					self.vocab_size    )

		lowerCAmelCase					:    Union[str, Any]      							= None
		if self.use_input_mask:
			lowerCAmelCase					:    Dict      							= random_attention_mask([self.batch_size, self.seq_length]    )

		lowerCAmelCase					:    Optional[int]      							= None
		if self.use_token_type_ids:
			lowerCAmelCase					:    List[str]      							= ids_tensor([self.batch_size, self.seq_length] ,					self.type_vocab_size    )

		lowerCAmelCase					:    Optional[Any]      							= None
		lowerCAmelCase					:    Optional[Any]      							= None
		lowerCAmelCase					:    Dict      							= None
		if self.use_labels:
			lowerCAmelCase					:    List[Any]      							= ids_tensor([self.batch_size] ,					self.type_sequence_label_size    )
			lowerCAmelCase					:    Any      							= ids_tensor([self.batch_size, self.seq_length] ,					self.num_labels    )
			lowerCAmelCase					:    Union[str, Any]      							= ids_tensor([self.batch_size] ,					self.num_choices    )

		lowerCAmelCase					:    Tuple      							= self.get_config()

		return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels



	def 				lowercase     (					self    ):
		return MobileBertConfig(
		    vocab_size=self.vocab_size ,					hidden_size=self.hidden_size ,					num_hidden_layers=self.num_hidden_layers ,					num_attention_heads=self.num_attention_heads ,					intermediate_size=self.intermediate_size ,					embedding_size=self.embedding_size ,					hidden_act=self.hidden_act ,					hidden_dropout_prob=self.hidden_dropout_prob ,					attention_probs_dropout_prob=self.attention_probs_dropout_prob ,					max_position_embeddings=self.max_position_embeddings ,					type_vocab_size=self.type_vocab_size ,					is_decoder=snake_case__ ,					initializer_range=self.initializer_range ,					)



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    Optional[int]      							= MobileBertModel(config=snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    int      							= model(snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__    )
		lowerCAmelCase					:    Optional[int]      							= model(snake_case__ ,					token_type_ids=snake_case__    )
		lowerCAmelCase					:    Optional[Any]      							= model(snake_case__    )

		self.parent.assertEqual(result.last_hidden_state.shape ,					(self.batch_size, self.seq_length, self.hidden_size)    )
		self.parent.assertEqual(result.pooler_output.shape ,					(self.batch_size, self.hidden_size)    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    int      							= MobileBertForMaskedLM(config=snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    str      							= model(snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__ ,					labels=snake_case__    )
		self.parent.assertEqual(result.logits.shape ,					(self.batch_size, self.seq_length, self.vocab_size)    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    Union[str, Any]      							= MobileBertForNextSentencePrediction(config=snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    str      							= model(
		    snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__ ,					labels=snake_case__ ,					)
		self.parent.assertEqual(result.logits.shape ,					(self.batch_size, 2)    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    List[Any]      							= MobileBertForPreTraining(config=snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    Tuple      							= model(
		    snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__ ,					labels=snake_case__ ,					next_sentence_label=snake_case__ ,					)
		self.parent.assertEqual(result.prediction_logits.shape ,					(self.batch_size, self.seq_length, self.vocab_size)    )
		self.parent.assertEqual(result.seq_relationship_logits.shape ,					(self.batch_size, 2)    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    Union[str, Any]      							= MobileBertForQuestionAnswering(config=snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    List[str]      							= model(
		    snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__ ,					start_positions=snake_case__ ,					end_positions=snake_case__ ,					)
		self.parent.assertEqual(result.start_logits.shape ,					(self.batch_size, self.seq_length)    )
		self.parent.assertEqual(result.end_logits.shape ,					(self.batch_size, self.seq_length)    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    Optional[int]      							= self.num_labels
		lowerCAmelCase					:    List[Any]      							= MobileBertForSequenceClassification(snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    Union[str, Any]      							= model(snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__ ,					labels=snake_case__    )
		self.parent.assertEqual(result.logits.shape ,					(self.batch_size, self.num_labels)    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    Union[str, Any]      							= self.num_labels
		lowerCAmelCase					:    int      							= MobileBertForTokenClassification(config=snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    Union[str, Any]      							= model(snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__ ,					labels=snake_case__    )
		self.parent.assertEqual(result.logits.shape ,					(self.batch_size, self.seq_length, self.num_labels)    )



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__ ,					snake_case__    ):
		lowerCAmelCase					:    List[str]      							= self.num_choices
		lowerCAmelCase					:    Any      							= MobileBertForMultipleChoice(config=snake_case__    )
		model.to(snake_case__    )
		model.eval()
		lowerCAmelCase					:    Dict      							= input_ids.unsqueeze(1    ).expand(-1 ,					self.num_choices ,					-1    ).contiguous()
		lowerCAmelCase					:    int      							= token_type_ids.unsqueeze(1    ).expand(-1 ,					self.num_choices ,					-1    ).contiguous()
		lowerCAmelCase					:    Union[str, Any]      							= input_mask.unsqueeze(1    ).expand(-1 ,					self.num_choices ,					-1    ).contiguous()
		lowerCAmelCase					:    List[str]      							= model(
		    snake_case__ ,					attention_mask=snake_case__ ,					token_type_ids=snake_case__ ,					labels=snake_case__ ,					)
		self.parent.assertEqual(result.logits.shape ,					(self.batch_size, self.num_choices)    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Any      							= self.prepare_config_and_inputs()
		(
		    (
		    lowerCAmelCase
		)							,   (
		    lowerCAmelCase
		)							,   (
		    lowerCAmelCase
		)							,   (
		    lowerCAmelCase
		)							,   (
		    lowerCAmelCase
		)							,   (
		    lowerCAmelCase
		)							,   (
		    lowerCAmelCase
		)							,   
		)					:    Optional[Any]      							= config_and_inputs
		lowerCAmelCase					:    List[Any]      							= {'input_ids': input_ids, 'token_type_ids': token_type_ids, 'attention_mask': input_mask}
		return config, inputs_dict



@require_torch
class  lowerCAmelCase				(				a      ,				a      ,				unittest.TestCase							):
	_lowerCamelCase   :					List[str]			      = (
	    (
	        MobileBertModel,
	        MobileBertForMaskedLM,
	        MobileBertForMultipleChoice,
	        MobileBertForNextSentencePrediction,
	        MobileBertForPreTraining,
	        MobileBertForQuestionAnswering,
	        MobileBertForSequenceClassification,
	        MobileBertForTokenClassification,
	    )
	    if is_torch_available()
	    else ()
	)
	_lowerCamelCase   :					Tuple			      = (
	    {
	        """feature-extraction""": MobileBertModel,
	        """fill-mask""": MobileBertForMaskedLM,
	        """question-answering""": MobileBertForQuestionAnswering,
	        """text-classification""": MobileBertForSequenceClassification,
	        """token-classification""": MobileBertForTokenClassification,
	        """zero-shot""": MobileBertForSequenceClassification,
	    }
	    if is_torch_available()
	    else {}
	)
	_lowerCamelCase   :					str			      = True



	def 				lowercase     (					self ,					snake_case__ ,					snake_case__ ,					snake_case__=False    ):
		lowerCAmelCase					:    int      							= super()._prepare_for_class(snake_case__ ,					snake_case__ ,					return_labels=snake_case__    )

		if return_labels:
			if model_class in get_values(snake_case__    ):
				lowerCAmelCase					:    str      							= torch.zeros(
				    (self.model_tester.batch_size, self.model_tester.seq_length) ,					dtype=torch.long ,					device=snake_case__    )
				lowerCAmelCase					:    Tuple      							= torch.zeros(
				    self.model_tester.batch_size ,					dtype=torch.long ,					device=snake_case__    )
		return inputs_dict



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    List[Any]      							= MobileBertModelTester(self    )
		lowerCAmelCase					:    Dict      							= ConfigTester(self ,					config_class=snake_case__ ,					hidden_size=37    )



	def 				lowercase     (					self    ):
		self.config_tester.run_common_tests()



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    int      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_model(*snake_case__    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Union[str, Any]      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_for_masked_lm(*snake_case__    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Tuple      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_for_multiple_choice(*snake_case__    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Union[str, Any]      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_for_next_sequence_prediction(*snake_case__    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    str      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_for_pretraining(*snake_case__    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    int      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_for_question_answering(*snake_case__    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Any      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_for_sequence_classification(*snake_case__    )



	def 				lowercase     (					self    ):
		lowerCAmelCase					:    Optional[Any]      							= self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_mobilebert_for_token_classification(*snake_case__    )


def 			__UpperCamelCase    (    _A    :				Optional[Any]				)    ->      Optional[int]:




	"""simple docstring"""





	return torch.tensor(
	    _A					,   dtype=torch.long					,   device=_A					,   )


_lowerCAmelCase  :		Union[str, Any]		     =	1E-3






@require_torch
@require_sentencepiece
@require_tokenizers
class  lowerCAmelCase				(				unittest.TestCase							):



	@slow
	def 				lowercase     (					self    ):
		lowerCAmelCase					:    List[str]      							= MobileBertModel.from_pretrained('google/mobilebert-uncased'    ).to(snake_case__    )
		lowerCAmelCase					:    List[Any]      							= _long_tensor([[101, 7110, 1005, 1056, 2023, 1_1333, 1_7413, 1029, 102]]    )
		with torch.no_grad():
			lowerCAmelCase					:    Tuple      							= model(snake_case__    )[0]
		lowerCAmelCase					:    List[Any]      							= torch.Size((1, 9, 512)    )
		self.assertEqual(output.shape ,					snake_case__    )
		lowerCAmelCase					:    Union[str, Any]      							= torch.tensor(
		    [
		        [
		            [-2.4_7_3_6_5_2_6e0_7, 8.2_6_9_1_6_5_6e0_4, 1.6_5_2_1_8_3_8e0_5],
		            [-5.7_5_4_1_7_0_4e-0_1, 3.9_0_5_6_0_2_2e0_0, 4.4_0_1_1_5_0_7e0_0],
		            [2.6_0_4_7_3_5_9e0_0, 1.5_6_7_7_6_5_2e0_0, -1.7_3_2_4_1_8_8e-0_1],
		        ]
		    ] ,					device=snake_case__ ,					)

		# MobileBERT results range from 10e0 to 10e8. Even a 0.0000001% difference with a value of 10e8 results in a
		# ~1 difference, it's therefore not a good idea to measure using addition.
		# Here, we instead divide the expected result with the result in order to obtain ~1. We then check that the
		# result is held between bounds: 1 - TOLERANCE < expected_result / result < 1 + TOLERANCE
		lowerCAmelCase					:    List[str]      							= torch.all((expected_slice / output[..., :3, :3]) >= 1 - TOLERANCE    )
		lowerCAmelCase					:    Dict      							= torch.all((expected_slice / output[..., :3, :3]) <= 1 + TOLERANCE    )

		self.assertTrue(lower_bound and upper_bound    )




'''simple docstring'''




import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


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



class  lowerCAmelCase				(				a							):



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

		if safety_checker is None:
			logger.warning(
			    f"You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"
			    ' that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered'
			    ' results in services or applications open to the public. Both the diffusers team and Hugging Face'
			    ' strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling'
			    ' it only for use-cases that involve analyzing network behavior or auditing its results. For more'
			    ' information, please have a look at https://github.com/huggingface/diffusers/pull/254 .'    )

		self.register_modules(
		    speech_model=snake_case__ ,					speech_processor=snake_case__ ,					vae=snake_case__ ,					text_encoder=snake_case__ ,					tokenizer=snake_case__ ,					unet=snake_case__ ,					scheduler=snake_case__ ,					feature_extractor=snake_case__ ,					)



	def 				lowercase     (					self ,					snake_case__ = "auto"    ):
		if slice_size == "auto":
			lowerCAmelCase					:    Union[str, Any]      							= self.unet.config.attention_head_dim // 2
		self.unet.set_attention_slice(snake_case__    )



	def 				lowercase     (					self    ):
		self.enable_attention_slicing(snake_case__    )



	@torch.no_grad()
	def __call__(					self ,					snake_case__ ,					snake_case__=1_6000 ,					snake_case__ = 512 ,					snake_case__ = 512 ,					snake_case__ = 50 ,					snake_case__ = 7.5 ,					snake_case__ = None ,					snake_case__ = 1 ,					snake_case__ = 0.0 ,					snake_case__ = None ,					snake_case__ = None ,					snake_case__ = "pil" ,					snake_case__ = True ,					snake_case__ = None ,					snake_case__ = 1 ,					**snake_case__ ,					):
		lowerCAmelCase					:    List[str]      							= self.speech_processor.feature_extractor(
		    snake_case__ ,					return_tensors='pt' ,					sampling_rate=snake_case__    ).input_features.to(self.device    )
		lowerCAmelCase					:    Optional[Any]      							= self.speech_model.generate(snake_case__ ,					max_length=48_0000    )

		lowerCAmelCase					:    str      							= self.speech_processor.tokenizer.batch_decode(snake_case__ ,					skip_special_tokens=snake_case__ ,					normalize=snake_case__    )[
		    0
		]

		if isinstance(snake_case__ ,					snake_case__    ):
			lowerCAmelCase					:    Optional[int]      							= 1
		elif isinstance(snake_case__ ,					snake_case__    ):
			lowerCAmelCase					:    Optional[int]      							= len(snake_case__    )
		else:
			raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(snake_case__    )}"    )

		if height % 8 != 0 or width % 8 != 0:
			raise ValueError(f"`height` and `width` have to be divisible by 8 but are {height} and {width}."    )

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

		# get prompt text embeddings
		lowerCAmelCase					:    str      							= self.tokenizer(
		    snake_case__ ,					padding='max_length' ,					max_length=self.tokenizer.model_max_length ,					return_tensors='pt' ,					)
		lowerCAmelCase					:    Tuple      							= text_inputs.input_ids

		if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
			lowerCAmelCase					:    str      							= self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]    )
			logger.warning(
			    'The following part of your input was truncated because CLIP can only handle sequences up to'
			    f" {self.tokenizer.model_max_length} tokens: {removed_text}"    )
			lowerCAmelCase					:    Union[str, Any]      							= text_input_ids[:, : self.tokenizer.model_max_length]
		lowerCAmelCase					:    Union[str, Any]      							= self.text_encoder(text_input_ids.to(self.device    )    )[0]

		# duplicate text embeddings for each generation per prompt, using mps friendly method
		lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase					:    int      							= text_embeddings.shape
		lowerCAmelCase					:    Any      							= text_embeddings.repeat(1 ,					snake_case__ ,					1    )
		lowerCAmelCase					:    Optional[int]      							= text_embeddings.view(bs_embed * num_images_per_prompt ,					snake_case__ ,					-1    )

		# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
		# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
		# corresponds to doing no classifier free guidance.
		lowerCAmelCase					:    List[str]      							= guidance_scale > 1.0
		# get unconditional embeddings for classifier free guidance
		if do_classifier_free_guidance:
			lowerCAmelCase					:    List[str]
			if negative_prompt is None:
				lowerCAmelCase					:    Any      							= [''] * batch_size
			elif type(snake_case__    ) is not type(snake_case__    ):
				raise TypeError(
				    f"`negative_prompt` should be the same type to `prompt`, but got {type(snake_case__    )} !="
				    f" {type(snake_case__    )}."    )
			elif isinstance(snake_case__ ,					snake_case__    ):
				lowerCAmelCase					:    Union[str, Any]      							= [negative_prompt]
			elif batch_size != len(snake_case__    ):
				raise ValueError(
				    f"`negative_prompt`: {negative_prompt} has batch size {len(snake_case__    )}, but `prompt`:"
				    f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
				    ' the batch size of `prompt`.'    )
			else:
				lowerCAmelCase					:    Dict      							= negative_prompt

			lowerCAmelCase					:    Optional[int]      							= text_input_ids.shape[-1]
			lowerCAmelCase					:    int      							= self.tokenizer(
			    snake_case__ ,					padding='max_length' ,					max_length=snake_case__ ,					truncation=snake_case__ ,					return_tensors='pt' ,					)
			lowerCAmelCase					:    Union[str, Any]      							= self.text_encoder(uncond_input.input_ids.to(self.device    )    )[0]

			# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
			lowerCAmelCase					:    List[Any]      							= uncond_embeddings.shape[1]
			lowerCAmelCase					:    List[str]      							= uncond_embeddings.repeat(1 ,					snake_case__ ,					1    )
			lowerCAmelCase					:    Optional[Any]      							= uncond_embeddings.view(batch_size * num_images_per_prompt ,					snake_case__ ,					-1    )

			# For classifier free guidance, we need to do two forward passes.
			# Here we concatenate the unconditional and text embeddings into a single batch
			# to avoid doing two forward passes
			lowerCAmelCase					:    List[str]      							= torch.cat([uncond_embeddings, text_embeddings]    )

		# get the initial random noise unless the user supplied it

		# Unlike in other pipelines, latents need to be generated in the target device
		# for 1-to-1 results reproducibility with the CompVis implementation.
		# However this currently doesn't work in `mps`.
		lowerCAmelCase					:    Union[str, Any]      							= (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
		lowerCAmelCase					:    Dict      							= text_embeddings.dtype
		if latents is None:
			if self.device.type == "mps":
				# randn does not exist on mps
				lowerCAmelCase					:    str      							= torch.randn(snake_case__ ,					generator=snake_case__ ,					device='cpu' ,					dtype=snake_case__    ).to(
				    self.device    )
			else:
				lowerCAmelCase					:    Tuple      							= torch.randn(snake_case__ ,					generator=snake_case__ ,					device=self.device ,					dtype=snake_case__    )
		else:
			if latents.shape != latents_shape:
				raise ValueError(f"Unexpected latents shape, got {latents.shape}, expected {latents_shape}"    )
			lowerCAmelCase					:    str      							= latents.to(self.device    )

		# set timesteps
		self.scheduler.set_timesteps(snake_case__    )

		# Some schedulers like PNDM have timesteps as arrays
		# It's more optimized to move all timesteps to correct device beforehand
		lowerCAmelCase					:    Union[str, Any]      							= self.scheduler.timesteps.to(self.device    )

		# scale the initial noise by the standard deviation required by the scheduler
		lowerCAmelCase					:    Any      							= latents * self.scheduler.init_noise_sigma

		# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
		# eta (η) is only used with the DDIMScheduler, it will be ignored for other schedulers.
		# eta corresponds to η in DDIM paper: https://arxiv.org/abs/2010.02502
		# and should be between [0, 1]
		lowerCAmelCase					:    Tuple      							= 'eta' in set(inspect.signature(self.scheduler.step    ).parameters.keys()    )
		lowerCAmelCase					:    Union[str, Any]      							= {}
		if accepts_eta:
			lowerCAmelCase					:    int      							= eta

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

			# predict the noise residual
			lowerCAmelCase					:    List[str]      							= self.unet(snake_case__ ,					snake_case__ ,					encoder_hidden_states=snake_case__    ).sample

			# perform guidance
			if do_classifier_free_guidance:
				lowerCAmelCase							,   lowerCAmelCase					:    Dict      							= noise_pred.chunk(2    )
				lowerCAmelCase					:    Tuple      							= noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

			# compute the previous noisy sample x_t -> x_t-1
			lowerCAmelCase					:    int      							= self.scheduler.step(snake_case__ ,					snake_case__ ,					snake_case__ ,					**snake_case__    ).prev_sample

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

		lowerCAmelCase					:    List[Any]      							= 1 / 0.1_8_2_1_5 * latents
		lowerCAmelCase					:    Dict      							= self.vae.decode(snake_case__    ).sample

		lowerCAmelCase					:    List[Any]      							= (image / 2 + 0.5).clamp(0 ,					1    )

		# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
		lowerCAmelCase					:    Union[str, Any]      							= image.cpu().permute(0 ,					2 ,					3 ,					1    ).float().numpy()

		if output_type == "pil":
			lowerCAmelCase					:    Dict      							= self.numpy_to_pil(snake_case__    )

		if not return_dict:
			return image

		return StableDiffusionPipelineOutput(images=snake_case__ ,					nsfw_content_detected=snake_case__    )
