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

  def __init__(		self :      Tuple     ,							__SCREAMING_SNAKE_CASE :      List[str]     ,							__SCREAMING_SNAKE_CASE :      Tuple=2     ,							__SCREAMING_SNAKE_CASE :      Optional[int]=True     ,							__SCREAMING_SNAKE_CASE :      Union[str, Any]=False     ,							__SCREAMING_SNAKE_CASE :      Any=10     ,							__SCREAMING_SNAKE_CASE :      Dict=3     ,							__SCREAMING_SNAKE_CASE :      str=32 * 8     ,							__SCREAMING_SNAKE_CASE :      Dict=32 * 8     ,							__SCREAMING_SNAKE_CASE :      Union[str, Any]=4     ,							__SCREAMING_SNAKE_CASE :      Union[str, Any]=64     ,							):
   __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 		_UpperCAmelCase   (		self :      Union[str, Any]       ):
   __a       =  floats_tensor([self.batch_size, self.num_channels, self.min_size, self.max_size]       ).to(
       A__       )

   __a       =  torch.ones([self.batch_size, self.min_size, self.max_size]     ,							device=A__       )

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

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

  def 		_UpperCAmelCase   (		self :      List[Any]       ):
   __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       =  1_28
   __a       =  self.hidden_dim
   __a       =  self.hidden_dim
   __a       =  self.hidden_dim
   return config

  def 		_UpperCAmelCase   (		self :      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 		_UpperCAmelCase   (		self :      Optional[Any]     ,							__SCREAMING_SNAKE_CASE :      List[str]     ,							__SCREAMING_SNAKE_CASE :      Optional[int]       ):
   __a       =  output.encoder_hidden_states
   __a       =  output.pixel_decoder_hidden_states
   __a       =  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_layers       )

  def 		_UpperCAmelCase   (		self :      Dict     ,							__SCREAMING_SNAKE_CASE :      Union[str, Any]     ,							__SCREAMING_SNAKE_CASE :      List[Any]     ,							__SCREAMING_SNAKE_CASE :      Optional[int]     ,							__SCREAMING_SNAKE_CASE :      Optional[Any]=False       ):
   with torch.no_grad():
    __a       =  MaskaFormerModel(config=A__       )
    model.to(A__       )
    model.eval()

    __a       =  model(pixel_values=A__     ,							pixel_mask=A__       )
    __a       =  model(A__     ,							output_hidden_states=A__       )

   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(A__     ,							A__       )

  def 		_UpperCAmelCase   (		self :      Tuple     ,							__SCREAMING_SNAKE_CASE :      Dict     ,							__SCREAMING_SNAKE_CASE :      Optional[int]     ,							__SCREAMING_SNAKE_CASE :      Any     ,							__SCREAMING_SNAKE_CASE :      Any     ,							__SCREAMING_SNAKE_CASE :      Optional[Any]       ):
   __a       =  MaskaFormerForUniversalSegmentation(config=A__       )
   model.to(A__       )
   model.eval()

   def comm_check_on_output(__SCREAMING_SNAKE_CASE :      Optional[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=A__     ,							pixel_mask=A__       )
    __a       =  model(A__       )

    comm_check_on_output(A__       )

    __a       =  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    A_ (       __a			,       __a			,       unittest.TestCase ):
  _SCREAMING_SNAKE_CASE    				=			(MaskaFormerModel, MaskaFormerForUniversalSegmentation) if is_torch_available() else ()
  _SCREAMING_SNAKE_CASE    				=			{"""feature-extraction""": MaskaFormerModel} if is_torch_available() else {}

  _SCREAMING_SNAKE_CASE    				=			False
  _SCREAMING_SNAKE_CASE    				=			False
  _SCREAMING_SNAKE_CASE    				=			False
  _SCREAMING_SNAKE_CASE    				=			False

  def 		_UpperCAmelCase   (		self :      Any       ):
   __a       =  MaskaFormerModelTester(self       )
   __a       =  ConfigTester(self     ,							config_class=A__     ,							has_text_modality=A__       )

  def 		_UpperCAmelCase   (		self :      int       ):
   self.config_tester.run_common_tests()

  def 		_UpperCAmelCase   (		self :      List[Any]       ):
   __a       ,     __a       =  self.model_tester.prepare_config_and_inputs_for_common()
   self.model_tester.create_and_check_maskaformer_model(A__     ,							**A__     ,							output_hidden_states=A__       )

  def 		_UpperCAmelCase   (		self :      List[Any]       ):
   __a       =  self.model_tester.prepare_config_and_inputs()
   self.model_tester.create_and_check_maskaformer_instance_segmentation_head_model(*A__       )

  @unittest.skip(reason="Mask2Former does not use inputs_embeds"       )
  def 		_UpperCAmelCase   (		self :      str       ):
   pass

  @unittest.skip(reason="Mask2Former does not have a get_input_embeddings method"       )
  def 		_UpperCAmelCase   (		self :      List[Any]       ):
   pass

  @unittest.skip(reason="Mask2Former is not a generative model"       )
  def 		_UpperCAmelCase   (		self :      Any       ):
   pass

  @unittest.skip(reason="Mask2Former does not use token embeddings"       )
  def 		_UpperCAmelCase   (		self :      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 		_UpperCAmelCase   (		self :      Tuple       ):
   pass

  @unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."       )
  def 		_UpperCAmelCase   (		self :      int       ):
   pass

  def 		_UpperCAmelCase   (		self :      Any       ):
   __a       ,     __a       =  self.model_tester.prepare_config_and_inputs_for_common()

   for model_class in self.all_model_classes:
    __a       =  model_class(A__       )
    __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]     ,							A__       )

  @slow
  def 		_UpperCAmelCase   (		self :      List[str]       ):
   for model_name in ["facebook/mask2former-swin-small-coco-instance"]:
    __a       =  MaskaFormerModel.from_pretrained(A__       )
    self.assertIsNotNone(A__       )

  def 		_UpperCAmelCase   (		self :      Tuple       ):
   __a       =  (self.model_tester.min_size,) * 2
   __a       =  {
       "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       =  self.model_tester.get_config()

   __a       =  MaskaFormerForUniversalSegmentation(A__       ).to(A__       )
   __a       =  model(**A__       )
   self.assertTrue(outputs.loss is not None       )

  def 		_UpperCAmelCase   (		self :      List[str]       ):
   __a       ,     __a       =  self.model_tester.prepare_config_and_inputs_for_common()
   self.model_tester.create_and_check_maskaformer_model(A__     ,							**A__     ,							output_hidden_states=A__       )

  def 		_UpperCAmelCase   (		self :      Any       ):
   __a       ,     __a       =  self.model_tester.prepare_config_and_inputs_for_common()

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

  def 		_UpperCAmelCase   (		self :      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(A__       )
   model.to(A__       )
   model.train()

   __a       =  model(A__     ,							mask_labels=A__     ,							class_labels=A__       ).loss
   loss.backward()

  def 		_UpperCAmelCase   (		self :      Dict       ):
   __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(A__       ).to(A__       )
   model.train()

   __a       =  model(A__     ,							mask_labels=A__     ,							class_labels=A__       )

   __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=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       )
SCREAMING_SNAKE_CASE :	int							     =		1e-4



def   __A		(			):



 """simple docstring"""


 __a       =  Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"					)
 return image






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

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

  @cached_property
  def 		_UpperCAmelCase   (		self :      Union[str, Any]       ):
   return MaskaFormerImageProcessor.from_pretrained(self.model_checkpoints       ) if is_vision_available() else None

  def 		_UpperCAmelCase   (		self :      Optional[Any]       ):
   __a       =  MaskaFormerModel.from_pretrained(self.model_checkpoints       ).to(A__       )
   __a       =  self.default_image_processor
   __a       =  prepare_img()
   __a       =  image_processor(A__     ,							return_tensors="pt"       ).to(A__       )
   __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(A__     ,							(1, 3, 3_84, 3_84)       )

   with torch.no_grad():
    __a       =  model(**A__       )

   __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(A__       )
   self.assertTrue(
       torch.allclose(
           outputs.encoder_last_hidden_state[0, 0, :3, :3]     ,							A__     ,							atol=A__       )       )

   __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(A__       )
   self.assertTrue(
       torch.allclose(
           outputs.pixel_decoder_last_hidden_state[0, 0, :3, :3]     ,							A__     ,							atol=A__       )       )

   __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(A__       )
   self.assertTrue(
       torch.allclose(
           outputs.transformer_decoder_last_hidden_state[0, :3, :3]     ,							A__     ,							atol=A__       )       )

  def 		_UpperCAmelCase   (		self :      Optional[Any]       ):
   __a       =  MaskaFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints       ).to(A__       ).eval()
   __a       =  self.default_image_processor
   __a       =  prepare_img()
   __a       =  image_processor(A__     ,							return_tensors="pt"       ).to(A__       )
   __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(A__     ,							(1, 3, 3_84, 3_84)       )

   with torch.no_grad():
    __a       =  model(**A__       )
   # 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(A__       ).to(A__       )
   self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]     ,							A__     ,							atol=A__       )       )
   # 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(A__       )
   self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]     ,							A__     ,							atol=A__       )       )

  def 		_UpperCAmelCase   (		self :      Any       ):
   __a       =  MaskaFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints       ).to(A__       ).eval()
   __a       =  self.default_image_processor

   __a       =  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       =  inputs["pixel_values"].to(A__       )
   __a       =  [el.to(A__       ) for el in inputs["mask_labels"]]
   __a       =  [el.to(A__       ) for el in inputs["class_labels"]]

   with torch.no_grad():
    __a       =  model(**A__       )

   self.assertTrue(outputs.loss is not None       )





