
import inspect
import re
from hashlib import shaaaa
from typing import Dict, List

from .arrow import arrow
from .audiofolder import audiofolder
from .csv import csv
from .imagefolder import imagefolder
from .json import json
from .pandas import pandas
from .parquet import parquet
from .sql import sql  # noqa F401
from .text import text


def      UpperCamelCase   (       __lowerCamelCase	:					List[str]       ):
       snake_case			:  int     							= []
       for line in lines:
              snake_case			:  List[Any]     							= re.sub(r"#.*"							,   ""							,   __lowerCamelCase       )  # remove comments
              if line:
                     filtered_lines.append(__lowerCamelCase       )
       snake_case			:  str     							= "\n".join(__lowerCamelCase       )

       # Make a hash from all this code
       snake_case			:  str     							= full_str.encode("utf-8"       )
       return shaaaa(__lowerCamelCase       ).hexdigest()


# get importable module names and hash for caching
__lowerCamelCase				=			{
    """csv""": (csv.__name__, _hash_python_lines(inspect.getsource(csv).splitlines())),
    """json""": (json.__name__, _hash_python_lines(inspect.getsource(json).splitlines())),
    """pandas""": (pandas.__name__, _hash_python_lines(inspect.getsource(pandas).splitlines())),
    """parquet""": (parquet.__name__, _hash_python_lines(inspect.getsource(parquet).splitlines())),
    """arrow""": (arrow.__name__, _hash_python_lines(inspect.getsource(arrow).splitlines())),
    """text""": (text.__name__, _hash_python_lines(inspect.getsource(text).splitlines())),
    """imagefolder""": (imagefolder.__name__, _hash_python_lines(inspect.getsource(imagefolder).splitlines())),
    """audiofolder""": (audiofolder.__name__, _hash_python_lines(inspect.getsource(audiofolder).splitlines())),
}

