


import copy
import tempfile
import unittest

from transformers import MaMaaaConfig, is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device
from transformers.utils import cached_property

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
	import torch

	from transformers import MaMaaaForConditionalGeneration, MaMaaaModel, MaMaaaTokenizer
	from transformers.models.mam_aaa.modeling_mam_aaa import MaMaaaDecoder, MaMaaaEncoder






def 			UpperCamelCase (     __magic_name__					:							List[str]   ,   __magic_name__					:							Union[str, Any]   ,   __magic_name__					:							str   ,   __magic_name__					:							Optional[Any]=None   ,   __magic_name__					:							List[Any]=None   ,   __magic_name__					:							Any=None   ,   __magic_name__					:							Optional[int]=None   ,   __magic_name__					:							Tuple=None   ,   )       ->  List[Any]:
		"""simple docstring"""




		if attention_mask is None:
				lowercase__					      =		input_ids.ne(config.pad_token_id					)
		if decoder_attention_mask is None:
				lowercase__					      =		decoder_input_ids.ne(config.pad_token_id					)
		if head_mask is None:
				lowercase__					      =		torch.ones(config.encoder_layers   ,   config.encoder_attention_heads   ,   device=A__					)
		if decoder_head_mask is None:
				lowercase__					      =		torch.ones(config.decoder_layers   ,   config.decoder_attention_heads   ,   device=A__					)
		if cross_attn_head_mask is None:
				lowercase__					      =		torch.ones(config.decoder_layers   ,   config.decoder_attention_heads   ,   device=A__					)
		return {
		    "input_ids": input_ids,
		    "decoder_input_ids": decoder_input_ids,
		    "attention_mask": attention_mask,
		    "decoder_attention_mask": attention_mask,
		    "head_mask": head_mask,
		    "decoder_head_mask": decoder_head_mask,
		    "cross_attn_head_mask": cross_attn_head_mask,
		}




