

from typing import Optional, Union

import torch
from torch import nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss

from ...activations import ACTaFN
from ...modeling_outputs import BaseModelOutputWithPoolingAndNoAttention, ImageClassifierOutputWithNoAttention
from ...modeling_utils import PreTrainedModel
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_mobilenet_va import MobileNetVaConfig


A_	:			int  	=      logging.get_logger(__name__)


# General docstring
A_	:			Tuple  	=      'MobileNetV1Config'

# Base docstring
A_	:			Any  	=      'google/mobilenet_v1_1.0_224'
A_	:			List[str]  	=      [1, 1024, 7, 7]

# Image classification docstring
A_	:			Dict  	=      'google/mobilenet_v1_1.0_224'
A_	:			List[Any]  	=      'tabby, tabby cat'


A_	:			Dict  	=      [
    'google/mobilenet_v1_1.0_224',
    'google/mobilenet_v1_0.75_192',
    # See all MobileNetV1 models at https://huggingface.co/models?filter=mobilenet_v1
]

def 						UpperCamelCase (lowercase_:			Tuple   ,				lowercase_:			List[str]   ,				lowercase_:			List[Any]=None )					->				Optional[Any]:
	A__							:   Optional[int]					       =		{}

	if isinstance(lowercase_   ,				lowercase_ ):
		A__							:   List[Any]					       =		model.mobilenet_va
	else:
		A__							:   Tuple					       =		model

	A__							:   Tuple					       =		"""MobilenetV1/Conv2d_0/"""
	A__							:   List[Any]					       =		backbone.conv_stem.convolution.weight
	A__							:   Dict					       =		backbone.conv_stem.normalization.bias
	A__							:   Dict					       =		backbone.conv_stem.normalization.weight
	A__							:   str					       =		backbone.conv_stem.normalization.running_mean
	A__							:   int					       =		backbone.conv_stem.normalization.running_var

	for i in range(13 ):
		A__							:   str					       =		i + 1
		A__							:   Union[str, Any]					       =		i * 2

		A__							:   int					       =		backbone.layer[pt_index]
		A__							:   Optional[Any]					       =		f"""MobilenetV1/Conv2d_{tf_index}_depthwise/"""
		A__							:   Tuple					       =		pointer.convolution.weight
		A__							:   int					       =		pointer.normalization.bias
		A__							:   Dict					       =		pointer.normalization.weight
		A__							:   Optional[Any]					       =		pointer.normalization.running_mean
		A__							:   List[str]					       =		pointer.normalization.running_var

		A__							:   Optional[int]					       =		backbone.layer[pt_index + 1]
		A__							:   str					       =		f"""MobilenetV1/Conv2d_{tf_index}_pointwise/"""
		A__							:   Dict					       =		pointer.convolution.weight
		A__							:   Dict					       =		pointer.normalization.bias
		A__							:   List[Any]					       =		pointer.normalization.weight
		A__							:   Any					       =		pointer.normalization.running_mean
		A__							:   Tuple					       =		pointer.normalization.running_var

	if isinstance(lowercase_   ,				lowercase_ ):
		A__							:   List[Any]					       =		"""MobilenetV1/Logits/Conv2d_1c_1x1/"""
		A__							:   int					       =		model.classifier.weight
		A__							:   Any					       =		model.classifier.bias

	return tf_to_pt_map

