


from __future__ import annotations

import inspect
import unittest

import numpy as np

from transformers import DeiTConfig
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 (
     TFDeiTForImageClassification,
     TFDeiTForImageClassificationWithTeacher,
     TFDeiTForMaskedImageModeling,
     TFDeiTModel,
 )
 from transformers.models.deit.modeling_tf_deit import TF_DEIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
 from PIL import Image

 from transformers import DeiTImageProcessor





class   __a      :
 def __init__(       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__=13    ,			lowerCAmelCase__=30    ,			lowerCAmelCase__=2    ,			lowerCAmelCase__=3    ,			lowerCAmelCase__=True    ,			lowerCAmelCase__=True    ,			lowerCAmelCase__=32    ,			lowerCAmelCase__=2    ,			lowerCAmelCase__=4    ,			lowerCAmelCase__=37    ,			lowerCAmelCase__="gelu"    ,			lowerCAmelCase__=0.1    ,			lowerCAmelCase__=0.1    ,			lowerCAmelCase__=10    ,			lowerCAmelCase__=0.0_2    ,			lowerCAmelCase__=3    ,			lowerCAmelCase__=None    ,			lowerCAmelCase__=2    ,			) ->    List[str]:




        '''simple docstring'''
        lowercase__:       List[str]        =       parent
        lowercase__:       Tuple        =       batch_size
        lowercase__:       Optional[Any]        =       image_size
        lowercase__:       int        =       patch_size
        lowercase__:       Union[str, Any]        =       num_channels
        lowercase__:       Dict        =       is_training
        lowercase__:       Any        =       use_labels
        lowercase__:       Optional[int]        =       hidden_size
        lowercase__:       Tuple        =       num_hidden_layers
        lowercase__:       List[str]        =       num_attention_heads
        lowercase__:       Any        =       intermediate_size
        lowercase__:       Union[str, Any]        =       hidden_act
        lowercase__:       Union[str, Any]        =       hidden_dropout_prob
        lowercase__:       Optional[Any]        =       attention_probs_dropout_prob
        lowercase__:       str        =       type_sequence_label_size
        lowercase__:       List[str]        =       initializer_range
        lowercase__:       Optional[Any]        =       scope
        lowercase__:       Dict        =       encoder_stride

        # in DeiT, the seq length equals the number of patches + 2 (we add 2 for the [CLS] and distilation tokens)
        lowercase__:       List[str]        =       (image_size // patch_size) ** 2
        lowercase__:       Tuple        =       num_patches + 2
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Optional[int]:




        '''simple docstring'''
        lowercase__:       List[Any]        =       floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]			)

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

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

        return config, pixel_values, labels
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Optional[Any]:




        '''simple docstring'''
        return DeiTConfig(
            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=lowerCAmelCase__    ,			initializer_range=self.initializer_range    ,			encoder_stride=self.encoder_stride    ,			)
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__			) ->    int:




        '''simple docstring'''
        lowercase__:       List[str]        =       TFDeiTModel(config=lowerCAmelCase__			)
        lowercase__:       Any        =       model(lowerCAmelCase__			)
        self.parent.assertEqual(result.last_hidden_state.shape    ,			(self.batch_size, self.seq_length, self.hidden_size)			)
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__			) ->    Union[str, Any]:




        '''simple docstring'''
        lowercase__:       Any        =       TFDeiTForMaskedImageModeling(config=lowerCAmelCase__			)
        lowercase__:       int        =       model(lowerCAmelCase__			)
        self.parent.assertEqual(
            result.reconstruction.shape    ,			(self.batch_size, self.num_channels, self.image_size, self.image_size)			)

        # test greyscale images
        lowercase__:       Tuple        =       1
        lowercase__:       Optional[int]        =       TFDeiTForMaskedImageModeling(lowerCAmelCase__			)

        lowercase__:       int        =       floats_tensor([self.batch_size, 1, self.image_size, self.image_size]			)
        lowercase__:       str        =       model(lowerCAmelCase__			)
        self.parent.assertEqual(result.reconstruction.shape    ,			(self.batch_size, 1, self.image_size, self.image_size)			)
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__			) ->    List[Any]:




        '''simple docstring'''
        lowercase__:       Any        =       self.type_sequence_label_size
        lowercase__:       Optional[Any]        =       TFDeiTForImageClassification(lowerCAmelCase__			)
        lowercase__:       List[Any]        =       model(lowerCAmelCase__    ,			labels=lowerCAmelCase__			)
        self.parent.assertEqual(result.logits.shape    ,			(self.batch_size, self.type_sequence_label_size)			)

        # test greyscale images
        lowercase__:       Optional[Any]        =       1
        lowercase__:       Union[str, Any]        =       TFDeiTForImageClassification(lowerCAmelCase__			)

        lowercase__:       List[str]        =       floats_tensor([self.batch_size, 1, self.image_size, self.image_size]			)
        lowercase__:       str        =       model(lowerCAmelCase__    ,			labels=lowerCAmelCase__			)
        self.parent.assertEqual(result.logits.shape    ,			(self.batch_size, self.type_sequence_label_size)			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        lowercase__:       Dict        =       self.prepare_config_and_inputs()
        lowercase__      ,						lowercase__      ,						lowercase__:       Dict        =       config_and_inputs
        lowercase__:       Optional[Any]        =       {'pixel_values': pixel_values}
        return config, inputs_dict





@require_tf
class   __a      (   __UpperCamelCase	,	__UpperCamelCase	,	unittest.TestCase				):
 __lowercase			:							Union[str, Any]       					=			(
     (
         TFDeiTModel,
         TFDeiTForImageClassification,
         TFDeiTForImageClassificationWithTeacher,
         TFDeiTForMaskedImageModeling,
     )
     if is_tf_available()
     else ()
 )
 __lowercase			:							Union[str, Any]       					=			(
     {
         'feature-extraction': TFDeiTModel,
         'image-classification': (TFDeiTForImageClassification, TFDeiTForImageClassificationWithTeacher),
     }
     if is_tf_available()
     else {}
 )

 __lowercase			:							Union[str, Any]       					=			False
 __lowercase			:							Tuple       					=			False
 __lowercase			:							int       					=			False
 __lowercase			:							Optional[int]       					=			False
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    List[Any]:




        '''simple docstring'''
        lowercase__:       List[Any]        =       TFDeiTModelTester(self			)
        lowercase__:       int        =       ConfigTester(self    ,			config_class=lowerCAmelCase__    ,			has_text_modality=lowerCAmelCase__    ,			hidden_size=37			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    List[str]:




        '''simple docstring'''
        self.config_tester.run_common_tests()
 @unittest.skip(reason='DeiT does not use inputs_embeds'			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    List[str]:




        '''simple docstring'''
        pass
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        lowercase__      ,						lowercase__:       int        =       self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
               lowercase__:       Tuple        =       model_class(lowerCAmelCase__			)
               self.assertIsInstance(model.get_input_embeddings()    ,			(tf.keras.layers.Layer)			)
               lowercase__:       Union[str, Any]        =       model.get_output_embeddings()
               self.assertTrue(x is None or isinstance(lowerCAmelCase__    ,			tf.keras.layers.Dense			)			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Union[str, Any]:




        '''simple docstring'''
        lowercase__      ,						lowercase__:       Union[str, Any]        =       self.model_tester.prepare_config_and_inputs_for_common()

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

               lowercase__:       Any        =       ['pixel_values']
               self.assertListEqual(arg_names[:1]    ,			lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        lowercase__:       Dict        =       self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    List[str]:




        '''simple docstring'''
        lowercase__:       Tuple        =       self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_masked_image_modeling(*lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    int:




        '''simple docstring'''
        lowercase__:       List[Any]        =       self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_image_classification(*lowerCAmelCase__			)
 def 		SCREAMING_SNAKE_CASE__       (       self    ,			lowerCAmelCase__    ,			lowerCAmelCase__    ,			lowerCAmelCase__=False			) ->    List[Any]:




        '''simple docstring'''
        lowercase__:       Union[str, Any]        =       super()._prepare_for_class(lowerCAmelCase__    ,			lowerCAmelCase__    ,			return_labels=lowerCAmelCase__			)

        if return_labels:
               if "labels" in inputs_dict and "labels" not in inspect.signature(model_class.call			).parameters:
                      del inputs_dict["labels"]

        return inputs_dict
 @slow
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        for model_name in TF_DEIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
               lowercase__:       Union[str, Any]        =       TFDeiTModel.from_pretrained(lowerCAmelCase__			)
               self.assertIsNotNone(lowerCAmelCase__			)





def        snake_case_					(  )  ->    Tuple:
       lowercase__:       Any        =       Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'					)
       return image





@require_tf
@require_vision
class   __a      (   unittest.TestCase				):
 @cached_property
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Tuple:




        '''simple docstring'''
        return (
            DeiTImageProcessor.from_pretrained('facebook/deit-base-distilled-patch16-224'			)
            if is_vision_available()
            else None
        )
 @slow
 def 		SCREAMING_SNAKE_CASE__       (       self			) ->    Dict:




        '''simple docstring'''
        lowercase__:       str        =       TFDeiTForImageClassificationWithTeacher.from_pretrained('facebook/deit-base-distilled-patch16-224'			)

        lowercase__:       str        =       self.default_image_processor
        lowercase__:       Dict        =       prepare_img()
        lowercase__:       Optional[int]        =       image_processor(images=lowerCAmelCase__    ,			return_tensors='tf'			)

        # forward pass
        lowercase__:       int        =       model(**lowerCAmelCase__			)

        # verify the logits
        lowercase__:       Tuple        =       tf.TensorShape((1, 1_000)			)
        self.assertEqual(outputs.logits.shape    ,			lowerCAmelCase__			)

        lowercase__:       List[Any]        =       tf.constant([-1.0_2_6_6, 0.1_9_1_2, -1.2_8_6_1]			)

        self.assertTrue(np.allclose(outputs.logits[0, :3]    ,			lowerCAmelCase__    ,			atol=1E-4			)			)






import argparse
import os
import re

import tensorflow as tf
import torch

from transformers import BertConfig, BertModel
from transformers.utils import logging


logging.set_verbosity_info()
__lowerCAmelCase			=							logging.get_logger(__name__)




def        snake_case_					(  snake_case      ,							snake_case      ,							snake_case					)  ->    Any:
       lowercase__:       Dict        =       os.path.abspath(snake_case					)
       logger.info(f'Converting TensorFlow checkpoint from {tf_path}'					)
       # Load weights from TF model
       lowercase__:       Optional[Any]        =       tf.train.list_variables(snake_case					)
       lowercase__:       List[Any]        =       []
       lowercase__:       Tuple        =       []
       lowercase__:       int        =       []
       for full_name, shape in init_vars:
              # logger.info(f"Loading TF weight {name} with shape {shape}")
              lowercase__:       Union[str, Any]        =       full_name.split('/'					)
              if full_name == "_CHECKPOINTABLE_OBJECT_GRAPH" or name[0] in ["global_step", "save_counter"]:
                     logger.info(f'Skipping non-model layer {full_name}'					)
                     continue
              if "optimizer" in full_name:
                     logger.info(f'Skipping optimization layer {full_name}'					)
                     continue
              if name[0] == "model":
                     # ignore initial 'model'
                     lowercase__:       str        =       name[1:]
              # figure out how many levels deep the name is
              lowercase__:       Optional[Any]        =       0
              for _name in name:
                     if _name.startswith('layer_with_weights'					):
                            depth += 1
                     else:
                            break
              layer_depth.append(snake_case					)
              # read data
              lowercase__:       Optional[Any]        =       tf.train.load_variable(snake_case      ,							snake_case					)
              names.append('/'.join(snake_case					)					)
              arrays.append(snake_case					)
       logger.info(f'Read a total of {len(snake_case					):,} layers'					)

       # Sanity check
       if len(set(snake_case					)					) != 1:
              raise ValueError(f'Found layer names with different depths (layer depth {list(set(snake_case					)					)})'					)
       lowercase__:       Any        =       list(set(snake_case					)					)[0]
       if layer_depth != 1:
              raise ValueError(
                  'The model contains more than just the embedding/encoder layers. This script does not handle MLM/NSP'
                  ' heads.'					)

       # convert layers
       logger.info('Converting weights...'					)
       for full_name, array in zip(snake_case      ,							snake_case					):
              lowercase__:       Optional[int]        =       full_name.split('/'					)
              lowercase__:       List[Any]        =       model
              lowercase__:       Any        =       []
              for i, m_name in enumerate(snake_case					):
                     if m_name == ".ATTRIBUTES":
                            # variable names end with .ATTRIBUTES/VARIABLE_VALUE
                            break
                     if m_name.startswith('layer_with_weights'					):
                            lowercase__:       Any        =       int(m_name.split('-'					)[-1]					)
                            if layer_num <= 2:
                                   # embedding layers
                                   # layer_num 0: word_embeddings
                                   # layer_num 1: position_embeddings
                                   # layer_num 2: token_type_embeddings
                                   continue
                            elif layer_num == 3:
                                   # embedding LayerNorm
                                   trace.extend(['embeddings', 'LayerNorm']					)
                                   lowercase__:       str        =       getattr(snake_case      ,							'embeddings'					)
                                   lowercase__:       int        =       getattr(snake_case      ,							'LayerNorm'					)
                            elif layer_num > 3 and layer_num < config.num_hidden_layers + 4:
                                   # encoder layers
                                   trace.extend(['encoder', 'layer', str(layer_num - 4					)]					)
                                   lowercase__:       int        =       getattr(snake_case      ,							'encoder'					)
                                   lowercase__:       List[str]        =       getattr(snake_case      ,							'layer'					)
                                   lowercase__:       Union[str, Any]        =       pointer[layer_num - 4]
                            elif layer_num == config.num_hidden_layers + 4:
                                   # pooler layer
                                   trace.extend(['pooler', 'dense']					)
                                   lowercase__:       Tuple        =       getattr(snake_case      ,							'pooler'					)
                                   lowercase__:       Tuple        =       getattr(snake_case      ,							'dense'					)
                     elif m_name == "embeddings":
                            trace.append('embeddings'					)
                            lowercase__:       Union[str, Any]        =       getattr(snake_case      ,							'embeddings'					)
                            if layer_num == 0:
                                   trace.append('word_embeddings'					)
                                   lowercase__:       Union[str, Any]        =       getattr(snake_case      ,							'word_embeddings'					)
                            elif layer_num == 1:
                                   trace.append('position_embeddings'					)
                                   lowercase__:       Dict        =       getattr(snake_case      ,							'position_embeddings'					)
                            elif layer_num == 2:
                                   trace.append('token_type_embeddings'					)
                                   lowercase__:       Optional[Any]        =       getattr(snake_case      ,							'token_type_embeddings'					)
                            else:
                                   raise ValueError(f'Unknown embedding layer with name {full_name}'					)
                            trace.append('weight'					)
                            lowercase__:       List[str]        =       getattr(snake_case      ,							'weight'					)
                     elif m_name == "_attention_layer":
                            # self-attention layer
                            trace.extend(['attention', 'self']					)
                            lowercase__:       int        =       getattr(snake_case      ,							'attention'					)
                            lowercase__:       Union[str, Any]        =       getattr(snake_case      ,							'self'					)
                     elif m_name == "_attention_layer_norm":
                            # output attention norm
                            trace.extend(['attention', 'output', 'LayerNorm']					)
                            lowercase__:       str        =       getattr(snake_case      ,							'attention'					)
                            lowercase__:       Optional[Any]        =       getattr(snake_case      ,							'output'					)
                            lowercase__:       List[Any]        =       getattr(snake_case      ,							'LayerNorm'					)
                     elif m_name == "_attention_output_dense":
                            # output attention dense
                            trace.extend(['attention', 'output', 'dense']					)
                            lowercase__:       Optional[Any]        =       getattr(snake_case      ,							'attention'					)
                            lowercase__:       Optional[int]        =       getattr(snake_case      ,							'output'					)
                            lowercase__:       Optional[Any]        =       getattr(snake_case      ,							'dense'					)
                     elif m_name == "_output_dense":
                            # output dense
                            trace.extend(['output', 'dense']					)
                            lowercase__:       Union[str, Any]        =       getattr(snake_case      ,							'output'					)
                            lowercase__:       List[Any]        =       getattr(snake_case      ,							'dense'					)
                     elif m_name == "_output_layer_norm":
                            # output dense
                            trace.extend(['output', 'LayerNorm']					)
                            lowercase__:       Any        =       getattr(snake_case      ,							'output'					)
                            lowercase__:       str        =       getattr(snake_case      ,							'LayerNorm'					)
                     elif m_name == "_key_dense":
                            # attention key
                            trace.append('key'					)
                            lowercase__:       Tuple        =       getattr(snake_case      ,							'key'					)
                     elif m_name == "_query_dense":
                            # attention query
                            trace.append('query'					)
                            lowercase__:       List[str]        =       getattr(snake_case      ,							'query'					)
                     elif m_name == "_value_dense":
                            # attention value
                            trace.append('value'					)
                            lowercase__:       Optional[int]        =       getattr(snake_case      ,							'value'					)
                     elif m_name == "_intermediate_dense":
                            # attention intermediate dense
                            trace.extend(['intermediate', 'dense']					)
                            lowercase__:       Any        =       getattr(snake_case      ,							'intermediate'					)
                            lowercase__:       str        =       getattr(snake_case      ,							'dense'					)
                     elif m_name == "_output_layer_norm":
                            # output layer norm
                            trace.append('output'					)
                            lowercase__:       Union[str, Any]        =       getattr(snake_case      ,							'output'					)
                     # weights & biases
                     elif m_name in ["bias", "beta"]:
                            trace.append('bias'					)
                            lowercase__:       str        =       getattr(snake_case      ,							'bias'					)
                     elif m_name in ["kernel", "gamma"]:
                            trace.append('weight'					)
                            lowercase__:       Tuple        =       getattr(snake_case      ,							'weight'					)
                     else:
                            logger.warning(f'Ignored {m_name}'					)
        # for certain layers reshape is necessary
              lowercase__:       Any        =       '.'.join(snake_case					)
              if re.match(R'(\S+)\.attention\.self\.(key|value|query)\.(bias|weight)'      ,							snake_case					) or re.match(
                  R'(\S+)\.attention\.output\.dense\.weight'      ,							snake_case					):
                     lowercase__:       Union[str, Any]        =       array.reshape(pointer.data.shape					)
              if "kernel" in full_name:
                     lowercase__:       str        =       array.transpose()
              if pointer.shape == array.shape:
                     lowercase__:       Union[str, Any]        =       torch.from_numpy(snake_case					)
              else:
                     raise ValueError(
                         f'Shape mismatch in layer {full_name}: Model expects shape {pointer.shape} but layer contains shape:'
                         f' {array.shape}'					)
              logger.info(f'Successfully set variable {full_name} to PyTorch layer {trace}'					)
       return model




def        snake_case_					(  snake_case      ,							snake_case      ,							snake_case					)  ->    Any:
       # Instantiate model
       logger.info(f'Loading model based on config from {config_path}...'					)
       lowercase__:       int        =       BertConfig.from_json_file(snake_case					)
       lowercase__:       Tuple        =       BertModel(snake_case					)

       # Load weights from checkpoint
       logger.info(f'Loading weights from checkpoint {tf_checkpoint_path}...'					)
       load_tfa_weights_in_bert(snake_case      ,							snake_case      ,							snake_case					)

       # Save pytorch-model
       logger.info(f'Saving PyTorch model to {pytorch_dump_path}...'					)
       torch.save(model.state_dict()      ,							snake_case					)


if __name__ == "__main__":
 __lowerCAmelCase			=							argparse.ArgumentParser()
 parser.add_argument(
     '''--tf_checkpoint_path''', type=str, required=True, help='''Path to the TensorFlow 2.x checkpoint path.'''
 )
 parser.add_argument(
     '''--bert_config_file''',
     type=str,
     required=True,
     help='''The config json file corresponding to the BERT model. This specifies the model architecture.''',
 )
 parser.add_argument(
     '''--pytorch_dump_path''',
     type=str,
     required=True,
     help='''Path to the output PyTorch model (must include filename).''',
 )
 __lowerCAmelCase			=							parser.parse_args()
 convert_tfa_checkpoint_to_pytorch(args.tf_checkpoint_path, args.bert_config_file, args.pytorch_dump_path)


