






import inspect
import unittest

from transformers import DataaVecVisionConfig
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_torch_multi_gpu, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

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


if is_torch_available():
	import torch
	from torch import nn

	from transformers import (
	    MODEL_MAPPING,
	    DataaVecVisionForImageClassification,
	    DataaVecVisionForSemanticSegmentation,
	    DataaVecVisionModel,
	)
	from transformers.models.dataavec.modeling_dataavec_vision import DATA2VEC_VISION_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
	from PIL import Image

	from transformers import BeitImageProcessor

class        _SCREAMING_SNAKE_CASE			:





	def __init__(						self							:   List[str]					,    __lowerCamelCase							:   Optional[Any]					,    __lowerCamelCase							:   List[Any]=100					,    __lowerCamelCase							:   Optional[Any]=13					,    __lowerCamelCase							:   int=30					,    __lowerCamelCase							:   List[str]=2					,    __lowerCamelCase							:   Optional[int]=3					,    __lowerCamelCase							:   Any=True					,    __lowerCamelCase							:   Union[str, Any]=True					,    __lowerCamelCase							:   int=32					,    __lowerCamelCase							:   List[str]=4					,    __lowerCamelCase							:   Dict=4					,    __lowerCamelCase							:   List[Any]=37					,    __lowerCamelCase							:   Optional[int]="gelu"					,    __lowerCamelCase							:   Optional[int]=0.1					,    __lowerCamelCase							:   Tuple=0.1					,    __lowerCamelCase							:   int=10					,    __lowerCamelCase							:   Union[str, Any]=0.02					,    __lowerCamelCase							:   List[Any]=3					,    __lowerCamelCase							:   Union[str, Any]=None					,    __lowerCamelCase							:   Optional[int]=[0, 1, 2, 3]					,    ):
		UpperCamelCase	:Optional[Any]		=      parent
		UpperCamelCase	:Optional[Any]		=      100
		UpperCamelCase	:str		=      batch_size
		UpperCamelCase	:Optional[int]		=      image_size
		UpperCamelCase	:Any		=      patch_size
		UpperCamelCase	:Optional[Any]		=      num_channels
		UpperCamelCase	:Optional[int]		=      is_training
		UpperCamelCase	:int		=      use_labels
		UpperCamelCase	:Union[str, Any]		=      hidden_size
		UpperCamelCase	:Optional[Any]		=      num_hidden_layers
		UpperCamelCase	:str		=      num_attention_heads
		UpperCamelCase	:Any		=      intermediate_size
		UpperCamelCase	:Union[str, Any]		=      hidden_act
		UpperCamelCase	:Union[str, Any]		=      hidden_dropout_prob
		UpperCamelCase	:Tuple		=      attention_probs_dropout_prob
		UpperCamelCase	:Any		=      type_sequence_label_size
		UpperCamelCase	:Optional[Any]		=      initializer_range
		UpperCamelCase	:str		=      scope
		UpperCamelCase	:str		=      out_indices
		UpperCamelCase	:Dict		=      num_labels

		# in BeiT, the seq length equals the number of patches + 1 (we add 1 for the [CLS] token)
		UpperCamelCase	:Any		=      (image_size // patch_size) ** 2
		UpperCamelCase	:int		=      num_patches + 1





	def 		_A					(						self							:   int      ):
		UpperCamelCase	:Dict		=      floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]      )

		UpperCamelCase	:int		=      None
		UpperCamelCase	:Dict		=      None
		if self.use_labels:
			UpperCamelCase	:List[str]		=      ids_tensor([self.batch_size]					,    self.type_sequence_label_size      )
			UpperCamelCase	:Tuple		=      ids_tensor([self.batch_size, self.image_size, self.image_size]					,    self.num_labels      )

		UpperCamelCase	:Optional[Any]		=      self.get_config()

		return config, pixel_values, labels, pixel_labels





	def 		_A					(						self							:   int      ):
		return DataaVecVisionConfig(
		    vocab_size=self.vocab_size					,    image_size=self.image_size					,    patch_size=self.patch_size					,    num_channels=self.num_channels					,    hidden_size=self.hidden_size					,    num_hidden_layers=self.num_hidden_layers					,    num_attention_heads=self.num_attention_heads					,    intermediate_size=self.intermediate_size					,    hidden_act=self.hidden_act					,    hidden_dropout_prob=self.hidden_dropout_prob					,    attention_probs_dropout_prob=self.attention_probs_dropout_prob					,    is_decoder=__snake_case					,    initializer_range=self.initializer_range					,    out_indices=self.out_indices					,    )





	def 		_A					(						self							:   List[str]					,    __lowerCamelCase							:   List[Any]					,    __lowerCamelCase							:   Union[str, Any]					,    __lowerCamelCase							:   Any					,    __lowerCamelCase							:   Optional[int]      ):
		UpperCamelCase	:Optional[int]		=      DataaVecVisionModel(config=__snake_case      )
		model.to(__snake_case      )
		model.eval()
		UpperCamelCase	:List[Any]		=      model(__snake_case      )
		# expected sequence length = num_patches + 1 (we add 1 for the [CLS] token)
		UpperCamelCase	:Optional[Any]		=      (self.image_size // self.patch_size) ** 2
		self.parent.assertEqual(result.last_hidden_state.shape					,    (self.batch_size, num_patches + 1, self.hidden_size)      )





	def 		_A					(						self							:   List[Any]					,    __lowerCamelCase							:   List[str]					,    __lowerCamelCase							:   Dict					,    __lowerCamelCase							:   List[str]					,    __lowerCamelCase							:   int      ):
		UpperCamelCase	:str		=      self.type_sequence_label_size
		UpperCamelCase	:Union[str, Any]		=      DataaVecVisionForImageClassification(__snake_case      )
		model.to(__snake_case      )
		model.eval()
		UpperCamelCase	:str		=      model(__snake_case					,    labels=__snake_case      )
		self.parent.assertEqual(result.logits.shape					,    (self.batch_size, self.type_sequence_label_size)      )





	def 		_A					(						self							:   int					,    __lowerCamelCase							:   Any					,    __lowerCamelCase							:   Optional[Any]					,    __lowerCamelCase							:   Optional[Any]					,    __lowerCamelCase							:   Optional[int]      ):
		UpperCamelCase	:Union[str, Any]		=      self.num_labels
		UpperCamelCase	:Dict		=      DataaVecVisionForSemanticSegmentation(__snake_case      )
		model.to(__snake_case      )
		model.eval()
		UpperCamelCase	:Dict		=      model(__snake_case      )
		self.parent.assertEqual(
		    result.logits.shape					,    (self.batch_size, self.num_labels, self.image_size * 2, self.image_size * 2)      )
		UpperCamelCase	:Union[str, Any]		=      model(__snake_case					,    labels=__snake_case      )
		self.parent.assertEqual(
		    result.logits.shape					,    (self.batch_size, self.num_labels, self.image_size * 2, self.image_size * 2)      )





	def 		_A					(						self							:   Optional[Any]      ):
		UpperCamelCase	:List[str]		=      self.prepare_config_and_inputs()
		UpperCamelCase	:Union[str, Any]		=      config_and_inputs
		UpperCamelCase	:Any		=      {'pixel_values': pixel_values}
		return config, inputs_dict

@require_torch
class        _SCREAMING_SNAKE_CASE			(     _lowercase						,  _lowercase						,  unittest.TestCase ):
	snake_case__				:	List[str]	      =    (
	    (DataaVecVisionModel, DataaVecVisionForImageClassification, DataaVecVisionForSemanticSegmentation)
	    if is_torch_available()
	    else ()
	)
	snake_case__				:	Dict	      =    (
	    {
	        """feature-extraction""": DataaVecVisionModel,
	        """image-classification""": DataaVecVisionForImageClassification,
	        """image-segmentation""": DataaVecVisionForSemanticSegmentation,
	    }
	    if is_torch_available()
	    else {}
	)

	snake_case__				:	Tuple	      =    False
	snake_case__				:	Union[str, Any]	      =    False
	snake_case__				:	str	      =    False





	def 		_A					(						self							:   Optional[Any]      ):
		UpperCamelCase	:int		=      DataaVecVisionModelTester(self      )
		UpperCamelCase	:Optional[Any]		=      ConfigTester(
		    self					,    config_class=__snake_case					,    has_text_modality=__snake_case					,    hidden_size=37      )





	def 		_A					(						self							:   Optional[int]      ):
		self.config_tester.run_common_tests()





	def 		_A					(						self							:   Dict      ):
		# Data2VecVision does not use inputs_embeds
		pass





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





	def 		_A					(						self							:   Tuple      ):
		UpperCamelCase	:Optional[int]		=      self.model_tester.prepare_config_and_inputs_for_common()

		for model_class in self.all_model_classes:
			UpperCamelCase	:Optional[int]		=      model_class(__snake_case      )
			self.assertIsInstance(model.get_input_embeddings()					,    (nn.Module)      )
			UpperCamelCase	:Tuple		=      model.get_output_embeddings()
			self.assertTrue(x is None or isinstance(__snake_case					,    nn.Linear      )      )





	def 		_A					(						self							:   Any      ):
		UpperCamelCase	:Dict		=      self.model_tester.prepare_config_and_inputs_for_common()

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

			UpperCamelCase	:int		=      ['pixel_values']
			self.assertListEqual(arg_names[:1]					,    __snake_case      )





	def 		_A					(						self							:   Dict      ):
		UpperCamelCase	:Union[str, Any]		=      self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_model(*__snake_case      )





	def 		_A					(						self							:   str      ):
		UpperCamelCase	:Dict		=      self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_for_image_segmentation(*__snake_case      )





	def 		_A					(						self							:   str      ):
		if not self.model_tester.is_training:
			return

		UpperCamelCase	:Tuple		=      self.model_tester.prepare_config_and_inputs_for_common()
		UpperCamelCase	:Tuple		=      True

		for model_class in self.all_model_classes:
			if model_class in [*get_values(__snake_case      )]:
				continue

			UpperCamelCase	:Union[str, Any]		=      model_class(__snake_case      )
			model.to(__snake_case      )
			model.train()
			UpperCamelCase	:Optional[Any]		=      self._prepare_for_class(__snake_case					,    __snake_case					,    return_labels=__snake_case      )
			UpperCamelCase	:str		=      model(**__snake_case      ).loss
			loss.backward()





	def 		_A					(						self							:   str      ):
		UpperCamelCase	:Tuple		=      self.model_tester.prepare_config_and_inputs_for_common()
		if not self.model_tester.is_training:
			return

		UpperCamelCase	:Union[str, Any]		=      False
		UpperCamelCase	:Union[str, Any]		=      True

		for model_class in self.all_model_classes:
			if model_class in [*get_values(__snake_case      )] or not model_class.supports_gradient_checkpointing:
				continue
			# TODO: remove the following 3 lines once we have a MODEL_FOR_SEMANTIC_SEGMENTATION_MAPPING
			# this can then be incorporated into _prepare_for_class in test_modeling_common.py
			elif model_class.__name__ == "Data2VecVisionForSemanticSegmentation":
				UpperCamelCase	:Any		=      inputs_dict['pixel_values'].shape
				UpperCamelCase	:List[str]		=      torch.zeros(
				    [self.model_tester.batch_size, height, width]					,    device=__snake_case      ).long()
			UpperCamelCase	:Optional[Any]		=      model_class(__snake_case      )
			model.gradient_checkpointing_enable()
			model.to(__snake_case      )
			model.train()
			UpperCamelCase	:Dict		=      self._prepare_for_class(__snake_case					,    __snake_case					,    return_labels=__snake_case      )
			UpperCamelCase	:Optional[Any]		=      model(**__snake_case      ).loss
			loss.backward()





	def 		_A					(						self							:   Union[str, Any]      ):
		UpperCamelCase	:int		=      self.model_tester.prepare_config_and_inputs_for_common()

		UpperCamelCase	:Union[str, Any]		=      _config_zero_init(__snake_case      )
		for model_class in self.all_model_classes:
			UpperCamelCase	:Union[str, Any]		=      model_class(config=__snake_case      )
			for name, param in model.named_parameters():
				# we skip lambda parameters as these require special initial values
				# determined by config.layer_scale_init_value
				if "lambda" in name:
					continue
				if param.requires_grad:
					self.assertIn(
					    ((param.data.mean() * 1E9).round() / 1E9).item()					,    [0.0, 1.0]					,    msg=F"""Parameter {name} of model {model_class} seems not properly initialized"""					,    )





	def 		_A					(						self							:   Union[str, Any]					,    __lowerCamelCase							:   Dict					,    __lowerCamelCase							:   Union[str, Any]					,    __lowerCamelCase							:   Any					,    __lowerCamelCase							:   List[Any]=2E-4					,    __lowerCamelCase							:   Dict="outputs"					,    __lowerCamelCase							:   Dict=None      ):
		# We override with a slightly higher tol value, as semseg models tend to diverge a bit more
		super().check_pt_tf_outputs(__snake_case					,    __snake_case					,    __snake_case					,    __snake_case					,    __snake_case					,    __snake_case      )





	def 		_A					(						self							:   Optional[Any]      ):
		UpperCamelCase	:List[Any]		=      self.model_tester.prepare_config_and_inputs()
		self.model_tester.create_and_check_for_image_classification(*__snake_case      )





	@slow
	def 		_A					(						self							:   Optional[int]      ):
		for model_name in DATA2VEC_VISION_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
			UpperCamelCase	:Any		=      DataaVecVisionModel.from_pretrained(__snake_case      )
			self.assertIsNotNone(__snake_case      )

def   SCREAMING_SNAKE_CASE_				(				)						->     Tuple:






	"""simple docstring"""



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

@require_torch
@require_vision
class        _SCREAMING_SNAKE_CASE			(     unittest.TestCase ):





	@cached_property
	def 		_A					(						self							:   List[Any]      ):
		return (
		    BeitImageProcessor.from_pretrained("""facebook/data2vec-vision-base-ft1k"""      ) if is_vision_available() else None
		)





	@slow
	def 		_A					(						self							:   List[Any]      ):
		UpperCamelCase	:Union[str, Any]		=      DataaVecVisionForImageClassification.from_pretrained("""facebook/data2vec-vision-base-ft1k"""      ).to(
		    __snake_case      )

		UpperCamelCase	:Any		=      self.default_image_processor
		UpperCamelCase	:int		=      prepare_img()
		UpperCamelCase	:Tuple		=      image_processor(images=__snake_case					,    return_tensors="""pt"""      ).to(__snake_case      )

		# forward pass
		with torch.no_grad():
			UpperCamelCase	:int		=      model(**__snake_case      )
		UpperCamelCase	:str		=      outputs.logits

		# verify the logits
		UpperCamelCase	:List[str]		=      torch.Size((1, 1_000)      )
		self.assertEqual(logits.shape					,    __snake_case      )

		UpperCamelCase	:Tuple		=      torch.tensor([0.3277, -0.1395, 0.0911]      ).to(__snake_case      )

		self.assertTrue(torch.allclose(logits[0, :3]					,    __snake_case					,    atol=1E-4      )      )

		UpperCamelCase	:Union[str, Any]		=      [model.config.labelaid[i] for i in ['remote control, remote', 'tabby, tabby cat']]
		self.assertEqual(logits[0].topk(2      ).indices.cpu().tolist()					,    __snake_case      )



import posixpath
from pathlib import Path
from unittest.mock import patch

import pytest
from fsspec.implementations.local import AbstractFileSystem, LocalFileSystem, stringify_path
from fsspec.registry import _registry as _fsspec_registry







class 						lowercase     (					_lowercase				):


		"""simple docstring"""




		a__       							=	"mock"


		def __init__(					self						,							*__snake_case						,							__snake_case						,							**__snake_case):
			super().__init__()
			_UpperCamelCase :		str								=    LocalFileSystem(*__snake_case						,							**__snake_case)
			_UpperCamelCase :		str								=    Path(__snake_case).resolve().as_posix() + '/'


		def 		A__       (					self						,							__snake_case						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		int								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.mkdir(__snake_case						,							*__snake_case						,							**__snake_case)


		def 		A__       (					self						,							__snake_case						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		List[str]								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.makedirs(__snake_case						,							*__snake_case						,							**__snake_case)


		def 		A__       (					self						,							__snake_case):
			_UpperCamelCase :		Dict								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.rmdir(__snake_case)


		def 		A__       (					self						,							__snake_case						,							__snake_case=True						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		Union[str, Any]								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			_UpperCamelCase :		Tuple								=    self._fs.ls(__snake_case						,							detail=__snake_case						,							*__snake_case						,							**__snake_case)
			if detail:
				return [{**info, "name": info["name"][len(self.local_root_dir) :]} for info in out]
			else:
				return [name[len(self.local_root_dir) :] for name in out]


		def 		A__       (					self						,							__snake_case						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		Optional[Any]								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			_UpperCamelCase :		Optional[Any]								=    dict(self._fs.info(__snake_case						,							*__snake_case						,							**__snake_case))
			_UpperCamelCase :		Dict								=    out['name'][len(self.local_root_dir) :]
			return out


		def 		A__       (					self						,							__snake_case						,							__snake_case						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		str								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			_UpperCamelCase :		Tuple								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.cp_file(__snake_case						,							__snake_case						,							*__snake_case						,							**__snake_case)


		def 		A__       (					self						,							__snake_case						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		Dict								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.rm_file(__snake_case						,							*__snake_case						,							**__snake_case)


		def 		A__       (					self						,							__snake_case						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		int								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.rm(__snake_case						,							*__snake_case						,							**__snake_case)


		def 		A__       (					self						,							__snake_case						,							*__snake_case						,							**__snake_case):
			_UpperCamelCase :		Dict								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs._open(__snake_case						,							*__snake_case						,							**__snake_case)


		def 		A__       (					self						,							__snake_case):
			_UpperCamelCase :		Dict								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.created(__snake_case)


		def 		A__       (					self						,							__snake_case):
			_UpperCamelCase :		Dict								=    posixpath.join(self.local_root_dir						,							self._strip_protocol(__snake_case))
			return self._fs.modified(__snake_case)






		@classmethod
		def 		A__       (					cls						,							__snake_case):
			_UpperCamelCase :		List[Any]								=    stringify_path(__snake_case)
			if path.startswith('mock://'):
				_UpperCamelCase :		int								=    path[7:]
			return path







class 						lowercase     (					_lowercase				):


		"""simple docstring"""




		a__       							=	"tmp"
		a__       							=	None


		def __init__(					self						,							*__snake_case						,							**__snake_case):
			assert self.tmp_dir is not None, "TmpDirFileSystem.tmp_dir is not set"
			super().__init__(*__snake_case						,							**__snake_case						,							local_root_dir=self.tmp_dir						,							auto_mkdir=__snake_case)






		@classmethod
		def 		A__       (					cls						,							__snake_case):
			_UpperCamelCase :		Optional[int]								=    stringify_path(__snake_case)
			if path.startswith('tmp://'):
				_UpperCamelCase :		Optional[Any]								=    path[6:]
			return path






@pytest.fixture
def        lowerCamelCase_  (				)    ->					Dict:




	'''simple docstring'''




	_UpperCamelCase :		List[str]								=    MockFileSystem
	_UpperCamelCase :		Optional[int]								=    TmpDirFileSystem
	yield
	del _fsspec_registry["mock"]
	del _fsspec_registry["tmp"]






@pytest.fixture
def        lowerCamelCase_  (				UpperCAmelCase_				:  Any				,							UpperCAmelCase_				:  Optional[int]						)    ->					Optional[int]:




	'''simple docstring'''




	_UpperCamelCase :		str								=    tmp_path_factory.mktemp('mockfs'						)
	return MockFileSystem(local_root_dir=UpperCAmelCase_				,							auto_mkdir=UpperCAmelCase_						)







@pytest.fixture
def        lowerCamelCase_  (				UpperCAmelCase_				:  Tuple				,							UpperCAmelCase_				:  Union[str, Any]						)    ->					int:




	'''simple docstring'''




	_UpperCamelCase :		Dict								=    tmp_path_factory.mktemp('tmpfs'						)
	with patch.object(UpperCAmelCase_				,							'tmp_dir'				,							UpperCAmelCase_						):
		yield TmpDirFileSystem()




