





"""simple docstring"""
import gc
import importlib.metadata
import tempfile
import unittest

from packaging import version

from transformers import (
    AutoModel,
    AutoModelForCausalLM,
    AutoModelForSeqaSeqLM,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    BitsAndBytesConfig,
    pipeline,
)
from transformers.testing_utils import (
    is_torch_available,
    require_accelerate,
    require_bitsandbytes,
    require_torch,
    require_torch_gpu,
    require_torch_multi_gpu,
    slow,
)






def 					A__				(			UpperCamelCase       ):
							if model.config.model_type == "gpt2":
														return model.transformer.h[0].mlp.c_fc
							return model.transformer.h[0].mlp.dense_ah_to_h


if is_torch_available():
		import torch
		import torch.nn as nn






		class 					_UpperCAmelCase   (	nn.Module ):


							def __init__(					self			:List[Any]     ,       __UpperCamelCase			:nn.Module     ,       __UpperCamelCase			:int		):
														super().__init__()
														A		=  module
														A		=  nn.Sequential(
														    nn.Linear(module.in_features     ,       __UpperCamelCase     ,       bias=__UpperCamelCase		)     ,       nn.Linear(__UpperCamelCase     ,       module.out_features     ,       bias=__UpperCamelCase		)     ,       )
														A		=  (2.0 / (5 * min(module.in_features     ,       module.out_features		))) ** 0.5
														nn.init.normal_(self.adapter[0].weight     ,       std=__UpperCamelCase		)
														nn.init.zeros_(self.adapter[1].weight		)
														self.adapter.to(module.weight.device		)


							def 	lowerCamelCase       (					self			:Tuple     ,       __UpperCamelCase			:List[Any]     ,       *__UpperCamelCase			:int     ,       **__UpperCamelCase			:List[Any]		):
														return self.module(__UpperCamelCase     ,       *__UpperCamelCase     ,       **__UpperCamelCase		) + self.adapter(__UpperCamelCase		)






@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class 					_UpperCAmelCase   (	unittest.TestCase ):
					# We keep the constants inside the init function and model loading inside setUp function

					# We need to test on relatively large models (aka >1b parameters otherwise the quantiztion may not work as expected)
					# Therefore here we use only bloom-1b3 to test our module
					UpperCamelCase      =							'''bigscience/bloom-1b7'''

					# Constant values
					UpperCamelCase      =							2.109_6595_5269_2574

					UpperCamelCase      =							'''Hello my name is'''
					UpperCamelCase      =							set()
					EXPECTED_OUTPUTS.add('''Hello my name is John and I am a professional photographer. I''' )
					EXPECTED_OUTPUTS.add('''Hello my name is John.\nI am a friend of your father.\n''' )
					EXPECTED_OUTPUTS.add('''Hello my name is John Doe, I am a student at the University''' )
					UpperCamelCase      =							1_0


					def 	lowerCamelCase       (					self			:Any		):
												# Models and tokenizer
												A		=  AutoTokenizer.from_pretrained(self.model_name		)







