
"""simple docstring"""


import inspect
import unittest

import numpy as np

from tests.test_modeling_common import floats_tensor
from transformers import MaskaFormerConfig, 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 MaskaFormerForUniversalSegmentation, MaskaFormerModel

		if is_vision_available():
				from transformers import MaskaFormerImageProcessor

if is_vision_available():
		from PIL import Image







class 					lowerCAmelCase_					:



						'''simple docstring'''


						def __init__(			self  :   Tuple	,A_  :   List[Any]	,A_  :   Tuple=2	,A_  :   Optional[int]=True	,A_  :   List[str]=False	,A_  :   Tuple=10	,A_  :   int=3	,A_  :   Union[str, Any]=32 * 8	,A_  :   Dict=32 * 8	,A_  :   Dict=4	,A_  :   Tuple=64	,)  -> List[Any]:
							A     = parent
							A     = batch_size
							A     = is_training
							A     = use_auxiliary_loss
							A     = num_queries
							A     = num_channels
							A     = min_size
							A     = max_size
							A     = num_labels
							A     = hidden_dim
							A     = hidden_dim


						def     _SCREAMING_SNAKE_CASE   (			self  :   Optional[int]   )  -> int:
							A     = floats_tensor([self.batch_size, self.num_channels, self.min_size, self.max_size]   ).to(
							    lowerCAmelCase__   )

							A     = torch.ones([self.batch_size, self.min_size, self.max_size]	,device=lowerCAmelCase__   )

							A     = (
							    torch.rand([self.batch_size, self.num_labels, self.min_size, self.max_size]	,device=lowerCAmelCase__   ) > 0.5
							).float()
							A     = (torch.rand((self.batch_size, self.num_labels)	,device=lowerCAmelCase__   ) > 0.5).long()

							A     = self.get_config()
							return config, pixel_values, pixel_mask, mask_labels, class_labels


						def     _SCREAMING_SNAKE_CASE   (			self  :   str   )  -> str:
							A     = MaskaFormerConfig(
							    hidden_size=self.hidden_dim	,)
							A     = self.num_queries
							A     = self.num_labels

							A     = [1, 1, 1, 1]
							A     = self.num_channels

							A     = 64
							A     = 128
							A     = self.hidden_dim
							A     = self.hidden_dim
							A     = self.hidden_dim
							return config


						def     _SCREAMING_SNAKE_CASE   (			self  :   Union[str, Any]   )  -> Dict:
							A ,   A ,   A ,   A ,   A     = self.prepare_config_and_inputs()
							A     = {'pixel_values': pixel_values, 'pixel_mask': pixel_mask}
							return config, inputs_dict


						def     _SCREAMING_SNAKE_CASE   (			self  :   List[Any]	,A_  :   Any	,A_  :   int   )  -> Optional[int]:
							A     = output.encoder_hidden_states
							A     = output.pixel_decoder_hidden_states
							A     = output.transformer_decoder_hidden_states

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


						def     _SCREAMING_SNAKE_CASE   (			self  :   Optional[Any]	,A_  :   Dict	,A_  :   List[str]	,A_  :   Dict	,A_  :   str=False   )  -> Tuple:




							with torch.no_grad():
								A     = MaskaFormerModel(config=lowerCAmelCase__   )
								model.to(lowerCAmelCase__   )
								model.eval()

								A     = model(pixel_values=lowerCAmelCase__	,pixel_mask=lowerCAmelCase__   )
								A     = model(lowerCAmelCase__	,output_hidden_states=lowerCAmelCase__   )

							self.parent.assertEqual(
							    output.transformer_decoder_last_hidden_state.shape	,(self.batch_size, self.num_queries, self.hidden_dim)	,)
							# 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(lowerCAmelCase__	,lowerCAmelCase__   )




						def     _SCREAMING_SNAKE_CASE   (			self  :   Union[str, Any]	,A_  :   str	,A_  :   int	,A_  :   List[str]	,A_  :   Optional[int]	,A_  :   Tuple   )  -> Tuple:
							A     = MaskaFormerForUniversalSegmentation(config=lowerCAmelCase__   )
							model.to(lowerCAmelCase__   )
							model.eval()

							def comm_check_on_output(A_  :   int   ):
								# 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     = model(pixel_values=lowerCAmelCase__	,pixel_mask=lowerCAmelCase__   )
								A     = model(lowerCAmelCase__   )

								comm_check_on_output(lowerCAmelCase__   )

								A     = model(
								    pixel_values=lowerCAmelCase__	,pixel_mask=lowerCAmelCase__	,mask_labels=lowerCAmelCase__	,class_labels=lowerCAmelCase__   )

							comm_check_on_output(lowerCAmelCase__   )

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