from datetime import datetime as dt
import os

from github import Github


SCREAMING_SNAKE_CASE :	Optional[Any]							     =		[
    """good first issue""",
    """good second issue""",
    """good difficult issue""",
    """feature request""",
    """new model""",
    """wip""",
]



def   __A		(			):



 """simple docstring"""


 __a       =  Github(os.environ["GITHUB_TOKEN"]					)
 __a       =  g.get_repo("huggingface/transformers"					)
 __a       =  repo.get_issues(state="open"					)

 for issue in open_issues:
  __a       =  sorted([comment for comment in issue.get_comments()]							,				key=lambda _A					: i.created_at							,				reverse=_A					)
  __a       =  comments[0] if len(_A					) > 0 else None
  if (
      last_comment is not None
      and last_comment.user.login == "github-actions[bot]"
      and (dt.utcnow() - issue.updated_at).days > 7
      and (dt.utcnow() - issue.created_at).days >= 30
      and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()					)
  ):
   # print(f"Would close issue {issue.number} since it has been 7 days of inactivity since bot mention.")
   issue.edit(state="closed"					)
  elif (
      (dt.utcnow() - issue.updated_at).days > 23
      and (dt.utcnow() - issue.created_at).days >= 30
      and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()					)
  ):
   # print(f"Would add stale comment to {issue.number}")
   issue.create_comment(
       "This issue has been automatically marked as stale because it has not had "
       "recent activity. If you think this still needs to be addressed "
       "please comment on this thread.\n\nPlease note that issues that do not follow the "
       "[contributing guidelines](https://github.com/huggingface/transformers/blob/main/CONTRIBUTING.md) "
       "are likely to be ignored."					)


if __name__ == "__main__":
  main()