# Used to infer the module to use based on the data files extensions
__lowerCamelCase				=			{
    """.csv""": ("""csv""", {}),
    """.tsv""": ("""csv""", {"""sep""": """\t"""}),
    """.json""": ("""json""", {}),
    """.jsonl""": ("""json""", {}),
    """.parquet""": ("""parquet""", {}),
    """.arrow""": ("""arrow""", {}),
    """.txt""": ("""text""", {}),
}
_EXTENSION_TO_MODULE.update({ext: ("""imagefolder""", {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ("""imagefolder""", {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext: ("""audiofolder""", {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ("""audiofolder""", {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
__lowerCamelCase				=			{"""imagefolder""", """audiofolder"""}

# Used to filter data files based on extensions given a module name
__lowerCamelCase				=			{}
for _ext, (_module, _) in _EXTENSION_TO_MODULE.items():
   _MODULE_TO_EXTENSIONS.setdefault(_module, []).append(_ext)

_MODULE_TO_EXTENSIONS["imagefolder"].append(""".zip""")




_MODULE_TO_EXTENSIONS["audiofolder"].append(""".zip""")







import inspect
import unittest

from transformers import ConvNextVaConfig
from transformers.models.auto import get_values
from transformers.models.auto.modeling_auto import MODEL_FOR_BACKBONE_MAPPING_NAMES, MODEL_MAPPING_NAMES
from transformers.testing_utils import require_torch, 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, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
   import torch

   from transformers import ConvNextVaBackbone, ConvNextVaForImageClassification, ConvNextVaModel
   from transformers.models.convnextva.modeling_convnextva import CONVNEXTV2_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
   from PIL import Image

   from transformers import AutoImageProcessor






class     UpperCAmelCase						:



   def __init__(self						:   int       ,		snake_case__						:   Optional[int]       ,		snake_case__						:   Optional[int]=13       ,		snake_case__						:   Tuple=32       ,		snake_case__						:   Optional[Any]=3       ,		snake_case__						:   Tuple=4       ,		snake_case__						:   List[Any]=[10, 20, 30, 40]       ,		snake_case__						:   Dict=[2, 2, 3, 2]       ,		snake_case__						:   Union[str, Any]=True       ,		snake_case__						:   Union[str, Any]=True       ,		snake_case__						:   List[str]=37       ,		snake_case__						:   List[Any]="gelu"       ,		snake_case__						:   Union[str, Any]=10       ,		snake_case__						:   int=0.02       ,		snake_case__						:   str=["stage2", "stage3", "stage4"]       ,		snake_case__						:   int=[2, 3, 4]       ,		snake_case__						:   List[Any]=None       ,		)						->      Optional[int]:


          '''simple docstring'''


          snake_case			:  Union[str, Any]     							= parent
          snake_case			:  Optional[int]     							= batch_size
          snake_case			:  Optional[Any]     							= image_size
          snake_case			:  Union[str, Any]     							= num_channels
          snake_case			:  Tuple     							= num_stages
          snake_case			:  Union[str, Any]     							= hidden_sizes
          snake_case			:  List[Any]     							= depths
          snake_case			:  Dict     							= is_training
          snake_case			:  List[Any]     							= use_labels
          snake_case			:  str     							= intermediate_size
          snake_case			:  Dict     							= hidden_act
          snake_case			:  Optional[int]     							= num_labels
          snake_case			:  Dict     							= initializer_range
          snake_case			:  List[Any]     							= out_features
          snake_case			:  int     							= out_indices
          snake_case			:  List[Any]     							= scope



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


          '''simple docstring'''


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

          snake_case			:  List[Any]     							= None
          if self.use_labels:
                 snake_case			:  List[Any]     							= ids_tensor([self.batch_size]       ,		self.num_labels					)

          snake_case			:  str     							= self.get_config()

          return config, pixel_values, labels



   def 							_SCREAMING_SNAKE_CASE  (self						:   Any					)						->      Dict:


          '''simple docstring'''


          return ConvNextVaConfig(
              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=snake_case__       ,		initializer_range=self.initializer_range       ,		out_features=self.out_features       ,		out_indices=self.out_indices       ,		num_labels=self.num_labels       ,		)



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


          '''simple docstring'''


          snake_case			:  List[str]     							= ConvNextVaModel(config=snake_case__					)
          model.to(snake_case__					)
          model.eval()
          snake_case			:  Any     							= model(snake_case__					)
          # 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 							_SCREAMING_SNAKE_CASE  (self						:   Union[str, Any]       ,		snake_case__						:   Optional[Any]       ,		snake_case__						:   str       ,		snake_case__						:   List[Any]					)						->      Tuple:


          '''simple docstring'''


          snake_case			:  str     							= ConvNextVaForImageClassification(snake_case__					)
          model.to(snake_case__					)
          model.eval()
          snake_case			:  Tuple     							= model(snake_case__       ,		labels=snake_case__					)
          self.parent.assertEqual(result.logits.shape       ,		(self.batch_size, self.num_labels)					)



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


          '''simple docstring'''


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

          # 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, self.hidden_sizes[1], 4, 4]					)

          # verify channels
          self.parent.assertEqual(len(model.channels					)       ,		len(config.out_features					)					)
          self.parent.assertListEqual(model.channels       ,		config.hidden_sizes[1:]					)

          # verify backbone works with out_features=None
          snake_case			:  Dict     							= None
          snake_case			:  Tuple     							= ConvNextVaBackbone(config=snake_case__					)
          model.to(snake_case__					)
          model.eval()
          snake_case			:  int     							= model(snake_case__					)

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

          # verify channels
          self.parent.assertEqual(len(model.channels					)       ,		1					)
          self.parent.assertListEqual(model.channels       ,		[config.hidden_sizes[-1]]					)



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


          '''simple docstring'''


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



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


          '''simple docstring'''


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






@require_torch
class     UpperCAmelCase						(		A_			,A_			,unittest.TestCase					):
   A__   :						Tuple			=	(
       (
           ConvNextVaModel,
           ConvNextVaForImageClassification,
           ConvNextVaBackbone,
       )
       if is_torch_available()
       else ()
   )
   A__   :						int			=	(
       {"feature-extraction": ConvNextVaModel, "image-classification": ConvNextVaForImageClassification}
       if is_torch_available()
       else {}
   )

   A__   :						str			=	False
   A__   :						Dict			=	False
   A__   :						Optional[int]			=	False
   A__   :						List[Any]			=	False
   A__   :						str			=	False



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


          '''simple docstring'''


          snake_case			:  Optional[Any]     							= ConvNextVaModelTester(self					)
          snake_case			:  int     							= ConfigTester(self       ,		config_class=snake_case__       ,		has_text_modality=snake_case__       ,		hidden_size=37					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   str					)						->      Tuple:


          '''simple docstring'''


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



   def 							_SCREAMING_SNAKE_CASE  (self						:   int					)						->      Dict:


          '''simple docstring'''


          return



   @unittest.skip(reason="ConvNextV2 does not use inputs_embeds"					)
   def 							_SCREAMING_SNAKE_CASE  (self						:   List[Any]					)						->      Optional[int]:


          '''simple docstring'''


          pass



   @unittest.skip(reason="ConvNextV2 does not support input and output embeddings"					)
   def 							_SCREAMING_SNAKE_CASE  (self						:   int					)						->      Optional[Any]:


          '''simple docstring'''


          pass



   @unittest.skip(reason="ConvNextV2 does not use feedforward chunking"					)
   def 							_SCREAMING_SNAKE_CASE  (self						:   Optional[int]					)						->      List[str]:


          '''simple docstring'''


          pass



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


          '''simple docstring'''


          if not self.model_tester.is_training:
                 return

          for model_class in self.all_model_classes:
                 snake_case							,   snake_case			:  str     							= self.model_tester.prepare_config_and_inputs_with_labels()
                 snake_case			:  Tuple     							= True

                 if model_class.__name__ in [
                     *get_values(snake_case__					),
                     *get_values(snake_case__					),
                 ]:
                        continue

                 snake_case			:  Union[str, Any]     							= model_class(snake_case__					)
                 model.to(snake_case__					)
                 model.train()
                 snake_case			:  str     							= self._prepare_for_class(snake_case__       ,		snake_case__       ,		return_labels=snake_case__					)
                 snake_case			:  List[str]     							= model(**snake_case__					).loss
                 loss.backward()



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


          '''simple docstring'''


          if not self.model_tester.is_training:
                 return

          for model_class in self.all_model_classes:
                 snake_case							,   snake_case			:  List[str]     							= self.model_tester.prepare_config_and_inputs_with_labels()
                 snake_case			:  List[Any]     							= False
                 snake_case			:  Any     							= True

                 if (
                     model_class.__name__
                     in [*get_values(snake_case__					), *get_values(snake_case__					)]
                     or not model_class.supports_gradient_checkpointing
                 ):
                        continue

                 snake_case			:  Union[str, Any]     							= model_class(snake_case__					)
                 model.to(snake_case__					)
                 model.gradient_checkpointing_enable()
                 model.train()
                 snake_case			:  List[str]     							= self._prepare_for_class(snake_case__       ,		snake_case__       ,		return_labels=snake_case__					)
                 snake_case			:  Any     							= model(**snake_case__					).loss
                 loss.backward()



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


          '''simple docstring'''


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

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

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



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


          '''simple docstring'''


          snake_case			:  Optional[Any]     							= self.model_tester.prepare_config_and_inputs()
          self.model_tester.create_and_check_model(*snake_case__					)



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


          '''simple docstring'''


          def check_hidden_states_output(snake_case__						:   Dict       ,		snake_case__						:   List[Any]       ,		snake_case__						:   Dict					):
                 snake_case			:  List[str]     							= model_class(snake_case__					)
                 model.to(snake_case__					)
                 model.eval()

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

                 snake_case			:  List[Any]     							= outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states

                 snake_case			:  Dict     							= self.model_tester.num_stages
                 self.assertEqual(len(snake_case__					)       ,		expected_num_stages + 1					)

                 # ConvNextV2'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]       ,		)

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

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

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

                 check_hidden_states_output(snake_case__       ,		snake_case__       ,		snake_case__					)



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


          '''simple docstring'''


          snake_case			:  Union[str, Any]     							= self.model_tester.prepare_config_and_inputs()
          self.model_tester.create_and_check_for_image_classification(*snake_case__					)



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


          '''simple docstring'''


          for model_name in CONVNEXTV2_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                 snake_case			:  Optional[int]     							= ConvNextVaModel.from_pretrained(snake_case__					)
                 self.assertIsNotNone(snake_case__					)



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






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



   @cached_property
   def 							_SCREAMING_SNAKE_CASE  (self						:   Tuple					)						->      Any:


          '''simple docstring'''


          return AutoImageProcessor.from_pretrained("facebook/convnextv2-tiny-1k-224"					) if is_vision_available() else None



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


          '''simple docstring'''


          snake_case			:  Union[str, Any]     							= ConvNextVaForImageClassification.from_pretrained("facebook/convnextv2-tiny-1k-224"					).to(snake_case__					)

          snake_case			:  List[str]     							= self.default_image_processor
          snake_case			:  str     							= prepare_img()
          snake_case			:  List[str]     							= preprocessor(images=snake_case__       ,		return_tensors="pt"					).to(snake_case__					)

          # forward pass
          with torch.no_grad():
                 snake_case			:  str     							= model(**snake_case__					)

          # verify the logits
          snake_case			:  Any     							= torch.Size((1, 10_00)					)
          self.assertEqual(outputs.logits.shape       ,		snake_case__					)

          snake_case			:  str     							= torch.tensor([0.9996, 0.1966, -0.4386]					).to(snake_case__					)
          self.assertTrue(torch.allclose(outputs.logits[0, :3]       ,		snake_case__       ,		atol=1e-4					)					)





