





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



					'''simple docstring'''







					def __init__(				self ,       snake_case_ ,       snake_case_=1_3 ,       snake_case_=3_2 ,       snake_case_=2 ,       snake_case_=3 ,       snake_case_=1_6 ,       snake_case_=[1, 2, 1] ,       snake_case_=[2, 2, 4] ,       snake_case_=2 ,       snake_case_=2.0 ,       snake_case_=True ,       snake_case_=0.0 ,       snake_case_=0.0 ,       snake_case_=0.1 ,       snake_case_="gelu" ,       snake_case_=False ,       snake_case_=True ,       snake_case_=0.0_2 ,       snake_case_=1e-5 ,       snake_case_=True ,       snake_case_=None ,       snake_case_=True ,       snake_case_=1_0 ,       snake_case_=8 ,       snake_case_=["stage1", "stage2", "stage3"] ,       snake_case_=[1, 2, 3] ,       )   ->			Any:







								'''simple docstring'''

								__lowercase 					=   parent
								__lowercase 					=   batch_size
								__lowercase 					=   image_size
								__lowercase 					=   patch_size
								__lowercase 					=   num_channels
								__lowercase 					=   embed_dim
								__lowercase 					=   depths
								__lowercase 					=   num_heads
								__lowercase 					=   window_size
								__lowercase 					=   mlp_ratio
								__lowercase 					=   qkv_bias
								__lowercase 					=   hidden_dropout_prob
								__lowercase 					=   attention_probs_dropout_prob
								__lowercase 					=   drop_path_rate
								__lowercase 					=   hidden_act
								__lowercase 					=   use_absolute_embeddings
								__lowercase 					=   patch_norm
								__lowercase 					=   layer_norm_eps
								__lowercase 					=   initializer_range
								__lowercase 					=   is_training
								__lowercase 					=   scope
								__lowercase 					=   use_labels
								__lowercase 					=   type_sequence_label_size
								__lowercase 					=   encoder_stride
								__lowercase 					=   out_features
								__lowercase 					=   out_indices






					def A      (				self						)   ->			int:







								'''simple docstring'''

								__lowercase 					=   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]						)

								__lowercase 					=   None
								if self.use_labels:
											__lowercase 					=   ids_tensor([self.batch_size] ,       self.type_sequence_label_size						)

								__lowercase 					=   self.get_config()

								return config, pixel_values, labels






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







								'''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 A      (				self ,       snake_case_ ,       snake_case_ ,       snake_case_						)   ->			Union[str, Any]:







								'''simple docstring'''

								__lowercase 					=   MaskFormerSwinModel(config=snake_case_						)
								model.to(snake_case_						)
								model.eval()
								__lowercase 					=   model(snake_case_						)

								__lowercase 					=   ((config.image_size // config.patch_size) ** 2) // (4 ** (len(config.depths						) - 1))
								__lowercase 					=   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 A      (				self ,       snake_case_ ,       snake_case_ ,       snake_case_						)   ->			List[Any]:







								'''simple docstring'''

								__lowercase 					=   MaskFormerSwinBackbone(config=snake_case_						)
								model.to(snake_case_						)
								model.eval()
								__lowercase 					=   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						) ,       [1_3, 1_6, 1_6, 1_6]						)

								# verify channels
								self.parent.assertEqual(len(model.channels						) ,       len(config.out_features						)						)
								self.parent.assertListEqual(model.channels ,       [1_6, 3_2, 6_4]						)

								# verify ValueError
								with self.parent.assertRaises(snake_case_						):
											__lowercase 					=   ['''stem''']
											__lowercase 					=   MaskFormerSwinBackbone(config=snake_case_						)






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







								'''simple docstring'''

								__lowercase 					=   self.prepare_config_and_inputs()
								__lowercase					, __lowercase					, __lowercase 					=   config_and_inputs
								__lowercase 					=   {'''pixel_values''': pixel_values}
								return config, inputs_dict




@require_torch
class 	lowerCamelCase_       (				lowerCAmelCase__     ,     lowerCAmelCase__     ,     unittest.TestCase				):



					'''simple docstring'''







					__UpperCAmelCase            =				(
					    (
					        MaskFormerSwinModel,
					        MaskFormerSwinBackbone,
					    )
					    if is_torch_available()
					    else ()
					)
					__UpperCAmelCase            =				{"feature-extraction": MaskFormerSwinModel} if is_torch_available() else {}
					__UpperCAmelCase            =				False
					__UpperCAmelCase            =				False
					__UpperCAmelCase            =				False
					__UpperCAmelCase            =				False
					__UpperCAmelCase            =				False






					def A      (				self						)   ->			Optional[int]:







								'''simple docstring'''

								__lowercase 					=   MaskFormerSwinModelTester(self						)
								__lowercase 					=   ConfigTester(self ,       config_class=snake_case_ ,       embed_dim=3_7						)






					@require_torch_multi_gpu
					@unittest.skip(
					    reason=(
					        '''`MaskFormerSwinModel` outputs `hidden_states_spatial_dimensions` which doesn\'t work well with'''
					        ''' `nn.DataParallel`'''
					    )						)
					def A      (				self						)   ->			int:







								'''simple docstring'''

								pass






					def A      (				self						)   ->			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 A      (				self						)   ->			Optional[Any]:







								'''simple docstring'''

								return






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







								'''simple docstring'''

								__lowercase 					=   self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*snake_case_						)






					def A      (				self						)   ->			Tuple:







								'''simple docstring'''

								__lowercase 					=   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 A      (				self						)   ->			Union[str, Any]:







								'''simple docstring'''

								pass






					@unittest.skip('''Swin does not support feedforward chunking'''						)
					def A      (				self						)   ->			Tuple:







								'''simple docstring'''

								pass






					def A      (				self						)   ->			Any:







								'''simple docstring'''

								__lowercase					, __lowercase 					=   self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
											__lowercase 					=   model_class(snake_case_						)
											self.assertIsInstance(model.get_input_embeddings() ,       (nn.Module)						)
											__lowercase 					=   model.get_output_embeddings()
											self.assertTrue(x is None or isinstance(snake_case_ ,       nn.Linear						)						)






					def A      (				self						)   ->			Any:







								'''simple docstring'''

								__lowercase					, __lowercase 					=   self.model_tester.prepare_config_and_inputs_for_common()

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

											__lowercase 					=   ['''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 A      (				self						)   ->			List[Any]:







								'''simple docstring'''

								pass






					@unittest.skip(reason='''MaskFormerSwin is only used as an internal backbone'''						)
					def A      (				self						)   ->			List[Any]:







								'''simple docstring'''

								pass






					def A      (				self ,       snake_case_ ,       snake_case_ ,       snake_case_ ,       snake_case_						)   ->			List[Any]:







								'''simple docstring'''

								__lowercase 					=   model_class(snake_case_						)
								model.to(snake_case_						)
								model.eval()

								with torch.no_grad():
											__lowercase 					=   model(**self._prepare_for_class(snake_case_ ,       snake_case_						)						)

								__lowercase 					=   outputs.hidden_states

								__lowercase 					=   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
								__lowercase 					=   (
								    config.patch_size
								    if isinstance(config.patch_size ,       collections.abc.Iterable						)
								    else (config.patch_size, config.patch_size)
								)

								__lowercase 					=   (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 A      (				self						)   ->			Union[str, Any]:







								'''simple docstring'''

								__lowercase					, __lowercase 					=   self.model_tester.prepare_config_and_inputs_for_common()

								__lowercase 					=   (
								    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:
											__lowercase 					=   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"]
											__lowercase 					=   True

											self.check_hidden_states_output(snake_case_ ,       snake_case_ ,       snake_case_ ,       snake_case_						)






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







								'''simple docstring'''

								__lowercase					, __lowercase 					=   self.model_tester.prepare_config_and_inputs_for_common()
								__lowercase 					=   3

								__lowercase 					=   (
								    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)
								)
								__lowercase 					=   (
								    config.patch_size
								    if isinstance(config.patch_size ,       collections.abc.Iterable						)
								    else (config.patch_size, config.patch_size)
								)

								__lowercase 					=   image_size[0] + patch_size[0] - (image_size[0] % patch_size[0])
								__lowercase 					=   image_size[1] + patch_size[1] - (image_size[1] % patch_size[1])

								for model_class in self.all_model_classes:
											__lowercase 					=   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"]
											__lowercase 					=   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 A      (				self						)   ->			Any:







								'''simple docstring'''

								pass






					@unittest.skip(reason='''This will be fixed once MaskFormerSwin is replaced by native Swin'''						)
					def A      (				self						)   ->			Tuple:







								'''simple docstring'''

								pass






					@unittest.skip(reason='''This will be fixed once MaskFormerSwin is replaced by native Swin'''						)
					def A      (				self						)   ->			List[Any]:







								'''simple docstring'''

								pass






					def A      (				self						)   ->			Optional[int]:







								'''simple docstring'''

								__lowercase					, __lowercase 					=   self.model_tester.prepare_config_and_inputs_for_common()

								def set_nan_tensor_to_zero(snake_case_						):
											__lowercase 					=   0
											return t

								def check_equivalence(snake_case_ ,       snake_case_ ,       snake_case_ ,       snake_case_={}						):
											with torch.no_grad():
														__lowercase 					=   model(**snake_case_ ,       return_dict=snake_case_ ,       **snake_case_						)
														__lowercase 					=   model(**snake_case_ ,       return_dict=snake_case_ ,       **snake_case_						).to_tuple()

														def recursive_check(snake_case_ ,       snake_case_						):
																	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:
											__lowercase 					=   model_class(snake_case_						)
											model.to(snake_case_						)
											model.eval()

											__lowercase 					=   self._prepare_for_class(snake_case_ ,       snake_case_						)
											__lowercase 					=   self._prepare_for_class(snake_case_ ,       snake_case_						)
											check_equivalence(snake_case_ ,       snake_case_ ,       snake_case_						)

											__lowercase 					=   self._prepare_for_class(snake_case_ ,       snake_case_ ,       return_labels=snake_case_						)
											__lowercase 					=   self._prepare_for_class(snake_case_ ,       snake_case_ ,       return_labels=snake_case_						)
											check_equivalence(snake_case_ ,       snake_case_ ,       snake_case_						)

											__lowercase 					=   self._prepare_for_class(snake_case_ ,       snake_case_						)
											__lowercase 					=   self._prepare_for_class(snake_case_ ,       snake_case_						)
											check_equivalence(snake_case_ ,       snake_case_ ,       snake_case_ ,       {'''output_hidden_states''': True}						)

											__lowercase 					=   self._prepare_for_class(snake_case_ ,       snake_case_ ,       return_labels=snake_case_						)
											__lowercase 					=   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 	lowerCamelCase_       (				unittest.TestCase     ,     lowerCAmelCase__				):



					'''simple docstring'''







					__UpperCAmelCase            =				(MaskFormerSwinBackbone,) if is_torch_available() else ()
					__UpperCAmelCase            =				MaskFormerSwinConfig






					def A      (				self						)   ->			Tuple:







								'''simple docstring'''

								__lowercase 					=   MaskFormerSwinModelTester(self						)






					def A      (				self						)   ->			int:







								'''simple docstring'''

								__lowercase					, __lowercase 					=   self.model_tester.prepare_config_and_inputs_for_common()
								__lowercase 					=   inputs_dict['''pixel_values'''].shape[0]

								for backbone_class in self.all_model_classes:
											__lowercase 					=   backbone_class(snake_case_						)
											backbone.to(snake_case_						)
											backbone.eval()

											__lowercase 					=   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
											__lowercase 					=   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)
																	__lowercase					, __lowercase					, __lowercase 					=   hidden_state.shape
																	self.assertTrue((h_batch_size, h_n_channels) ,       (batch_size, n_channels)						)

            # Test output_attentions=True
											if self.has_attentions:
														__lowercase 					=   backbone(**snake_case_ ,       output_attentions=snake_case_						)
														self.assertIsNotNone(outputs.attentions						)






import numpy as np
import torch
from torch.utils.data import DataLoader

from accelerate.utils.dataclasses import DistributedType
class 	lowerCamelCase_       :



					'''simple docstring'''







					def __init__(				self ,       snake_case_=2 ,       snake_case_=3 ,       snake_case_=6_4 ,       snake_case_=None						)   ->			List[str]:







								'''simple docstring'''

								__lowercase 					=   np.random.default_rng(snake_case_						)
								__lowercase 					=   length
								__lowercase 					=   rng.normal(size=(length,)						).astype(np.floataa						)
								__lowercase 					=   a * self.x + b + rng.normal(scale=0.1 ,       size=(length,)						).astype(np.floataa						)






					def __len__(				self						)   ->			Dict:







								'''simple docstring'''

								return self.length






					def __getitem__(				self ,       snake_case_						)   ->			Union[str, Any]:







								'''simple docstring'''

								return {"x": self.x[i], "y": self.y[i]}




class 	lowerCamelCase_       (				torch.nn.Module				):



					'''simple docstring'''







					def __init__(				self ,       snake_case_=0 ,       snake_case_=0 ,       snake_case_=False						)   ->			Optional[Any]:







								'''simple docstring'''

								super().__init__()
								__lowercase 					=   torch.nn.Parameter(torch.tensor([2, 3]						).float()						)
								__lowercase 					=   torch.nn.Parameter(torch.tensor([2, 3]						).float()						)
								__lowercase 					=   True






					def A      (				self ,       snake_case_=None						)   ->			List[Any]:







								'''simple docstring'''

								if self.first_batch:
											print(F'Model dtype: {self.a.dtype}, {self.b.dtype}. Input dtype: {x.dtype}'						)
											__lowercase 					=   False
								return x * self.a[0] + self.b[0]




class 	lowerCamelCase_       (				torch.nn.Module				):



					'''simple docstring'''







					def __init__(				self ,       snake_case_=0 ,       snake_case_=0 ,       snake_case_=False						)   ->			List[str]:







								'''simple docstring'''

								super().__init__()
								__lowercase 					=   torch.nn.Parameter(torch.tensor(snake_case_						).float()						)
								__lowercase 					=   torch.nn.Parameter(torch.tensor(snake_case_						).float()						)
								__lowercase 					=   True






					def A      (				self ,       snake_case_=None						)   ->			str:







								'''simple docstring'''

								if self.first_batch:
											print(F'Model dtype: {self.a.dtype}, {self.b.dtype}. Input dtype: {x.dtype}'						)
											__lowercase 					=   False
								return x * self.a + self.b




def 		lowercase_  (     _UpperCamelCase       ,			_UpperCamelCase = 16  ):


			'''simple docstring'''







			from datasets import load_dataset
			from transformers import AutoTokenizer

			__lowercase 					=   AutoTokenizer.from_pretrained('''bert-base-cased'''  )
			__lowercase 					=   {'''train''': '''tests/test_samples/MRPC/train.csv''', '''validation''': '''tests/test_samples/MRPC/dev.csv'''}
			__lowercase 					=   load_dataset('''csv'''       ,			data_files=_UpperCamelCase  )
			__lowercase 					=   datasets['''train'''].unique('''label'''  )

			__lowercase 					=   {v: i for i, v in enumerate(_UpperCamelCase  )}

			def tokenize_function(_UpperCamelCase  ):
						# max_length=None => use the model max length (it's actually the default)
						__lowercase 					=   tokenizer(
						    examples['''sentence1''']       ,			examples['''sentence2''']       ,			truncation=_UpperCamelCase       ,			max_length=_UpperCamelCase       ,			padding='''max_length'''  )
						if "label" in examples:
									__lowercase 					=   [label_to_id[l] for l in examples['''label''']]
						return outputs

			# Apply the method we just defined to all the examples in all the splits of the dataset
			__lowercase 					=   datasets.map(
			    _UpperCamelCase       ,			batched=_UpperCamelCase       ,			remove_columns=['''sentence1''', '''sentence2''', '''label''']       ,			)

			def collate_fn(_UpperCamelCase  ):
						# On TPU it's best to pad everything to the same length or training will be very slow.
						if accelerator.distributed_type == DistributedType.TPU:
									return tokenizer.pad(_UpperCamelCase       ,			padding='''max_length'''       ,			max_length=1_28       ,			return_tensors='''pt'''  )
						return tokenizer.pad(_UpperCamelCase       ,			padding='''longest'''       ,			return_tensors='''pt'''  )

			# Instantiate dataloaders.
			__lowercase 					=   DataLoader(tokenized_datasets['''train''']       ,			shuffle=_UpperCamelCase       ,			collate_fn=_UpperCamelCase       ,			batch_size=2  )
			__lowercase 					=   DataLoader(tokenized_datasets['''validation''']       ,			shuffle=_UpperCamelCase       ,			collate_fn=_UpperCamelCase       ,			batch_size=1  )

			return train_dataloader, eval_dataloader


