
import collections
import inspect
import unittest

from transformers import DinatConfig
from transformers.testing_utils import require_natten, require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_backbone_common import BackboneTesterMixin
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 DinatBackbone, DinatForImageClassification, DinatModel
  from transformers.models.dinat.modeling_dinat import DINAT_PRETRAINED_MODEL_ARCHIVE_LIST

if is_vision_available():
  from PIL import Image

  from transformers import AutoImageProcessor

class 		__magic_name__				:


  """simple docstring"""





  def __init__(					self	,		a__	,		a__=13	,		a__=64	,		a__=4	,		a__=3	,		a__=16	,		a__=[1, 2, 1]	,		a__=[2, 4, 8]	,		a__=3	,		a__=[[3], [1, 2], [1]]	,		a__=2.0	,		a__=True	,		a__=0.0	,		a__=0.0	,		a__=0.1	,		a__="gelu"	,		a__=True	,		a__=0.02	,		a__=1E-5	,		a__=True	,		a__=None	,		a__=True	,		a__=10	,		a__=["stage1", "stage2"]	,		a__=[1, 2]	,		):
    _lowerCamelCase      =						parent
    _lowerCamelCase      =						batch_size
    _lowerCamelCase      =						image_size
    _lowerCamelCase      =						patch_size
    _lowerCamelCase      =						num_channels
    _lowerCamelCase      =						embed_dim
    _lowerCamelCase      =						depths
    _lowerCamelCase      =						num_heads
    _lowerCamelCase      =						kernel_size
    _lowerCamelCase      =						dilations
    _lowerCamelCase      =						mlp_ratio
    _lowerCamelCase      =						qkv_bias
    _lowerCamelCase      =						hidden_dropout_prob
    _lowerCamelCase      =						attention_probs_dropout_prob
    _lowerCamelCase      =						drop_path_rate
    _lowerCamelCase      =						hidden_act
    _lowerCamelCase      =						patch_norm
    _lowerCamelCase      =						layer_norm_eps
    _lowerCamelCase      =						initializer_range
    _lowerCamelCase      =						is_training
    _lowerCamelCase      =						scope
    _lowerCamelCase      =						use_labels
    _lowerCamelCase      =						num_labels
    _lowerCamelCase      =						out_features
    _lowerCamelCase      =						out_indices


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase      =						floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]		)

    _lowerCamelCase      =						None
    if self.use_labels:
      _lowerCamelCase      =						ids_tensor([self.batch_size]	,		self.num_labels		)

    _lowerCamelCase      =						self.get_config()

    return config, pixel_values, labels


  def     _UpperCAmelCase  (					self		):
    return DinatConfig(
        num_labels=self.num_labels	,		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	,		kernel_size=self.kernel_size	,		dilations=self.dilations	,		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	,		patch_norm=self.patch_norm	,		layer_norm_eps=self.layer_norm_eps	,		initializer_range=self.initializer_range	,		out_features=self.out_features	,		out_indices=self.out_indices	,		)


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__		):
    _lowerCamelCase      =						DinatModel(config=a__		)
    model.to(a__		)
    model.eval()
    _lowerCamelCase      =						model(a__		)

    _lowerCamelCase      =						_lowerCamelCase      =						(config.image_size // config.patch_size) // (2 ** (len(config.depths		) - 1))
    _lowerCamelCase      =						int(config.embed_dim * 2 ** (len(config.depths		) - 1)		)

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


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__		):
    _lowerCamelCase      =						DinatForImageClassification(a__		)
    model.to(a__		)
    model.eval()
    _lowerCamelCase      =						model(a__	,		labels=a__		)
    self.parent.assertEqual(result.logits.shape	,		(self.batch_size, self.num_labels)		)

    # test greyscale images
    _lowerCamelCase      =						1
    _lowerCamelCase      =						DinatForImageClassification(a__		)
    model.to(a__		)
    model.eval()

    _lowerCamelCase      =						floats_tensor([self.batch_size, 1, self.image_size, self.image_size]		)
    _lowerCamelCase      =						model(a__		)
    self.parent.assertEqual(result.logits.shape	,		(self.batch_size, self.num_labels)		)


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__		):
    _lowerCamelCase      =						DinatBackbone(config=a__		)
    model.to(a__		)
    model.eval()
    _lowerCamelCase      =						model(a__		)

    # verify hidden states
    self.parent.assertEqual(len(result.feature_maps		)	,		len(config.out_features		)		)
    self.parent.assertListEqual(list(result.feature_maps[0].shape		)	,		[self.batch_size, model.channels[0], 16, 16]		)

    # verify channels
    self.parent.assertEqual(len(model.channels		)	,		len(config.out_features		)		)

    # verify backbone works with out_features=None
    _lowerCamelCase      =						None
    _lowerCamelCase      =						DinatBackbone(config=a__		)
    model.to(a__		)
    model.eval()
    _lowerCamelCase      =						model(a__		)

    # verify feature maps
    self.parent.assertEqual(len(result.feature_maps		)	,		1		)
    self.parent.assertListEqual(list(result.feature_maps[0].shape		)	,		[self.batch_size, model.channels[-1], 4, 4]		)

    # verify channels
    self.parent.assertEqual(len(model.channels		)	,		1		)


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







