





'''simple docstring'''

import inspect
import os

import torch
from transformers import AutoModel
from transformers.testing_utils import mockenv_context
from transformers.trainer_utils import set_seed

import accelerate
from accelerate.accelerator import Accelerator
from accelerate.state import AcceleratorState
from accelerate.test_utils.testing import (
    AccelerateTestCase,
    TempDirTestCase,
    execute_subprocess_async,
    require_cuda,
    require_fsdp,
    require_multi_gpu,
    slow,
)
from accelerate.utils.constants import (
    FSDP_AUTO_WRAP_POLICY,
    FSDP_BACKWARD_PREFETCH,
    FSDP_SHARDING_STRATEGY,
    FSDP_STATE_DICT_TYPE,
)
from accelerate.utils.dataclasses import FullyShardedDataParallelPlugin
from accelerate.utils.other import patch_environment


set_seed(42)

__lowerCAmelCase							:	List[Any]			="bert-base-cased"
__lowerCAmelCase							:	Dict			="fp16"
__lowerCAmelCase							:	Optional[int]			="bf16"
__lowerCAmelCase							:	Tuple			=[FPaa, BFaa]






@require_fsdp
@require_cuda
class    UpperCAmelCase	(       A_				):
 def      UpperCAmelCase_							(				self      :Tuple							)->							int:
     super().setUp()

     A__				=			dict(
         ACCELERATE_USE_FSDP="true"						,       MASTER_ADDR="localhost"						,       MASTER_PORT="10999"						,       RANK="0"						,       LOCAL_RANK="0"						,       WORLD_SIZE="1"						,       )
 def      UpperCAmelCase_							(				self      :int							)->							Optional[int]:
     from torch.distributed.fsdp.fully_sharded_data_parallel import ShardingStrategy

     for i, strategy in enumerate(snake_case__							):
         A__				=			self.dist_env.copy()
         A__				=			F"{i + 1}"
         A__				=			strategy
         with mockenv_context(**snake_case__							):
             A__				=			FullyShardedDataParallelPlugin()
             self.assertEqual(fsdp_plugin.sharding_strategy						,       ShardingStrategy(i + 1							)							)
 def      UpperCAmelCase_							(				self      :Union[str, Any]							)->							Any:
     from torch.distributed.fsdp.fully_sharded_data_parallel import BackwardPrefetch

     for i, prefetch_policy in enumerate(snake_case__							):
         A__				=			self.dist_env.copy()
         A__				=			prefetch_policy
         with mockenv_context(**snake_case__							):
             A__				=			FullyShardedDataParallelPlugin()
             if prefetch_policy == "NO_PREFETCH":
                 self.assertIsNone(fsdp_plugin.backward_prefetch							)
             else:
                 self.assertEqual(fsdp_plugin.backward_prefetch						,       BackwardPrefetch(i + 1							)							)
 def      UpperCAmelCase_							(				self      :List[str]							)->							Tuple:
     from torch.distributed.fsdp.fully_sharded_data_parallel import StateDictType

     for i, state_dict_type in enumerate(snake_case__							):
         A__				=			self.dist_env.copy()
         A__				=			state_dict_type
         with mockenv_context(**snake_case__							):
             A__				=			FullyShardedDataParallelPlugin()
             self.assertEqual(fsdp_plugin.state_dict_type						,       StateDictType(i + 1							)							)
             if state_dict_type == "FULL_STATE_DICT":
                 self.assertTrue(fsdp_plugin.state_dict_config.offload_to_cpu							)
                 self.assertTrue(fsdp_plugin.state_dict_config.ranka_only							)
 def      UpperCAmelCase_							(				self      :str							)->							List[str]:
     A__				=			AutoModel.from_pretrained(snake_case__							)
     for policy in FSDP_AUTO_WRAP_POLICY:
         A__				=			self.dist_env.copy()
         A__				=			policy
         if policy == "TRANSFORMER_BASED_WRAP":
             A__				=			"BertLayer"
         elif policy == "SIZE_BASED_WRAP":
             A__				=			"2000"
         with mockenv_context(**snake_case__							):
             A__				=			FullyShardedDataParallelPlugin()
             fsdp_plugin.set_auto_wrap_policy(snake_case__							)
             if policy == "NO_WRAP":
                 self.assertIsNone(fsdp_plugin.auto_wrap_policy							)
             else:
                 self.assertIsNotNone(fsdp_plugin.auto_wrap_policy							)

     A__				=			self.dist_env.copy()
     A__				=			"TRANSFORMER_BASED_WRAP"
     A__				=			"T5Layer"
     with mockenv_context(**snake_case__							):
         A__				=			FullyShardedDataParallelPlugin()
         with self.assertRaises(snake_case__							) as cm:
             fsdp_plugin.set_auto_wrap_policy(snake_case__							)
         self.assertTrue("Could not find the transformer layer class to wrap in the model." in str(cm.exception							)							)

     A__				=			self.dist_env.copy()
     A__				=			"SIZE_BASED_WRAP"
     A__				=			"0"
     with mockenv_context(**snake_case__							):
         A__				=			FullyShardedDataParallelPlugin()
         fsdp_plugin.set_auto_wrap_policy(snake_case__							)
         self.assertIsNone(fsdp_plugin.auto_wrap_policy							)
 def      UpperCAmelCase_							(				self      :Optional[int]							)->							Optional[Any]:
     from torch.distributed.fsdp.fully_sharded_data_parallel import MixedPrecision
     from torch.distributed.fsdp.sharded_grad_scaler import ShardedGradScaler

     for mp_dtype in dtypes:
         A__				=			self.dist_env.copy()
         A__				=			mp_dtype
         with mockenv_context(**snake_case__							):
             A__				=			Accelerator()
             if mp_dtype == "fp16":
                 A__				=			torch.floataa
             elif mp_dtype == "bf16":
                 A__				=			torch.bfloataa
             A__				=			MixedPrecision(param_dtype=snake_case__						,       reduce_dtype=snake_case__						,       buffer_dtype=snake_case__							)
             self.assertEqual(accelerator.state.fsdp_plugin.mixed_precision_policy						,       snake_case__							)
             if mp_dtype == FPaa:
                 self.assertTrue(isinstance(accelerator.scaler						,       snake_case__							)							)
             elif mp_dtype == BFaa:
                 self.assertIsNone(accelerator.scaler							)
             AcceleratorState._reset_state(snake_case__							)



 def      UpperCAmelCase_							(				self      :Any							)->							int:
     from torch.distributed.fsdp.fully_sharded_data_parallel import CPUOffload

     for flag in [True, False]:
         A__				=			self.dist_env.copy()
         A__				=			str(snake_case__							).lower()
         with mockenv_context(**snake_case__							):
             A__				=			FullyShardedDataParallelPlugin()
             self.assertEqual(fsdp_plugin.cpu_offload						,       CPUOffload(offload_params=snake_case__							)							)






