from __future__ import annotations

import inspect
import unittest
from typing import List, Tuple

from transformers import ConvNextConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
    import tensorflow as tf

    from transformers import TFConvNextForImageClassification, TFConvNextModel


if is_vision_available():
    from PIL import Image

    from transformers import ConvNextImageProcessor






class     __UpperCAmelCase:






 """simple docstring"""




 def __init__(    self ,       __magic_name__ ,       __magic_name__=13 ,       __magic_name__=32 ,       __magic_name__=3 ,       __magic_name__=4 ,       __magic_name__=[10, 20, 30, 40] ,       __magic_name__=[2, 2, 3, 2] ,       __magic_name__=True ,       __magic_name__=True ,       __magic_name__=37 ,       __magic_name__="gelu" ,       __magic_name__=10 ,       __magic_name__=0.02 ,       __magic_name__=3 ,       __magic_name__=None ,       ):

        """simple docstring"""

        A_							:       List[str]	      =				parent
        A_							:       Union[str, Any]	      =				batch_size
        A_							:       Optional[Any]	      =				image_size
        A_							:       List[str]	      =				num_channels
        A_							:       int	      =				num_stages
        A_							:       Union[str, Any]	      =				hidden_sizes
        A_							:       Union[str, Any]	      =				depths
        A_							:       Union[str, Any]	      =				is_training
        A_							:       Union[str, Any]	      =				use_labels
        A_							:       List[Any]	      =				intermediate_size
        A_							:       List[str]	      =				hidden_act
        A_							:       Any	      =				type_sequence_label_size
        A_							:       Any	      =				initializer_range
        A_							:       Any	      =				scope
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       str	      =				floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]      )

        A_							:       str	      =				None
        if self.use_labels:
               A_							:       Any	      =				ids_tensor([self.batch_size] ,       self.type_sequence_label_size      )

        A_							:       List[str]	      =				self.get_config()

        return config, pixel_values, labels
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        return ConvNextConfig(
            num_channels=self.num_channels ,       hidden_sizes=self.hidden_sizes ,       depths=self.depths ,       num_stages=self.num_stages ,       hidden_act=self.hidden_act ,       is_decoder=__magic_name__ ,       initializer_range=self.initializer_range ,       )
 def  UpperCAmelCase   (    self ,       __magic_name__ ,       __magic_name__ ,       __magic_name__      ):

        """simple docstring"""

        A_							:       int	      =				TFConvNextModel(config=__magic_name__      )
        A_							:       Union[str, Any]	      =				model(__magic_name__ ,       training=__magic_name__      )
        # expected last hidden states: B, C, H // 32, W // 32
        self.parent.assertEqual(
            result.last_hidden_state.shape ,       (self.batch_size, self.hidden_sizes[-1], self.image_size // 32, self.image_size // 32) ,       )
 def  UpperCAmelCase   (    self ,       __magic_name__ ,       __magic_name__ ,       __magic_name__      ):

        """simple docstring"""

        A_							:       Any	      =				self.type_sequence_label_size
        A_							:       Tuple	      =				TFConvNextForImageClassification(__magic_name__      )
        A_							:       List[str]	      =				model(__magic_name__ ,       labels=__magic_name__ ,       training=__magic_name__      )
        self.parent.assertEqual(result.logits.shape ,       (self.batch_size, self.type_sequence_label_size)      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       Dict	      =				self.prepare_config_and_inputs()
        A_      ,							A_      ,							A_							:       List[str]	      =				config_and_inputs
        A_							:       int	      =				{'''pixel_values''': pixel_values}
        return config, inputs_dict






@require_tf
class     __UpperCAmelCase(    A__      ,	A__      ,	unittest.TestCase				):






 """simple docstring"""




 __magic_name__          =   (TFConvNextModel, TFConvNextForImageClassification) if is_tf_available() else ()
 __magic_name__          =   (
     {"""feature-extraction""": TFConvNextModel, """image-classification""": TFConvNextForImageClassification}
     if is_tf_available()
     else {}
 )

 __magic_name__          =   False
 __magic_name__          =   False
 __magic_name__          =   False
 __magic_name__          =   False
 __magic_name__          =   False
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       List[str]	      =				TFConvNextModelTester(self      )
        A_							:       Optional[int]	      =				ConfigTester(
            self ,       config_class=__magic_name__ ,       has_text_modality=__magic_name__ ,       hidden_size=37 ,       )
 @unittest.skip(reason='''ConvNext does not use inputs_embeds'''      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        pass
 @unittest.skipIf(
     not is_tf_available() or len(tf.config.list_physical_devices('''GPU'''      )      ) == 0 ,       reason='''TF does not support backprop for grouped convolutions on CPU.''' ,       )
 @slow
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        super().test_keras_fit()
 @unittest.skip(reason='''ConvNext does not support input and output embeddings'''      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        pass
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

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

        for model_class in self.all_model_classes:
               A_							:       Any	      =				model_class(__magic_name__      )
               A_							:       str	      =				inspect.signature(model.call      )
               # signature.parameters is an OrderedDict => so arg_names order is deterministic
               A_							:       Any	      =				[*signature.parameters.keys()]

               A_							:       Any	      =				['''pixel_values''']
               self.assertListEqual(arg_names[:1] ,       __magic_name__      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       List[str]	      =				self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*__magic_name__      )
 @unittest.skipIf(
     not is_tf_available() or len(tf.config.list_physical_devices('''GPU'''      )      ) == 0 ,       reason='''TF does not support backprop for grouped convolutions on CPU.''' ,       )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        super().test_dataset_conversion()
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        def check_hidden_states_output(__magic_name__ ,       __magic_name__ ,       __magic_name__      ):
               A_							:       int	      =				model_class(__magic_name__      )

               A_							:       Any	      =				model(**self._prepare_for_class(__magic_name__ ,       __magic_name__      )      )
               A_							:       List[str]	      =				outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

               A_							:       List[Any]	      =				self.model_tester.num_stages
               self.assertEqual(len(__magic_name__      ) ,       expected_num_stages + 1      )

               # ConvNext's feature maps are of shape (batch_size, num_channels, height, width)
               self.assertListEqual(
                   list(hidden_states[0].shape[-2:]      ) ,       [self.model_tester.image_size // 4, self.model_tester.image_size // 4] ,       )

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

        for model_class in self.all_model_classes:
               A_							:       Union[str, Any]	      =				True
               check_hidden_states_output(__magic_name__ ,       __magic_name__ ,       __magic_name__      )

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

               check_hidden_states_output(__magic_name__ ,       __magic_name__ ,       __magic_name__      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

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

        def check_equivalence(__magic_name__ ,       __magic_name__ ,       __magic_name__ ,       __magic_name__={}      ):
               A_							:       Optional[Any]	      =				model(__magic_name__ ,       return_dict=__magic_name__ ,       **__magic_name__      )
               A_							:       Any	      =				model(__magic_name__ ,       return_dict=__magic_name__ ,       **__magic_name__      ).to_tuple()

               def recursive_check(__magic_name__ ,       __magic_name__      ):
                      if isinstance(__magic_name__ ,       (List, Tuple)      ):
                             for tuple_iterable_value, dict_iterable_value in zip(__magic_name__ ,       __magic_name__      ):
                                    recursive_check(__magic_name__ ,       __magic_name__      )
                      elif tuple_object is None:
                             return
                      else:
                             self.assertTrue(
                                 all(tf.equal(__magic_name__ ,       __magic_name__      )      ) ,       msg=(
                                     '''Tuple and dict output are not equal. Difference:'''
                                     f""" {tf.math.reduce_max(tf.abs(tuple_object - dict_object      )      )}"""
                                 ) ,       )

                      recursive_check(__magic_name__ ,       __magic_name__      )

        for model_class in self.all_model_classes:
               A_							:       Optional[int]	      =				model_class(__magic_name__      )

               A_							:       Tuple	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__      )
               A_							:       Tuple	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__      )
               check_equivalence(__magic_name__ ,       __magic_name__ ,       __magic_name__      )

               A_							:       Any	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__ ,       return_labels=__magic_name__      )
               A_							:       List[str]	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__ ,       return_labels=__magic_name__      )
               check_equivalence(__magic_name__ ,       __magic_name__ ,       __magic_name__      )

               A_							:       Any	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__      )
               A_							:       List[str]	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__      )
               check_equivalence(__magic_name__ ,       __magic_name__ ,       __magic_name__ ,       {'''output_hidden_states''': True}      )

               A_							:       List[str]	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__ ,       return_labels=__magic_name__      )
               A_							:       Dict	      =				self._prepare_for_class(__magic_name__ ,       __magic_name__ ,       return_labels=__magic_name__      )
               check_equivalence(__magic_name__ ,       __magic_name__ ,       __magic_name__ ,       {'''output_hidden_states''': True}      )
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       Optional[int]	      =				self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_image_classification(*__magic_name__      )
 @slow
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       Optional[Any]	      =				TFConvNextModel.from_pretrained('''facebook/convnext-tiny-224'''      )
        self.assertIsNotNone(__magic_name__      )


def       a__      (					)		->	List[str]:
       A_							:       Union[str, Any]	      =				Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''							)
       return image