class 					_UpperCAmelCase   (	lowercase_ ):


					def 	lowerCamelCase       (					self			:Optional[int]		):
												super().setUp()

												# Models and tokenizer
												A		=  AutoModelForCausalLM.from_pretrained(
												    self.model_name     ,       torch_dtype=torch.floataa     ,       device_map="auto"		)
												A		=  AutoModelForCausalLM.from_pretrained(self.model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)


					def 	lowerCamelCase       (					self			:Tuple		):
												del self.model_fpaa
												del self.model_abit

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


					def 	lowerCamelCase       (					self			:List[Any]		):
												A		=  self.model_abit.config

												self.assertTrue(hasattr(__UpperCamelCase     ,       "quantization_config"		)		)

												A		=  config.to_dict()
												A		=  config.to_diff_dict()

												A		=  config.to_json_string()


					def 	lowerCamelCase       (					self			:int		):
												from bitsandbytes.nn import Paramsabit

												A		=  self.model_fpaa.get_memory_footprint()
												A		=  self.model_abit.get_memory_footprint()

												self.assertAlmostEqual(mem_fpaa / mem_abit     ,       self.EXPECTED_RELATIVE_DIFFERENCE		)
												A		=  get_some_linear_layer(self.model_abit		)
												self.assertTrue(linear.weight.__class__ == Paramsabit		)


					def 	lowerCamelCase       (					self			:int		):
												from transformers import TaPreTrainedModel

												self.model_fpaa.get_memory_footprint()
												self.model_abit.get_memory_footprint()

												for name, module in self.model_abit.named_modules():
																			if isinstance(__UpperCamelCase     ,       torch.nn.Linear		):
																										if name not in ["lm_head"] + TaPreTrainedModel._keep_in_fpaa_modules:
																																	# 4-bit parameters are packed in uint8 variables
																																	self.assertTrue(module.weight.dtype == torch.uinta		)


					def 	lowerCamelCase       (					self			:Dict		):
												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		)
												A		=  self.model_abit.generate(input_ids=encoded_input["input_ids"].to(0		)     ,       max_new_tokens=10		)

												self.assertIn(self.tokenizer.decode(output_sequences[0]     ,       skip_special_tokens=__UpperCamelCase		)     ,       self.EXPECTED_OUTPUTS		)


					def 	lowerCamelCase       (					self			:List[Any]		):
												A		=  BitsAndBytesConfig()
												A		=  True

												A		=  AutoModelForCausalLM.from_pretrained(
												    self.model_name     ,       quantization_config=__UpperCamelCase     ,       device_map="auto"		)

												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		)
												A		=  model_abit_from_config.generate(
												    input_ids=encoded_input["input_ids"].to(0		)     ,       max_new_tokens=10		)

												self.assertIn(self.tokenizer.decode(output_sequences[0]     ,       skip_special_tokens=__UpperCamelCase		)     ,       self.EXPECTED_OUTPUTS		)


					def 	lowerCamelCase       (					self			:Any		):
												with self.assertRaises(__UpperCamelCase		), tempfile.TemporaryDirectory() as tmpdirname:
																			self.model_abit.save_pretrained(__UpperCamelCase		)


					def 	lowerCamelCase       (					self			:Optional[Any]		):
												A		=  BitsAndBytesConfig()

												with self.assertRaises(__UpperCamelCase		):
																			A		=  AutoModelForCausalLM.from_pretrained(
																			    self.model_name     ,       quantization_config=__UpperCamelCase     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"     ,       bnb_abit_quant_type="nf4"     ,       )


					def 	lowerCamelCase       (					self			:Tuple		):
												with self.assertRaises(__UpperCamelCase		):
																			# Tries with `str`
																			self.model_abit.to("cpu"		)

												with self.assertRaises(__UpperCamelCase		):
																			# Tries with a `dtype``
																			self.model_abit.to(torch.floataa		)

												with self.assertRaises(__UpperCamelCase		):
																			# Tries with a `device`
																			self.model_abit.to(torch.device("cuda:0"		)		)

												with self.assertRaises(__UpperCamelCase		):
																			# Tries with a `device`
																			self.model_abit.float()

												with self.assertRaises(__UpperCamelCase		):
																			# Tries with a `device`
																			self.model_abit.half()

												# Test if we did not break anything
												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		)

												A		=  self.model_fpaa.to(torch.floataa		)
												A		=  self.model_fpaa.generate(input_ids=encoded_input["input_ids"].to(0		)     ,       max_new_tokens=10		)

												# Check this does not throw an error
												A		=  self.model_fpaa.to("cpu"		)

												# Check this does not throw an error
												A		=  self.model_fpaa.half()

												# Check this does not throw an error
												A		=  self.model_fpaa.float()


					def 	lowerCamelCase       (					self			:Optional[int]		):
												A		=  AutoModelForSeqaSeqLM.from_pretrained("t5-small"     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)
												self.assertTrue(model.decoder.block[0].layer[2].DenseReluDense.wo.weight.dtype == torch.floataa		)