class 							A   :
					'''simple docstring'''






					def __init__(self   :						Optional[Any]				, _UpperCAmelCase   :						str				, _UpperCAmelCase   :						Dict=13				, _UpperCAmelCase   :						Dict=7				, _UpperCAmelCase   :						Optional[int]=True				, _UpperCAmelCase   :						Dict=False				, _UpperCAmelCase   :						int=99				, _UpperCAmelCase   :						Tuple=16				, _UpperCAmelCase   :						str=2				, _UpperCAmelCase   :						List[str]=4				, _UpperCAmelCase   :						Dict=4				, _UpperCAmelCase   :						Optional[Any]="relu"				, _UpperCAmelCase   :						str=0.1				, _UpperCAmelCase   :						Union[str, Any]=0.1				, _UpperCAmelCase   :						int=0.0				, _UpperCAmelCase   :						Dict=0.0				, _UpperCAmelCase   :						str=20				, _UpperCAmelCase   :						Union[str, Any]=2				, _UpperCAmelCase   :						Tuple=1				, _UpperCAmelCase   :						List[str]=0				, )						->       Dict:




							"""simple docstring"""


							lowercase__					      =		parent
							lowercase__					      =		batch_size
							lowercase__					      =		seq_length
							lowercase__					      =		is_training
							lowercase__					      =		use_labels
							lowercase__					      =		vocab_size
							lowercase__					      =		hidden_size
							lowercase__					      =		num_hidden_layers
							lowercase__					      =		num_attention_heads
							lowercase__					      =		intermediate_size
							lowercase__					      =		hidden_act
							lowercase__					      =		hidden_dropout_prob
							lowercase__					      =		attention_probs_dropout_prob
							lowercase__					      =		encoder_layerdrop
							lowercase__					      =		decoder_layerdrop
							lowercase__					      =		max_position_embeddings
							lowercase__					      =		eos_token_id
							lowercase__					      =		pad_token_id
							lowercase__					      =		bos_token_id

					def   lowerCamelCase__					(self   :						Optional[int]		)						->       Optional[int]:




							"""simple docstring"""


							lowercase__					      =		ids_tensor([self.batch_size, self.seq_length]				, self.vocab_size		)
							lowercase__					      =		self.eos_token_id  # Eos Token
							lowercase__					      =		ids_tensor([self.batch_size, self.seq_length]				, self.vocab_size		)

							# we need to clamp the input ids here to avoid having pad token in between
							# this is because for M2M100 the position_ids are prepared such that
							# all pad tokens have pos id = 2 and rest are between 2..seq_length
							# and the seq_length here is seq_length - num_pad_tokens
							# but when using past, there is no way of knowing if the past input ids had
							# pad tokens in them, which results in incorrect seq_lenth and which in turn results in
							# position_ids being off by num_pad_tokens in past input
							lowercase__					      =		input_ids.clamp(self.pad_token_id + 1		)
							lowercase__					      =		decoder_input_ids.clamp(self.pad_token_id + 1		)

							lowercase__					      =		self.get_config()
							lowercase__					      =		prepare_mam_aaa_inputs_dict(lowercase__				, lowercase__				, lowercase__		)
							return config, inputs_dict

					def   lowerCamelCase__					(self   :						Optional[Any]		)						->       str:




							"""simple docstring"""


							return MaMaaaConfig(
							    vocab_size=self.vocab_size				, d_model=self.hidden_size				, encoder_layers=self.num_hidden_layers				, decoder_layers=self.num_hidden_layers				, encoder_attention_heads=self.num_attention_heads				, decoder_attention_heads=self.num_attention_heads				, encoder_ffn_dim=self.intermediate_size				, decoder_ffn_dim=self.intermediate_size				, dropout=self.hidden_dropout_prob				, attention_dropout=self.attention_probs_dropout_prob				, encoder_layerdrop=self.encoder_layerdrop				, decoder_layerdrop=self.decoder_layerdrop				, max_position_embeddings=self.max_position_embeddings				, eos_token_id=self.eos_token_id				, bos_token_id=self.bos_token_id				, pad_token_id=self.pad_token_id				, )

					def   lowerCamelCase__					(self   :						Optional[int]		)						->       str:




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.prepare_config_and_inputs()
							return config, inputs_dict

					def   lowerCamelCase__					(self   :						Tuple				, _UpperCAmelCase   :						Optional[Any]				, _UpperCAmelCase   :						Any		)						->       Union[str, Any]:




							"""simple docstring"""


							lowercase__					      =		MaMaaaModel(config=lowercase__		).get_decoder().to(lowercase__		).eval()
							lowercase__					      =		inputs_dict["""input_ids"""]
							lowercase__					      =		inputs_dict["""attention_mask"""]
							lowercase__					      =		inputs_dict["""head_mask"""]

							# first forward pass
							lowercase__					      =		model(lowercase__				, attention_mask=lowercase__				, head_mask=lowercase__				, use_cache=lowercase__		)

							lowercase__      ,     lowercase__					      =		outputs.to_tuple()

							# create hypothetical multiple next token and extent to next_input_ids
							lowercase__					      =		ids_tensor((self.batch_size, 3)				, config.vocab_size		)
							lowercase__					      =		ids_tensor((self.batch_size, 3)				, 2		)

							# append to next input_ids and
							lowercase__					      =		torch.cat([input_ids, next_tokens]				, dim=-1		)
							lowercase__					      =		torch.cat([attention_mask, next_attn_mask]				, dim=-1		)

							lowercase__					      =		model(lowercase__				, attention_mask=lowercase__		)["""last_hidden_state"""]
							lowercase__					      =		model(lowercase__				, attention_mask=lowercase__				, past_key_values=lowercase__		)[
							    """last_hidden_state"""
							]

							# select random slice
							lowercase__					      =		ids_tensor((1,)				, output_from_past.shape[-1]		).item()
							lowercase__					      =		output_from_no_past[:, -3:, random_slice_idx].detach()
							lowercase__					      =		output_from_past[:, :, random_slice_idx].detach()

							self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]		)

							# test that outputs are equal for slice
							self.parent.assertTrue(torch.allclose(lowercase__				, lowercase__				, atol=1E-2		)		)


					def   lowerCamelCase__					(self   :						List[Any]				, _UpperCAmelCase   :						Any				, _UpperCAmelCase   :						Optional[Any]		)						->       Any:




							"""simple docstring"""


							lowercase__					      =		MaMaaaModel(config=lowercase__		).to(lowercase__		).eval()
							lowercase__					      =		model(**lowercase__		)

							lowercase__					      =		outputs.encoder_last_hidden_state
							lowercase__					      =		outputs.last_hidden_state

							with tempfile.TemporaryDirectory() as tmpdirname:
									lowercase__					      =		model.get_encoder()
									encoder.save_pretrained(lowercase__		)
									lowercase__					      =		MaMaaaEncoder.from_pretrained(lowercase__		).to(lowercase__		)

							lowercase__					      =		encoder(inputs_dict["""input_ids"""]				, attention_mask=inputs_dict["""attention_mask"""]		)[
							    0
							]

							self.parent.assertTrue((encoder_last_hidden_state_a - encoder_last_hidden_state).abs().max().item() < 1E-3		)

							with tempfile.TemporaryDirectory() as tmpdirname:
									lowercase__					      =		model.get_decoder()
									decoder.save_pretrained(lowercase__		)
									lowercase__					      =		MaMaaaDecoder.from_pretrained(lowercase__		).to(lowercase__		)

							lowercase__					      =		decoder(
							    input_ids=inputs_dict["""decoder_input_ids"""]				, attention_mask=inputs_dict["""decoder_attention_mask"""]				, encoder_hidden_states=lowercase__				, encoder_attention_mask=inputs_dict["""attention_mask"""]				, )[0]

							self.parent.assertTrue((last_hidden_state_a - last_hidden_state).abs().max().item() < 1E-3		)






