

from typing import Optional

import torch
import torch.utils.checkpoint
from torch import Tensor, nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss

from ...activations import ACTaFN
from ...modeling_outputs import (
    BackboneOutput,
    BaseModelOutputWithNoAttention,
    BaseModelOutputWithPoolingAndNoAttention,
    ImageClassifierOutputWithNoAttention,
)
from ...modeling_utils import PreTrainedModel
from ...utils import (
    add_code_sample_docstrings,
    add_start_docstrings,
    add_start_docstrings_to_model_forward,
    logging,
    replace_return_docstrings,
)
from ...utils.backbone_utils import BackboneMixin
from .configuration_resnet import ResNetConfig


A_	:			Dict  	=      logging.get_logger(__name__)

# General docstring
A_	:			Tuple  	=      'ResNetConfig'

# Base docstring
A_	:			Optional[Any]  	=      'microsoft/resnet-50'
A_	:			Any  	=      [1, 2048, 7, 7]

# Image classification docstring
A_	:			List[Any]  	=      'microsoft/resnet-50'
A_	:			Optional[int]  	=      'tiger cat'

A_	:			Tuple  	=      [
    'microsoft/resnet-50',
    # See all resnet models at https://huggingface.co/models?filter=resnet
]



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__ = 3   ,	A__ = 1   ,	A__ = "relu"      ):
						super().__init__()
						A__							:   int					       =		nn.Convad(
						    A__   ,	A__   ,	kernel_size=A__   ,	stride=A__   ,	padding=kernel_size // 2   ,	bias=A__      )
						A__							:   Optional[int]					       =		nn.BatchNormad(A__      )
						A__							:   Dict					       =		ACTaFN[activation] if activation is not None else nn.Identity()






					def   __A      (			self   ,	A__      ):
						A__							:   Any					       =		self.convolution(A__      )
						A__							:   Any					       =		self.normalization(A__      )
						A__							:   int					       =		self.activation(A__      )
						return hidden_state



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__()
						A__							:   int					       =		ResNetConvLayer(
						    config.num_channels   ,	config.embedding_size   ,	kernel_size=7   ,	stride=2   ,	activation=config.hidden_act      )
						A__							:   Tuple					       =		nn.MaxPoolad(kernel_size=3   ,	stride=2   ,	padding=1      )
						A__							:   Optional[int]					       =		config.num_channels






					def   __A      (			self   ,	A__      ):
						A__							:   Tuple					       =		pixel_values.shape[1]
						if num_channels != self.num_channels:
							raise ValueError(
							    """Make sure that the channel dimension of the pixel values match with the one set in the configuration."""      )
						A__							:   Any					       =		self.embedder(A__      )
						A__							:   Union[str, Any]					       =		self.pooler(A__      )
						return embedding



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__ = 2      ):
						super().__init__()
						A__							:   Optional[int]					       =		nn.Convad(A__   ,	A__   ,	kernel_size=1   ,	stride=A__   ,	bias=A__      )
						A__							:   Dict					       =		nn.BatchNormad(A__      )






					def   __A      (			self   ,	A__      ):
						A__							:   int					       =		self.convolution(A__      )
						A__							:   Optional[Any]					       =		self.normalization(A__      )
						return hidden_state



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__ = 1   ,	A__ = "relu"      ):
						super().__init__()
						A__							:   Any					       =		in_channels != out_channels or stride != 1
						A__							:   List[str]					       =		(
						    ResNetShortCut(A__   ,	A__   ,	stride=A__      ) if should_apply_shortcut else nn.Identity()
						)
						A__							:   List[Any]					       =		nn.Sequential(
						    ResNetConvLayer(A__   ,	A__   ,	stride=A__      )   ,	ResNetConvLayer(A__   ,	A__   ,	activation=A__      )   ,	)
						A__							:   List[Any]					       =		ACTaFN[activation]






					def   __A      (			self   ,	A__      ):
						A__							:   Optional[Any]					       =		hidden_state
						A__							:   Any					       =		self.layer(A__      )
						A__							:   Any					       =		self.shortcut(A__      )
						hidden_state += residual
						A__							:   Dict					       =		self.activation(A__      )
						return hidden_state



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__ = 1   ,	A__ = "relu"   ,	A__ = 4      ):
						super().__init__()
						A__							:   str					       =		in_channels != out_channels or stride != 1
						A__							:   List[Any]					       =		out_channels // reduction
						A__							:   Tuple					       =		(
						    ResNetShortCut(A__   ,	A__   ,	stride=A__      ) if should_apply_shortcut else nn.Identity()
						)
						A__							:   List[Any]					       =		nn.Sequential(
						    ResNetConvLayer(A__   ,	A__   ,	kernel_size=1      )   ,	ResNetConvLayer(A__   ,	A__   ,	stride=A__      )   ,	ResNetConvLayer(A__   ,	A__   ,	kernel_size=1   ,	activation=A__      )   ,	)
						A__							:   Dict					       =		ACTaFN[activation]






					def   __A      (			self   ,	A__      ):
						A__							:   List[str]					       =		hidden_state
						A__							:   Tuple					       =		self.layer(A__      )
						A__							:   List[Any]					       =		self.shortcut(A__      )
						hidden_state += residual
						A__							:   Any					       =		self.activation(A__      )
						return hidden_state



