


"""simple docstring"""




import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from transformers import MaMaaaTokenizer, is_torch_available
from transformers.testing_utils import (
    get_tests_dir,
    nested_simplify,
    require_sentencepiece,
    require_tokenizers,
    require_torch,
    slow,
)
from transformers.utils import is_sentencepiece_available


if is_sentencepiece_available():
   from transformers.models.mam_aaa.tokenization_mam_aaa import VOCAB_FILES_NAMES, save_json

from ...test_tokenization_common import TokenizerTesterMixin


if is_sentencepiece_available():
   __SCREAMING_SNAKE_CASE      		=    get_tests_dir("""fixtures/test_sentencepiece.model""")


if is_torch_available():
   from transformers.models.mam_aaa.modeling_mam_aaa import shift_tokens_right

__SCREAMING_SNAKE_CASE      		=    12_80_22
__SCREAMING_SNAKE_CASE      		=    12_80_28





@require_sentencepiece
class __snake_case		(       __snake_case					,							unittest.TestCase   ):







   """simple docstring"""




   lowerCAmelCase_					: Optional[int]  						=      MaMaaaTokenizer
   lowerCAmelCase_					: str  						=      False
   lowerCAmelCase_					: int  						=      False
   lowerCAmelCase_					: str  						=      True




   def SCREAMING_SNAKE_CASE_   (		self				:List[str]							):
          super().setUp()

          _a							 =					["</s>", "<unk>", "▁This", "▁is", "▁a", "▁t", "est", "\u0120", "<pad>"]
          _a							 =					dict(zip(_SCREAMING_SNAKE_CASE					,	range(len(_SCREAMING_SNAKE_CASE							)							)							)							)
          _a							 =					Path(self.tmpdirname							)
          save_json(_SCREAMING_SNAKE_CASE					,	save_dir / VOCAB_FILES_NAMES["vocab_file"]							)
          if not (save_dir / VOCAB_FILES_NAMES["spm_file"]).exists():
                 copyfile(_SCREAMING_SNAKE_CASE					,	save_dir / VOCAB_FILES_NAMES["spm_file"]							)

          _a							 =					MaMaaaTokenizer.from_pretrained(self.tmpdirname							)
          tokenizer.save_pretrained(self.tmpdirname							)




   def SCREAMING_SNAKE_CASE_   (		self				:str					,	**UpperCamelCase__				:List[str]							):
          return MaMaaaTokenizer.from_pretrained(self.tmpdirname					,	**_SCREAMING_SNAKE_CASE							)




   def SCREAMING_SNAKE_CASE_   (		self				:Optional[int]					,	UpperCamelCase__				:Optional[Any]							):
          return (
              "This is a test",
              "This is a test",
          )




   def SCREAMING_SNAKE_CASE_   (		self				:List[Any]							):
          _a							 =					"</s>"
          _a							 =					0

          self.assertEqual(self.get_tokenizer()._convert_token_to_id(_SCREAMING_SNAKE_CASE							)					,	_SCREAMING_SNAKE_CASE							)
          self.assertEqual(self.get_tokenizer()._convert_id_to_token(_SCREAMING_SNAKE_CASE							)					,	_SCREAMING_SNAKE_CASE							)




   def SCREAMING_SNAKE_CASE_   (		self				:Optional[Any]							):
          _a							 =					self.get_tokenizer()
          _a							 =					list(tokenizer.get_vocab().keys()							)

          self.assertEqual(vocab_keys[0]					,	"</s>"							)
          self.assertEqual(vocab_keys[1]					,	"<unk>"							)
          self.assertEqual(vocab_keys[-1]					,	"<s>"							)
          self.assertEqual(len(_SCREAMING_SNAKE_CASE							)					,	tokenizer.vocab_size + len(tokenizer.get_added_vocab()							)							)




   @unittest.skip("Skip this test while all models are still to be uploaded."							)
   def SCREAMING_SNAKE_CASE_   (		self				:str							):
          pass




   def SCREAMING_SNAKE_CASE_   (		self				:Any							):
          _a							 =					self.get_tokenizer()

          _a							 =					tokenizer.tokenize("This is a test"							)
          self.assertListEqual(_SCREAMING_SNAKE_CASE					,	["▁This", "▁is", "▁a", "▁t", "est"]							)

          self.assertListEqual(
              tokenizer.convert_tokens_to_ids(_SCREAMING_SNAKE_CASE							)					,	[2, 3, 4, 5, 6]					,	)

          _a							 =					tokenizer.convert_ids_to_tokens([2, 3, 4, 5, 6]							)
          self.assertListEqual(_SCREAMING_SNAKE_CASE					,	["▁This", "▁is", "▁a", "▁t", "est"]							)

          _a							 =					tokenizer.convert_tokens_to_string(_SCREAMING_SNAKE_CASE							)
          self.assertEqual(_SCREAMING_SNAKE_CASE					,	"This is a test"							)




   @slow
   def SCREAMING_SNAKE_CASE_   (		self				:str							):
          _a							 =					{"input_ids": [[128_022, 110_108, 397, 11, 38_272, 2_247, 124_811, 285, 18_105, 1_586, 207, 7, 39_534, 4_428, 397, 1_019, 18_105, 1_586, 207, 7, 41_337, 16_786, 241, 7, 20_214, 17, 125_690, 10_398, 7, 44_378, 58_069, 68_342, 7_798, 7_343, 11, 299, 33_310, 4, 158, 37_350, 94_077, 4_569, 299, 33_310, 90, 4, 52_840, 290, 4, 31_270, 112, 299, 682, 4, 52_840, 39_953, 14_079, 193, 52_519, 90_894, 17_894, 120_697, 11, 40_445, 551, 17, 1_019, 52_519, 90_894, 17_756, 963, 11, 40_445, 480, 17, 9_792, 1_120, 5_173, 1_393, 6_240, 16_786, 241, 120_996, 28, 1_245, 1_393, 118_240, 11_123, 1_019, 93_612, 2_691, 10_618, 98_058, 120_409, 1_928, 279, 4, 40_683, 367, 178, 207, 1_019, 103, 103_121, 506, 65_296, 5, 2], [128_022, 21_217, 367, 117, 125_450, 128, 719, 7, 7_308, 40, 93_612, 12_669, 1_116, 16_704, 71, 17_785, 3_699, 15_592, 35, 144, 9_584, 241, 11_943, 713, 950, 799, 2_247, 88_427, 150, 149, 118_813, 120_706, 1_019, 106_906, 81_518, 28, 1_224, 22_799, 397, 5, 2, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], [128_022, 1_658, 123_311, 5_155, 5_578, 4_722, 279, 14_947, 2_366, 1_120, 1_197, 14, 1_348, 9_232, 5, 2, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]], "attention_mask": [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]]}  # noqa: E501
          # fmt: on

          self.tokenizer_integration_test_util(
              expected_encoding=_SCREAMING_SNAKE_CASE					,	model_name="facebook/m2m100_418M"					,	revision="c168bae485c864188cf9aa0e4108b0b6934dc91e"					,	)






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







   """simple docstring"""




   lowerCAmelCase_					: int  						=      'facebook/m2m100_418M'
   lowerCAmelCase_					: Tuple  						=      [
       'In my opinion, there are two levels of response from the French government.',
       'NSA Affair Emphasizes Complete Lack of Debate on Intelligence',
   ]
   lowerCAmelCase_					: Dict  						=      [
       'Selon moi, il y a deux niveaux de réponse de la part du gouvernement français.',
       'L\'affaire NSA souligne l\'absence totale de débat sur le renseignement',
   ]

   # fmt: off
   lowerCAmelCase_					: List[str]  						=      [EN_CODE, 593, 1_949, 115_781, 4, 71_586, 4_234, 60_633, 126_233, 432, 123_808, 15_592, 1_197, 117_132, 120_618, 5, 2]




   @classmethod
   def SCREAMING_SNAKE_CASE_   (		cls				:Any							):
          _a							 =					MaMaaaTokenizer.from_pretrained(
              cls.checkpoint_name					,	src_lang="en"					,	tgt_lang="fr"							)
          _a							 =					1
          return cls




   def SCREAMING_SNAKE_CASE_   (		self				:str							):
          self.assertEqual(self.tokenizer.get_lang_id("ar"							)					,	128_006							)
          self.assertEqual(self.tokenizer.get_lang_id("en"							)					,	128_022							)
          self.assertEqual(self.tokenizer.get_lang_id("ro"							)					,	128_076							)
          self.assertEqual(self.tokenizer.get_lang_id("mr"							)					,	128_063							)




   def SCREAMING_SNAKE_CASE_   (		self				:Tuple							):
          _a							 =					self.tokenizer.get_vocab()
          self.assertEqual(len(_SCREAMING_SNAKE_CASE							)					,	self.tokenizer.vocab_size							)
          self.assertEqual(vocab["<unk>"]					,	3							)
          self.assertIn(self.tokenizer.get_lang_token("en"							)					,	_SCREAMING_SNAKE_CASE							)




   def SCREAMING_SNAKE_CASE_   (		self				:Optional[Any]							):
          _a							 =					"en"
          _a							 =					self.tokenizer.batch_encode_plus(self.src_text							).input_ids[0]
          self.assertListEqual(self.expected_src_tokens					,	_SCREAMING_SNAKE_CASE							)




   def SCREAMING_SNAKE_CASE_   (		self				:Union[str, Any]							):
          self.assertIn(_SCREAMING_SNAKE_CASE					,	self.tokenizer.all_special_ids							)
          # fmt: off
          _a							 =					[FR_CODE, 5_364, 82, 8_642, 4, 294, 47, 8, 14_028, 136, 3_286, 9_706, 6, 90_797, 6, 144_012, 162, 88_128, 30_061, 5, 2]
          # fmt: on
          _a							 =					self.tokenizer.decode(_SCREAMING_SNAKE_CASE					,	skip_special_tokens=_SCREAMING_SNAKE_CASE							)
          _a							 =					self.tokenizer.decode(generated_ids[1:]					,	skip_special_tokens=_SCREAMING_SNAKE_CASE							)
          self.assertEqual(_SCREAMING_SNAKE_CASE					,	_SCREAMING_SNAKE_CASE							)
          self.assertNotIn(self.tokenizer.eos_token					,	_SCREAMING_SNAKE_CASE							)




   def SCREAMING_SNAKE_CASE_   (		self				:Optional[Any]							):
          _a							 =					tempfile.mkdtemp()
          _a							 =					self.tokenizer.lang_token_to_id
          self.tokenizer.save_pretrained(_SCREAMING_SNAKE_CASE							)
          _a							 =					MaMaaaTokenizer.from_pretrained(_SCREAMING_SNAKE_CASE							)
          self.assertDictEqual(new_tok.lang_token_to_id					,	_SCREAMING_SNAKE_CASE							)




   @require_torch
   def SCREAMING_SNAKE_CASE_   (		self				:Union[str, Any]							):
          _a							 =					"en"
          _a							 =					"fr"

          _a							 =					self.tokenizer(self.src_text					,	text_target=self.tgt_text					,	padding=_SCREAMING_SNAKE_CASE					,	return_tensors="pt"							)

          _a							 =					shift_tokens_right(
              batch["labels"]					,	self.tokenizer.pad_token_id					,	self.tokenizer.eos_token_id							)

          for k in batch:
                 _a							 =					batch[k].tolist()
          # batch = {k: v.tolist() for k,v in batch.items()}
          # fairseq batch: https://gist.github.com/sshleifer/cba08bc2109361a74ac3760a7e30e4f4
          # batch.decoder_inputs_ids[0][0] ==
          assert batch.input_ids[1][0] == EN_CODE
          assert batch.input_ids[1][-1] == 2
          assert batch.labels[1][0] == FR_CODE
          assert batch.labels[1][-1] == 2
          assert batch.decoder_input_ids[1][:2] == [2, FR_CODE]




   @require_torch
   def SCREAMING_SNAKE_CASE_   (		self				:Any							):
          _a							 =					"mr"
          self.assertListEqual(self.tokenizer.prefix_tokens					,	[self.tokenizer.get_lang_id("mr"							)]							)
          self.assertListEqual(self.tokenizer.suffix_tokens					,	[self.tokenizer.eos_token_id]							)

          _a							 =					"zh"
          self.assertListEqual(self.tokenizer.prefix_tokens					,	[self.tokenizer.get_lang_id("zh"							)]							)
          self.assertListEqual(self.tokenizer.suffix_tokens					,	[self.tokenizer.eos_token_id]							)




   @require_torch
   def SCREAMING_SNAKE_CASE_   (		self				:Tuple							):
          _a							 =					"mr"
          self.tokenizer._switch_to_target_mode()
          self.assertListEqual(self.tokenizer.prefix_tokens					,	[self.tokenizer.get_lang_id("mr"							)]							)
          self.assertListEqual(self.tokenizer.suffix_tokens					,	[self.tokenizer.eos_token_id]							)
          self.tokenizer._switch_to_input_mode()
          self.assertListEqual(self.tokenizer.prefix_tokens					,	[self.tokenizer.get_lang_id(self.tokenizer.src_lang							)]							)

          _a							 =					"zh"
          self.tokenizer._switch_to_target_mode()
          self.assertListEqual(self.tokenizer.prefix_tokens					,	[self.tokenizer.get_lang_id("zh"							)]							)
          self.assertListEqual(self.tokenizer.suffix_tokens					,	[self.tokenizer.eos_token_id]							)
          self.tokenizer._switch_to_input_mode()
          self.assertListEqual(self.tokenizer.prefix_tokens					,	[self.tokenizer.get_lang_id(self.tokenizer.src_lang							)]							)




   @require_torch
   def SCREAMING_SNAKE_CASE_   (		self				:Any							):
          _a							 =					self.tokenizer._build_translation_inputs("A test"					,	return_tensors="pt"					,	src_lang="en"					,	tgt_lang="ar"							)

          self.assertEqual(
              nested_simplify(_SCREAMING_SNAKE_CASE							)					,	{
                  # en_XX, A, test, EOS
                  "input_ids": [[128_022, 58, 4_183, 2]],
                  "attention_mask": [[1, 1, 1, 1]],
                  # ar_AR
                  "forced_bos_token_id": 128_006,
              }					,	)








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


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