@require_torch
class 							A   (	lowerCamelCase__			,  lowerCamelCase__			,  lowerCamelCase__			,  unittest.TestCase							):
					'''simple docstring'''






					A__      =			(
					    (
					        MaMaaaModel,
					        MaMaaaForConditionalGeneration,
					    )
					    if is_torch_available()
					    else ()
					)
					A__      =			(MaMaaaForConditionalGeneration,) if is_torch_available() else ()
					A__      =			(
					    {
					        'conversational': MaMaaaForConditionalGeneration,
					        'feature-extraction': MaMaaaModel,
					        'summarization': MaMaaaForConditionalGeneration,
					        'text2text-generation': MaMaaaForConditionalGeneration,
					        'translation': MaMaaaForConditionalGeneration,
					    }
					    if is_torch_available()
					    else {}
					)
					A__      =			True
					A__      =			True
					A__      =			False
					A__      =			False

					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						List[Any]				, _UpperCAmelCase   :						str				, _UpperCAmelCase   :						List[str]				, _UpperCAmelCase   :						Optional[int]				, _UpperCAmelCase   :						str		)						->       List[str]:




							"""simple docstring"""


							if pipeline_test_casse_name == "TranslationPipelineTests":
									# Get `ValueError: Translation requires a `src_lang` and a `tgt_lang` for this model`.
									# `M2M100Config` was never used in pipeline tests: cannot create a simple tokenizer.
									return True

							return False

					def   lowerCamelCase__					(self   :						List[str]		)						->       Dict:




							"""simple docstring"""


							lowercase__					      =		MaMaaaModelTester(self		)
							lowercase__					      =		ConfigTester(self				, config_class=lowercase__		)

					def   lowerCamelCase__					(self   :						List[Any]		)						->       Tuple:




							"""simple docstring"""


							self.config_tester.run_common_tests()

					def   lowerCamelCase__					(self   :						Union[str, Any]		)						->       int:




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.model_tester.prepare_config_and_inputs()
							for model_class in self.all_model_classes:
									lowercase__					      =		model_class(lowercase__		)

									with tempfile.TemporaryDirectory() as tmpdirname:
											model.save_pretrained(lowercase__		)
											lowercase__      ,     lowercase__					      =		model_class.from_pretrained(lowercase__				, output_loading_info=lowercase__		)
									self.assertEqual(info["""missing_keys"""]				, []		)

					def   lowerCamelCase__					(self   :						str		)						->       Optional[int]:




							"""simple docstring"""


							lowercase__					      =		self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_decoder_model_past_large_inputs(*lowercase__		)

					def   lowerCamelCase__					(self   :						Dict		)						->       Tuple:




							"""simple docstring"""


							lowercase__					      =		self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.check_encoder_decoder_model_standalone(*lowercase__		)

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




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in (MaMaaaModel, MaMaaaForConditionalGeneration):
									lowercase__					      =		model_class(lowercase__		)
									model.to(lowercase__		)
									model.eval()

									lowercase__					      =		copy.deepcopy(self._prepare_for_class(lowercase__				, lowercase__		)		)

									if not self.is_encoder_decoder:
											lowercase__					      =		inputs["""input_ids"""]
											del inputs["input_ids"]
									else:
											lowercase__					      =		inputs["""input_ids"""]
											lowercase__					      =		inputs.get("""decoder_input_ids"""				, lowercase__		)
											del inputs["input_ids"]
											inputs.pop("""decoder_input_ids"""				, lowercase__		)

									lowercase__					      =		model.get_input_embeddings()
									if not self.is_encoder_decoder:
											lowercase__					      =		wte(lowercase__		)
									else:
											lowercase__					      =		wte(lowercase__		)
											lowercase__					      =		wte(lowercase__		)

									with torch.no_grad():
											model(**lowercase__		)[0]


					def   lowerCamelCase__					(self   :						int		)						->       List[Any]:




							"""simple docstring"""


							lowercase__      ,     lowercase__					      =		self.model_tester.prepare_config_and_inputs()
							lowercase__					      =		input_dict["""input_ids"""]
							lowercase__					      =		input_ids.ne(1		).to(lowercase__		)
							lowercase__					      =		MaMaaaForConditionalGeneration(lowercase__		).eval().to(lowercase__		)
							if torch_device == "cuda":
									model.half()
							model.generate(lowercase__				, attention_mask=lowercase__		)
							model.generate(num_beams=4				, do_sample=lowercase__				, early_stopping=lowercase__				, num_return_sequences=3		)