class    _a						(nn.Module    ):


					'''simple docstring'''






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

						A__							:   Tuple					       =		ResNetBottleNeckLayer if config.layer_type == """bottleneck""" else ResNetBasicLayer

						A__							:   int					       =		nn.Sequential(
						    # downsampling is done in the first layer with stride of 2
						    layer(A__   ,	A__   ,	stride=A__   ,	activation=config.hidden_act      )   ,	*[layer(A__   ,	A__   ,	activation=config.hidden_act      ) for _ in range(depth - 1      )]   ,	)






					def   __A      (			self   ,	A__      ):
						A__							:   Optional[int]					       =		input
						for layer in self.layers:
							A__							:   Dict					       =		layer(A__      )
						return hidden_state



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__()
						A__							:   Dict					       =		nn.ModuleList([]      )
						# based on `downsample_in_first_stage` the first layer of the first stage may or may not downsample the input
						self.stages.append(
						    ResNetStage(
						        A__   ,	config.embedding_size   ,	config.hidden_sizes[0]   ,	stride=2 if config.downsample_in_first_stage else 1   ,	depth=config.depths[0]   ,	)      )
						A__							:   Tuple					       =		zip(config.hidden_sizes   ,	config.hidden_sizes[1:]      )
						for (in_channels, out_channels), depth in zip(A__   ,	config.depths[1:]      ):
							self.stages.append(ResNetStage(A__   ,	A__   ,	A__   ,	depth=A__      )      )






					def   __A      (			self   ,	A__   ,	A__ = False   ,	A__ = True      ):
						A__							:   Optional[int]					       =		() if output_hidden_states else None

						for stage_module in self.stages:
							if output_hidden_states:
								A__							:   Tuple					       =		hidden_states + (hidden_state,)

							A__							:   Optional[int]					       =		stage_module(A__      )

						if output_hidden_states:
							A__							:   str					       =		hidden_states + (hidden_state,)

						if not return_dict:
							return tuple(v for v in [hidden_state, hidden_states] if v is not None      )

						return BaseModelOutputWithNoAttention(
						    last_hidden_state=A__   ,	hidden_states=A__   ,	)



class    _a						(__magic_name__    ):


					'''simple docstring'''






					UpperCAmelCase__:       str  							=	ResNetConfig
					UpperCAmelCase__:       Optional[Any]  							=	'''resnet'''
					UpperCAmelCase__:       List[Any]  							=	'''pixel_values'''
					UpperCAmelCase__:       List[str]  							=	True






					def   __A      (			self   ,	A__      ):
						if isinstance(A__   ,	nn.Convad      ):
							nn.init.kaiming_normal_(module.weight   ,	mode="""fan_out"""   ,	nonlinearity="""relu"""      )
						elif isinstance(A__   ,	(nn.BatchNormad, nn.GroupNorm)      ):
							nn.init.constant_(module.weight   ,	1      )
							nn.init.constant_(module.bias   ,	0      )






					def   __A      (			self   ,	A__   ,	A__=False      ):
						if isinstance(A__   ,	A__      ):
							A__							:   str					       =		value



