



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    snake_case__				:







						"""simple docstring"""




						def __init__(   self					,   __lowercase					,   __lowercase=2					,   __lowercase=True					,   __lowercase=False					,   __lowercase=1_0					,   __lowercase=3					,   __lowercase=3_2 * 4					,   __lowercase=3_2 * 6					,   __lowercase=4					,   __lowercase=3_2					,   )					->      int:
									"""simple docstring"""




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

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      List[Any]:
									"""simple docstring"""




									a__   :						int						       =  floats_tensor([self.batch_size, self.num_channels, self.min_size, self.max_size]  ).to(
									    __SCREAMING_SNAKE_CASE  )

									a__   :						Optional[Any]						       =  torch.ones([self.batch_size, self.min_size, self.max_size]					,   device=__SCREAMING_SNAKE_CASE  )

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

									a__   :						Tuple						       =  self.get_config()
									return config, pixel_values, pixel_mask, mask_labels, class_labels

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      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_2_8					,   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 							SCREAMING_SNAKE_CASE__(   self  )					->      List[Any]:
									"""simple docstring"""




									a__       ,						a__       ,						a__       ,						a__       ,						a__   :						List[Any]						       =  self.prepare_config_and_inputs()
									a__   :						Tuple						       =  {"""pixel_values""": pixel_values, """pixel_mask""": pixel_mask}
									return config, inputs_dict

						def 							SCREAMING_SNAKE_CASE__(   self					,   __lowercase					,   __lowercase  )					->      List[Any]:
									"""simple docstring"""




									a__   :						List[str]						       =  output.encoder_hidden_states
									a__   :						Optional[Any]						       =  output.pixel_decoder_hidden_states
									a__   :						List[Any]						       =  output.transformer_decoder_hidden_states

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

						def 							SCREAMING_SNAKE_CASE__(   self					,   __lowercase					,   __lowercase					,   __lowercase					,   __lowercase=False  )					->      int:
									"""simple docstring"""




									with torch.no_grad():
												a__   :						Optional[Any]						       =  MaskFormerModel(config=__SCREAMING_SNAKE_CASE  )
												model.to(__SCREAMING_SNAKE_CASE  )
												model.eval()

												a__   :						Union[str, Any]						       =  model(pixel_values=__SCREAMING_SNAKE_CASE					,   pixel_mask=__SCREAMING_SNAKE_CASE  )
												a__   :						Optional[int]						       =  model(__SCREAMING_SNAKE_CASE					,   output_hidden_states=__SCREAMING_SNAKE_CASE  )
									# 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(__SCREAMING_SNAKE_CASE					,   __SCREAMING_SNAKE_CASE  )

						def 							SCREAMING_SNAKE_CASE__(   self					,   __lowercase					,   __lowercase					,   __lowercase					,   __lowercase					,   __lowercase  )					->      Optional[int]:
									"""simple docstring"""




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

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

												comm_check_on_output(__SCREAMING_SNAKE_CASE  )

												a__   :						List[Any]						       =  model(
												    pixel_values=__SCREAMING_SNAKE_CASE					,   pixel_mask=__SCREAMING_SNAKE_CASE					,   mask_labels=__SCREAMING_SNAKE_CASE					,   class_labels=__SCREAMING_SNAKE_CASE  )

									comm_check_on_output(__SCREAMING_SNAKE_CASE  )

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