def 			UpperCamelCase (     __magic_name__					:							str					)       ->  List[Any]:
		"""simple docstring"""




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


A	:		Union[str, Any]    =       1e-4







@require_torch
@require_sentencepiece
@require_tokenizers
@slow
class 							A   (	unittest.TestCase							):
					'''simple docstring'''






					@cached_property
					def   lowerCamelCase__					(self   :						List[Any]		)						->       Optional[Any]:




							"""simple docstring"""


							return MaMaaaTokenizer.from_pretrained("""facebook/m2m100_418M"""		)

					def   lowerCamelCase__					(self   :						Optional[int]		)						->       Dict:




							"""simple docstring"""


							lowercase__					      =		MaMaaaModel.from_pretrained("""facebook/m2m100_418M"""		).to(lowercase__		)
							lowercase__					      =		_long_tensor([[12_8028, 98, 12, 3_0527, 2732, 159, 7755, 6_1904, 3_9144, 38, 2]]		)
							lowercase__					      =		_long_tensor([[2, 12_8028, 98, 12, 3_0527, 2732, 159, 7755, 6_1904, 3_9144, 38]]		)
							lowercase__					      =		prepare_mam_aaa_inputs_dict(model.config				, lowercase__				, lowercase__		)
							with torch.no_grad():
									lowercase__					      =		model(**lowercase__		)[0]
							lowercase__					      =		torch.Size((1, 11, 1024)		)
							self.assertEqual(output.shape				, lowercase__		)
							# change to expected output here
							lowercase__					      =		torch.tensor(
							    [[-0.7_780, -0.1_676, 0.1_038], [-6.7_556, -1.3_992, 0.0_567], [-7.5_383, -0.5_920, -0.2_779]]				, device=lowercase__		)
							self.assertTrue(torch.allclose(output[:, :3, :3]				, lowercase__				, atol=lowercase__		)		)

					def   lowerCamelCase__					(self   :						Union[str, Any]		)						->       int:




							"""simple docstring"""


							lowercase__					      =		MaMaaaForConditionalGeneration.from_pretrained("""facebook/m2m100_418M"""		).to(lowercase__		)

							# change to intended input
							lowercase__					      =		_long_tensor([[12_8028, 98, 12, 3_0527, 2732, 159, 7755, 6_1904, 3_9144, 38, 2]]		)
							lowercase__					      =		_long_tensor([[2, 12_8028, 98, 12, 3_0527, 2732, 159, 7755, 6_1904, 3_9144, 38]]		)
							lowercase__					      =		prepare_mam_aaa_inputs_dict(model.config				, lowercase__				, lowercase__		)
							with torch.no_grad():
									lowercase__					      =		model(**lowercase__		)[0]
							lowercase__					      =		torch.Size((1, 11, model.config.vocab_size)		)
							self.assertEqual(output.shape				, lowercase__		)
							# change to expected output here
							lowercase__					      =		torch.tensor(
							    [[-1.0_448, -1.0_411, 3.7_992], [-3.2_191, -3.2_386, -1.3_451], [-3.6_210, -3.5_993, 0.4_925]]				, device=lowercase__		)
							self.assertTrue(torch.allclose(output[:, :3, :3]				, lowercase__				, atol=lowercase__		)		)


					def   lowerCamelCase__					(self   :						Union[str, Any]		)						->       Union[str, Any]:




							"""simple docstring"""


							lowercase__					      =		MaMaaaForConditionalGeneration.from_pretrained("""facebook/m2m100_418M"""		).to(lowercase__		)
							lowercase__					      =		MaMaaaTokenizer.from_pretrained("""facebook/m2m100_418M"""				, src_lang="""fr"""				, tgt_lang="""en"""		)

							lowercase__					      =		[
							    """L\'affaire NSA souligne l\'absence totale de débat sur le renseignement""",
							    """Selon moi, il y a deux niveaux de réponse de la part du gouvernement français.""",
							    """Lorsque François Hollande téléphone à Barack Obama ou quand le ministre des affaires étrangères Laurent"""
							    """ Fabius convoque l\'ambassadeur des Etats-Unis, ils réagissent à une vraie découverte, qui est celle de"""
							    """ l\'ampleur de la surveillance américaine sur l\'ensemble des communications en France.""",
							]

							# The below article tests that we don't add any hypotheses outside of the top n_beams
							lowercase__					      =		tokenizer(lowercase__				, padding=lowercase__				, return_tensors="""pt"""		)

							lowercase__					      =		model.generate(
							    input_ids=dct["""input_ids"""].to(lowercase__		)				, attention_mask=dct["""attention_mask"""].to(lowercase__		)				, num_beams=5				, forced_bos_token_id=tokenizer.get_lang_id("""en"""		)				, )

							lowercase__					      =		[
							    """The NSA case highlights the total absence of intelligence debate""",
							    """I think there are two levels of response from the French government.""",
							    """When François Hollande calls Barack Obama or when Foreign Minister Laurent Fabius calls the U.S."""
							    """ Ambassador, they respond to a real discovery, which is that of the scale of U.S. surveillance on all"""
							    """ communications in France.""",
							]

							lowercase__					      =		tokenizer.batch_decode(
							    hypotheses_batch.tolist()				, clean_up_tokenization_spaces=lowercase__				, skip_special_tokens=lowercase__		)
							assert generated == expected_en





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

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


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