@require_fsdp
@require_multi_gpu
@slow
class    UpperCAmelCase	(       A_				):
 def      UpperCAmelCase_							(				self      :Union[str, Any]							)->							int:
     super().setUp()
     A__				=			0.8_2
     A__				=			[
         "fsdp_shard_grad_op_transformer_based_wrap",
         "fsdp_full_shard_transformer_based_wrap",
     ]
     A__				=			{
         "multi_gpu_fp16": 32_00,
         "fsdp_shard_grad_op_transformer_based_wrap_fp16": 20_00,
         "fsdp_full_shard_transformer_based_wrap_fp16": 19_00,
         # Disabling below test as it overwhelms the RAM memory usage
         # on CI self-hosted runner leading to tests getting killed.
         # "fsdp_full_shard_cpu_offload_transformer_based_wrap_fp32": 1500,  # fp16 was leading to indefinite hang
     }
     A__				=			1_60
     A__				=			1_60

     A__				=			inspect.getfile(accelerate.test_utils							)
     A__				=			os.path.sep.join(mod_file.split(os.path.sep							)[:-1] + ["scripts", "external_deps"]							)
 def      UpperCAmelCase_							(				self      :int							)->							Union[str, Any]:
     A__				=			os.path.join(self.test_scripts_folder						,       "test_performance.py"							)
     A__				=			["accelerate", "launch", "--num_processes=2", "--num_machines=1", "--machine_rank=0", "--use_fsdp"]
     for config in self.performance_configs:
         A__				=			cmd.copy()
         for i, strategy in enumerate(snake_case__							):
             if strategy.lower() in config:
                 cmd_config.append(F"--fsdp_sharding_strategy={i+1}"							)
                 break

         if "fp32" in config:
             cmd_config.append("--mixed_precision=no"							)
         else:
             cmd_config.append("--mixed_precision=fp16"							)

         if "cpu_offload" in config:
             cmd_config.append("--fsdp_offload_params=True"							)

         for policy in FSDP_AUTO_WRAP_POLICY:
             if policy.lower() in config:
                 cmd_config.append(F"--fsdp_auto_wrap_policy={policy}"							)
                 break

         if policy == "TRANSFORMER_BASED_WRAP":
             cmd_config.append("--fsdp_transformer_layer_cls_to_wrap=BertLayer"							)
         elif policy == "SIZE_BASED_WRAP":
             cmd_config.append("--fsdp_min_num_params=2000"							)

         cmd_config.extend(
             [
                 self.test_file_path,
                 F"--output_dir={self.tmpdir}",
                 F"--performance_lower_bound={self.performance_lower_bound}",
             ]							)
         with patch_environment(omp_num_threads=1							):
             execute_subprocess_async(snake_case__						,       env=os.environ.copy()							)
 def      UpperCAmelCase_							(				self      :str							)->							Optional[Any]:
     A__				=			os.path.join(self.test_scripts_folder						,       "test_checkpointing.py"							)
     A__				=			[
         "accelerate",
         "launch",
         "--num_processes=2",
         "--num_machines=1",
         "--machine_rank=0",
         "--use_fsdp",
         "--mixed_precision=fp16",
         "--fsdp_transformer_layer_cls_to_wrap=BertLayer",
     ]

     for i, strategy in enumerate(snake_case__							):
         A__				=			cmd.copy()
         cmd_config.append(F"--fsdp_sharding_strategy={i+1}"							)
         if strategy != "FULL_SHARD":
             continue
         A__				=			len(snake_case__							)
         for state_dict_type in FSDP_STATE_DICT_TYPE:
             A__				=			cmd_config[:state_dict_config_index]
             cmd_config.append(F"--fsdp_state_dict_type={state_dict_type}"							)
             cmd_config.extend(
                 [
                     self.test_file_path,
                     F"--output_dir={self.tmpdir}",
                     "--partial_train_epoch=1",
                 ]							)
             with patch_environment(omp_num_threads=1							):
                 execute_subprocess_async(snake_case__						,       env=os.environ.copy()							)

             A__				=			cmd_config[:-1]
             A__				=			os.path.join(self.tmpdir						,       "epoch_0"							)
             cmd_config.extend(
                 [
                     F"--resume_from_checkpoint={resume_from_checkpoint}",
                 ]							)
             with patch_environment(omp_num_threads=1							):
                 execute_subprocess_async(snake_case__						,       env=os.environ.copy()							)



 def      UpperCAmelCase_							(				self      :Optional[Any]							)->							str:
     A__				=			os.path.join(self.test_scripts_folder						,       "test_peak_memory_usage.py"							)
     A__				=			[
         "accelerate",
         "launch",
         "--num_processes=2",
         "--num_machines=1",
         "--machine_rank=0",
     ]
     for spec, peak_mem_upper_bound in self.peak_memory_usage_upper_bound.items():
         A__				=			cmd.copy()
         if "fp16" in spec:
             cmd_config.extend(["--mixed_precision=fp16"]							)
         else:
             cmd_config.extend(["--mixed_precision=no"]							)

         if "multi_gpu" in spec:
             continue
         else:
             cmd_config.extend(["--use_fsdp"]							)
             for i, strategy in enumerate(snake_case__							):
                 if strategy.lower() in spec:
                     cmd_config.append(F"--fsdp_sharding_strategy={i+1}"							)
                     break

             if "cpu_offload" in spec:
                 cmd_config.append("--fsdp_offload_params=True"							)

             for policy in FSDP_AUTO_WRAP_POLICY:
                 if policy.lower() in spec:
                     cmd_config.append(F"--fsdp_auto_wrap_policy={policy}"							)
                     break

             if policy == "TRANSFORMER_BASED_WRAP":
                 cmd_config.append("--fsdp_transformer_layer_cls_to_wrap=BertLayer"							)
             elif policy == "SIZE_BASED_WRAP":
                 cmd_config.append("--fsdp_min_num_params=2000"							)

         cmd_config.extend(
             [
                 self.test_file_path,
                 F"--output_dir={self.tmpdir}",
                 F"--peak_memory_upper_bound={peak_mem_upper_bound}",
                 F"--n_train={self.n_train}",
                 F"--n_val={self.n_val}",
             ]							)
         with patch_environment(omp_num_threads=1							):
             execute_subprocess_async(snake_case__						,       env=os.environ.copy()							)