@require_torch
class 					lowerCAmelCase_					(   _lowercase		,       _lowercase		,       unittest.TestCase     ):



						'''simple docstring'''


						_lowerCamelCase:     Optional[int]   =     (MaskaFormerModel, MaskaFormerForUniversalSegmentation) if is_torch_available() else ()
						_lowerCamelCase:     List[Any]   =     {'feature-extraction': MaskaFormerModel} if is_torch_available() else {}

						_lowerCamelCase:     Any   =     False
						_lowerCamelCase:     List[str]   =     False
						_lowerCamelCase:     Optional[int]   =     False
						_lowerCamelCase:     int   =     False


						def     _SCREAMING_SNAKE_CASE   (			self  :   Union[str, Any]   )  -> str:
							A     = MaskaFormerModelTester(self   )
							A     = ConfigTester(self	,config_class=lowerCAmelCase__	,has_text_modality=lowerCAmelCase__   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   int   )  -> List[str]:
							self.config_tester.run_common_tests()


						def     _SCREAMING_SNAKE_CASE   (			self  :   int   )  -> List[str]:
							A ,   A     = self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.create_and_check_maskaformer_model(lowerCAmelCase__	,**lowerCAmelCase__	,output_hidden_states=lowerCAmelCase__   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   List[str]   )  -> str:
							A     = self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_maskaformer_instance_segmentation_head_model(*lowerCAmelCase__   )


						@unittest.skip(reason='Mask2Former does not use inputs_embeds'   )
						def     _SCREAMING_SNAKE_CASE   (			self  :   Union[str, Any]   )  -> List[str]:
							pass


						@unittest.skip(reason='Mask2Former does not have a get_input_embeddings method'   )
						def     _SCREAMING_SNAKE_CASE   (			self  :   Union[str, Any]   )  -> Tuple:
							pass


						@unittest.skip(reason='Mask2Former is not a generative model'   )
						def     _SCREAMING_SNAKE_CASE   (			self  :   Optional[int]   )  -> int:
							pass


						@unittest.skip(reason='Mask2Former does not use token embeddings'   )
						def     _SCREAMING_SNAKE_CASE   (			self  :   int   )  -> List[str]:
							pass


						@require_torch_multi_gpu
						@unittest.skip(
						    reason='Mask2Former has some layers using `add_module` which doesn\'t work well with `nn.DataParallel`'   )
						def     _SCREAMING_SNAKE_CASE   (			self  :   Dict   )  -> List[Any]:
							pass


						@unittest.skip('Will be fixed soon by reducing the size of the model used for common tests.'   )
						def     _SCREAMING_SNAKE_CASE   (			self  :   Optional[int]   )  -> List[Any]:
							pass


						def     _SCREAMING_SNAKE_CASE   (			self  :   List[Any]   )  -> Any:
							A ,   A     = self.model_tester.prepare_config_and_inputs_for_common()

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

								A     = ['pixel_values']
								self.assertListEqual(arg_names[:1]	,lowerCAmelCase__   )


						@slow
						def     _SCREAMING_SNAKE_CASE   (			self  :   List[Any]   )  -> str:




							for model_name in ["facebook/mask2former-swin-small-coco-instance"]:
								A     = MaskaFormerModel.from_pretrained(lowerCAmelCase__   )
								self.assertIsNotNone(lowerCAmelCase__   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   int   )  -> Tuple:
							A     = (self.model_tester.min_size,) * 2
							A     = {
							    'pixel_values': torch.randn((2, 3, *size)	,device=lowerCAmelCase__   ),
							    'mask_labels': torch.randn((2, 10, *size)	,device=lowerCAmelCase__   ),
							    'class_labels': torch.zeros(2	,10	,device=lowerCAmelCase__   ).long(),
							}
							A     = self.model_tester.get_config()

							A     = MaskaFormerForUniversalSegmentation(lowerCAmelCase__   ).to(lowerCAmelCase__   )
							A     = model(**lowerCAmelCase__   )
							self.assertTrue(outputs.loss is not None   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   Tuple   )  -> Optional[Any]:
							A ,   A     = self.model_tester.prepare_config_and_inputs_for_common()
							self.model_tester.create_and_check_maskaformer_model(lowerCAmelCase__	,**lowerCAmelCase__	,output_hidden_states=lowerCAmelCase__   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   str   )  -> Optional[int]:
							A ,   A     = self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
								A     = model_class(lowerCAmelCase__   ).to(lowerCAmelCase__   )
								A     = model(**lowerCAmelCase__	,output_attentions=lowerCAmelCase__   )
								self.assertTrue(outputs.attentions is not None   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   Tuple   )  -> Dict:




							if not self.model_tester.is_training:
								return

							A     = self.all_model_classes[1]
							A ,   A ,   A ,   A ,   A     = self.model_tester.prepare_config_and_inputs()

							A     = model_class(lowerCAmelCase__   )
							model.to(lowerCAmelCase__   )
							model.train()

							A     = model(lowerCAmelCase__	,mask_labels=lowerCAmelCase__	,class_labels=lowerCAmelCase__   ).loss
							loss.backward()




						def     _SCREAMING_SNAKE_CASE   (			self  :   Optional[Any]   )  -> Optional[Any]:
							A     = self.all_model_classes[1]
							A ,   A ,   A ,   A ,   A     = self.model_tester.prepare_config_and_inputs()
							A     = True
							A     = True

							A     = model_class(lowerCAmelCase__   ).to(lowerCAmelCase__   )
							model.train()

							A     = model(lowerCAmelCase__	,mask_labels=lowerCAmelCase__	,class_labels=lowerCAmelCase__   )

							A     = outputs.encoder_hidden_states[0]
							encoder_hidden_states.retain_grad()

							A     = outputs.pixel_decoder_hidden_states[0]
							pixel_decoder_hidden_states.retain_grad()

							A     = outputs.transformer_decoder_hidden_states[0]
							transformer_decoder_hidden_states.retain_grad()

							A     = outputs.attentions[0]
							attentions.retain_grad()

							outputs.loss.backward(retain_graph=lowerCAmelCase__   )

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