def 						UpperCamelCase (lowercase_:			int   ,				lowercase_:			Any   ,				lowercase_:			Dict )					->				List[Any]:
	try:
		import numpy as np
		import tensorflow as tf
	except ImportError:
		logger.error(
		    """Loading a TensorFlow models in PyTorch, requires TensorFlow to be installed. Please see """
		    """https://www.tensorflow.org/install/ for installation instructions.""" )
		raise

	# Load weights from TF model
	A__							:   int					       =		tf.train.list_variables(lowercase_ )
	A__							:   Optional[int]					       =		{}
	for name, shape in init_vars:
		logger.info(f"""Loading TF weight {name} with shape {shape}""" )
		A__							:   int					       =		tf.train.load_variable(lowercase_   ,				lowercase_ )
		A__							:   Optional[int]					       =		array

	# Build TF to PyTorch weights loading map
	A__							:   Dict					       =		_build_tf_to_pytorch_map(lowercase_   ,				lowercase_   ,				lowercase_ )

	for name, pointer in tf_to_pt_map.items():
		logger.info(f"""Importing {name}""" )
		if name not in tf_weights:
			logger.info(f"""{name} not in tf pre-trained weights, skipping""" )
			continue

		A__							:   str					       =		tf_weights[name]

		if "depthwise_weights" in name:
			logger.info("""Transposing depthwise""" )
			A__							:   Tuple					       =		np.transpose(lowercase_   ,				(2, 3, 0, 1) )
		elif "weights" in name:
			logger.info("""Transposing""" )
			if len(pointer.shape ) == 2:  # copying into linear layer
				A__							:   Optional[int]					       =		array.squeeze().transpose()
			else:
				A__							:   int					       =		np.transpose(lowercase_   ,				(3, 2, 0, 1) )

		if pointer.shape != array.shape:
			raise ValueError(f"""Pointer shape {pointer.shape} and array shape {array.shape} mismatched""" )

		logger.info(f"""Initialize PyTorch weight {name} {array.shape}""" )
		A__							:   int					       =		torch.from_numpy(lowercase_ )

		tf_weights.pop(lowercase_   ,				lowercase_ )
		tf_weights.pop(name + """/RMSProp"""   ,				lowercase_ )
		tf_weights.pop(name + """/RMSProp_1"""   ,				lowercase_ )
		tf_weights.pop(name + """/ExponentialMovingAverage"""   ,				lowercase_ )

	logger.info(f"""Weights not copied to PyTorch model: {', '.join(tf_weights.keys() )}""" )
	return model

def 						UpperCamelCase (lowercase_:			torch.Tensor   ,				lowercase_:			nn.Convad )					->				torch.Tensor:
	A__       ,       A__							:   List[Any]					       =		features.shape[-2:]
	A__       ,       A__							:   Tuple					       =		conv_layer.stride
	A__       ,       A__							:   str					       =		conv_layer.kernel_size

	if in_height % stride_height == 0:
		A__							:   Tuple					       =		max(kernel_height - stride_height   ,				0 )
	else:
		A__							:   Any					       =		max(kernel_height - (in_height % stride_height)   ,				0 )

	if in_width % stride_width == 0:
		A__							:   Optional[Any]					       =		max(kernel_width - stride_width   ,				0 )
	else:
		A__							:   Tuple					       =		max(kernel_width - (in_width % stride_width)   ,				0 )

	A__							:   Optional[Any]					       =		pad_along_width // 2
	A__							:   List[Any]					       =		pad_along_width - pad_left
	A__							:   Optional[int]					       =		pad_along_height // 2
	A__							:   Union[str, Any]					       =		pad_along_height - pad_top

	A__							:   Tuple					       =		(pad_left, pad_right, pad_top, pad_bottom)
	return nn.functional.pad(lowercase_   ,				lowercase_   ,				"""constant"""   ,				0.0 )



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__ = 1   ,	A__ = 1   ,	A__ = False   ,	A__ = True   ,	A__ = True   ,	):
						super().__init__()
						A__							:   Dict					       =		config

						if in_channels % groups != 0:
							raise ValueError(F"""Input channels ({in_channels}) are not divisible by {groups} groups."""      )
						if out_channels % groups != 0:
							raise ValueError(F"""Output channels ({out_channels}) are not divisible by {groups} groups."""      )

						A__							:   int					       =		0 if config.tf_padding else int((kernel_size - 1) / 2      )

						A__							:   Any					       =		nn.Convad(
						    in_channels=A__   ,	out_channels=A__   ,	kernel_size=A__   ,	stride=A__   ,	padding=A__   ,	groups=A__   ,	bias=A__   ,	padding_mode="""zeros"""   ,	)

						if use_normalization:
							A__							:   Any					       =		nn.BatchNormad(
							    num_features=A__   ,	eps=config.layer_norm_eps   ,	momentum=0.9_9_9_7   ,	affine=A__   ,	track_running_stats=A__   ,	)
						else:
							A__							:   int					       =		None

						if use_activation:
							if isinstance(A__   ,	A__      ):
								A__							:   Optional[Any]					       =		ACTaFN[use_activation]
							elif isinstance(config.hidden_act   ,	A__      ):
								A__							:   Optional[Any]					       =		ACTaFN[config.hidden_act]
							else:
								A__							:   str					       =		config.hidden_act
						else:
							A__							:   Optional[int]					       =		None






					def   __A      (			self   ,	A__      ):
						if self.config.tf_padding:
							A__							:   str					       =		apply_tf_padding(A__   ,	self.convolution      )
						A__							:   Optional[int]					       =		self.convolution(A__      )
						if self.normalization is not None:
							A__							:   Union[str, Any]					       =		self.normalization(A__      )
						if self.activation is not None:
							A__							:   str					       =		self.activation(A__      )
						return features



