





"""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 					__UpperCAmelCase							:
				def __init__(		self    :			str	, a_    :			Dict	, a_    :			Dict=2	, a_    :			List[Any]=True	, a_    :			str=False	, a_    :			Dict=10	, a_    :			Union[str, Any]=3	, a_    :			str=32 * 4	, a_    :			str=32 * 6	, a_    :			Dict=4	, a_    :			Union[str, Any]=32	, )	->	str:



							'''simple docstring'''



							a__  :	List[Any]		 =			parent
							a__  :	Tuple		 =			batch_size
							a__  :	List[str]		 =			is_training
							a__  :	List[str]		 =			use_auxiliary_loss
							a__  :	Dict		 =			num_queries
							a__  :	List[Any]		 =			num_channels
							a__  :	str		 =			min_size
							a__  :	List[str]		 =			max_size
							a__  :	Optional[Any]		 =			num_labels
							a__  :	Optional[Any]		 =			mask_feature_size
				def       UpperCAmelCase      (		self    :			Optional[int]    )	->	int:



							'''simple docstring'''



							a__  :	Optional[Any]		 =			floats_tensor([self.batch_size, self.num_channels, self.min_size, self.max_size]    ).to(
							    a_    )

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

							a__  :	List[str]		 =			(
							    torch.rand([self.batch_size, self.num_labels, self.min_size, self.max_size]	, device=a_    ) > 0.5
							).float()
							a__  :	Optional[int]		 =			(torch.rand((self.batch_size, self.num_labels)	, device=a_    ) > 0.5).long()

							a__  :	Tuple		 =			self.get_config()
							return config, pixel_values, pixel_mask, mask_labels, class_labels
				def       UpperCAmelCase      (		self    :			List[Any]    )	->	Union[str, Any]:



							'''simple docstring'''



							return MaskFormerConfig.from_backbone_and_decoder_configs(
							    backbone_config=SwinConfig(
							        depths=[1, 1, 1, 1]	, )	, decoder_config=DetrConfig(
							        decoder_ffn_dim=1_28	, 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       UpperCAmelCase      (		self    :			Optional[int]    )	->	Optional[Any]:



							'''simple docstring'''



							a__ ,	a__ ,	a__ ,	a__ ,	a__  :	List[str]		 =			self.prepare_config_and_inputs()
							a__  :	Dict		 =			{"pixel_values": pixel_values, "pixel_mask": pixel_mask}
							return config, inputs_dict
				def       UpperCAmelCase      (		self    :			Union[str, Any]	, a_    :			Dict	, a_    :			Optional[int]    )	->	List[str]:



							'''simple docstring'''



							a__  :	List[Any]		 =			output.encoder_hidden_states
							a__  :	str		 =			output.pixel_decoder_hidden_states
							a__  :	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       UpperCAmelCase      (		self    :			Optional[Any]	, a_    :			str	, a_    :			Union[str, Any]	, a_    :			Any	, a_    :			Optional[Any]=False    )	->	Any:



							'''simple docstring'''



							with torch.no_grad():
										a__  :	Union[str, Any]		 =			MaskFormerModel(config=a_    )
										model.to(a_    )
										model.eval()

										a__  :	Any		 =			model(pixel_values=a_	, pixel_mask=a_    )
										a__  :	Tuple		 =			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       UpperCAmelCase      (		self    :			Optional[int]	, a_    :			Dict	, a_    :			Optional[int]	, a_    :			Any	, a_    :			Tuple	, a_    :			Optional[int]    )	->	Tuple:



							'''simple docstring'''



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

							def comm_check_on_output(a_    :			List[Any]    ):
										# 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():
										a__  :	Optional[Any]		 =			model(pixel_values=a_	, pixel_mask=a_    )
										a__  :	List[Any]		 =			model(a_    )

										comm_check_on_output(a_    )

										a__  :	List[Any]		 =			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 					__UpperCAmelCase							(						_UpperCamelCase						,  _UpperCamelCase						,  unittest.TestCase  ):
				__lowerCamelCase				:	str				     =				(MaskFormerModel, MaskFormerForInstanceSegmentation) if is_torch_available() else ()
				__lowerCamelCase				:	Union[str, Any]				     =				(
				    {"feature-extraction": MaskFormerModel, "image-segmentation": MaskFormerForInstanceSegmentation}
				    if is_torch_available()
				    else {}
				)

				__lowerCamelCase				:	Any				     =				False
				__lowerCamelCase				:	int				     =				False
				__lowerCamelCase				:	str				     =				False
				__lowerCamelCase				:	int				     =				False
				def       UpperCAmelCase      (		self    :			Tuple    )	->	List[Any]:



							'''simple docstring'''



							a__  :	int		 =			MaskFormerModelTester(self    )
							a__  :	str		 =			ConfigTester(self	, config_class=a_	, has_text_modality=a_    )
				def       UpperCAmelCase      (		self    :			List[str]    )	->	Optional[int]:



							'''simple docstring'''



							self.config_tester.run_common_tests()
				def       UpperCAmelCase      (		self    :			Any    )	->	Union[str, Any]:



							'''simple docstring'''



							a__ ,	a__  :	Any		 =			self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.create_and_check_maskformer_model(a_	, **a_	, output_hidden_states=a_    )
				def       UpperCAmelCase      (		self    :			Optional[Any]    )	->	Tuple:



							'''simple docstring'''



							a__  :	Dict		 =			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       UpperCAmelCase      (		self    :			Dict    )	->	Optional[Any]:



							'''simple docstring'''



							pass
				@unittest.skip(reason="MaskFormer does not have a get_input_embeddings method"    )
				def       UpperCAmelCase      (		self    :			List[Any]    )	->	int:



							'''simple docstring'''



							pass
				@unittest.skip(reason="MaskFormer is not a generative model"    )
				def       UpperCAmelCase      (		self    :			Optional[Any]    )	->	Optional[int]:



							'''simple docstring'''



							pass
				@unittest.skip(reason="MaskFormer does not use token embeddings"    )
				def       UpperCAmelCase      (		self    :			Optional[int]    )	->	List[Any]:



							'''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       UpperCAmelCase      (		self    :			List[str]    )	->	List[str]:



							'''simple docstring'''



							pass
				@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."    )
				def       UpperCAmelCase      (		self    :			List[str]    )	->	Union[str, Any]:



							'''simple docstring'''



							pass
				def       UpperCAmelCase      (		self    :			str    )	->	Optional[Any]:



							'''simple docstring'''



							a__ ,	a__  :	Optional[int]		 =			self.model_tester.prepare_config_and_inputs_for_common()

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

										a__  :	List[Any]		 =			["pixel_values"]
										self.assertListEqual(arg_names[:1]	, a_    )
				@slow
				def       UpperCAmelCase      (		self    :			str    )	->	List[str]:



							'''simple docstring'''



							for model_name in ["facebook/maskformer-swin-small-coco"]:
										a__  :	List[Any]		 =			MaskFormerModel.from_pretrained(a_    )
										self.assertIsNotNone(a_    )
				def       UpperCAmelCase      (		self    :			Union[str, Any]    )	->	str:



							'''simple docstring'''



							a__  :	Any		 =			(self.model_tester.min_size,) * 2
							a__  :	Tuple		 =			{
							    "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(),
							}

							a__  :	Any		 =			MaskFormerForInstanceSegmentation(MaskFormerConfig()    ).to(a_    )
							a__  :	str		 =			model(**a_    )
							self.assertTrue(outputs.loss is not None    )
				def       UpperCAmelCase      (		self    :			Tuple    )	->	Any:



							'''simple docstring'''



							a__ ,	a__  :	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       UpperCAmelCase      (		self    :			Optional[int]    )	->	Tuple:



							'''simple docstring'''



							a__ ,	a__  :	Tuple		 =			self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
										a__  :	Any		 =			model_class(a_    ).to(a_    )
										a__  :	Union[str, Any]		 =			model(**a_	, output_attentions=a_    )
										self.assertTrue(outputs.attentions is not None    )
				def       UpperCAmelCase      (		self    :			List[str]    )	->	Tuple:



							'''simple docstring'''



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

							a__  :	int		 =			model_class(a_    )
							model.to(a_    )
							model.train()

							a__  :	int		 =			model(a_	, mask_labels=a_	, class_labels=a_    ).loss
							loss.backward()






				def       UpperCAmelCase      (		self    :			Union[str, Any]    )	->	List[Any]:



							'''simple docstring'''



							a__  :	Optional[int]		 =			self.all_model_classes[1]
							a__ ,	a__ ,	a__ ,	a__ ,	a__  :	Tuple		 =			self.model_tester.prepare_config_and_inputs()
							a__  :	int		 =			True
							a__  :	Tuple		 =			True

							a__  :	List[str]		 =			model_class(a_    )
							model.to(a_    )
							model.train()

							a__  :	Optional[int]		 =			model(a_	, mask_labels=a_	, class_labels=a_    )

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

							a__  :	List[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
							a__  :	Optional[int]		 =			outputs.transformer_decoder_hidden_states[0]
							transformer_decoder_hidden_states.retain_grad()

							a__  :	Optional[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    )


__UpperCAmelCase						=			1e-4






def      lowercase__ (					)    ->			Dict:



			'''simple docstring'''
			a__  :	Optional[Any]		 =			Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"			)
			return image





@require_vision
@slow
class 					__UpperCAmelCase							(						unittest.TestCase  ):
				@cached_property
				def       UpperCAmelCase      (		self    :			Union[str, Any]    )	->	int:



							'''simple docstring'''



							return (
							    MaskFormerImageProcessor.from_pretrained("facebook/maskformer-swin-small-coco"    )
							    if is_vision_available()
							    else None
							)
				def       UpperCAmelCase      (		self    :			Optional[int]    )	->	Union[str, Any]:



							'''simple docstring'''



							a__  :	Optional[int]		 =			MaskFormerModel.from_pretrained("facebook/maskformer-swin-small-coco"    ).to(a_    )
							a__  :	List[str]		 =			self.default_image_processor
							a__  :	Union[str, Any]		 =			prepare_img()
							a__  :	Optional[int]		 =			image_processor(a_	, return_tensors="pt"    ).to(a_    )
							a__  :	Union[str, Any]		 =			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, 8_00, 10_88)    )

							with torch.no_grad():
										a__  :	List[str]		 =			model(**a_    )

							a__  :	Optional[Any]		 =			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_    )    )

							a__  :	List[str]		 =			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_    )    )

							a__  :	Tuple		 =			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       UpperCAmelCase      (		self    :			str    )	->	Any:



							'''simple docstring'''



							a__  :	Any		 =			(
							    MaskFormerForInstanceSegmentation.from_pretrained("facebook/maskformer-swin-small-coco"    )
							    .to(a_    )
							    .eval()
							)
							a__  :	List[Any]		 =			self.default_image_processor
							a__  :	List[str]		 =			prepare_img()
							a__  :	List[str]		 =			image_processor(a_	, return_tensors="pt"    ).to(a_    )
							a__  :	Tuple		 =			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, 8_00, 10_88)    )

							with torch.no_grad():
										a__  :	List[str]		 =			model(**a_    )
							# masks_queries_logits
							a__  :	List[Any]		 =			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)	, )
							a__  :	Optional[int]		 =			[
							    [-1.373_7124, -1.772_4937, -1.936_4233],
							    [-1.597_7281, -1.986_7939, -2.152_3695],
							    [-1.579_5398, -1.926_9832, -2.09_3942],
							]
							a__  :	List[str]		 =			torch.tensor(a_    ).to(a_    )
							self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]	, a_	, atol=a_    )    )
							# class_queries_logits
							a__  :	Any		 =			outputs.class_queries_logits
							self.assertEqual(
							    class_queries_logits.shape	, (1, model.config.decoder_config.num_queries, model.config.num_labels + 1)    )
							a__  :	int		 =			torch.tensor(
							    [
							        [1.6_512E00, -5.2_572E00, -3.3_519E00],
							        [3.6_169E-02, -5.9_025E00, -2.9_313E00],
							        [1.0_766E-04, -7.7_630E00, -5.1_263E00],
							    ]    ).to(a_    )
							self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]	, a_	, atol=a_    )    )
				def       UpperCAmelCase      (		self    :			Any    )	->	Dict:



							'''simple docstring'''



							a__  :	Dict		 =			(
							    MaskFormerForInstanceSegmentation.from_pretrained("facebook/maskformer-resnet101-coco-stuff"    )
							    .to(a_    )
							    .eval()
							)
							a__  :	Any		 =			self.default_image_processor
							a__  :	Any		 =			prepare_img()
							a__  :	Any		 =			image_processor(a_	, return_tensors="pt"    ).to(a_    )
							a__  :	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, 8_00, 10_88)    )

							with torch.no_grad():
										a__  :	List[str]		 =			model(**a_    )
							# masks_queries_logits
							a__  :	Dict		 =			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)	, )
							a__  :	List[str]		 =			[[-0.9046, -2.6366, -4.6062], [-3.4179, -5.7890, -8.8057], [-4.9179, -7.6560, -10.7711]]
							a__  :	Dict		 =			torch.tensor(a_    ).to(a_    )
							self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]	, a_	, atol=a_    )    )
							# class_queries_logits
							a__  :	Optional[int]		 =			outputs.class_queries_logits
							self.assertEqual(
							    class_queries_logits.shape	, (1, model.config.decoder_config.num_queries, model.config.num_labels + 1)    )
							a__  :	Dict		 =			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       UpperCAmelCase      (		self    :			Optional[Any]    )	->	List[str]:



							'''simple docstring'''



							a__  :	str		 =			(
							    MaskFormerForInstanceSegmentation.from_pretrained("facebook/maskformer-swin-small-coco"    )
							    .to(a_    )
							    .eval()
							)
							a__  :	str		 =			self.default_image_processor

							a__  :	str		 =			image_processor(
							    [np.zeros((3, 8_00, 13_33)    ), np.zeros((3, 8_00, 13_33)    )]	, segmentation_maps=[np.zeros((3_84, 3_84)    ).astype(np.floataa    ), np.zeros((3_84, 3_84)    ).astype(np.floataa    )]	, return_tensors="pt"	, )

							a__  :	List[str]		 =			inputs["pixel_values"].to(a_    )
							a__  :	Dict		 =			[el.to(a_    ) for el in inputs["mask_labels"]]
							a__  :	Optional[Any]		 =			[el.to(a_    ) for el in inputs["class_labels"]]

							with torch.no_grad():
										a__  :	int		 =			model(**a_    )

							self.assertTrue(outputs.loss is not None    )






"""simple docstring"""






import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class 					__UpperCAmelCase							:
				def       UpperCAmelCase      (		self    :			List[str]    )	->	List[str]:



							'''simple docstring'''



							torch.manual_seed(0    )
							a__  :	List[str]		 =			TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	int		 =			AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	Any		 =			UNetaDConditionModel(
							    sample_size=32	, layers_per_block=1	, block_out_channels=[32, 64]	, down_block_types=[
							        "ResnetDownsampleBlock2D",
							        "SimpleCrossAttnDownBlock2D",
							    ]	, mid_block_type="UNetMidBlock2DSimpleCrossAttn"	, up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]	, in_channels=3	, out_channels=6	, cross_attention_dim=32	, encoder_hid_dim=32	, attention_head_dim=8	, addition_embed_type="text"	, addition_embed_type_num_heads=2	, cross_attention_norm="group_norm"	, resnet_time_scale_shift="scale_shift"	, act_fn="gelu"	, )
							unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							torch.manual_seed(0    )
							a__  :	str		 =			DDPMScheduler(
							    num_train_timesteps=10_00	, beta_schedule="squaredcos_cap_v2"	, beta_start=0.0001	, beta_end=0.02	, thresholding=a_	, dynamic_thresholding_ratio=0.95	, sample_max_value=1.0	, prediction_type="epsilon"	, variance_type="learned_range"	, )

							torch.manual_seed(0    )
							a__  :	Optional[int]		 =			IFWatermarker()

							return {
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "unet": unet,
							    "scheduler": scheduler,
							    "watermarker": watermarker,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
				def       UpperCAmelCase      (		self    :			Optional[Any]    )	->	Union[str, Any]:



							'''simple docstring'''



							torch.manual_seed(0    )
							a__  :	List[Any]		 =			TaEncoderModel.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	List[str]		 =			AutoTokenizer.from_pretrained("hf-internal-testing/tiny-random-t5"    )

							torch.manual_seed(0    )
							a__  :	Dict		 =			UNetaDConditionModel(
							    sample_size=32	, layers_per_block=[1, 2]	, block_out_channels=[32, 64]	, down_block_types=[
							        "ResnetDownsampleBlock2D",
							        "SimpleCrossAttnDownBlock2D",
							    ]	, mid_block_type="UNetMidBlock2DSimpleCrossAttn"	, up_block_types=["SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"]	, in_channels=6	, out_channels=6	, cross_attention_dim=32	, encoder_hid_dim=32	, attention_head_dim=8	, addition_embed_type="text"	, addition_embed_type_num_heads=2	, cross_attention_norm="group_norm"	, resnet_time_scale_shift="scale_shift"	, act_fn="gelu"	, class_embed_type="timestep"	, mid_block_scale_factor=1.414	, time_embedding_act_fn="gelu"	, time_embedding_dim=32	, )
							unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							torch.manual_seed(0    )
							a__  :	Optional[int]		 =			DDPMScheduler(
							    num_train_timesteps=10_00	, beta_schedule="squaredcos_cap_v2"	, beta_start=0.0001	, beta_end=0.02	, thresholding=a_	, dynamic_thresholding_ratio=0.95	, sample_max_value=1.0	, prediction_type="epsilon"	, variance_type="learned_range"	, )

							torch.manual_seed(0    )
							a__  :	Tuple		 =			DDPMScheduler(
							    num_train_timesteps=10_00	, beta_schedule="squaredcos_cap_v2"	, beta_start=0.0001	, beta_end=0.02	, )

							torch.manual_seed(0    )
							a__  :	Optional[Any]		 =			IFWatermarker()

							return {
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "unet": unet,
							    "scheduler": scheduler,
							    "image_noising_scheduler": image_noising_scheduler,
							    "watermarker": watermarker,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
				def       UpperCAmelCase      (		self    :			List[Any]    )	->	Tuple:



							'''simple docstring'''



							a__  :	Dict		 =			self.get_dummy_components()
							a__  :	Any		 =			self.pipeline_class(**a_    )
							pipe.to(a_    )
							pipe.set_progress_bar_config(disable=a_    )

							a__  :	Any		 =			self.get_dummy_inputs(a_    )

							a__  :	Optional[int]		 =			inputs["prompt"]
							a__  :	List[Any]		 =			inputs["generator"]
							a__  :	Optional[int]		 =			inputs["num_inference_steps"]
							a__  :	Any		 =			inputs["output_type"]

							if "image" in inputs:
										a__  :	Any		 =			inputs["image"]
							else:
										a__  :	Dict		 =			None

							if "mask_image" in inputs:
										a__  :	Optional[int]		 =			inputs["mask_image"]
							else:
										a__  :	Any		 =			None

							if "original_image" in inputs:
										a__  :	List[Any]		 =			inputs["original_image"]
							else:
										a__  :	str		 =			None

							a__ ,	a__  :	Optional[int]		 =			pipe.encode_prompt(a_    )

							# inputs with prompt converted to embeddings
							a__  :	Union[str, Any]		 =			{
							    "prompt_embeds": prompt_embeds,
							    "negative_prompt_embeds": negative_prompt_embeds,
							    "generator": generator,
							    "num_inference_steps": num_inference_steps,
							    "output_type": output_type,
							}

							if image is not None:
										a__  :	Dict		 =			image

							if mask_image is not None:
										a__  :	Any		 =			mask_image

							if original_image is not None:
										a__  :	Optional[int]		 =			original_image

							# set all optional components to None
							for optional_component in pipe._optional_components:
										setattr(a_	, a_	, a_    )

							a__  :	int		 =			pipe(**a_    )[0]

							with tempfile.TemporaryDirectory() as tmpdir:
										pipe.save_pretrained(a_    )
										a__  :	List[str]		 =			self.pipeline_class.from_pretrained(a_    )
										pipe_loaded.to(a_    )
										pipe_loaded.set_progress_bar_config(disable=a_    )

							pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							for optional_component in pipe._optional_components:
										self.assertTrue(
										    getattr(a_	, a_    ) is None	, F"`{optional_component}` did not stay set to None after loading."	, )

							a__  :	Union[str, Any]		 =			self.get_dummy_inputs(a_    )

							a__  :	str		 =			inputs["generator"]
							a__  :	Dict		 =			inputs["num_inference_steps"]
							a__  :	Optional[int]		 =			inputs["output_type"]

							# inputs with prompt converted to embeddings
							a__  :	List[Any]		 =			{
							    "prompt_embeds": prompt_embeds,
							    "negative_prompt_embeds": negative_prompt_embeds,
							    "generator": generator,
							    "num_inference_steps": num_inference_steps,
							    "output_type": output_type,
							}

							if image is not None:
										a__  :	Dict		 =			image

							if mask_image is not None:
										a__  :	Any		 =			mask_image

							if original_image is not None:
										a__  :	Dict		 =			original_image

							a__  :	Optional[Any]		 =			pipe_loaded(**a_    )[0]

							a__  :	int		 =			np.abs(to_np(a_    ) - to_np(a_    )    ).max()
							self.assertLess(a_	, 1E-4    )






				def       UpperCAmelCase      (		self    :			int    )	->	Any:



							'''simple docstring'''



							a__  :	Dict		 =			self.get_dummy_components()
							a__  :	Dict		 =			self.pipeline_class(**a_    )
							pipe.to(a_    )
							pipe.set_progress_bar_config(disable=a_    )

							a__  :	List[str]		 =			self.get_dummy_inputs(a_    )
							a__  :	Dict		 =			pipe(**a_    )[0]

							with tempfile.TemporaryDirectory() as tmpdir:
										pipe.save_pretrained(a_    )
										a__  :	str		 =			self.pipeline_class.from_pretrained(a_    )
										pipe_loaded.to(a_    )
										pipe_loaded.set_progress_bar_config(disable=a_    )

							pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()    )  # For reproducibility tests

							a__  :	Optional[int]		 =			self.get_dummy_inputs(a_    )
							a__  :	Optional[int]		 =			pipe_loaded(**a_    )[0]

							a__  :	List[Any]		 =			np.abs(to_np(a_    ) - to_np(a_    )    ).max()
							self.assertLess(a_	, 1E-4    )