_lowercase				 =					1e-4






def    _snake_case					(		):
	A     = Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'  )
	return image







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



						'''simple docstring'''


						@cached_property
						def     _SCREAMING_SNAKE_CASE   (			self  :   List[str]   )  -> Any:
							return "facebook/mask2former-swin-small-coco-instance"


						@cached_property
						def     _SCREAMING_SNAKE_CASE   (			self  :   List[str]   )  -> List[str]:
							return MaskaFormerImageProcessor.from_pretrained(self.model_checkpoints   ) if is_vision_available() else None


						def     _SCREAMING_SNAKE_CASE   (			self  :   List[str]   )  -> Tuple:
							A     = MaskaFormerModel.from_pretrained(self.model_checkpoints   ).to(lowerCAmelCase__   )
							A     = self.default_image_processor
							A     = prepare_img()
							A     = image_processor(lowerCAmelCase__	,return_tensors='pt'   ).to(lowerCAmelCase__   )
							A     = 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(lowerCAmelCase__	,(1, 3, 384, 384)   )

							with torch.no_grad():
								A     = model(**lowerCAmelCase__   )

							A     = torch.tensor(
							    [[-0.27_90, -1.07_17, -1.16_68], [-0.51_28, -0.31_28, -0.49_87], [-0.58_32, 0.19_71, -0.01_97]]   ).to(lowerCAmelCase__   )
							self.assertTrue(
							    torch.allclose(
							        outputs.encoder_last_hidden_state[0, 0, :3, :3]	,lowerCAmelCase__	,atol=lowerCAmelCase__   )   )

							A     = torch.tensor(
							    [[0.89_73, 1.18_47, 1.17_76], [1.19_34, 1.50_40, 1.51_28], [1.11_53, 1.44_86, 1.49_51]]   ).to(lowerCAmelCase__   )
							self.assertTrue(
							    torch.allclose(
							        outputs.pixel_decoder_last_hidden_state[0, 0, :3, :3]	,lowerCAmelCase__	,atol=lowerCAmelCase__   )   )

							A     = torch.tensor(
							    [[2.11_52, 1.70_00, -0.86_03], [1.58_08, 1.80_04, -0.93_53], [1.60_43, 1.74_95, -0.59_99]]   ).to(lowerCAmelCase__   )
							self.assertTrue(
							    torch.allclose(
							        outputs.transformer_decoder_last_hidden_state[0, :3, :3]	,lowerCAmelCase__	,atol=lowerCAmelCase__   )   )


						def     _SCREAMING_SNAKE_CASE   (			self  :   Tuple   )  -> List[Any]:
							A     = MaskaFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints   ).to(lowerCAmelCase__   ).eval()
							A     = self.default_image_processor
							A     = prepare_img()
							A     = image_processor(lowerCAmelCase__	,return_tensors='pt'   ).to(lowerCAmelCase__   )
							A     = 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(lowerCAmelCase__	,(1, 3, 384, 384)   )

							with torch.no_grad():
								A     = model(**lowerCAmelCase__   )
							# masks_queries_logits
							A     = outputs.masks_queries_logits
							self.assertEqual(
							    masks_queries_logits.shape	,(1, model.config.num_queries, inputs_shape[-2] // 4, inputs_shape[-1] // 4)   )
							A     = [
							    [-8.78_39, -9.00_56, -8.81_21],
							    [-7.41_04, -7.03_13, -6.54_01],
							    [-6.61_05, -6.34_27, -6.46_75],
							]
							A     = torch.tensor(lowerCAmelCase__   ).to(lowerCAmelCase__   )
							self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]	,lowerCAmelCase__	,atol=lowerCAmelCase__   )   )
							# class_queries_logits
							A     = outputs.class_queries_logits
							self.assertEqual(class_queries_logits.shape	,(1, model.config.num_queries, model.config.num_labels + 1)   )
							A     = torch.tensor(
							    [
							        [1.83_24, -8.08_35, -4.19_22],
							        [0.84_50, -9.00_50, -3.60_53],
							        [0.30_45, -7.72_93, -3.02_75],
							    ]   ).to(lowerCAmelCase__   )
							self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]	,lowerCAmelCase__	,atol=lowerCAmelCase__   )   )




						def     _SCREAMING_SNAKE_CASE   (			self  :   Tuple   )  -> int:
							A     = MaskaFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints   ).to(lowerCAmelCase__   ).eval()
							A     = self.default_image_processor

							A     = 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'	,)

							A     = inputs['pixel_values'].to(lowerCAmelCase__   )
							A     = [el.to(lowerCAmelCase__   ) for el in inputs['mask_labels']]
							A     = [el.to(lowerCAmelCase__   ) for el in inputs['class_labels']]

							with torch.no_grad():
								A     = model(**lowerCAmelCase__   )

							self.assertTrue(outputs.loss is not None   )