class    _a						(__magic_name__    ):


					'''simple docstring'''






					UpperCAmelCase__:       Any  							=	MobileNetVaConfig
					UpperCAmelCase__:       Dict  							=	load_tf_weights_in_mobilenet_va
					UpperCAmelCase__:       List[str]  							=	'''mobilenet_v1'''
					UpperCAmelCase__:       Any  							=	'''pixel_values'''
					UpperCAmelCase__:       Optional[Any]  							=	False






					def   __A      (			self   ,	A__      ):
						if isinstance(A__   ,	(nn.Linear, nn.Convad)      ):
							module.weight.data.normal_(mean=0.0   ,	std=self.config.initializer_range      )
							if module.bias is not None:
								module.bias.data.zero_()
						elif isinstance(A__   ,	nn.BatchNormad      ):
							module.bias.data.zero_()
							module.weight.data.fill_(1.0      )



A_	:			List[Any]  	=      r'\n    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass. Use it\n    as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and\n    behavior.\n\n    Parameters:\n        config ([`MobileNetV1Config`]): Model configuration class with all the parameters of the model.\n            Initializing with a config file does not load the weights associated with the model, only the\n            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.\n'

A_	:			Union[str, Any]  	=      r'\n    Args:\n        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):\n            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See\n            [`MobileNetV1ImageProcessor.__call__`] for details.\n        output_hidden_states (`bool`, *optional*):\n            Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for\n            more detail.\n        return_dict (`bool`, *optional*):\n            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.\n'



@add_start_docstrings(
    '''The bare MobileNetV1 model outputting raw hidden-states without any specific head on top.'''		,       __magic_name__		,       )
class    _a						(__magic_name__    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__ = True      ):
						super().__init__(A__      )
						A__							:   Optional[Any]					       =		config

						A__							:   int					       =		32
						A__							:   Any					       =		max(int(depth * config.depth_multiplier      )   ,	config.min_depth      )

						A__							:   int					       =		MobileNetVaConvLayer(
						    A__   ,	in_channels=config.num_channels   ,	out_channels=A__   ,	kernel_size=3   ,	stride=2   ,	)

						A__							:   Dict					       =		[1, 2, 1, 2, 1, 2, 1, 1, 1, 1, 1, 2, 1]

						A__							:   Optional[Any]					       =		nn.ModuleList()
						for i in range(13      ):
							A__							:   Optional[int]					       =		out_channels

							if strides[i] == 2 or i == 0:
								depth *= 2
								A__							:   int					       =		max(int(depth * config.depth_multiplier      )   ,	config.min_depth      )

							self.layer.append(
							    MobileNetVaConvLayer(
							        A__   ,	in_channels=A__   ,	out_channels=A__   ,	kernel_size=3   ,	stride=strides[i]   ,	groups=A__   ,	)      )

							self.layer.append(
							    MobileNetVaConvLayer(
							        A__   ,	in_channels=A__   ,	out_channels=A__   ,	kernel_size=1   ,	)      )

						A__							:   Optional[Any]					       =		nn.AdaptiveAvgPoolad((1, 1)      ) if add_pooling_layer else None

						# Initialize weights and apply final processing
						self.post_init()






					def   __A      (			self   ,	A__      ):
						raise NotImplementedError






					@add_start_docstrings_to_model_forward(A__      )
					@add_code_sample_docstrings(
					    checkpoint=_CHECKPOINT_FOR_DOC   ,	output_type=A__   ,	config_class=_CONFIG_FOR_DOC   ,	modality="""vision"""   ,	expected_output=_EXPECTED_OUTPUT_SHAPE   ,	)
					def   __A      (			self   ,	A__ = None   ,	A__ = None   ,	A__ = None   ,	):
						A__							:   Union[str, Any]					       =		(
						    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
						)
						A__							:   List[str]					       =		return_dict if return_dict is not None else self.config.use_return_dict

						if pixel_values is None:
							raise ValueError("""You have to specify pixel_values"""      )

						A__							:   Optional[Any]					       =		self.conv_stem(A__      )

						A__							:   Any					       =		() if output_hidden_states else None

						for i, layer_module in enumerate(self.layer      ):
							A__							:   List[str]					       =		layer_module(A__      )

							if output_hidden_states:
								A__							:   Tuple					       =		all_hidden_states + (hidden_states,)

						A__							:   Any					       =		hidden_states

						if self.pooler is not None:
							A__							:   Dict					       =		torch.flatten(self.pooler(A__      )   ,	start_dim=1      )
						else:
							A__							:   str					       =		None

						if not return_dict:
							return tuple(v for v in [last_hidden_state, pooled_output, all_hidden_states] if v is not None      )

						return BaseModelOutputWithPoolingAndNoAttention(
						    last_hidden_state=A__   ,	pooler_output=A__   ,	hidden_states=A__   ,	)