@require_torch
class    snake_case__				(__UpperCAmelCase					,	__UpperCAmelCase					,	unittest.TestCase		):







						"""simple docstring"""




						__lowerCAmelCase     :Union[str, Any]						=     (MaskFormerModel, MaskFormerForInstanceSegmentation) if is_torch_available() else ()
						__lowerCAmelCase     :Any						=     (
						    {'''feature-extraction''': MaskFormerModel, '''image-segmentation''': MaskFormerForInstanceSegmentation}
						    if is_torch_available()
						    else {}
						)

						__lowerCAmelCase     :Optional[Any]						=     False
						__lowerCAmelCase     :Any						=     False
						__lowerCAmelCase     :Union[str, Any]						=     False
						__lowerCAmelCase     :Dict						=     False

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Tuple:
									"""simple docstring"""




									a__   :						Dict						       =  MaskFormerModelTester(self  )
									a__   :						Optional[int]						       =  ConfigTester(self					,   config_class=__SCREAMING_SNAKE_CASE					,   has_text_modality=__SCREAMING_SNAKE_CASE  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Any:
									"""simple docstring"""




									self.config_tester.run_common_tests()

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Optional[Any]:
									"""simple docstring"""




									a__       ,						a__   :						Union[str, Any]						       =  self.model_tester.prepare_config_and_inputs_for_common()
									self.model_tester.create_and_check_maskformer_model(__SCREAMING_SNAKE_CASE					,   **__SCREAMING_SNAKE_CASE					,   output_hidden_states=__SCREAMING_SNAKE_CASE  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Union[str, Any]:
									"""simple docstring"""




									a__   :						Tuple						       =  self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_maskformer_instance_segmentation_head_model(*__SCREAMING_SNAKE_CASE  )

						@unittest.skip(reason="""MaskFormer does not use inputs_embeds"""  )
						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Any:
									"""simple docstring"""




									pass

						@unittest.skip(reason="""MaskFormer does not have a get_input_embeddings method"""  )
						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Any:
									"""simple docstring"""




									pass

						@unittest.skip(reason="""MaskFormer is not a generative model"""  )
						def 							SCREAMING_SNAKE_CASE__(   self  )					->      int:
									"""simple docstring"""




									pass

						@unittest.skip(reason="""MaskFormer does not use token embeddings"""  )
						def 							SCREAMING_SNAKE_CASE__(   self  )					->      List[str]:
									"""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 							SCREAMING_SNAKE_CASE__(   self  )					->      Optional[int]:
									"""simple docstring"""




									pass

						@unittest.skip("""Will be fixed soon by reducing the size of the model used for common tests."""  )
						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Optional[Any]:
									"""simple docstring"""




									pass

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      str:
									"""simple docstring"""




									a__       ,						a__   :						List[Any]						       =  self.model_tester.prepare_config_and_inputs_for_common()

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

												a__   :						Tuple						       =  ["""pixel_values"""]
												self.assertListEqual(arg_names[:1]					,   __SCREAMING_SNAKE_CASE  )

						@slow
						def 							SCREAMING_SNAKE_CASE__(   self  )					->      str:
									"""simple docstring"""




									for model_name in ["facebook/maskformer-swin-small-coco"]:
												a__   :						Tuple						       =  MaskFormerModel.from_pretrained(__SCREAMING_SNAKE_CASE  )
												self.assertIsNotNone(__SCREAMING_SNAKE_CASE  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      int:
									"""simple docstring"""




									a__   :						Optional[Any]						       =  (self.model_tester.min_size,) * 2
									a__   :						int						       =  {
									    """pixel_values""": torch.randn((2, 3, *size)					,   device=__SCREAMING_SNAKE_CASE  ),
									    """mask_labels""": torch.randn((2, 1_0, *size)					,   device=__SCREAMING_SNAKE_CASE  ),
									    """class_labels""": torch.zeros(2					,   1_0					,   device=__SCREAMING_SNAKE_CASE  ).long(),
									}

									a__   :						Dict						       =  MaskFormerForInstanceSegmentation(MaskFormerConfig()  ).to(__SCREAMING_SNAKE_CASE  )
									a__   :						int						       =  model(**__SCREAMING_SNAKE_CASE  )
									self.assertTrue(outputs.loss is not None  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Tuple:
									"""simple docstring"""




									a__       ,						a__   :						Optional[int]						       =  self.model_tester.prepare_config_and_inputs_for_common()
									self.model_tester.create_and_check_maskformer_model(__SCREAMING_SNAKE_CASE					,   **__SCREAMING_SNAKE_CASE					,   output_hidden_states=__SCREAMING_SNAKE_CASE  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      List[str]:
									"""simple docstring"""




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

									for model_class in self.all_model_classes:
												a__   :						Optional[int]						       =  model_class(__SCREAMING_SNAKE_CASE  ).to(__SCREAMING_SNAKE_CASE  )
												a__   :						int						       =  model(**__SCREAMING_SNAKE_CASE					,   output_attentions=__SCREAMING_SNAKE_CASE  )
												self.assertTrue(outputs.attentions is not None  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Any:
									"""simple docstring"""




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

									a__   :						str						       =  model_class(__SCREAMING_SNAKE_CASE  )
									model.to(__SCREAMING_SNAKE_CASE  )
									model.train()

									a__   :						Union[str, Any]						       =  model(__SCREAMING_SNAKE_CASE					,   mask_labels=__SCREAMING_SNAKE_CASE					,   class_labels=__SCREAMING_SNAKE_CASE  ).loss
									loss.backward()

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Dict:
									"""simple docstring"""




									a__   :						Optional[int]						       =  self.all_model_classes[1]
									a__       ,						a__       ,						a__       ,						a__       ,						a__   :						str						       =  self.model_tester.prepare_config_and_inputs()
									a__   :						Union[str, Any]						       =  True
									a__   :						Dict						       =  True

									a__   :						Any						       =  model_class(__SCREAMING_SNAKE_CASE  )
									model.to(__SCREAMING_SNAKE_CASE  )
									model.train()

									a__   :						List[Any]						       =  model(__SCREAMING_SNAKE_CASE					,   mask_labels=__SCREAMING_SNAKE_CASE					,   class_labels=__SCREAMING_SNAKE_CASE  )

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

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

									outputs.loss.backward(retain_graph=__SCREAMING_SNAKE_CASE  )

									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		:      int					=1E-4

def 	lowerCAmelCase_     (  )							->       List[Any]:







			"""simple docstring"""
			a__   :						Tuple						       =  Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png""")
			return image





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







						"""simple docstring"""




						@cached_property
						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Any:
									"""simple docstring"""




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

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      str:
									"""simple docstring"""




									a__   :						int						       =  MaskFormerModel.from_pretrained("""facebook/maskformer-swin-small-coco"""  ).to(__SCREAMING_SNAKE_CASE  )
									a__   :						int						       =  self.default_image_processor
									a__   :						Any						       =  prepare_img()
									a__   :						List[Any]						       =  image_processor(__SCREAMING_SNAKE_CASE					,   return_tensors="""pt"""  ).to(__SCREAMING_SNAKE_CASE  )
									a__   :						str						       =  inputs["""pixel_values"""].shape
									# check size is divisible by 32
									self.assertTrue((inputs_shape[-1] % 3_2) == 0 and (inputs_shape[-2] % 3_2) == 0  )
									# check size
									self.assertEqual(__SCREAMING_SNAKE_CASE					,   (1, 3, 8_0_0, 1_0_8_8)  )

									with torch.no_grad():
												a__   :						Optional[Any]						       =  model(**__SCREAMING_SNAKE_CASE  )

									a__   :						Tuple						       =  torch.tensor(
									    [[-0.0_4_8_2, 0.9_2_2_8, 0.4_9_5_1], [-0.2_5_4_7, 0.8_0_1_7, 0.8_5_2_7], [-0.0_0_6_9, 0.3_3_8_5, -0.0_0_8_9]]  ).to(__SCREAMING_SNAKE_CASE  )
									self.assertTrue(
									    torch.allclose(
									        outputs.encoder_last_hidden_state[0, 0, :3, :3]					,   __SCREAMING_SNAKE_CASE					,   atol=__SCREAMING_SNAKE_CASE  )  )

									a__   :						Any						       =  torch.tensor(
									    [[-0.8_4_2_2, -0.8_4_3_4, -0.9_7_1_8], [-1.0_1_4_4, -0.5_5_6_5, -0.4_1_9_5], [-1.0_0_3_8, -0.4_4_8_4, -0.1_9_6_1]]  ).to(__SCREAMING_SNAKE_CASE  )
									self.assertTrue(
									    torch.allclose(
									        outputs.pixel_decoder_last_hidden_state[0, 0, :3, :3]					,   __SCREAMING_SNAKE_CASE					,   atol=__SCREAMING_SNAKE_CASE  )  )

									a__   :						Tuple						       =  torch.tensor(
									    [[0.2_8_5_2, -0.0_1_5_9, 0.9_7_3_5], [0.6_2_5_4, 0.1_8_5_8, 0.8_5_2_9], [-0.0_6_8_0, -0.4_1_1_6, 1.8_4_1_3]]  ).to(__SCREAMING_SNAKE_CASE  )
									self.assertTrue(
									    torch.allclose(
									        outputs.transformer_decoder_last_hidden_state[0, :3, :3]					,   __SCREAMING_SNAKE_CASE					,   atol=__SCREAMING_SNAKE_CASE  )  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      str:
									"""simple docstring"""




									a__   :						Optional[Any]						       =  (
									    MaskFormerForInstanceSegmentation.from_pretrained("""facebook/maskformer-swin-small-coco"""  )
									    .to(__SCREAMING_SNAKE_CASE  )
									    .eval()
									)
									a__   :						Optional[Any]						       =  self.default_image_processor
									a__   :						List[str]						       =  prepare_img()
									a__   :						Union[str, Any]						       =  image_processor(__SCREAMING_SNAKE_CASE					,   return_tensors="""pt"""  ).to(__SCREAMING_SNAKE_CASE  )
									a__   :						Union[str, Any]						       =  inputs["""pixel_values"""].shape
									# check size is divisible by 32
									self.assertTrue((inputs_shape[-1] % 3_2) == 0 and (inputs_shape[-2] % 3_2) == 0  )
									# check size
									self.assertEqual(__SCREAMING_SNAKE_CASE					,   (1, 3, 8_0_0, 1_0_8_8)  )

									with torch.no_grad():
												a__   :						List[Any]						       =  model(**__SCREAMING_SNAKE_CASE  )
									# masks_queries_logits
									a__   :						List[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)					,   )
									a__   :						int						       =  [
									    [-1.3_7_3_7_1_2_4, -1.7_7_2_4_9_3_7, -1.9_3_6_4_2_3_3],
									    [-1.5_9_7_7_2_8_1, -1.9_8_6_7_9_3_9, -2.1_5_2_3_6_9_5],
									    [-1.5_7_9_5_3_9_8, -1.9_2_6_9_8_3_2, -2.0_9_3_9_4_2],
									]
									a__   :						List[str]						       =  torch.tensor(__SCREAMING_SNAKE_CASE  ).to(__SCREAMING_SNAKE_CASE  )
									self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]					,   __SCREAMING_SNAKE_CASE					,   atol=__SCREAMING_SNAKE_CASE  )  )
									# 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__   :						Union[str, Any]						       =  torch.tensor(
									    [
									        [1.6512E00, -5.2572E00, -3.3519E00],
									        [3.6169E-02, -5.9025E00, -2.9313E00],
									        [1.0766E-04, -7.7630E00, -5.1263E00],
									    ]  ).to(__SCREAMING_SNAKE_CASE  )
									self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]					,   __SCREAMING_SNAKE_CASE					,   atol=__SCREAMING_SNAKE_CASE  )  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Union[str, Any]:
									"""simple docstring"""




									a__   :						Optional[Any]						       =  (
									    MaskFormerForInstanceSegmentation.from_pretrained("""facebook/maskformer-resnet101-coco-stuff"""  )
									    .to(__SCREAMING_SNAKE_CASE  )
									    .eval()
									)
									a__   :						List[str]						       =  self.default_image_processor
									a__   :						str						       =  prepare_img()
									a__   :						str						       =  image_processor(__SCREAMING_SNAKE_CASE					,   return_tensors="""pt"""  ).to(__SCREAMING_SNAKE_CASE  )
									a__   :						int						       =  inputs["""pixel_values"""].shape
									# check size is divisible by 32
									self.assertTrue((inputs_shape[-1] % 3_2) == 0 and (inputs_shape[-2] % 3_2) == 0  )
									# check size
									self.assertEqual(__SCREAMING_SNAKE_CASE					,   (1, 3, 8_0_0, 1_0_8_8)  )

									with torch.no_grad():
												a__   :						Tuple						       =  model(**__SCREAMING_SNAKE_CASE  )
									# 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__   :						int						       =  [[-0.9_0_4_6, -2.6_3_6_6, -4.6_0_6_2], [-3.4_1_7_9, -5.7_8_9_0, -8.8_0_5_7], [-4.9_1_7_9, -7.6_5_6_0, -1_0.7_7_1_1]]
									a__   :						str						       =  torch.tensor(__SCREAMING_SNAKE_CASE  ).to(__SCREAMING_SNAKE_CASE  )
									self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]					,   __SCREAMING_SNAKE_CASE					,   atol=__SCREAMING_SNAKE_CASE  )  )
									# class_queries_logits
									a__   :						int						       =  outputs.class_queries_logits
									self.assertEqual(
									    class_queries_logits.shape					,   (1, model.config.decoder_config.num_queries, model.config.num_labels + 1)  )
									a__   :						List[str]						       =  torch.tensor(
									    [[4.7_1_8_8, -3.2_5_8_5, -2.8_8_5_7], [6.6_8_7_1, -2.9_1_8_1, -1.2_4_8_7], [7.2_4_4_9, -2.2_7_6_4, -2.1_8_7_4]]  ).to(__SCREAMING_SNAKE_CASE  )
									self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]					,   __SCREAMING_SNAKE_CASE					,   atol=__SCREAMING_SNAKE_CASE  )  )

						def 							SCREAMING_SNAKE_CASE__(   self  )					->      Optional[int]:
									"""simple docstring"""




									a__   :						Optional[int]						       =  (
									    MaskFormerForInstanceSegmentation.from_pretrained("""facebook/maskformer-swin-small-coco"""  )
									    .to(__SCREAMING_SNAKE_CASE  )
									    .eval()
									)
									a__   :						int						       =  self.default_image_processor

									a__   :						int						       =  image_processor(
									    [np.zeros((3, 8_0_0, 1_3_3_3)  ), np.zeros((3, 8_0_0, 1_3_3_3)  )]					,   segmentation_maps=[np.zeros((3_8_4, 3_8_4)  ).astype(np.floataa  ), np.zeros((3_8_4, 3_8_4)  ).astype(np.floataa  )]					,   return_tensors="""pt"""					,   )

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

									with torch.no_grad():
												a__   :						Tuple						       =  model(**__SCREAMING_SNAKE_CASE  )

									self.assertTrue(outputs.loss is not None  )