'''simple docstring'''

from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType
class 							__lowerCAmelCase	(       __magic_name__					):






       """simple docstring"""
       _snake_case      :	Dict									=       ['image_processor', 'tokenizer']
       _snake_case      :	int									=       'BlipImageProcessor'
       _snake_case      :	Tuple									=       ('BertTokenizer', 'BertTokenizerFast')





       def __init__(				self  :						Union[str, Any]			,					lowerCAmelCase__  :						List[Any]			,					lowerCAmelCase__  :						List[str]	)       ->	int:

            '''simple docstring'''




            _UpperCamelCase							       =			False
            super().__init__(lowerCAmelCase__			,					lowerCAmelCase__	)
            _UpperCamelCase							       =			self.image_processor





       def __call__(				self  :						int			,					lowerCAmelCase__  :						ImageInput = None			,					lowerCAmelCase__  :						Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None			,					lowerCAmelCase__  :						bool = True			,					lowerCAmelCase__  :						Union[bool, str, PaddingStrategy] = False			,					lowerCAmelCase__  :						Union[bool, str, TruncationStrategy] = None			,					lowerCAmelCase__  :						Optional[int] = None			,					lowerCAmelCase__  :						int = 0			,					lowerCAmelCase__  :						Optional[int] = None			,					lowerCAmelCase__  :						Optional[bool] = None			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = False			,					lowerCAmelCase__  :						bool = True			,					lowerCAmelCase__  :						Optional[Union[str, TensorType]] = None			,					**lowerCAmelCase__  :						Optional[Any]			,					)       ->	BatchEncoding:

            '''simple docstring'''




            if images is None and text is None:
                 raise ValueError('''You have to specify either images or text.'''	)

            # Get only text
            if images is None:
                 _UpperCamelCase							       =			self.tokenizer
                 _UpperCamelCase							       =			self.tokenizer(
                     text=lowerCAmelCase__			,					add_special_tokens=lowerCAmelCase__			,					padding=lowerCAmelCase__			,					truncation=lowerCAmelCase__			,					max_length=lowerCAmelCase__			,					stride=lowerCAmelCase__			,					pad_to_multiple_of=lowerCAmelCase__			,					return_attention_mask=lowerCAmelCase__			,					return_overflowing_tokens=lowerCAmelCase__			,					return_special_tokens_mask=lowerCAmelCase__			,					return_offsets_mapping=lowerCAmelCase__			,					return_token_type_ids=lowerCAmelCase__			,					return_length=lowerCAmelCase__			,					verbose=lowerCAmelCase__			,					return_tensors=lowerCAmelCase__			,					**lowerCAmelCase__			,					)
                 return text_encoding

            # add pixel_values
            _UpperCamelCase							       =			self.image_processor(lowerCAmelCase__			,					return_tensors=lowerCAmelCase__	)

            if text is not None:
                 _UpperCamelCase							       =			self.tokenizer(
                     text=lowerCAmelCase__			,					add_special_tokens=lowerCAmelCase__			,					padding=lowerCAmelCase__			,					truncation=lowerCAmelCase__			,					max_length=lowerCAmelCase__			,					stride=lowerCAmelCase__			,					pad_to_multiple_of=lowerCAmelCase__			,					return_attention_mask=lowerCAmelCase__			,					return_overflowing_tokens=lowerCAmelCase__			,					return_special_tokens_mask=lowerCAmelCase__			,					return_offsets_mapping=lowerCAmelCase__			,					return_token_type_ids=lowerCAmelCase__			,					return_length=lowerCAmelCase__			,					verbose=lowerCAmelCase__			,					return_tensors=lowerCAmelCase__			,					**lowerCAmelCase__			,					)
            else:
                 _UpperCamelCase							       =			None

            if text_encoding is not None:
                 encoding_image_processor.update(lowerCAmelCase__	)

            return encoding_image_processor





       def snake_case__		(				self  :						Any			,					*lowerCAmelCase__  :						Union[str, Any]			,					**lowerCAmelCase__  :						Dict	)       ->	List[str]:

            '''simple docstring'''




            return self.tokenizer.batch_decode(*lowerCAmelCase__			,					**lowerCAmelCase__	)





       def snake_case__		(				self  :						Tuple			,					*lowerCAmelCase__  :						int			,					**lowerCAmelCase__  :						Optional[int]	)       ->	Optional[int]:

            '''simple docstring'''




            return self.tokenizer.decode(*lowerCAmelCase__			,					**lowerCAmelCase__	)







       @property
       def snake_case__		(				self  :						Any	)       ->	str:

            '''simple docstring'''




            _UpperCamelCase							       =			self.tokenizer.model_input_names
            _UpperCamelCase							       =			self.image_processor.model_input_names
            return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names	)	)