@add_start_docstrings(
    '''
    MobileNetV1 model with an image classification head on top (a linear layer on top of the pooled features), e.g. for
    ImageNet.
    '''		,       __magic_name__		,       )
class    _a						(__magic_name__    ):


					'''simple docstring'''






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

						A__							:   Dict					       =		config.num_labels
						A__							:   List[Any]					       =		MobileNetVaModel(A__      )

						A__							:   Union[str, Any]					       =		self.mobilenet_va.layer[-1].convolution.out_channels

						# Classifier head
						A__							:   Union[str, Any]					       =		nn.Dropout(config.classifier_dropout_prob   ,	inplace=A__      )
						A__							:   Optional[Any]					       =		nn.Linear(A__   ,	config.num_labels      ) if config.num_labels > 0 else nn.Identity()

						# Initialize weights and apply final processing
						self.post_init()






					@add_start_docstrings_to_model_forward(A__      )
					@add_code_sample_docstrings(
					    checkpoint=_IMAGE_CLASS_CHECKPOINT   ,	output_type=A__   ,	config_class=_CONFIG_FOR_DOC   ,	expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT   ,	)
					def   __A      (			self   ,	A__ = None   ,	A__ = None   ,	A__ = None   ,	A__ = None   ,	):
						A__							:   List[Any]					       =		return_dict if return_dict is not None else self.config.use_return_dict

						A__							:   List[str]					       =		self.mobilenet_va(A__   ,	output_hidden_states=A__   ,	return_dict=A__      )

						A__							:   List[Any]					       =		outputs.pooler_output if return_dict else outputs[1]

						A__							:   Union[str, Any]					       =		self.classifier(self.dropout(A__      )      )

						A__							:   str					       =		None
						if labels is not None:
							if self.config.problem_type is None:
								if self.num_labels == 1:
									A__							:   Dict					       =		"""regression"""
								elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
									A__							:   Union[str, Any]					       =		"""single_label_classification"""
								else:
									A__							:   List[Any]					       =		"""multi_label_classification"""

							if self.config.problem_type == "regression":
								A__							:   int					       =		MSELoss()
								if self.num_labels == 1:
									A__							:   Optional[int]					       =		loss_fct(logits.squeeze()   ,	labels.squeeze()      )
								else:
									A__							:   Tuple					       =		loss_fct(A__   ,	A__      )
							elif self.config.problem_type == "single_label_classification":
								A__							:   Tuple					       =		CrossEntropyLoss()
								A__							:   Optional[int]					       =		loss_fct(logits.view(-1   ,	self.num_labels      )   ,	labels.view(-1      )      )
							elif self.config.problem_type == "multi_label_classification":
								A__							:   Optional[int]					       =		BCEWithLogitsLoss()
								A__							:   Tuple					       =		loss_fct(A__   ,	A__      )

						if not return_dict:
							A__							:   Optional[int]					       =		(logits,) + outputs[2:]
							return ((loss,) + output) if loss is not None else output

						return ImageClassifierOutputWithNoAttention(
						    loss=A__   ,	logits=A__   ,	hidden_states=outputs.hidden_states   ,	)






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