import collections
import inspect
import unittest
from typing import Dict, List, Tuple

from transformers import MaskFormerSwinConfig
from transformers.testing_utils import require_torch, require_torch_multi_gpu, torch_device
from transformers.utils import is_torch_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
   import torch
   from torch import nn

   from transformers import MaskFormerSwinBackbone
   from transformers.models.maskformer import MaskFormerSwinModel






class   UpperCAmelCase		:



   def __init__(self		:					Dict    ,						snake_case__		:					Dict    ,						snake_case__		:					Any=13    ,						snake_case__		:					Any=32    ,						snake_case__		:					Optional[Any]=2    ,						snake_case__		:					Union[str, Any]=3    ,						snake_case__		:					List[Any]=16    ,						snake_case__		:					int=[1, 2, 1]    ,						snake_case__		:					Dict=[2, 2, 4]    ,						snake_case__		:					Dict=2    ,						snake_case__		:					Tuple=2.0    ,						snake_case__		:					Optional[int]=True    ,						snake_case__		:					Union[str, Any]=0.0    ,						snake_case__		:					Any=0.0    ,						snake_case__		:					Union[str, Any]=0.1    ,						snake_case__		:					int="gelu"    ,						snake_case__		:					Optional[int]=False    ,						snake_case__		:					List[Any]=True    ,						snake_case__		:					List[str]=0.02    ,						snake_case__		:					int=1e-5    ,						snake_case__		:					List[str]=True    ,						snake_case__		:					Union[str, Any]=None    ,						snake_case__		:					List[Any]=True    ,						snake_case__		:					Optional[Any]=10    ,						snake_case__		:					Optional[Any]=8    ,						snake_case__		:					Any=["stage1", "stage2", "stage3"]    ,						snake_case__		:					Tuple=[1, 2, 3]    ,						)		->	Union[str, Any]:


          '''simple docstring'''


          snake_case       :   Any							=      parent
          snake_case       :   Optional[int]							=      batch_size
          snake_case       :   Union[str, Any]							=      image_size
          snake_case       :   Dict							=      patch_size
          snake_case       :   Optional[Any]							=      num_channels
          snake_case       :   Union[str, Any]							=      embed_dim
          snake_case       :   int							=      depths
          snake_case       :   List[str]							=      num_heads
          snake_case       :   Union[str, Any]							=      window_size
          snake_case       :   Union[str, Any]							=      mlp_ratio
          snake_case       :   List[Any]							=      qkv_bias
          snake_case       :   List[Any]							=      hidden_dropout_prob
          snake_case       :   Union[str, Any]							=      attention_probs_dropout_prob
          snake_case       :   Union[str, Any]							=      drop_path_rate
          snake_case       :   int							=      hidden_act
          snake_case       :   Optional[int]							=      use_absolute_embeddings
          snake_case       :   int							=      patch_norm
          snake_case       :   Union[str, Any]							=      layer_norm_eps
          snake_case       :   Any							=      initializer_range
          snake_case       :   Optional[Any]							=      is_training
          snake_case       :   Tuple							=      scope
          snake_case       :   Optional[int]							=      use_labels
          snake_case       :   Optional[Any]							=      type_sequence_label_size
          snake_case       :   Union[str, Any]							=      encoder_stride
          snake_case       :   Any							=      out_features
          snake_case       :   Tuple							=      out_indices



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[Any] )		->	Dict:


          '''simple docstring'''


          snake_case       :   Optional[Any]							=      floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size] )

          snake_case       :   int							=      None
          if self.use_labels:
                 snake_case       :   Tuple							=      ids_tensor([self.batch_size]    ,						self.type_sequence_label_size )

          snake_case       :   Dict							=      self.get_config()

          return config, pixel_values, labels



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


          '''simple docstring'''


          return MaskFormerSwinConfig(
              image_size=self.image_size    ,						patch_size=self.patch_size    ,						num_channels=self.num_channels    ,						embed_dim=self.embed_dim    ,						depths=self.depths    ,						num_heads=self.num_heads    ,						window_size=self.window_size    ,						mlp_ratio=self.mlp_ratio    ,						qkv_bias=self.qkv_bias    ,						hidden_dropout_prob=self.hidden_dropout_prob    ,						attention_probs_dropout_prob=self.attention_probs_dropout_prob    ,						drop_path_rate=self.drop_path_rate    ,						hidden_act=self.hidden_act    ,						use_absolute_embeddings=self.use_absolute_embeddings    ,						path_norm=self.patch_norm    ,						layer_norm_eps=self.layer_norm_eps    ,						initializer_range=self.initializer_range    ,						encoder_stride=self.encoder_stride    ,						out_features=self.out_features    ,						out_indices=self.out_indices    ,						)



   def 				_SCREAMING_SNAKE_CASE   (self		:					Tuple    ,						snake_case__		:					List[Any]    ,						snake_case__		:					List[str]    ,						snake_case__		:					Tuple )		->	Optional[Any]:


          '''simple docstring'''


          snake_case       :   Union[str, Any]							=      MaskFormerSwinModel(config=snake_case__ )
          model.to(snake_case__ )
          model.eval()
          snake_case       :   List[Any]							=      model(snake_case__ )

          snake_case       :   Dict							=      ((config.image_size // config.patch_size) ** 2) // (4 ** (len(config.depths ) - 1))
          snake_case       :   int							=      int(config.embed_dim * 2 ** (len(config.depths ) - 1) )

          self.parent.assertEqual(result.last_hidden_state.shape    ,						(self.batch_size, expected_seq_len, expected_dim) )



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


          '''simple docstring'''


          snake_case       :   Optional[int]							=      MaskFormerSwinBackbone(config=snake_case__ )
          model.to(snake_case__ )
          model.eval()
          snake_case       :   List[Any]							=      model(snake_case__ )

          # verify feature maps
          self.parent.assertEqual(len(result.feature_maps )    ,						len(config.out_features ) )
          self.parent.assertListEqual(list(result.feature_maps[0].shape )    ,						[13, 16, 16, 16] )

          # verify channels
          self.parent.assertEqual(len(model.channels )    ,						len(config.out_features ) )
          self.parent.assertListEqual(model.channels    ,						[16, 32, 64] )

          # verify ValueError
          with self.parent.assertRaises(snake_case__ ):
                 snake_case       :   Tuple							=      ["stem"]
                 snake_case       :   List[Any]							=      MaskFormerSwinBackbone(config=snake_case__ )



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


          '''simple docstring'''


          snake_case       :   Union[str, Any]							=      self.prepare_config_and_inputs()
          snake_case				,					snake_case				,					snake_case       :   List[Any]							=      config_and_inputs
          snake_case       :   int							=      {"pixel_values": pixel_values}
          return config, inputs_dict






@require_torch
class   UpperCAmelCase		(						A_       ,A_       ,unittest.TestCase ):
   A__					:		List[str]						     =     (
       (
           MaskFormerSwinModel,
           MaskFormerSwinBackbone,
       )
       if is_torch_available()
       else ()
   )
   A__					:		str						     =     {"feature-extraction": MaskFormerSwinModel} if is_torch_available() else {}
   A__					:		Optional[Any]						     =     False
   A__					:		List[Any]						     =     False
   A__					:		List[str]						     =     False
   A__					:		List[str]						     =     False
   A__					:		Union[str, Any]						     =     False



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[int] )		->	List[str]:


          '''simple docstring'''


          snake_case       :   str							=      MaskFormerSwinModelTester(self )
          snake_case       :   Optional[int]							=      ConfigTester(self    ,						config_class=snake_case__    ,						embed_dim=37 )



   @require_torch_multi_gpu
   @unittest.skip(
       reason=(
           "`MaskFormerSwinModel` outputs `hidden_states_spatial_dimensions` which doesn't work well with"
           " `nn.DataParallel`"
       ) )
   def 				_SCREAMING_SNAKE_CASE   (self		:					str )		->	Optional[Any]:


          '''simple docstring'''


          pass



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


          '''simple docstring'''


          self.create_and_test_config_common_properties()
          self.config_tester.create_and_test_config_to_json_string()
          self.config_tester.create_and_test_config_to_json_file()
          self.config_tester.create_and_test_config_from_and_save_pretrained()
          self.config_tester.create_and_test_config_with_num_labels()
          self.config_tester.check_config_can_be_init_without_params()
          self.config_tester.check_config_arguments_init()



   def 				_SCREAMING_SNAKE_CASE   (self		:					Tuple )		->	List[Any]:


          '''simple docstring'''


          return



   def 				_SCREAMING_SNAKE_CASE   (self		:					Dict )		->	str:


          '''simple docstring'''


          snake_case       :   str							=      self.model_tester.prepare_config_and_inputs()
          self.model_tester.create_and_check_model(*snake_case__ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					int )		->	Dict:


          '''simple docstring'''


          snake_case       :   Any							=      self.model_tester.prepare_config_and_inputs()
          self.model_tester.create_and_check_backbone(*snake_case__ )



   @unittest.skip("Swin does not use inputs_embeds" )
   def 				_SCREAMING_SNAKE_CASE   (self		:					int )		->	Any:


          '''simple docstring'''


          pass



   @unittest.skip("Swin does not support feedforward chunking" )
   def 				_SCREAMING_SNAKE_CASE   (self		:					List[str] )		->	Dict:


          '''simple docstring'''


          pass



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


          '''simple docstring'''


          snake_case				,					snake_case       :   Any							=      self.model_tester.prepare_config_and_inputs_for_common()

          for model_class in self.all_model_classes:
                 snake_case       :   int							=      model_class(snake_case__ )
                 self.assertIsInstance(model.get_input_embeddings()    ,						(nn.Module) )
                 snake_case       :   List[Any]							=      model.get_output_embeddings()
                 self.assertTrue(x is None or isinstance(snake_case__    ,						nn.Linear ) )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[Any] )		->	Dict:


          '''simple docstring'''


          snake_case				,					snake_case       :   str							=      self.model_tester.prepare_config_and_inputs_for_common()

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

                 snake_case       :   Tuple							=      ["pixel_values"]
                 self.assertListEqual(arg_names[:1]    ,						snake_case__ )



   @unittest.skip(reason="MaskFormerSwin is only used as backbone and doesn't support output_attentions" )
   def 				_SCREAMING_SNAKE_CASE   (self		:					List[Any] )		->	str:


          '''simple docstring'''


          pass



   @unittest.skip(reason="MaskFormerSwin is only used as an internal backbone" )
   def 				_SCREAMING_SNAKE_CASE   (self		:					Tuple )		->	Any:


          '''simple docstring'''


          pass



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[Any]    ,						snake_case__		:					List[Any]    ,						snake_case__		:					str    ,						snake_case__		:					List[Any]    ,						snake_case__		:					Tuple )		->	Optional[int]:


          '''simple docstring'''


          snake_case       :   Tuple							=      model_class(snake_case__ )
          model.to(snake_case__ )
          model.eval()

          with torch.no_grad():
                 snake_case       :   Any							=      model(**self._prepare_for_class(snake_case__    ,						snake_case__ ) )

          snake_case       :   int							=      outputs.hidden_states

          snake_case       :   Union[str, Any]							=      getattr(
              self.model_tester    ,						"expected_num_hidden_layers"    ,						len(self.model_tester.depths ) + 1 )
          self.assertEqual(len(snake_case__ )    ,						snake_case__ )

          # Swin has a different seq_length
          snake_case       :   Any							=      (
              config.patch_size
              if isinstance(config.patch_size    ,						collections.abc.Iterable )
              else (config.patch_size, config.patch_size)
          )

          snake_case       :   Tuple							=      (image_size[1] // patch_size[1]) * (image_size[0] // patch_size[0])

          self.assertListEqual(
              list(hidden_states[0].shape[-2:] )    ,						[num_patches, self.model_tester.embed_dim]    ,						)



   def 				_SCREAMING_SNAKE_CASE   (self		:					Dict )		->	Union[str, Any]:


          '''simple docstring'''


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

          snake_case       :   int							=      (
              self.model_tester.image_size
              if isinstance(self.model_tester.image_size    ,						collections.abc.Iterable )
              else (self.model_tester.image_size, self.model_tester.image_size)
          )

          for model_class in self.all_model_classes:
                 snake_case       :   int							=      True
                 self.check_hidden_states_output(snake_case__    ,						snake_case__    ,						snake_case__    ,						snake_case__ )

                 # check that output_hidden_states also work using config
                 del inputs_dict["output_hidden_states"]
                 snake_case       :   Dict							=      True

                 self.check_hidden_states_output(snake_case__    ,						snake_case__    ,						snake_case__    ,						snake_case__ )



   def 				_SCREAMING_SNAKE_CASE   (self		:					int )		->	Any:


          '''simple docstring'''


          snake_case				,					snake_case       :   Union[str, Any]							=      self.model_tester.prepare_config_and_inputs_for_common()
          snake_case       :   Any							=      3

          snake_case       :   List[str]							=      (
              self.model_tester.image_size
              if isinstance(self.model_tester.image_size    ,						collections.abc.Iterable )
              else (self.model_tester.image_size, self.model_tester.image_size)
          )
          snake_case       :   Tuple							=      (
              config.patch_size
              if isinstance(config.patch_size    ,						collections.abc.Iterable )
              else (config.patch_size, config.patch_size)
          )

          snake_case       :   str							=      image_size[0] + patch_size[0] - (image_size[0] % patch_size[0])
          snake_case       :   str							=      image_size[1] + patch_size[1] - (image_size[1] % patch_size[1])

          for model_class in self.all_model_classes:
                 snake_case       :   str							=      True
                 self.check_hidden_states_output(snake_case__    ,						snake_case__    ,						snake_case__    ,						(padded_height, padded_width) )

                 # check that output_hidden_states also work using config
                 del inputs_dict["output_hidden_states"]
                 snake_case       :   Optional[Any]							=      True
                 self.check_hidden_states_output(snake_case__    ,						snake_case__    ,						snake_case__    ,						(padded_height, padded_width) )



   @unittest.skip(reason="MaskFormerSwin doesn't have pretrained checkpoints" )
   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[int] )		->	str:


          '''simple docstring'''


          pass



   @unittest.skip(reason="This will be fixed once MaskFormerSwin is replaced by native Swin" )
   def 				_SCREAMING_SNAKE_CASE   (self		:					str )		->	int:


          '''simple docstring'''


          pass



   @unittest.skip(reason="This will be fixed once MaskFormerSwin is replaced by native Swin" )
   def 				_SCREAMING_SNAKE_CASE   (self		:					int )		->	str:


          '''simple docstring'''


          pass



   def 				_SCREAMING_SNAKE_CASE   (self		:					Any )		->	Any:


          '''simple docstring'''


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

          def set_nan_tensor_to_zero(snake_case__		:					Union[str, Any] ):
                 snake_case       :   Any							=      0
                 return t

          def check_equivalence(snake_case__		:					Union[str, Any]    ,						snake_case__		:					int    ,						snake_case__		:					List[str]    ,						snake_case__		:					Optional[int]={} ):
                 with torch.no_grad():
                        snake_case       :   Optional[Any]							=      model(**snake_case__    ,						return_dict=snake_case__    ,						**snake_case__ )
                        snake_case       :   Tuple							=      model(**snake_case__    ,						return_dict=snake_case__    ,						**snake_case__ ).to_tuple()

                        def recursive_check(snake_case__		:					List[str]    ,						snake_case__		:					Optional[Any] ):
                               if isinstance(snake_case__    ,						(List, Tuple) ):
                                      for tuple_iterable_value, dict_iterable_value in zip(snake_case__    ,						snake_case__ ):
                                             recursive_check(snake_case__    ,						snake_case__ )
                               elif isinstance(snake_case__    ,						snake_case__ ):
                                      for tuple_iterable_value, dict_iterable_value in zip(
                                          tuple_object.values()    ,						dict_object.values() ):
                                             recursive_check(snake_case__    ,						snake_case__ )
                               elif tuple_object is None:
                                      return
                               else:
                                      self.assertTrue(
                                          torch.allclose(
                                              set_nan_tensor_to_zero(snake_case__ )    ,						set_nan_tensor_to_zero(snake_case__ )    ,						atol=1e-5 )    ,						msg=(
                                              "Tuple and dict output are not equal. Difference:"
                                              f""" {torch.max(torch.abs(tuple_object - dict_object ) )}. Tuple has `nan`:"""
                                              f""" {torch.isnan(snake_case__ ).any()} and `inf`: {torch.isinf(snake_case__ )}. Dict has"""
                                              f""" `nan`: {torch.isnan(snake_case__ ).any()} and `inf`: {torch.isinf(snake_case__ )}."""
                                          )    ,						)

                        recursive_check(snake_case__    ,						snake_case__ )

          for model_class in self.all_model_classes:
                 snake_case       :   Optional[int]							=      model_class(snake_case__ )
                 model.to(snake_case__ )
                 model.eval()

                 snake_case       :   Union[str, Any]							=      self._prepare_for_class(snake_case__    ,						snake_case__ )
                 snake_case       :   Tuple							=      self._prepare_for_class(snake_case__    ,						snake_case__ )
                 check_equivalence(snake_case__    ,						snake_case__    ,						snake_case__ )

                 snake_case       :   Tuple							=      self._prepare_for_class(snake_case__    ,						snake_case__    ,						return_labels=snake_case__ )
                 snake_case       :   Optional[Any]							=      self._prepare_for_class(snake_case__    ,						snake_case__    ,						return_labels=snake_case__ )
                 check_equivalence(snake_case__    ,						snake_case__    ,						snake_case__ )

                 snake_case       :   Dict							=      self._prepare_for_class(snake_case__    ,						snake_case__ )
                 snake_case       :   List[Any]							=      self._prepare_for_class(snake_case__    ,						snake_case__ )
                 check_equivalence(snake_case__    ,						snake_case__    ,						snake_case__    ,						{"output_hidden_states": True} )

                 snake_case       :   Any							=      self._prepare_for_class(snake_case__    ,						snake_case__    ,						return_labels=snake_case__ )
                 snake_case       :   List[str]							=      self._prepare_for_class(snake_case__    ,						snake_case__    ,						return_labels=snake_case__ )
                 check_equivalence(snake_case__    ,						snake_case__    ,						snake_case__    ,						{"output_hidden_states": True} )