import logging
import random

import ray

from transformers import RagConfig, RagRetriever, RagTokenizer
from transformers.models.rag.retrieval_rag import CustomHFIndex


__snake_case						:List[Any]      		=     logging.getLogger(__name__)

class     _A       :



       def __init__(		self							:     List[str]):


           '''simple docstring'''

           __a						       =				False



       def 		_lowerCamelCase			(		self							:     Any     ,		__SCREAMING_SNAKE_CASE							:     Optional[Any]     ,		__SCREAMING_SNAKE_CASE							:     Optional[Any]     ,		__SCREAMING_SNAKE_CASE							:     List[str]     ,		__SCREAMING_SNAKE_CASE							:     Union[str, Any]):


           '''simple docstring'''

           if not self.initialized:
               __a						       =				RagRetriever(
                   __SCREAMING_SNAKE_CASE     ,		question_encoder_tokenizer=__SCREAMING_SNAKE_CASE     ,		generator_tokenizer=__SCREAMING_SNAKE_CASE     ,		index=__SCREAMING_SNAKE_CASE     ,		init_retrieval=__SCREAMING_SNAKE_CASE     ,		)
               __a						       =				True



       def 		_lowerCamelCase			(		self							:     List[str]):


           '''simple docstring'''

           self.retriever.index.init_index()





       def 		_lowerCamelCase			(		self							:     Tuple     ,		__SCREAMING_SNAKE_CASE							:     Union[str, Any]     ,		__SCREAMING_SNAKE_CASE							:     Union[str, Any]):


           '''simple docstring'''

           __a       ,   __a						       =				self.retriever._main_retrieve(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE)
           return doc_ids, retrieved_doc_embeds