A_	:			int  	=      logging.get_logger(__name__)  # pylint: disable=invalid-name



class    _a						(__magic_name__    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	):
						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=A__   ,	speech_processor=A__   ,	vae=A__   ,	text_encoder=A__   ,	tokenizer=A__   ,	unet=A__   ,	scheduler=A__   ,	feature_extractor=A__   ,	)






					def   __A      (			self   ,	A__ = "auto"      ):
						if slice_size == "auto":
							A__							:   Optional[Any]					       =		self.unet.config.attention_head_dim // 2
						self.unet.set_attention_slice(A__      )






					def   __A      (			self      ):
						self.enable_attention_slicing(A__      )






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

						A__							:   Union[str, Any]					       =		self.speech_processor.tokenizer.batch_decode(A__   ,	skip_special_tokens=A__   ,	normalize=A__      )[
						    0
						]

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

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

						# get prompt text embeddings
						A__							:   Optional[int]					       =		self.tokenizer(
						    A__   ,	padding="""max_length"""   ,	max_length=self.tokenizer.model_max_length   ,	return_tensors="""pt"""   ,	)
						A__							:   Any					       =		text_inputs.input_ids

						if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
							A__							:   str					       =		self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]      )
							logger.warning(
							    """The following part of your input was truncated because CLIP can only handle sequences up to"""
							    F""" {self.tokenizer.model_max_length} tokens: {removed_text}"""      )
							A__							:   Dict					       =		text_input_ids[:, : self.tokenizer.model_max_length]
						A__							:   int					       =		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__							:   List[str]					       =		text_embeddings.shape
						A__							:   Dict					       =		text_embeddings.repeat(1   ,	A__   ,	1      )
						A__							:   Union[str, Any]					       =		text_embeddings.view(bs_embed * num_images_per_prompt   ,	A__   ,	-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__							:   int					       =		guidance_scale > 1.0
						# get unconditional embeddings for classifier free guidance
						if do_classifier_free_guidance:
							A__							:   List[str]
							if negative_prompt is None:
								A__							:   Union[str, Any]					       =		[""""""] * batch_size
							elif type(A__      ) is not type(A__      ):
								raise TypeError(
								    F"""`negative_prompt` should be the same type to `prompt`, but got {type(A__      )} !="""
								    F""" {type(A__      )}."""      )
							elif isinstance(A__   ,	A__      ):
								A__							:   Union[str, Any]					       =		[negative_prompt]
							elif batch_size != len(A__      ):
								raise ValueError(
								    F"""`negative_prompt`: {negative_prompt} has batch size {len(A__      )}, but `prompt`:"""
								    F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
								    """ the batch size of `prompt`."""      )
							else:
								A__							:   int					       =		negative_prompt

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

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

						# set timesteps
						self.scheduler.set_timesteps(A__      )

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

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

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

							# predict the noise residual
							A__							:   Union[str, Any]					       =		self.unet(A__   ,	A__   ,	encoder_hidden_states=A__      ).sample

							# perform guidance
							if do_classifier_free_guidance:
								A__       ,       A__							:   List[Any]					       =		noise_pred.chunk(2      )
								A__							:   Dict					       =		noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

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

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

						A__							:   str					       =		1 / 0.1_8_2_1_5 * latents
						A__							:   Optional[Any]					       =		self.vae.decode(A__      ).sample

						A__							:   Tuple					       =		(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__							:   Any					       =		image.cpu().permute(0   ,	2   ,	3   ,	1      ).float().numpy()

						if output_type == "pil":
							A__							:   Optional[Any]					       =		self.numpy_to_pil(A__      )

						if not return_dict:
							return image

						return StableDiffusionPipelineOutput(images=A__   ,	nsfw_content_detected=A__      )