@require_torch
class   UpperCAmelCase		(						unittest.TestCase       ,A_ ):
   A__					:		int						     =     (MaskFormerSwinBackbone,) if is_torch_available() else ()
   A__					:		int						     =     MaskFormerSwinConfig



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


          '''simple docstring'''


          snake_case       :   Union[str, Any]							=      MaskFormerSwinModelTester(self )



   def 				_SCREAMING_SNAKE_CASE   (self		:					Optional[int] )		->	Optional[Any]:


          '''simple docstring'''


          snake_case				,					snake_case       :   Dict							=      self.model_tester.prepare_config_and_inputs_for_common()
          snake_case       :   Optional[int]							=      inputs_dict["pixel_values"].shape[0]

          for backbone_class in self.all_model_classes:
                 snake_case       :   Optional[int]							=      backbone_class(snake_case__ )
                 backbone.to(snake_case__ )
                 backbone.eval()

                 snake_case       :   Union[str, Any]							=      backbone(**snake_case__ )

                 # Test default outputs and verify feature maps
                 self.assertIsInstance(outputs.feature_maps    ,						snake_case__ )
                 self.assertTrue(len(outputs.feature_maps ) == len(backbone.channels ) )
                 for feature_map, n_channels in zip(outputs.feature_maps    ,						backbone.channels ):
                        self.assertTrue(feature_map.shape[:2]    ,						(batch_size, n_channels) )
                 self.assertIsNone(outputs.hidden_states )
                 self.assertIsNone(outputs.attentions )

                 # Test output_hidden_states=True
                 snake_case       :   Optional[int]							=      backbone(**snake_case__    ,						output_hidden_states=snake_case__ )
                 self.assertIsNotNone(outputs.hidden_states )
                 self.assertTrue(len(outputs.hidden_states )    ,						len(backbone.stage_names ) )
                 # We skip the stem layer
                 for hidden_states, n_channels in zip(outputs.hidden_states[1:]    ,						backbone.channels ):
                        for hidden_state in hidden_states:
                               # Hidden states are in the format (batch_size, (height * width), n_channels)
                               snake_case				,					snake_case				,					snake_case       :   Dict							=      hidden_state.shape
                               self.assertTrue((h_batch_size, h_n_channels)    ,						(batch_size, n_channels) )

            # Test output_attentions=True
                 if self.has_attentions:
                        snake_case       :   Optional[Any]							=      backbone(**snake_case__    ,						output_attentions=snake_case__ )
                        self.assertIsNotNone(outputs.attentions )





