


'''simple docstring'''
import inspect
import unittest

import numpy as np

from tests.test_modeling_common import floats_tensor
from transformers import DetrConfig, MaskFormerConfig, SwinConfig, is_torch_available, is_vision_available
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
from transformers.utils import cached_property

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import MaskFormerForInstanceSegmentation, MaskFormerModel

		if is_vision_available():
				from transformers import MaskFormerImageProcessor

if is_vision_available():
		from PIL import Image
class    lowercase	:

		"""simple docstring"""

		def __init__(		self    ,   UpperCamelCase_    ,   UpperCamelCase_=2    ,   UpperCamelCase_=True    ,   UpperCamelCase_=False    ,   UpperCamelCase_=10    ,   UpperCamelCase_=3    ,   UpperCamelCase_=32 * 4    ,   UpperCamelCase_=32 * 6    ,   UpperCamelCase_=4    ,   UpperCamelCase_=32    ,   ):

							'''simple docstring'''


							UpperCamelCase__   :List[Any]												= parent
							UpperCamelCase__   :List[Any]												= batch_size
							UpperCamelCase__   :Optional[Any]												= is_training
							UpperCamelCase__   :int												= use_auxiliary_loss
							UpperCamelCase__   :Dict												= num_queries
							UpperCamelCase__   :Optional[Any]												= num_channels
							UpperCamelCase__   :Any												= min_size
							UpperCamelCase__   :Union[str, Any]												= max_size
							UpperCamelCase__   :int												= num_labels
							UpperCamelCase__   :Tuple												= mask_feature_size




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :str												= floats_tensor([self.batch_size, self.num_channels, self.min_size, self.max_size]							).to(
							    a_							)

							UpperCamelCase__   :str												= torch.ones([self.batch_size, self.min_size, self.max_size]    ,   device=a_							)

							UpperCamelCase__   :List[Any]												= (
							    torch.rand([self.batch_size, self.num_labels, self.min_size, self.max_size]    ,   device=a_							) > 0.5
							).float()
							UpperCamelCase__   :Any												= (torch.rand((self.batch_size, self.num_labels)    ,   device=a_							) > 0.5).long()

							UpperCamelCase__   :Optional[Any]												= self.get_config()
							return config, pixel_values, pixel_mask, mask_labels, class_labels




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							return MaskFormerConfig.from_backbone_and_decoder_configs(
							    backbone_config=SwinConfig(
							        depths=[1, 1, 1, 1]    ,   )    ,   decoder_config=DetrConfig(
							        decoder_ffn_dim=128    ,   num_queries=self.num_queries    ,   decoder_attention_heads=2    ,   d_model=self.mask_feature_size    ,   )    ,   mask_feature_size=self.mask_feature_size    ,   fpn_feature_size=self.mask_feature_size    ,   num_channels=self.num_channels    ,   num_labels=self.num_labels    ,   )




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :Tuple												= self.prepare_config_and_inputs()
							UpperCamelCase__   :Dict												= {'''pixel_values''': pixel_values, '''pixel_mask''': pixel_mask}
							return config, inputs_dict




		def     lowerCAmelCase__				(		self    ,   UpperCamelCase_    ,   UpperCamelCase_							):

							'''simple docstring'''


							UpperCamelCase__   :Any												= output.encoder_hidden_states
							UpperCamelCase__   :List[str]												= output.pixel_decoder_hidden_states
							UpperCamelCase__   :int												= output.transformer_decoder_hidden_states

							self.parent.assertTrue(len(a_							)    ,   len(config.backbone_config.depths							)							)
							self.parent.assertTrue(len(a_							)    ,   len(config.backbone_config.depths							)							)
							self.parent.assertTrue(len(a_							)    ,   config.decoder_config.decoder_layers							)




		def     lowerCAmelCase__				(		self    ,   UpperCamelCase_    ,   UpperCamelCase_    ,   UpperCamelCase_    ,   UpperCamelCase_=False							):

							'''simple docstring'''


							with torch.no_grad():
												UpperCamelCase__   :Optional[int]												= MaskFormerModel(config=a_							)
												model.to(a_							)
												model.eval()

												UpperCamelCase__   :Optional[Any]												= model(pixel_values=a_    ,   pixel_mask=a_							)
												UpperCamelCase__   :int												= model(a_    ,   output_hidden_states=a_							)
							# the correct shape of output.transformer_decoder_hidden_states ensure the correcteness of the
							# encoder and pixel decoder
							self.parent.assertEqual(
							    output.transformer_decoder_last_hidden_state.shape    ,   (self.batch_size, self.num_queries, self.mask_feature_size)    ,   )
							# let's ensure the other two hidden state exists
							self.parent.assertTrue(output.pixel_decoder_last_hidden_state is not None							)
							self.parent.assertTrue(output.encoder_last_hidden_state is not None							)

							if output_hidden_states:
												self.check_output_hidden_state(a_    ,   a_							)




		def     lowerCAmelCase__				(		self    ,   UpperCamelCase_    ,   UpperCamelCase_    ,   UpperCamelCase_    ,   UpperCamelCase_    ,   UpperCamelCase_							):

							'''simple docstring'''


							UpperCamelCase__   :Any												= MaskFormerForInstanceSegmentation(config=a_							)
							model.to(a_							)
							model.eval()

							def comm_check_on_output(UpperCamelCase_							):
												# let's still check that all the required stuff is there
												self.parent.assertTrue(result.transformer_decoder_last_hidden_state is not None							)
												self.parent.assertTrue(result.pixel_decoder_last_hidden_state is not None							)
												self.parent.assertTrue(result.encoder_last_hidden_state is not None							)
												# okay, now we need to check the logits shape
												# due to the encoder compression, masks have a //4 spatial size
												self.parent.assertEqual(
												    result.masks_queries_logits.shape    ,   (self.batch_size, self.num_queries, self.min_size // 4, self.max_size // 4)    ,   )
												# + 1 for null class
												self.parent.assertEqual(
												    result.class_queries_logits.shape    ,   (self.batch_size, self.num_queries, self.num_labels + 1)							)

							with torch.no_grad():
												UpperCamelCase__   :Optional[Any]												= model(pixel_values=a_    ,   pixel_mask=a_							)
												UpperCamelCase__   :Optional[int]												= model(a_							)

												comm_check_on_output(a_							)

												UpperCamelCase__   :List[str]												= model(
												    pixel_values=a_    ,   pixel_mask=a_    ,   mask_labels=a_    ,   class_labels=a_							)

							comm_check_on_output(a_							)

							self.parent.assertTrue(result.loss is not None							)
							self.parent.assertEqual(result.loss.shape    ,   torch.Size([1]							)							)



@require_torch
class    lowercase	(							__snake_case	,		__snake_case	,		unittest.TestCase					):

		"""simple docstring"""

		_a     =				(MaskFormerModel, MaskFormerForInstanceSegmentation) if is_torch_available() else ()
		_a     =				(
		    {'feature-extraction': MaskFormerModel, 'image-segmentation': MaskFormerForInstanceSegmentation}
		    if is_torch_available()
		    else {}
		)

		_a     =				False
		_a     =				False
		_a     =				False
		_a     =				False




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :Optional[Any]												= MaskFormerModelTester(self							)
							UpperCamelCase__   :int												= ConfigTester(self    ,   config_class=a_    ,   has_text_modality=a_							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							self.config_tester.run_common_tests()




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :str												= self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.create_and_check_maskformer_model(a_    ,   **a_    ,   output_hidden_states=a_							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :Optional[int]												= self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_maskformer_instance_segmentation_head_model(*a_							)




		@unittest.skip(reason='''MaskFormer does not use inputs_embeds'''							)
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							pass




		@unittest.skip(reason='''MaskFormer does not have a get_input_embeddings method'''							)
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							pass




		@unittest.skip(reason='''MaskFormer is not a generative model'''							)
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							pass




		@unittest.skip(reason='''MaskFormer does not use token embeddings'''							)
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							pass




		@require_torch_multi_gpu
		@unittest.skip(
		    reason='''MaskFormer has some layers using `add_module` which doesn\'t work well with `nn.DataParallel`'''							)
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							pass




		@unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''							)
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							pass




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :str												= self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
												UpperCamelCase__   :Optional[int]												= model_class(a_							)
												UpperCamelCase__   :Optional[int]												= inspect.signature(model.forward							)
												# signature.parameters is an OrderedDict => so arg_names order is deterministic
												UpperCamelCase__   :Union[str, Any]												= [*signature.parameters.keys()]

												UpperCamelCase__   :str												= ['''pixel_values''']
												self.assertListEqual(arg_names[:1]    ,   a_							)




		@slow
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							for model_name in ["facebook/maskformer-swin-small-coco"]:
												UpperCamelCase__   :Optional[Any]												= MaskFormerModel.from_pretrained(a_							)
												self.assertIsNotNone(a_							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :List[Any]												= (self.model_tester.min_size,) * 2
							UpperCamelCase__   :List[str]												= {
							    '''pixel_values''': torch.randn((2, 3, *size)    ,   device=a_							),
							    '''mask_labels''': torch.randn((2, 10, *size)    ,   device=a_							),
							    '''class_labels''': torch.zeros(2    ,   10    ,   device=a_							).long(),
							}

							UpperCamelCase__   :List[Any]												= MaskFormerForInstanceSegmentation(MaskFormerConfig()							).to(a_							)
							UpperCamelCase__   :Optional[Any]												= model(**a_							)
							self.assertTrue(outputs.loss is not None							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :int												= self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.create_and_check_maskformer_model(a_    ,   **a_    ,   output_hidden_states=a_							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :Union[str, Any]												= self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
												UpperCamelCase__   :int												= model_class(a_							).to(a_							)
												UpperCamelCase__   :str												= model(**a_    ,   output_attentions=a_							)
												self.assertTrue(outputs.attentions is not None							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							if not self.model_tester.is_training:
												return
							# only MaskFormerForInstanceSegmentation has the loss
							UpperCamelCase__   :Union[str, Any]												= self.all_model_classes[1]
							UpperCamelCase__   :List[Any]												= self.model_tester.prepare_config_and_inputs()

							UpperCamelCase__   :Union[str, Any]												= model_class(a_							)
							model.to(a_							)
							model.train()

							UpperCamelCase__   :Any												= model(a_    ,   mask_labels=a_    ,   class_labels=a_							).loss
							loss.backward()




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :Dict												= self.all_model_classes[1]
							UpperCamelCase__   :str												= self.model_tester.prepare_config_and_inputs()
							UpperCamelCase__   :str												= True
							UpperCamelCase__   :str												= True

							UpperCamelCase__   :Optional[int]												= model_class(a_							)
							model.to(a_							)
							model.train()

							UpperCamelCase__   :Union[str, Any]												= model(a_    ,   mask_labels=a_    ,   class_labels=a_							)

							UpperCamelCase__   :Optional[Any]												= outputs.encoder_hidden_states[0]
							encoder_hidden_states.retain_grad()

							UpperCamelCase__   :Union[str, Any]												= outputs.pixel_decoder_hidden_states[0]
							pixel_decoder_hidden_states.retain_grad()
							# we requires_grad=True in inputs_embeds (line 2152), the original implementation don't
							UpperCamelCase__   :int												= outputs.transformer_decoder_hidden_states[0]
							transformer_decoder_hidden_states.retain_grad()

							UpperCamelCase__   :Any												= outputs.attentions[0]
							attentions.retain_grad()

							outputs.loss.backward(retain_graph=a_							)

							self.assertIsNotNone(encoder_hidden_states.grad							)
							self.assertIsNotNone(pixel_decoder_hidden_states.grad							)
							self.assertIsNotNone(transformer_decoder_hidden_states.grad							)
							self.assertIsNotNone(attentions.grad							)



__snake_case							      =     1e-4


def 			a      (			)  ->	int:



					'''simple docstring'''






					UpperCamelCase__   :Any												= Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''   )
					return image







@require_vision
@slow
class    lowercase	(							unittest.TestCase					):

		"""simple docstring"""

		@cached_property
		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							return (
							    MaskFormerImageProcessor.from_pretrained('''facebook/maskformer-swin-small-coco'''							)
							    if is_vision_available()
							    else None
							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :Optional[int]												= MaskFormerModel.from_pretrained('''facebook/maskformer-swin-small-coco'''							).to(a_							)
							UpperCamelCase__   :Tuple												= self.default_image_processor
							UpperCamelCase__   :List[str]												= prepare_img()
							UpperCamelCase__   :Dict												= image_processor(a_    ,   return_tensors='''pt'''							).to(a_							)
							UpperCamelCase__   :str												= inputs['''pixel_values'''].shape
							# check size is divisible by 32
							self.assertTrue((inputs_shape[-1] % 32) == 0 and (inputs_shape[-2] % 32) == 0							)
							# check size
							self.assertEqual(a_    ,   (1, 3, 800, 1088)							)

							with torch.no_grad():
												UpperCamelCase__   :Tuple												= model(**a_							)

							UpperCamelCase__   :Dict												= torch.tensor(
							    [[-0.0482, 0.9228, 0.4951], [-0.2547, 0.8017, 0.8527], [-0.0069, 0.3385, -0.0089]]							).to(a_							)
							self.assertTrue(
							    torch.allclose(
							        outputs.encoder_last_hidden_state[0, 0, :3, :3]    ,   a_    ,   atol=a_							)							)

							UpperCamelCase__   :int												= torch.tensor(
							    [[-0.8422, -0.8434, -0.9718], [-1.0144, -0.5565, -0.4195], [-1.0038, -0.4484, -0.1961]]							).to(a_							)
							self.assertTrue(
							    torch.allclose(
							        outputs.pixel_decoder_last_hidden_state[0, 0, :3, :3]    ,   a_    ,   atol=a_							)							)

							UpperCamelCase__   :Any												= torch.tensor(
							    [[0.2852, -0.0159, 0.9735], [0.6254, 0.1858, 0.8529], [-0.0680, -0.4116, 1.8413]]							).to(a_							)
							self.assertTrue(
							    torch.allclose(
							        outputs.transformer_decoder_last_hidden_state[0, :3, :3]    ,   a_    ,   atol=a_							)							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :int												= (
							    MaskFormerForInstanceSegmentation.from_pretrained('''facebook/maskformer-swin-small-coco'''							)
							    .to(a_							)
							    .eval()
							)
							UpperCamelCase__   :str												= self.default_image_processor
							UpperCamelCase__   :Union[str, Any]												= prepare_img()
							UpperCamelCase__   :Tuple												= image_processor(a_    ,   return_tensors='''pt'''							).to(a_							)
							UpperCamelCase__   :str												= inputs['''pixel_values'''].shape
							# check size is divisible by 32
							self.assertTrue((inputs_shape[-1] % 32) == 0 and (inputs_shape[-2] % 32) == 0							)
							# check size
							self.assertEqual(a_    ,   (1, 3, 800, 1088)							)

							with torch.no_grad():
												UpperCamelCase__   :List[Any]												= model(**a_							)
							# masks_queries_logits
							UpperCamelCase__   :Tuple												= outputs.masks_queries_logits
							self.assertEqual(
							    masks_queries_logits.shape    ,   (1, model.config.decoder_config.num_queries, inputs_shape[-2] // 4, inputs_shape[-1] // 4)    ,   )
							UpperCamelCase__   :Any												= [
							    [-1.3737124, -1.7724937, -1.9364233],
							    [-1.5977281, -1.9867939, -2.1523695],
							    [-1.5795398, -1.9269832, -2.093942],
							]
							UpperCamelCase__   :List[Any]												= torch.tensor(a_							).to(a_							)
							self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]    ,   a_    ,   atol=a_							)							)
							# class_queries_logits
							UpperCamelCase__   :Tuple												= outputs.class_queries_logits
							self.assertEqual(
							    class_queries_logits.shape    ,   (1, model.config.decoder_config.num_queries, model.config.num_labels + 1)							)
							UpperCamelCase__   :str												= torch.tensor(
							    [
							        [1.65_12e00, -5.25_72e00, -3.35_19e00],
							        [3.61_69e-02, -5.90_25e00, -2.93_13e00],
							        [1.07_66e-04, -7.76_30e00, -5.12_63e00],
							    ]							).to(a_							)
							self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]    ,   a_    ,   atol=a_							)							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :List[str]												= (
							    MaskFormerForInstanceSegmentation.from_pretrained('''facebook/maskformer-resnet101-coco-stuff'''							)
							    .to(a_							)
							    .eval()
							)
							UpperCamelCase__   :Dict												= self.default_image_processor
							UpperCamelCase__   :Optional[Any]												= prepare_img()
							UpperCamelCase__   :List[Any]												= image_processor(a_    ,   return_tensors='''pt'''							).to(a_							)
							UpperCamelCase__   :Optional[int]												= inputs['''pixel_values'''].shape
							# check size is divisible by 32
							self.assertTrue((inputs_shape[-1] % 32) == 0 and (inputs_shape[-2] % 32) == 0							)
							# check size
							self.assertEqual(a_    ,   (1, 3, 800, 1088)							)

							with torch.no_grad():
												UpperCamelCase__   :List[str]												= model(**a_							)
							# masks_queries_logits
							UpperCamelCase__   :str												= outputs.masks_queries_logits
							self.assertEqual(
							    masks_queries_logits.shape    ,   (1, model.config.decoder_config.num_queries, inputs_shape[-2] // 4, inputs_shape[-1] // 4)    ,   )
							UpperCamelCase__   :int												= [[-0.9046, -2.6366, -4.6062], [-3.4179, -5.7890, -8.8057], [-4.9179, -7.6560, -10.7711]]
							UpperCamelCase__   :str												= torch.tensor(a_							).to(a_							)
							self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]    ,   a_    ,   atol=a_							)							)
							# class_queries_logits
							UpperCamelCase__   :str												= outputs.class_queries_logits
							self.assertEqual(
							    class_queries_logits.shape    ,   (1, model.config.decoder_config.num_queries, model.config.num_labels + 1)							)
							UpperCamelCase__   :int												= torch.tensor(
							    [[4.7188, -3.2585, -2.8857], [6.6871, -2.9181, -1.2487], [7.2449, -2.2764, -2.1874]]							).to(a_							)
							self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]    ,   a_    ,   atol=a_							)							)




		def     lowerCAmelCase__				(		self							):

							'''simple docstring'''


							UpperCamelCase__   :Union[str, Any]												= (
							    MaskFormerForInstanceSegmentation.from_pretrained('''facebook/maskformer-swin-small-coco'''							)
							    .to(a_							)
							    .eval()
							)
							UpperCamelCase__   :Dict												= self.default_image_processor

							UpperCamelCase__   :Any												= image_processor(
							    [np.zeros((3, 800, 1333)							), np.zeros((3, 800, 1333)							)]    ,   segmentation_maps=[np.zeros((384, 384)							).astype(np.floataa							), np.zeros((384, 384)							).astype(np.floataa							)]    ,   return_tensors='''pt'''    ,   )

							UpperCamelCase__   :Any												= inputs['''pixel_values'''].to(a_							)
							UpperCamelCase__   :int												= [el.to(a_							) for el in inputs['''mask_labels''']]
							UpperCamelCase__   :int												= [el.to(a_							) for el in inputs['''class_labels''']]

							with torch.no_grad():
												UpperCamelCase__   :List[Any]												= model(**a_							)

							self.assertTrue(outputs.loss is not None							)



"""simple docstring"""
import inspect
from typing import Callable, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

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


SCREAMING_SNAKE_CASE :    List[str]									=				logging.get_logger(__name__)  # pylint: disable=invalid-name



class 			_UpperCAmelCase (							__snake_case				):






		'''simple docstring'''

		def __init__(self				,       a_				,       a_				,       a_				,       a_				,       a_				,       a_				,       a_				,       ):





			'''simple docstring'''

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



		def 				SCREAMING_SNAKE_CASE  (self				,       a_ = "auto"					):





			'''simple docstring'''

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



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			self.enable_attention_slicing(a_					)



		@torch.no_grad()
		def __call__(self				,       a_				,       a_ = 5_12				,       a_ = 5_12				,       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_ = None				,       **a_				,       ):





			'''simple docstring'''

			if isinstance(a_				,       a_					):
				__snake_case	:						Any						=				1
			elif isinstance(a_				,       a_					):
				__snake_case	:						Any						=				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
			__snake_case	:						int						=				self.tokenizer(
			    a_				,       padding='''max_length'''				,       max_length=self.tokenizer.model_max_length				,       return_tensors='''pt'''				,       )
			__snake_case	:						int						=				text_inputs.input_ids

			if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
				__snake_case	:						List[Any]						=				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}"""					)
				__snake_case	:						Optional[int]						=				text_input_ids[:, : self.tokenizer.model_max_length]

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

			# duplicate text embeddings for each generation per prompt, using mps friendly method
			__snake_case	,							__snake_case	,							__snake_case	:						Union[str, Any]						=				text_embeddings.shape
			__snake_case	:						Optional[int]						=				text_embeddings.repeat(1				,       a_				,       1					)
			__snake_case	:						Dict						=				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.
			__snake_case	:						Dict						=				guidance_scale > 1.0
			# get unconditional embeddings for classifier free guidance
			if do_classifier_free_guidance:
				__snake_case	:						List[str]
				if negative_prompt is None:
					__snake_case	:						List[Any]						=				['''''']
				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_					):
					__snake_case	:						List[str]						=				[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:
					__snake_case	:						Optional[int]						=				negative_prompt

				__snake_case	:						Optional[int]						=				text_input_ids.shape[-1]
				__snake_case	:						List[Any]						=				self.tokenizer(
				    a_				,       padding='''max_length'''				,       max_length=a_				,       truncation=a_				,       return_tensors='''pt'''				,       )
				__snake_case	:						str						=				self.text_encoder(uncond_input.input_ids.to(self.device					)					)[0]

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

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

			# 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
			__snake_case	:						Optional[Any]						=				self.scheduler.timesteps.to(self.device					)

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

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

				# predict the noise residual
				__snake_case	:						int						=				self.unet(a_				,       a_				,       encoder_hidden_states=a_					).sample

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

				# compute the previous noisy sample x_t -> x_t-1
				__snake_case	:						Optional[Any]						=				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_					)

			__snake_case	:						Union[str, Any]						=				1 / 0.1_8215 * latents
			__snake_case	:						Optional[Any]						=				self.vae.decode(a_					).sample

			__snake_case	:						Union[str, Any]						=				(image / 2 + 0.5).clamp(0				,       1					)

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

			if self.safety_checker is not None:
				__snake_case	:						Optional[int]						=				self.feature_extractor(self.numpy_to_pil(a_					)				,       return_tensors='''pt'''					).to(
				    self.device					)
				__snake_case	,							__snake_case	:						List[Any]						=				self.safety_checker(
				    images=a_				,       clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype					)					)
			else:
				__snake_case	:						Union[str, Any]						=				None

			if output_type == "pil":
				__snake_case	:						Union[str, Any]						=				self.numpy_to_pil(a_					)

			if not return_dict:
				return (image, has_nsfw_concept)

			return StableDiffusionPipelineOutput(images=a_				,       nsfw_content_detected=a_					)