class 							A   (	UpperCAmelCase__							):
					'''simple docstring'''






					def __init__(self   :						List[str]				, _UpperCAmelCase   :						AutoencoderKL				, _UpperCAmelCase   :						CLIPTextModel				, _UpperCAmelCase   :						CLIPTokenizer				, _UpperCAmelCase   :						UNetaDConditionModel				, _UpperCAmelCase   :						Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]				, _UpperCAmelCase   :						StableDiffusionSafetyChecker				, _UpperCAmelCase   :						CLIPImageProcessor				, )						->       Dict:




							"""simple docstring"""


							super().__init__()
							self.register_modules(
							    vae=_UpperCAmelCase				, text_encoder=_UpperCAmelCase				, tokenizer=_UpperCAmelCase				, unet=_UpperCAmelCase				, scheduler=_UpperCAmelCase				, safety_checker=_UpperCAmelCase				, feature_extractor=_UpperCAmelCase				, )

					def   lowerCamelCase__					(self   :						Optional[Any]				, _UpperCAmelCase   :						Optional[Union[str, int]] = "auto"		)						->       Tuple:




							"""simple docstring"""


							if slice_size == "auto":
									# half the attention head size is usually a good trade-off between
									# speed and memory
									lowercase__					      =		self.unet.config.attention_head_dim // 2
							self.unet.set_attention_slice(_UpperCAmelCase		)

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




							"""simple docstring"""


							self.enable_attention_slicing(_UpperCAmelCase		)


					@torch.no_grad()
					def __call__(self   :						Any				, _UpperCAmelCase   :						Union[str, List[str]]				, _UpperCAmelCase   :						int = 512				, _UpperCAmelCase   :						int = 512				, _UpperCAmelCase   :						int = 50				, _UpperCAmelCase   :						float = 7.5				, _UpperCAmelCase   :						Optional[Union[str, List[str]]] = None				, _UpperCAmelCase   :						Optional[int] = 1				, _UpperCAmelCase   :						float = 0.0				, _UpperCAmelCase   :						Optional[torch.Generator] = None				, _UpperCAmelCase   :						Optional[torch.FloatTensor] = None				, _UpperCAmelCase   :						Optional[str] = "pil"				, _UpperCAmelCase   :						bool = True				, _UpperCAmelCase   :						Optional[Callable[[int, int, torch.FloatTensor], None]] = None				, _UpperCAmelCase   :						int = 1				, _UpperCAmelCase   :						Optional[torch.FloatTensor] = None				, **_UpperCAmelCase   :						Any				, )						->       Tuple:




							"""simple docstring"""


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

							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(_UpperCAmelCase				, _UpperCAmelCase		) or callback_steps <= 0)
							):
									raise ValueError(
									    f'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
									    f''' {type(_UpperCAmelCase		)}.'''		)

							# get prompt text embeddings
							lowercase__					      =		self.tokenizer(
							    _UpperCAmelCase				, padding="""max_length"""				, max_length=self.tokenizer.model_max_length				, return_tensors="""pt"""				, )
							lowercase__					      =		text_inputs.input_ids

							if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
									lowercase__					      =		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}'''		)
									lowercase__					      =		text_input_ids[:, : self.tokenizer.model_max_length]

							if text_embeddings is None:
									lowercase__					      =		self.text_encoder(text_input_ids.to(self.device		)		)[0]

							# duplicate text embeddings for each generation per prompt, using mps friendly method
							lowercase__      ,     lowercase__      ,     lowercase__					      =		text_embeddings.shape
							lowercase__					      =		text_embeddings.repeat(1				, _UpperCAmelCase				, 1		)
							lowercase__					      =		text_embeddings.view(bs_embed * num_images_per_prompt				, _UpperCAmelCase				, -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.
							lowercase__					      =		guidance_scale > 1.0
							# get unconditional embeddings for classifier free guidance
							if do_classifier_free_guidance:
									lowercase__					      =		42
									if negative_prompt is None:
											lowercase__					      =		[""""""]
									elif type(_UpperCAmelCase		) is not type(_UpperCAmelCase		):
											raise TypeError(
											    f'''`negative_prompt` should be the same type to `prompt`, but got {type(_UpperCAmelCase		)} !='''
											    f''' {type(_UpperCAmelCase		)}.'''		)
									elif isinstance(_UpperCAmelCase				, _UpperCAmelCase		):
											lowercase__					      =		[negative_prompt]
									elif batch_size != len(_UpperCAmelCase		):
											raise ValueError(
											    f'''`negative_prompt`: {negative_prompt} has batch size {len(_UpperCAmelCase		)}, but `prompt`:'''
											    f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
											    """ the batch size of `prompt`."""		)
									else:
											lowercase__					      =		negative_prompt

									lowercase__					      =		text_input_ids.shape[-1]
									lowercase__					      =		self.tokenizer(
									    _UpperCAmelCase				, padding="""max_length"""				, max_length=_UpperCAmelCase				, truncation=_UpperCAmelCase				, return_tensors="""pt"""				, )
									lowercase__					      =		self.text_encoder(uncond_input.input_ids.to(self.device		)		)[0]

									# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
									lowercase__					      =		uncond_embeddings.shape[1]
									lowercase__					      =		uncond_embeddings.repeat(_UpperCAmelCase				, _UpperCAmelCase				, 1		)
									lowercase__					      =		uncond_embeddings.view(batch_size * num_images_per_prompt				, _UpperCAmelCase				, -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
									lowercase__					      =		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`.
							lowercase__					      =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
							lowercase__					      =		(batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
							lowercase__					      =		text_embeddings.dtype
							if latents is None:
									if self.device.type == "mps":
											# randn does not exist on mps
											lowercase__					      =		torch.randn(
											    _UpperCAmelCase				, generator=_UpperCAmelCase				, device="""cpu"""				, dtype=_UpperCAmelCase		).to(self.device		)
											lowercase__					      =		torch.randn(_UpperCAmelCase				, generator=_UpperCAmelCase				, device="""cpu"""				, dtype=_UpperCAmelCase		).to(
											    self.device		)
									else:
											lowercase__					      =		torch.randn(
											    _UpperCAmelCase				, generator=_UpperCAmelCase				, device=self.device				, dtype=_UpperCAmelCase		)
											lowercase__					      =		torch.randn(_UpperCAmelCase				, generator=_UpperCAmelCase				, device=self.device				, dtype=_UpperCAmelCase		)
							else:
									if latents_reference.shape != latents_shape:
											raise ValueError(f'''Unexpected latents shape, got {latents.shape}, expected {latents_shape}'''		)
									lowercase__					      =		latents_reference.to(self.device		)
									lowercase__					      =		latents.to(self.device		)

							# This is the key part of the pipeline where we
							# try to ensure that the generated images w/ the same seed
							# but different sizes actually result in similar images
							lowercase__					      =		(latents_shape[3] - latents_shape_reference[3]) // 2
							lowercase__					      =		(latents_shape[2] - latents_shape_reference[2]) // 2
							lowercase__					      =		latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
							lowercase__					      =		latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
							lowercase__					      =		0 if dx < 0 else dx
							lowercase__					      =		0 if dy < 0 else dy
							lowercase__					      =		max(-dx				, 0		)
							lowercase__					      =		max(-dy				, 0		)
							# import pdb
							# pdb.set_trace()
							lowercase__					      =		latents_reference[:, :, dy : dy + h, dx : dx + w]

							# set timesteps
							self.scheduler.set_timesteps(_UpperCAmelCase		)

							# Some schedulers like PNDM have timesteps as arrays
							# It's more optimized to move all timesteps to correct device beforehand
							lowercase__					      =		self.scheduler.timesteps.to(self.device		)

							# scale the initial noise by the standard deviation required by the scheduler
							lowercase__					      =		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]
							lowercase__					      =		"""eta""" in set(inspect.signature(self.scheduler.step		).parameters.keys()		)
							lowercase__					      =		{}
							if accepts_eta:
									lowercase__					      =		eta

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

									# predict the noise residual
									lowercase__					      =		self.unet(_UpperCAmelCase				, _UpperCAmelCase				, encoder_hidden_states=_UpperCAmelCase		).sample

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

									# compute the previous noisy sample x_t -> x_t-1
									lowercase__					      =		self.scheduler.step(_UpperCAmelCase				, _UpperCAmelCase				, _UpperCAmelCase				, **_UpperCAmelCase		).prev_sample

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

							lowercase__					      =		1 / 0.18_215 * latents
							lowercase__					      =		self.vae.decode(_UpperCAmelCase		).sample

							lowercase__					      =		(image / 2 + 0.5).clamp(0				, 1		)

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

							if self.safety_checker is not None:
									lowercase__					      =		self.feature_extractor(self.numpy_to_pil(_UpperCAmelCase		)				, return_tensors="""pt"""		).to(
									    self.device		)
									lowercase__      ,     lowercase__					      =		self.safety_checker(
									    images=_UpperCAmelCase				, clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype		)		)
							else:
									lowercase__					      =		None

							if output_type == "pil":
									lowercase__					      =		self.numpy_to_pil(_UpperCAmelCase		)

							if not return_dict:
									return (image, has_nsfw_concept)

							return StableDiffusionPipelineOutput(images=_UpperCAmelCase				, nsfw_content_detected=_UpperCAmelCase		)