@require_tf
@require_vision
class     __UpperCAmelCase(    unittest.TestCase				):






 """simple docstring"""




 @cached_property
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        return ConvNextImageProcessor.from_pretrained('''facebook/convnext-tiny-224'''      ) if is_vision_available() else None
 @slow
 def  UpperCAmelCase   (    self      ):

        """simple docstring"""

        A_							:       List[str]	      =				TFConvNextForImageClassification.from_pretrained('''facebook/convnext-tiny-224'''      )

        A_							:       List[Any]	      =				self.default_image_processor
        A_							:       List[str]	      =				prepare_img()
        A_							:       str	      =				image_processor(images=__magic_name__ ,       return_tensors='''tf'''      )

        # forward pass
        A_							:       Union[str, Any]	      =				model(**__magic_name__      )

        # verify the logits
        A_							:       str	      =				tf.TensorShape((1, 1000)      )
        self.assertEqual(outputs.logits.shape ,       __magic_name__      )

        A_							:       int	      =				tf.constant([-0.02_60, -0.47_39, 0.19_11]      )

        tf.debugging.assert_near(outputs.logits[0, :3] ,       __magic_name__ ,       atol=1e-4      )



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

    _lowerCAmelCase        =       True
except ImportError:
    _lowerCAmelCase        =       False


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