@require_bitsandbytes
@require_accelerate
@require_torch
@require_torch_gpu
@slow
class 					_UpperCAmelCase   (	unittest.TestCase ):


					@classmethod
					def 	lowerCamelCase       (					cls			:Optional[Any]		):
												A		=  "t5-small"
												A		=  "google/flan-t5-small"  # flan-t5 uses dense-act instead of dense-relu-dense
												A		=  AutoTokenizer.from_pretrained(cls.model_name		)
												A		=  "Translate in German: Hello, my dog is cute"


					def 	lowerCamelCase       (					self			:Dict		):
												gc.collect()
												torch.cuda.empty_cache()


					def 	lowerCamelCase       (					self			:str		):
												from transformers import TaForConditionalGeneration

												A		=  TaForConditionalGeneration._keep_in_fpaa_modules
												A		=  None

												# test with `t5-small`
												A		=  TaForConditionalGeneration.from_pretrained(self.model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)
												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		).to(0		)
												A		=  model.generate(**__UpperCamelCase		)

												# test with `flan-t5-small`
												A		=  TaForConditionalGeneration.from_pretrained(
												    self.dense_act_model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)
												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		).to(0		)
												A		=  model.generate(**__UpperCamelCase		)
												A		=  modules


					def 	lowerCamelCase       (					self			:Union[str, Any]		):
												import bitsandbytes as bnb

												from transformers import TaForConditionalGeneration

												# test with `t5-small`
												A		=  TaForConditionalGeneration.from_pretrained(self.model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)

												# there was a bug with decoders - this test checks that it is fixed
												self.assertTrue(isinstance(model.decoder.block[0].layer[0].SelfAttention.q     ,       bnb.nn.Linearabit		)		)

												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		).to(0		)
												A		=  model.generate(**__UpperCamelCase		)

												# test with `flan-t5-small`
												A		=  TaForConditionalGeneration.from_pretrained(
												    self.dense_act_model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)
												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		).to(0		)
												A		=  model.generate(**__UpperCamelCase		)







class 					_UpperCAmelCase   (	lowercase_ ):


					def 	lowerCamelCase       (					self			:List[Any]		):
												super().setUp()
												# model_name
												A		=  "bigscience/bloom-560m"
												A		=  "t5-small"

												# Different types of model

												A		=  AutoModel.from_pretrained(self.model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)
												# Sequence classification model
												A		=  AutoModelForSequenceClassification.from_pretrained(
												    self.model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)
												# CausalLM model
												A		=  AutoModelForCausalLM.from_pretrained(self.model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)
												# Seq2seq model
												A		=  AutoModelForSeqaSeqLM.from_pretrained(
												    self.seq_to_seq_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="auto"		)


					def 	lowerCamelCase       (					self			:Any		):
												del self.base_model
												del self.sequence_model
												del self.model_abit
												del self.seq_to_seq_model

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


					def 	lowerCamelCase       (					self			:Optional[Any]		):
												from bitsandbytes.nn import Paramsabit

												self.assertTrue(self.base_model.h[-1].mlp.dense_ah_to_h.weight.__class__ == Paramsabit		)

												# Other heads should be nn.Parameter
												self.assertTrue(self.model_abit.lm_head.weight.__class__ == torch.nn.Parameter		)
												self.assertTrue(self.sequence_model.score.weight.__class__ == torch.nn.Parameter		)
												self.assertTrue(self.seq_to_seq_model.lm_head.weight.__class__ == torch.nn.Parameter		)