class     _A       (	__UpperCAmelCase    ):



       def __init__(		self							:     Optional[Any]     ,		__SCREAMING_SNAKE_CASE							:     List[Any]     ,		__SCREAMING_SNAKE_CASE							:     List[str]     ,		__SCREAMING_SNAKE_CASE							:     List[str]     ,		__SCREAMING_SNAKE_CASE							:     int     ,		__SCREAMING_SNAKE_CASE							:     Union[str, Any]=None):


           '''simple docstring'''

           if index is not None and index.is_initialized() and len(__SCREAMING_SNAKE_CASE) > 0:
               raise ValueError(
                   '''When using Ray for distributed fine-tuning, '''
                   '''you\'ll need to provide the paths instead, '''
                   '''as the dataset and the index are loaded '''
                   '''separately. More info in examples/rag/use_own_knowledge_dataset.py ''')
           super().__init__(
               __SCREAMING_SNAKE_CASE     ,		question_encoder_tokenizer=__SCREAMING_SNAKE_CASE     ,		generator_tokenizer=__SCREAMING_SNAKE_CASE     ,		index=__SCREAMING_SNAKE_CASE     ,		init_retrieval=__SCREAMING_SNAKE_CASE     ,		)
           __a						       =				retrieval_workers
           if len(self.retrieval_workers) > 0:
               ray.get(
                   [
                       worker.create_rag_retriever.remote(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE)
                       for worker in self.retrieval_workers
                   ])



       def 		_lowerCamelCase			(		self							:     List[Any]):


           '''simple docstring'''

           logger.info('''initializing retrieval''')

           if len(self.retrieval_workers) > 0:
               ray.get([worker.init_retrieval.remote() for worker in self.retrieval_workers])
           else:
               # Non-distributed training. Load index into this same process.
               self.index.init_index()



       def 		_lowerCamelCase			(		self							:     Dict     ,		__SCREAMING_SNAKE_CASE							:     Optional[int]     ,		__SCREAMING_SNAKE_CASE							:     List[Any]):


           '''simple docstring'''

           if len(self.retrieval_workers) > 0:
               # Select a random retrieval actor.
               __a						       =				self.retrieval_workers[random.randint(0     ,		len(self.retrieval_workers) - 1)]
               __a       ,   __a						       =				ray.get(random_worker.retrieve.remote(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE))
           else:
               __a       ,   __a						       =				self._main_retrieve(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE)
           return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(__SCREAMING_SNAKE_CASE)



       @classmethod
       def 		_lowerCamelCase			(		cls							:     Any     ,		__SCREAMING_SNAKE_CASE							:     Tuple     ,		__SCREAMING_SNAKE_CASE							:     Tuple=None     ,		**__SCREAMING_SNAKE_CASE							:     Optional[int]):


           '''simple docstring'''

           return super(__SCREAMING_SNAKE_CASE     ,		cls).get_tokenizers(__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		**__SCREAMING_SNAKE_CASE)





       @classmethod
       def 		_lowerCamelCase			(		cls							:     Tuple     ,		__SCREAMING_SNAKE_CASE							:     Optional[int]     ,		__SCREAMING_SNAKE_CASE							:     str     ,		__SCREAMING_SNAKE_CASE							:     str=None     ,		**__SCREAMING_SNAKE_CASE							:     List[Any]):


           '''simple docstring'''

           __a						       =				kwargs.pop('''config'''     ,		__SCREAMING_SNAKE_CASE) or RagConfig.from_pretrained(__SCREAMING_SNAKE_CASE     ,		**__SCREAMING_SNAKE_CASE)
           __a						       =				RagTokenizer.from_pretrained(__SCREAMING_SNAKE_CASE     ,		config=__SCREAMING_SNAKE_CASE)
           __a						       =				rag_tokenizer.question_encoder
           __a						       =				rag_tokenizer.generator
           if indexed_dataset is not None:
               __a						       =				'''custom'''
               __a						       =				CustomHFIndex(config.retrieval_vector_size     ,		__SCREAMING_SNAKE_CASE)
           else:
               __a						       =				cls._build_index(__SCREAMING_SNAKE_CASE)
           return cls(
               __SCREAMING_SNAKE_CASE     ,		question_encoder_tokenizer=__SCREAMING_SNAKE_CASE     ,		generator_tokenizer=__SCREAMING_SNAKE_CASE     ,		retrieval_workers=__SCREAMING_SNAKE_CASE     ,		index=__SCREAMING_SNAKE_CASE     ,		)