@require_natten
@require_torch
class 		__magic_name__				( lowercase_						,lowercase_						,unittest.TestCase					):


  """simple docstring"""





  _UpperCamelCase						 = (
      (
          DinatModel,
          DinatForImageClassification,
          DinatBackbone,
      )
      if is_torch_available()
      else ()
  )
  _UpperCamelCase						 = (
      {"feature-extraction": DinatModel, "image-classification": DinatForImageClassification}
      if is_torch_available()
      else {}
  )
  _UpperCamelCase						 = False

  _UpperCamelCase						 = False
  _UpperCamelCase						 = False
  _UpperCamelCase						 = False
  _UpperCamelCase						 = False


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase      =						DinatModelTester(self		)
    _lowerCamelCase      =						ConfigTester(self	,		config_class=a__	,		embed_dim=37		)


  def     _UpperCAmelCase  (					self		):
    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     _UpperCAmelCase  (					self		):
    return


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase      =						self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_model(*a__		)


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase      =						self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_for_image_classification(*a__		)


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase      =						self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_backbone(*a__		)


  @unittest.skip(reason='''Dinat does not use inputs_embeds'''		)
  def     _UpperCAmelCase  (					self		):
    pass


  @unittest.skip(reason='''Dinat does not use feedforward chunking'''		)
  def     _UpperCAmelCase  (					self		):
    pass


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase    ,    _lowerCamelCase      =						self.model_tester.prepare_config_and_inputs_for_common()

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


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase    ,    _lowerCamelCase      =						self.model_tester.prepare_config_and_inputs_for_common()

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

      _lowerCamelCase      =						['''pixel_values''']
      self.assertListEqual(arg_names[:1]	,		a__		)


  def     _UpperCAmelCase  (					self		):
    self.skipTest('''Dinat\'s attention operation is handled entirely by NATTEN.'''		)


  def     _UpperCAmelCase  (					self	,		a__	,		a__	,		a__	,		a__		):
    _lowerCamelCase      =						model_class(a__		)
    model.to(a__		)
    model.eval()

    with torch.no_grad():
      _lowerCamelCase      =						model(**self._prepare_for_class(a__	,		a__		)		)

    _lowerCamelCase      =						outputs.hidden_states

    _lowerCamelCase      =						getattr(
        self.model_tester	,		'''expected_num_hidden_layers'''	,		len(self.model_tester.depths		) + 1		)
    self.assertEqual(len(a__		)	,		a__		)

    # Dinat has a different seq_length
    _lowerCamelCase      =						(
        config.patch_size
        if isinstance(config.patch_size	,		collections.abc.Iterable		)
        else (config.patch_size, config.patch_size)
    )

    _lowerCamelCase      =						image_size[0] // patch_size[0]
    _lowerCamelCase      =						image_size[1] // patch_size[1]

    self.assertListEqual(
        list(hidden_states[0].shape[-3:]		)	,		[height, width, self.model_tester.embed_dim]	,		)

    if model_class.__name__ != "DinatBackbone":
      _lowerCamelCase      =						outputs.reshaped_hidden_states
      self.assertEqual(len(a__		)	,		a__		)

      _lowerCamelCase    ,    _lowerCamelCase    ,    _lowerCamelCase    ,    _lowerCamelCase      =						reshaped_hidden_states[0].shape
      _lowerCamelCase      =						(
          reshaped_hidden_states[0].view(a__	,		a__	,		a__	,		a__		).permute(0	,		2	,		3	,		1		)
      )
      self.assertListEqual(
          list(reshaped_hidden_states.shape[-3:]		)	,		[height, width, self.model_tester.embed_dim]	,		)


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase    ,    _lowerCamelCase      =						self.model_tester.prepare_config_and_inputs_for_common()

    _lowerCamelCase      =						(
        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:
      _lowerCamelCase      =						True
      self.check_hidden_states_output(a__	,		a__	,		a__	,		a__		)

      # check that output_hidden_states also work using config
      del inputs_dict["output_hidden_states"]
      _lowerCamelCase      =						True

      self.check_hidden_states_output(a__	,		a__	,		a__	,		a__		)


  @slow
  def     _UpperCAmelCase  (					self		):
    for model_name in DINAT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
      _lowerCamelCase      =						DinatModel.from_pretrained(a__		)
      self.assertIsNotNone(a__		)


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase    ,    _lowerCamelCase      =						self.model_tester.prepare_config_and_inputs_for_common()

    _lowerCamelCase      =						_config_zero_init(a__		)
    for model_class in self.all_model_classes:
      _lowerCamelCase      =						model_class(config=a__		)
      for name, param in model.named_parameters():
        if "embeddings" not in name and 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'''	,		)







@require_natten
@require_vision
@require_torch
class 		__magic_name__				( unittest.TestCase					):


  """simple docstring"""





  @cached_property
  def     _UpperCAmelCase  (					self		):
    return AutoImageProcessor.from_pretrained('''shi-labs/dinat-mini-in1k-224'''		) if is_vision_available() else None


  @slow
  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase      =						DinatForImageClassification.from_pretrained('''shi-labs/dinat-mini-in1k-224'''		).to(a__		)
    _lowerCamelCase      =						self.default_image_processor

    _lowerCamelCase      =						Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''		)
    _lowerCamelCase      =						image_processor(images=a__	,		return_tensors='''pt'''		).to(a__		)

    # forward pass
    with torch.no_grad():
      _lowerCamelCase      =						model(**a__		)

    # verify the logits
    _lowerCamelCase      =						torch.Size((1, 10_00)		)
    self.assertEqual(outputs.logits.shape	,		a__		)
    _lowerCamelCase      =						torch.tensor([-0.1545, -0.7667, 0.4642]		).to(a__		)
    self.assertTrue(torch.allclose(outputs.logits[0, :3]	,		a__	,		atol=1E-4		)		)







@require_torch
@require_natten
class 		__magic_name__				( unittest.TestCase						,lowercase_					):


  """simple docstring"""





  _UpperCamelCase						 = (DinatBackbone,) if is_torch_available() else ()
  _UpperCamelCase						 = DinatConfig


  def     _UpperCAmelCase  (					self		):
    _lowerCamelCase      =						DinatModelTester(self		)






import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
  import torch_xla.core.xla_model as xm  # noqa: F401

  _UpperCAmelCase    			=					True
except ImportError:
  _UpperCAmelCase    			=					False


# Cache this result has it's a C FFI call which can be pretty time-consuming
_UpperCAmelCase    			=					torch.distributed.is_available()


def 		_lowerCamelCase       (   _a  ):
  """simple docstring"""




  _lowerCamelCase      =						importlib.util.find_spec(_a  ) is not None
  if package_exists:
    try:
      _lowerCamelCase      =						importlib.metadata.metadata(_a  )
      return True
    except importlib.metadata.PackageNotFoundError:
      return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _torch_distributed_available


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  try:
    pass
  except ImportError:
    print(
        '''Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not'''
        ''' detected. If you see \"ValueError: Invalid backend: \'ccl\'\" error, please install Intel(R) oneCCL'''
        ''' Bindings for PyTorch*.'''  )
  return (
      importlib.util.find_spec('''torch_ccl'''  ) is not None
      or importlib.util.find_spec('''oneccl_bindings_for_pytorch'''  ) is not None
  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return importlib.metadata.version('''oneccl_bind_pt'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''transformer_engine'''  )


@lru_cache
def 		_lowerCamelCase       (   _a=True  ):
  """simple docstring"""




  if torch.cuda.is_available():
    return False
  if _tpu_available and check_device:
    try:
      # Will raise a RuntimeError if no XLA configuration is found
      _lowerCamelCase      =						xm.xla_device()
      return True
    except RuntimeError:
      return False
  return _tpu_available


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''deepspeed'''  )


def 		_lowerCamelCase       (   _a=False  ):
  """simple docstring"""




  if is_tpu_available():
    return not ignore_tpu
  if torch.cuda.is_available():
    return torch.cuda.is_bfaa_supported()
  if is_npu_available():
    return False
  return True


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  _lowerCamelCase      =						_is_package_available('''bitsandbytes'''  )
  if package_exists:
    _lowerCamelCase      =						version.parse(importlib.metadata.version('''bitsandbytes'''  )  )
    return compare_versions(_a			,   '''>='''			,   '''0.39.0'''  )
  return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  _lowerCamelCase      =						_is_package_available('''bitsandbytes'''  )
  if package_exists:
    _lowerCamelCase      =						version.parse(importlib.metadata.version('''bitsandbytes'''  )  )
    return compare_versions(_a			,   '''>='''			,   '''0.37.2'''  )
  return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''bitsandbytes'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  if strtobool(os.environ.get('''ACCELERATE_USE_MEGATRON_LM'''			,   '''False'''  )  ) == 1:
    _lowerCamelCase      =						importlib.util.find_spec('''megatron'''  ) is not None
    if package_exists:
      try:
        _lowerCamelCase      =						parse(importlib.metadata.version('''megatron-lm'''  )  )
        return compare_versions(_a			,   '''>='''			,   '''2.2.0'''  )
      except Exception as e:
        warnings.warn(F'''Parse Megatron version failed. Exception:{e}'''  )
        return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''safetensors'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''transformers'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''datasets'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''aim'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''tensorboard'''  ) or _is_package_available('''tensorboardX'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''wandb'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''comet_ml'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''boto3'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  if _is_package_available('''rich'''  ):
    if "ACCELERATE_DISABLE_RICH" in os.environ:
      warnings.warn(
          '''`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`.'''  )
      return not parse_flag_from_env('''ACCELERATE_DISABLE_RICH'''			,   _a  )
    return parse_flag_from_env('''ACCELERATE_ENABLE_RICH'''			,   _a  )
  return False


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''sagemaker'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''tqdm'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return _is_package_available('''mlflow'''  )


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  return is_torch_version('''>='''			,   '''1.12'''  ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()


def 		_lowerCamelCase       (   ):
  """simple docstring"""




  def get_major_and_minor_from_version(_a  ):
    return str(version.parse(_a  ).major  ) + "." + str(version.parse(_a  ).minor  )

  _lowerCamelCase      =						importlib.metadata.version('''torch'''  )
  if importlib.util.find_spec('''intel_extension_for_pytorch'''  ) is None:
    return False
  _lowerCamelCase      =						'''N/A'''
  try:
    _lowerCamelCase      =						importlib.metadata.version('''intel_extension_for_pytorch'''  )
  except importlib.metadata.PackageNotFoundError:
    return False
  _lowerCamelCase      =						get_major_and_minor_from_version(_torch_version  )
  _lowerCamelCase      =						get_major_and_minor_from_version(_ipex_version  )
  if torch_major_and_minor != ipex_major_and_minor:
    warnings.warn(
        F'''Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'''
        F''' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'''  )
    return False
  return True


@lru_cache
def 		_lowerCamelCase       (   _a=False  ):
  """simple docstring"""




  if importlib.util.find_spec('''torch'''  ) is None or importlib.util.find_spec('''torch_npu'''  ) is None:
    return False

  import torch
  import torch_npu  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no NPU is found
      _lowerCamelCase      =						torch.npu.device_count()
      return torch.npu.is_available()
    except RuntimeError:
      return False
  return hasattr(_a			,   '''npu'''  ) and torch.npu.is_available()






@lru_cache
def 		_lowerCamelCase       (   _a=False  ):
  """simple docstring"""




  if not parse_flag_from_env('''ACCELERATE_USE_XPU'''			,   default=_a  ):
    return False
  if is_ipex_available():
    import torch

    if is_torch_version('''<='''			,   '''1.12'''  ):
      return False
  else:
    return False

  import intel_extension_for_pytorch  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no XPU  is found
      _lowerCamelCase      =						torch.xpu.device_count()
      return torch.xpu.is_available()
    except RuntimeError:
      return False
  return hasattr(_a			,   '''xpu'''  ) and torch.xpu.is_available()