class 					_UpperCAmelCase   (	lowercase_ ):


					def 	lowerCamelCase       (					self			:str		):
												super().setUp()


					def 	lowerCamelCase       (					self			:Tuple		):
												del self.pipe

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


					def 	lowerCamelCase       (					self			:List[str]		):
												A		=  pipeline(
												    "text-generation"     ,       model=self.model_name     ,       model_kwargs={"device_map": "auto", "load_in_4bit": True, "torch_dtype": torch.floataa}     ,       max_new_tokens=self.MAX_NEW_TOKENS     ,       )

												# Real second forward pass
												A		=  self.pipe(self.input_text		)
												self.assertIn(pipeline_output[0]["generated_text"]     ,       self.EXPECTED_OUTPUTS		)







@require_torch_multi_gpu
class 					_UpperCAmelCase   (	lowercase_ ):


					def 	lowerCamelCase       (					self			:Optional[int]		):
												super().setUp()


					def 	lowerCamelCase       (					self			:Optional[int]		):
												A		=  AutoModelForCausalLM.from_pretrained(
												    self.model_name     ,       load_in_abit=__UpperCamelCase     ,       device_map="balanced"		)

												# Check correct device map
												self.assertEqual(set(model_parallel.hf_device_map.values()		)     ,       {0, 1}		)

												# Check that inference pass works on the model
												A		=  self.tokenizer(self.input_text     ,       return_tensors="pt"		)

												# Second real batch
												A		=  model_parallel.generate(input_ids=encoded_input["input_ids"].to(0		)     ,       max_new_tokens=10		)
												self.assertIn(self.tokenizer.decode(output_parallel[0]     ,       skip_special_tokens=__UpperCamelCase		)     ,       self.EXPECTED_OUTPUTS		)







class 					_UpperCAmelCase   (	lowercase_ ):


					def 	lowerCamelCase       (					self			:Optional[Any]		):
												A		=  "facebook/opt-350m"
												super().setUp()


					def 	lowerCamelCase       (					self			:Optional[int]		):
												if version.parse(importlib.metadata.version("bitsandbytes"		)		) < version.parse("0.37.0"		):
																			return

												# Step 1: freeze all parameters
												A		=  AutoModelForCausalLM.from_pretrained(self.model_name     ,       load_in_abit=__UpperCamelCase		)

												self.assertEqual(set(model.hf_device_map.values()		)     ,       {torch.cuda.current_device()}		)

												for param in model.parameters():
																			A		=  False  # freeze the model - train adapters later
																			if param.ndim == 1:
																										# cast the small parameters (e.g. layernorm) to fp32 for stability
																										A		=  param.data.to(torch.floataa		)

        # Step 2: add adapters
												for _, module in model.named_modules():
																			if "OPTAttention" in repr(type(__UpperCamelCase		)		):
																										A		=  LoRALayer(module.q_proj     ,       rank=16		)
																										A		=  LoRALayer(module.k_proj     ,       rank=16		)
																										A		=  LoRALayer(module.v_proj     ,       rank=16		)

        # Step 3: dummy batch
												A		=  self.tokenizer("Test batch "     ,       return_tensors="pt"		).to(0		)

												# Step 4: Check if the gradient is not None
												with torch.cuda.amp.autocast():
																			A		=  model.forward(**__UpperCamelCase		)
																			out.logits.norm().backward()

												for module in model.modules():
																			if isinstance(__UpperCamelCase     ,       __UpperCamelCase		):
																										self.assertTrue(module.adapter[1].weight.grad is not None		)
																										self.assertTrue(module.adapter[1].weight.grad.norm().item() > 0		)
																			elif isinstance(__UpperCamelCase     ,       nn.Embedding		):
																										self.assertTrue(module.weight.grad is None		)







class 					_UpperCAmelCase   (	lowercase_ ):
					UpperCamelCase      =							'''gpt2-xl'''
					UpperCamelCase      =							3.3191_8548_5415_2187









