
'''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  _UpperCAmelCase :

     """simple docstring"""

     def __init__(      self							,		lowerCAmelCase_							,		lowerCAmelCase_=2							,		lowerCAmelCase_=True							,		lowerCAmelCase_=False							,		lowerCAmelCase_=10							,		lowerCAmelCase_=3							,		lowerCAmelCase_=32 * 8							,		lowerCAmelCase_=32 * 8							,		lowerCAmelCase_=4							,		lowerCAmelCase_=64							,		):

       '''simple docstring'''

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

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

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

       a_      :	Any      =							self.get_config()
       return config, pixel_values, pixel_mask, mask_labels, class_labels


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Tuple      =							MaskaFormerConfig(
           hidden_size=self.hidden_dim							,		)
       a_      :	Optional[Any]      =							self.num_queries
       a_      :	Optional[int]      =							self.num_labels

       a_      :	Any      =							[1, 1, 1, 1]
       a_      :	Union[str, Any]      =							self.num_channels

       a_      :	Optional[int]      =							64
       a_      :	Dict      =							1_28
       a_      :	Dict      =							self.hidden_dim
       a_      :	Dict      =							self.hidden_dim
       a_      :	int      =							self.hidden_dim
       return config


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      , a_      , a_      , a_      :	Dict      =							self.prepare_config_and_inputs()
       a_      :	str      =							{"""pixel_values""": pixel_values, """pixel_mask""": pixel_mask}
       return config, inputs_dict


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Dict      =							output.encoder_hidden_states
       a_      :	List[Any]      =							output.pixel_decoder_hidden_states
       a_      :	Any      =							output.transformer_decoder_hidden_states

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


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_=False ):

       '''simple docstring'''

       with torch.no_grad():
         a_      :	List[Any]      =							MaskaFormerModel(config=__lowercase )
         model.to(__lowercase )
         model.eval()

         a_      :	List[Any]      =							model(pixel_values=__lowercase							,		pixel_mask=__lowercase )
         a_      :	Dict      =							model(__lowercase							,		output_hidden_states=__lowercase )

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






     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	Tuple      =							MaskaFormerForUniversalSegmentation(config=__lowercase )
       model.to(__lowercase )
       model.eval()

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

         comm_check_on_output(__lowercase )

         a_      :	Optional[int]      =							model(
             pixel_values=__lowercase							,		pixel_mask=__lowercase							,		mask_labels=__lowercase							,		class_labels=__lowercase )

       comm_check_on_output(__lowercase )

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







@require_torch
class  _UpperCAmelCase ( lowercase__	,lowercase__	,unittest.TestCase    ):

     """simple docstring"""

     a_    	=					(MaskaFormerModel, MaskaFormerForUniversalSegmentation) if is_torch_available() else ()
     a_    	=					{"""feature-extraction""": MaskaFormerModel} if is_torch_available() else {}

     a_    	=					False
     a_    	=					False
     a_    	=					False
     a_    	=					False


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Any      =							MaskaFormerModelTester(self )
       a_      :	List[str]      =							ConfigTester(self							,		config_class=__lowercase							,		has_text_modality=__lowercase )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       self.config_tester.run_common_tests()


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	Union[str, Any]      =							self.model_tester.prepare_config_and_inputs_for_common()
       self.model_tester.create_and_check_maskaformer_model(__lowercase							,		**__lowercase							,		output_hidden_states=__lowercase )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Tuple      =							self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_maskaformer_instance_segmentation_head_model(*__lowercase )


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

       '''simple docstring'''

       pass


     @unittest.skip(reason="""Mask2Former does not have a get_input_embeddings method""" )
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       pass


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

       '''simple docstring'''

       pass


     @unittest.skip(reason="""Mask2Former does not use token embeddings""" )
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       pass


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

       '''simple docstring'''

       pass


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

       '''simple docstring'''

       pass


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

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

         a_      :	List[Any]      =							["""pixel_values"""]
         self.assertListEqual(arg_names[:1]							,		__lowercase )


     @slow
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       for model_name in ["facebook/mask2former-swin-small-coco-instance"]:
         a_      :	Union[str, Any]      =							MaskaFormerModel.from_pretrained(__lowercase )
         self.assertIsNotNone(__lowercase )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

       a_      :	List[Any]      =							MaskaFormerForUniversalSegmentation(__lowercase ).to(__lowercase )
       a_      :	Dict      =							model(**__lowercase )
       self.assertTrue(outputs.loss is not None )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	Any      =							self.model_tester.prepare_config_and_inputs_for_common()
       self.model_tester.create_and_check_maskaformer_model(__lowercase							,		**__lowercase							,		output_hidden_states=__lowercase )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      , a_      :	List[str]      =							self.model_tester.prepare_config_and_inputs_for_common()

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


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       if not self.model_tester.is_training:
         return

       a_      :	Optional[Any]      =							self.all_model_classes[1]
       a_      , a_      , a_      , a_      , a_      :	Any      =							self.model_tester.prepare_config_and_inputs()

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

       a_      :	Dict      =							model(__lowercase							,		mask_labels=__lowercase							,		class_labels=__lowercase ).loss
       loss.backward()






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

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

       a_      :	int      =							model(__lowercase							,		mask_labels=__lowercase							,		class_labels=__lowercase )

       a_      :	int      =							outputs.encoder_hidden_states[0]
       encoder_hidden_states.retain_grad()

       a_      :	Dict      =							outputs.pixel_decoder_hidden_states[0]
       pixel_decoder_hidden_states.retain_grad()

       a_      :	Tuple      =							outputs.transformer_decoder_hidden_states[0]
       transformer_decoder_hidden_states.retain_grad()

       a_      :	str      =							outputs.attentions[0]
       attentions.retain_grad()

       outputs.loss.backward(retain_graph=__lowercase )

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







__snake_case:  Tuple								=			1E-4






def 	_snake_case						(      ):
  """simple docstring"""







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




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

     """simple docstring"""

     @cached_property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return "facebook/mask2former-swin-small-coco-instance"


     @cached_property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       return MaskaFormerImageProcessor.from_pretrained(self.model_checkpoints ) if is_vision_available() else None


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	str      =							MaskaFormerModel.from_pretrained(self.model_checkpoints ).to(__lowercase )
       a_      :	Dict      =							self.default_image_processor
       a_      :	str      =							prepare_img()
       a_      :	Tuple      =							image_processor(__lowercase							,		return_tensors="""pt""" ).to(__lowercase )
       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(__lowercase							,		(1, 3, 3_84, 3_84) )

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

       a_      :	int      =							torch.tensor(
           [[-0.2790, -1.0717, -1.1668], [-0.5128, -0.3128, -0.4987], [-0.5832, 0.1971, -0.0197]] ).to(__lowercase )
       self.assertTrue(
           torch.allclose(
               outputs.encoder_last_hidden_state[0, 0, :3, :3]							,		__lowercase							,		atol=__lowercase ) )

       a_      :	List[str]      =							torch.tensor(
           [[0.8973, 1.1847, 1.1776], [1.1934, 1.5040, 1.5128], [1.1153, 1.4486, 1.4951]] ).to(__lowercase )
       self.assertTrue(
           torch.allclose(
               outputs.pixel_decoder_last_hidden_state[0, 0, :3, :3]							,		__lowercase							,		atol=__lowercase ) )

       a_      :	str      =							torch.tensor(
           [[2.1152, 1.7000, -0.8603], [1.5808, 1.8004, -0.9353], [1.6043, 1.7495, -0.5999]] ).to(__lowercase )
       self.assertTrue(
           torch.allclose(
               outputs.transformer_decoder_last_hidden_state[0, :3, :3]							,		__lowercase							,		atol=__lowercase ) )


     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	Dict      =							MaskaFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints ).to(__lowercase ).eval()
       a_      :	Tuple      =							self.default_image_processor
       a_      :	Union[str, Any]      =							prepare_img()
       a_      :	Tuple      =							image_processor(__lowercase							,		return_tensors="""pt""" ).to(__lowercase )
       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(__lowercase							,		(1, 3, 3_84, 3_84) )

       with torch.no_grad():
         a_      :	Any      =							model(**__lowercase )
       # masks_queries_logits
       a_      :	int      =							outputs.masks_queries_logits
       self.assertEqual(
           masks_queries_logits.shape							,		(1, model.config.num_queries, inputs_shape[-2] // 4, inputs_shape[-1] // 4) )
       a_      :	List[str]      =							[
           [-8.7839, -9.0056, -8.8121],
           [-7.4104, -7.0313, -6.5401],
           [-6.6105, -6.3427, -6.4675],
       ]
       a_      :	Union[str, Any]      =							torch.tensor(__lowercase ).to(__lowercase )
       self.assertTrue(torch.allclose(masks_queries_logits[0, 0, :3, :3]							,		__lowercase							,		atol=__lowercase ) )
       # class_queries_logits
       a_      :	str      =							outputs.class_queries_logits
       self.assertEqual(class_queries_logits.shape							,		(1, model.config.num_queries, model.config.num_labels + 1) )
       a_      :	Tuple      =							torch.tensor(
           [
               [1.8324, -8.0835, -4.1922],
               [0.8450, -9.0050, -3.6053],
               [0.3045, -7.7293, -3.0275],
           ] ).to(__lowercase )
       self.assertTrue(torch.allclose(outputs.class_queries_logits[0, :3, :3]							,		__lowercase							,		atol=__lowercase ) )






     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	int      =							MaskaFormerForUniversalSegmentation.from_pretrained(self.model_checkpoints ).to(__lowercase ).eval()
       a_      :	str      =							self.default_image_processor

       a_      :	Optional[int]      =							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_      :	Optional[int]      =							inputs["""pixel_values"""].to(__lowercase )
       a_      :	Dict      =							[el.to(__lowercase ) for el in inputs["""mask_labels"""]]
       a_      :	int      =							[el.to(__lowercase ) for el in inputs["""class_labels"""]]

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

       self.assertTrue(outputs.loss is not None )

'''simple docstring'''

from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


__snake_case:  Optional[Any]								=			logging.get_logger(__name__)


__snake_case:  Tuple								=			TypeVar("DatasetType", Dataset, IterableDataset)






def 	_snake_case						(      A_       :    List[DatasetType]     ,		A_       :    Optional[List[float]] = None     ,		A_       :    Optional[int] = None     ,		A_       :    Optional[DatasetInfo] = None     ,		A_       :    Optional[NamedSplit] = None     ,		A_       :    Literal["first_exhausted", "all_exhausted"] = "first_exhausted"     ,		):
  """simple docstring"""







  from .arrow_dataset import Dataset
  from .iterable_dataset import IterableDataset

  if not datasets:
    raise ValueError("""Unable to interleave an empty list of datasets."""   )
  for i, dataset in enumerate(A_   ):
    if not isinstance(A_     ,		(Dataset, IterableDataset)   ):
      if isinstance(A_     ,		(DatasetDict, IterableDatasetDict)   ):
        if not dataset:
          raise ValueError(
              f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '''
              """is an empty dataset dictionary."""   )
        raise ValueError(
            f'''Dataset at position {i} has at least one split: {list(A_   )}\n'''
            f'''Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(A_   )   )}\']'''   )
      raise ValueError(
          f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A_   ).__name__}.'''   )
    if i == 0:
      a_      , a_      :	Union[str, Any]      =							(
          (Dataset, IterableDataset) if isinstance(A_     ,		A_   ) else (IterableDataset, Dataset)
      )
    elif not isinstance(A_     ,		A_   ):
      raise ValueError(
          f'''Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'''   )
  if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
    raise ValueError(f'''{stopping_strategy} is not supported. Please enter a valid stopping_strategy.'''   )
  if dataset_type is Dataset:
    return _interleave_map_style_datasets(
        A_     ,		A_     ,		A_     ,		info=A_     ,		split=A_     ,		stopping_strategy=A_   )
  else:
    return _interleave_iterable_datasets(
        A_     ,		A_     ,		A_     ,		info=A_     ,		split=A_     ,		stopping_strategy=A_   )






def 	_snake_case						(      A_       :    List[DatasetType]     ,		A_       :    Optional[DatasetInfo] = None     ,		A_       :    Optional[NamedSplit] = None     ,		A_       :    int = 0     ,		):
  """simple docstring"""







  if not dsets:
    raise ValueError("""Unable to concatenate an empty list of datasets."""   )
  for i, dataset in enumerate(A_   ):
    if not isinstance(A_     ,		(Dataset, IterableDataset)   ):
      if isinstance(A_     ,		(DatasetDict, IterableDatasetDict)   ):
        if not dataset:
          raise ValueError(
              f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} '''
              """is an empty dataset dictionary."""   )
        raise ValueError(
            f'''Dataset at position {i} has at least one split: {list(A_   )}\n'''
            f'''Please pick one to interleave with the other datasets, for example: dataset[\'{next(iter(A_   )   )}\']'''   )
      raise ValueError(
          f'''Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A_   ).__name__}.'''   )
    if i == 0:
      a_      , a_      :	str      =							(
          (Dataset, IterableDataset) if isinstance(A_     ,		A_   ) else (IterableDataset, Dataset)
      )
    elif not isinstance(A_     ,		A_   ):
      raise ValueError(
          f'''Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects.'''   )
  if dataset_type is Dataset:
    return _concatenate_map_style_datasets(A_     ,		info=A_     ,		split=A_     ,		axis=A_   )
  else:
    return _concatenate_iterable_datasets(A_     ,		info=A_     ,		split=A_     ,		axis=A_   )