def       a__      (					a							)		->	Optional[Any]:
       # Check we're not importing a "pkg_name" directory somewhere but the actual library by trying to grab the version
       A_							:       str	      =				importlib.util.find_spec(a							) is not None
       if package_exists:
              try:
                     A_							:       str	      =				importlib.metadata.metadata(a							)
                     return True
              except importlib.metadata.PackageNotFoundError:
                     return False


def       a__      (					)		->	bool:
       return _torch_distributed_available


def       a__      (					)		->	int:
       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       a__      (					)		->	List[str]:
       return importlib.metadata.version('''oneccl_bind_pt'''							)


def       a__      (					)		->	List[str]:
       return _is_package_available('''transformer_engine'''							)


@lru_cache
def       a__      (					a=True							)		->	Union[str, Any]:
       # Due to bugs on the amp series GPUs, we disable torch-xla on them
       if torch.cuda.is_available():
              return False
       if _tpu_available and check_device:
              try:
                     # Will raise a RuntimeError if no XLA configuration is found
                     A_							:       Tuple	      =				xm.xla_device()
                     return True
              except RuntimeError:
                     return False
       return _tpu_available


def       a__      (					)		->	Dict:
       return _is_package_available('''deepspeed'''							)


def       a__      (					a=False							)		->	Optional[int]:
       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       a__      (					)		->	Optional[Any]:
       A_							:       Optional[int]	      =				_is_package_available('''bitsandbytes'''							)
       if package_exists:
              A_							:       Any	      =				version.parse(importlib.metadata.version('''bitsandbytes'''							)							)
              return compare_versions(a  ,   '''>='''  ,   '''0.39.0'''							)
       return False


def       a__      (					)		->	Any:
       A_							:       List[str]	      =				_is_package_available('''bitsandbytes'''							)
       if package_exists:
              A_							:       Tuple	      =				version.parse(importlib.metadata.version('''bitsandbytes'''							)							)
              return compare_versions(a  ,   '''>='''  ,   '''0.37.2'''							)
       return False


def       a__      (					)		->	List[str]:
       return _is_package_available('''bitsandbytes'''							)


def       a__      (					)		->	List[str]:
       if strtobool(os.environ.get('''ACCELERATE_USE_MEGATRON_LM'''  ,   '''False'''							)							) == 1:
              A_							:       str	      =				importlib.util.find_spec('''megatron'''							) is not None
              if package_exists:
                     try:
                            A_							:       Dict	      =				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       a__      (					)		->	Tuple:
       return _is_package_available('''safetensors'''							)


def       a__      (					)		->	Tuple:
       return _is_package_available('''transformers'''							)


def       a__      (					)		->	int:
       return _is_package_available('''datasets'''							)


def       a__      (					)		->	Optional[int]:
       return _is_package_available('''aim'''							)


def       a__      (					)		->	Optional[Any]:
       return _is_package_available('''tensorboard'''							) or _is_package_available('''tensorboardX'''							)


def       a__      (					)		->	Optional[int]:
       return _is_package_available('''wandb'''							)


def       a__      (					)		->	Tuple:
       return _is_package_available('''comet_ml'''							)


def       a__      (					)		->	Tuple:
       return _is_package_available('''boto3'''							)


def       a__      (					)		->	Tuple:
       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       a__      (					)		->	Any:
       return _is_package_available('''sagemaker'''							)


def       a__      (					)		->	Union[str, Any]:
       return _is_package_available('''tqdm'''							)


def       a__      (					)		->	List[str]:
       return _is_package_available('''mlflow'''							)


def       a__      (					)		->	Tuple:
       return is_torch_version('''>='''  ,   '''1.12'''							) and torch.backends.mps.is_available() and torch.backends.mps.is_built()


def       a__      (					)		->	Tuple:
       def get_major_and_minor_from_version(a							):
              return str(version.parse(a							).major							) + "." + str(version.parse(a							).minor							)

       A_							:       Tuple	      =				importlib.metadata.version('''torch'''							)
       if importlib.util.find_spec('''intel_extension_for_pytorch'''							) is None:
              return False
       A_							:       Optional[int]	      =				'''N/A'''
       try:
              A_							:       List[Any]	      =				importlib.metadata.version('''intel_extension_for_pytorch'''							)
       except importlib.metadata.PackageNotFoundError:
              return False
       A_							:       Tuple	      =				get_major_and_minor_from_version(_torch_version							)
       A_							:       int	      =				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       a__      (					a=False							)		->	Optional[int]:
       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
                     A_							:       Dict	      =				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       a__      (					a=False							)		->	Optional[Any]:
       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
                     A_							:       Tuple	      =				torch.xpu.device_count()
                     return torch.xpu.is_available()
              except RuntimeError:
                     return False
       return hasattr(a  ,   '''xpu'''							) and torch.xpu.is_available()