"""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


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






class 					_UpperCAmelCase   (	lowercase_ ):


					def __init__(					self			:Dict     ,       __UpperCamelCase			:WhisperForConditionalGeneration     ,       __UpperCamelCase			:WhisperProcessor     ,       __UpperCamelCase			:AutoencoderKL     ,       __UpperCamelCase			:CLIPTextModel     ,       __UpperCamelCase			:CLIPTokenizer     ,       __UpperCamelCase			:UNetaDConditionModel     ,       __UpperCamelCase			:Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]     ,       __UpperCamelCase			:StableDiffusionSafetyChecker     ,       __UpperCamelCase			:CLIPImageProcessor     ,       ):
												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=__UpperCamelCase     ,       speech_processor=__UpperCamelCase     ,       vae=__UpperCamelCase     ,       text_encoder=__UpperCamelCase     ,       tokenizer=__UpperCamelCase     ,       unet=__UpperCamelCase     ,       scheduler=__UpperCamelCase     ,       feature_extractor=__UpperCamelCase     ,       )


					def 	lowerCamelCase       (					self			:Any     ,       __UpperCamelCase			:Optional[Union[str, int]] = "auto"		):
												if slice_size == "auto":
																			A		=  self.unet.config.attention_head_dim // 2
												self.unet.set_attention_slice(__UpperCamelCase		)


					def 	lowerCamelCase       (					self			:Tuple		):
												self.enable_attention_slicing(__UpperCamelCase		)


					@torch.no_grad()
					def __call__(					self			:Optional[Any]     ,       __UpperCamelCase			:Any     ,       __UpperCamelCase			:Dict=1_60_00     ,       __UpperCamelCase			:int = 5_12     ,       __UpperCamelCase			:int = 5_12     ,       __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			:Dict     ,       ):
												A		=  self.speech_processor.feature_extractor(
												    __UpperCamelCase     ,       return_tensors="pt"     ,       sampling_rate=__UpperCamelCase		).input_features.to(self.device		)
												A		=  self.speech_model.generate(__UpperCamelCase     ,       max_length=48_00_00		)

												A		=  self.speech_processor.tokenizer.batch_decode(__UpperCamelCase     ,       skip_special_tokens=__UpperCamelCase     ,       normalize=__UpperCamelCase		)[
												    0
												]

												if isinstance(__UpperCamelCase     ,       __UpperCamelCase		):
																			A		=  1
												elif isinstance(__UpperCamelCase     ,       __UpperCamelCase		):
																			A		=  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
												A		=  self.tokenizer(
												    __UpperCamelCase     ,       padding="max_length"     ,       max_length=self.tokenizer.model_max_length     ,       return_tensors="pt"     ,       )
												A		=  text_inputs.input_ids

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

												# duplicate text embeddings for each generation per prompt, using mps friendly method
												A,	A,	A		=  text_embeddings.shape
												A		=  text_embeddings.repeat(1     ,       __UpperCamelCase     ,       1		)
												A		=  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.
												A		=  guidance_scale > 1.0
												# get unconditional embeddings for classifier free guidance
												if do_classifier_free_guidance:
																			A		=  42
																			if negative_prompt is None:
																										A		=  [""] * batch_size
																			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		):
																										A		=  [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:
																										A		=  negative_prompt

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

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

												# 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
												A		=  self.scheduler.timesteps.to(self.device		)

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

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

																			# predict the noise residual
																			A		=  self.unet(__UpperCamelCase     ,       __UpperCamelCase     ,       encoder_hidden_states=__UpperCamelCase		).sample

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

																			# compute the previous noisy sample x_t -> x_t-1
																			A		=  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		)

												A		=  1 / 0.18_215 * latents
												A		=  self.vae.decode(__UpperCamelCase		).sample

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

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

												if output_type == "pil":
																			A		=  self.numpy_to_pil(__UpperCamelCase		)

												if not return_dict:
																			return image

												return StableDiffusionPipelineOutput(images=__UpperCamelCase     ,       nsfw_content_detected=__UpperCamelCase		)