A_	:			Union[str, 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 ([`ResNetConfig`]): 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_	:			Optional[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            [`ConvNextImageProcessor.__call__`] for details.\n\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 ResNet model outputting raw features without any specific head on top.'''		,       __magic_name__		,       )
class    _a						(__magic_name__    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__(A__      )
						A__							:   Tuple					       =		config
						A__							:   Union[str, Any]					       =		ResNetEmbeddings(A__      )
						A__							:   List[Any]					       =		ResNetEncoder(A__      )
						A__							:   Tuple					       =		nn.AdaptiveAvgPoolad((1, 1)      )
						# Initialize weights and apply final processing
						self.post_init()






					@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__   ,	A__ = None   ,	A__ = None      ):
						A__							:   str					       =		(
						    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
						)
						A__							:   List[Any]					       =		return_dict if return_dict is not None else self.config.use_return_dict

						A__							:   Optional[int]					       =		self.embedder(A__      )

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

						A__							:   Any					       =		encoder_outputs[0]

						A__							:   List[str]					       =		self.pooler(A__      )

						if not return_dict:
							return (last_hidden_state, pooled_output) + encoder_outputs[1:]

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



@add_start_docstrings(
    '''
    ResNet 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__							:   Optional[int]					       =		ResNetModel(A__      )
						# classification head
						A__							:   Optional[Any]					       =		nn.Sequential(
						    nn.Flatten()   ,	nn.Linear(config.hidden_sizes[-1]   ,	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__							:   str					       =		return_dict if return_dict is not None else self.config.use_return_dict

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

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

						A__							:   List[Any]					       =		self.classifier(A__      )

						A__							:   List[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__							:   Tuple					       =		"""single_label_classification"""
								else:
									A__							:   Any					       =		"""multi_label_classification"""
							if self.config.problem_type == "regression":
								A__							:   Tuple					       =		MSELoss()
								if self.num_labels == 1:
									A__							:   Optional[Any]					       =		loss_fct(logits.squeeze()   ,	labels.squeeze()      )
								else:
									A__							:   Optional[Any]					       =		loss_fct(A__   ,	A__      )
							elif self.config.problem_type == "single_label_classification":
								A__							:   Tuple					       =		CrossEntropyLoss()
								A__							:   Union[str, Any]					       =		loss_fct(logits.view(-1   ,	self.num_labels      )   ,	labels.view(-1      )      )
							elif self.config.problem_type == "multi_label_classification":
								A__							:   Optional[Any]					       =		BCEWithLogitsLoss()
								A__							:   Dict					       =		loss_fct(A__   ,	A__      )

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

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



@add_start_docstrings(
    '''
    ResNet backbone, to be used with frameworks like DETR and MaskFormer.
    '''		,       __magic_name__		,       )
class    _a						(__magic_name__		,       __magic_name__    ):


					'''simple docstring'''






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

						A__							:   Union[str, Any]					       =		[config.embedding_size] + config.hidden_sizes
						A__							:   int					       =		ResNetEmbeddings(A__      )
						A__							:   Union[str, Any]					       =		ResNetEncoder(A__      )

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






					@add_start_docstrings_to_model_forward(A__      )
					@replace_return_docstrings(output_type=A__   ,	config_class=_CONFIG_FOR_DOC      )
					def   __A      (			self   ,	A__   ,	A__ = None   ,	A__ = None      ):
						A__							:   Any					       =		return_dict if return_dict is not None else self.config.use_return_dict
						A__							:   Union[str, Any]					       =		(
						    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
						)

						A__							:   Tuple					       =		self.embedder(A__      )

						A__							:   str					       =		self.encoder(A__   ,	output_hidden_states=A__   ,	return_dict=A__      )

						A__							:   Tuple					       =		outputs.hidden_states

						A__							:   Tuple					       =		()
						for idx, stage in enumerate(self.stage_names      ):
							if stage in self.out_features:
								feature_maps += (hidden_states[idx],)

						if not return_dict:
							A__							:   Optional[Any]					       =		(feature_maps,)
							if output_hidden_states:
								output += (outputs.hidden_states,)
							return output

						return BackboneOutput(
						    feature_maps=A__   ,	hidden_states=outputs.hidden_states if output_hidden_states else None   ,	attentions=A__   ,	)






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__      )