class 					A__     (     __snake_case ):

     '''simple docstring'''






     def __init__( self      :    Tuple   ,						_SCREAMING_SNAKE_CASE      :    WhisperForConditionalGeneration   ,						_SCREAMING_SNAKE_CASE      :    WhisperProcessor   ,						_SCREAMING_SNAKE_CASE      :    AutoencoderKL   ,						_SCREAMING_SNAKE_CASE      :    CLIPTextModel   ,						_SCREAMING_SNAKE_CASE      :    CLIPTokenizer   ,						_SCREAMING_SNAKE_CASE      :    UNetaDConditionModel   ,						_SCREAMING_SNAKE_CASE      :    Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]   ,						_SCREAMING_SNAKE_CASE      :    StableDiffusionSafetyChecker   ,						_SCREAMING_SNAKE_CASE      :    CLIPImageProcessor   ,						):






           """simple docstring"""




           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=_SCREAMING_SNAKE_CASE   ,						speech_processor=_SCREAMING_SNAKE_CASE   ,						vae=_SCREAMING_SNAKE_CASE   ,						text_encoder=_SCREAMING_SNAKE_CASE   ,						tokenizer=_SCREAMING_SNAKE_CASE   ,						unet=_SCREAMING_SNAKE_CASE   ,						scheduler=_SCREAMING_SNAKE_CASE   ,						feature_extractor=_SCREAMING_SNAKE_CASE   ,						)







     def   _SCREAMING_SNAKE_CASE     ( self      :    str   ,						_SCREAMING_SNAKE_CASE      :    Optional[Union[str, int]] = "auto" ):






           """simple docstring"""




           if slice_size == "auto":
                 UpperCamelCase					       =  self.unet.config.attention_head_dim // 2
           self.unet.set_attention_slice(_SCREAMING_SNAKE_CASE )







     def   _SCREAMING_SNAKE_CASE     ( self      :    int ):






           """simple docstring"""




           self.enable_attention_slicing(_SCREAMING_SNAKE_CASE )







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






           """simple docstring"""




           UpperCamelCase					       =  self.speech_processor.feature_extractor(
               _SCREAMING_SNAKE_CASE   ,						return_tensors='pt'   ,						sampling_rate=_SCREAMING_SNAKE_CASE ).input_features.to(self.device )
           UpperCamelCase					       =  self.speech_model.generate(_SCREAMING_SNAKE_CASE   ,						max_length=48_0000 )

           UpperCamelCase					       =  self.speech_processor.tokenizer.batch_decode(_SCREAMING_SNAKE_CASE   ,						skip_special_tokens=_SCREAMING_SNAKE_CASE   ,						normalize=_SCREAMING_SNAKE_CASE )[
               0
           ]

           if isinstance(_SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE ):
                 UpperCamelCase					       =  1
           elif isinstance(_SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE ):
                 UpperCamelCase					       =  len(_SCREAMING_SNAKE_CASE )
           else:
                 raise ValueError(f'`prompt` has to be of type `str` or `list` but is {type(_SCREAMING_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(_SCREAMING_SNAKE_CASE   ,						_SCREAMING_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(_SCREAMING_SNAKE_CASE )}.' )

           # get prompt text embeddings
           UpperCamelCase					       =  self.tokenizer(
               _SCREAMING_SNAKE_CASE   ,						padding='max_length'   ,						max_length=self.tokenizer.model_max_length   ,						return_tensors='pt'   ,						)
           UpperCamelCase					       =  text_inputs.input_ids

           if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
                 UpperCamelCase					       =  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}' )
                 UpperCamelCase					       =  text_input_ids[:, : self.tokenizer.model_max_length]
           UpperCamelCase					       =  self.text_encoder(text_input_ids.to(self.device ) )[0]

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

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

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

           # set timesteps
           self.scheduler.set_timesteps(_SCREAMING_SNAKE_CASE )

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

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

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

                 # predict the noise residual
                 UpperCamelCase					       =  self.unet(_SCREAMING_SNAKE_CASE   ,						_SCREAMING_SNAKE_CASE   ,						encoder_hidden_states=_SCREAMING_SNAKE_CASE ).sample

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

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

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

           UpperCamelCase					       =  1 / 0.1_8_2_1_5 * latents
           UpperCamelCase					       =  self.vae.decode(_SCREAMING_SNAKE_CASE ).sample

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

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

           if output_type == "pil":
                 UpperCamelCase					       =  self.numpy_to_pil(_SCREAMING_SNAKE_CASE )

           if not return_dict:
                 return image

           return StableDiffusionPipelineOutput(images=_SCREAMING_SNAKE_CASE   ,						nsfw_content_detected=_SCREAMING_SNAKE_CASE )



