





"""simple docstring"""





import os
import unittest
from tempfile import TemporaryDirectory

import torch
import torch.nn as nn

from accelerate.utils import (
    OffloadedWeightsLoader,
    extract_submodules_state_dict,
    load_offloaded_weight,
    offload_state_dict,
    offload_weight,
)




class 							snake_case							(			nn.Module ):

      '''simple docstring'''



      def __init__(							self       :      List[str] ):


        '''simple docstring'''




        super().__init__()
        __UpperCAmelCase    :						List[Any]       =						nn.Linear(3						,				4 )
        __UpperCAmelCase    :						Optional[Any]       =						nn.BatchNormad(4 )
        __UpperCAmelCase    :						Optional[Any]       =						nn.Linear(4						,				5 )







      def       A_					(							self       :      Optional[Any]						,				__lowercase       :      List[Any] ):


        '''simple docstring'''




        return self.lineara(self.batchnorm(self.lineara(__lowercase ) ) )




class 							snake_case							(			unittest.TestCase ):

      '''simple docstring'''



      def       A_					(							self       :      List[str] ):


        '''simple docstring'''




        __UpperCAmelCase    :						str       =						ModelForTest()
        with TemporaryDirectory() as tmp_dir:
          offload_state_dict(__lowercase						,				model.state_dict() )
          __UpperCAmelCase    :						List[Any]       =						os.path.join(__lowercase						,				'''index.json''' )
          self.assertTrue(os.path.isfile(__lowercase ) )
          # TODO: add tests on what is inside the index

          for key in ["linear1.weight", "linear1.bias", "linear2.weight", "linear2.bias"]:
            __UpperCAmelCase    :						Optional[Any]       =						os.path.join(__lowercase						,				f'''{key}.dat''' )
            self.assertTrue(os.path.isfile(__lowercase ) )
            # TODO: add tests on the fact weights are properly loaded

      def       A_					(							self       :      str ):


        '''simple docstring'''




        __UpperCAmelCase    :						Optional[Any]       =						[torch.floataa, torch.floataa, torch.bfloataa]

        for dtype in dtypes:
          __UpperCAmelCase    :						str       =						torch.randn(2						,				3						,				dtype=__lowercase )
          with TemporaryDirectory() as tmp_dir:
            __UpperCAmelCase    :						Optional[Any]       =						offload_weight(__lowercase						,				'''weight'''						,				__lowercase						,				{} )
            __UpperCAmelCase    :						List[str]       =						os.path.join(__lowercase						,				'''weight.dat''' )
            self.assertTrue(os.path.isfile(__lowercase ) )
            self.assertDictEqual(__lowercase						,				{'''weight''': {'''shape''': [2, 3], '''dtype''': str(__lowercase ).split('''.''' )[1]}} )

            __UpperCAmelCase    :						List[Any]       =						load_offloaded_weight(__lowercase						,				index['''weight'''] )
            self.assertTrue(torch.equal(__lowercase						,				__lowercase ) )

      def       A_					(							self       :      List[str] ):


        '''simple docstring'''




        __UpperCAmelCase    :						int       =						ModelForTest()
        __UpperCAmelCase    :						List[str]       =						model.state_dict()
        __UpperCAmelCase    :						Optional[Any]       =						{k: v for k, v in state_dict.items() if '''linear2''' not in k}
        __UpperCAmelCase    :						str       =						{k: v for k, v in state_dict.items() if '''linear2''' in k}

        with TemporaryDirectory() as tmp_dir:
          offload_state_dict(__lowercase						,				__lowercase )
          __UpperCAmelCase    :						int       =						OffloadedWeightsLoader(state_dict=__lowercase						,				save_folder=__lowercase )

          # Every key is there with the right value
          self.assertEqual(sorted(__lowercase )						,				sorted(state_dict.keys() ) )
          for key, param in state_dict.items():
            self.assertTrue(torch.allclose(__lowercase						,				weight_map[key] ) )

        __UpperCAmelCase    :						Any       =						{k: v for k, v in state_dict.items() if '''weight''' in k}
        __UpperCAmelCase    :						List[Any]       =						{k: v for k, v in state_dict.items() if '''weight''' not in k}

        with TemporaryDirectory() as tmp_dir:
          offload_state_dict(__lowercase						,				__lowercase )
          __UpperCAmelCase    :						Optional[Any]       =						OffloadedWeightsLoader(state_dict=__lowercase						,				save_folder=__lowercase )

          # Every key is there with the right value
          self.assertEqual(sorted(__lowercase )						,				sorted(state_dict.keys() ) )
          for key, param in state_dict.items():
            self.assertTrue(torch.allclose(__lowercase						,				weight_map[key] ) )

        with TemporaryDirectory() as tmp_dir:
          offload_state_dict(__lowercase						,				__lowercase )
          # Duplicates are removed
          __UpperCAmelCase    :						Dict       =						OffloadedWeightsLoader(state_dict=__lowercase						,				save_folder=__lowercase )

          # Every key is there with the right value
          self.assertEqual(sorted(__lowercase )						,				sorted(state_dict.keys() ) )
          for key, param in state_dict.items():
            self.assertTrue(torch.allclose(__lowercase						,				weight_map[key] ) )







      def       A_					(							self       :      Union[str, Any] ):


        '''simple docstring'''




        __UpperCAmelCase    :						Dict       =						{'''a.1''': 0, '''a.10''': 1, '''a.2''': 2}
        __UpperCAmelCase    :						Dict       =						extract_submodules_state_dict(__lowercase						,				['''a.1''', '''a.2'''] )
        self.assertDictEqual(__lowercase						,				{'''a.1''': 0, '''a.2''': 2} )

        __UpperCAmelCase    :						List[str]       =						{'''a.1.a''': 0, '''a.10.a''': 1, '''a.2.a''': 2}
        __UpperCAmelCase    :						Union[str, Any]       =						extract_submodules_state_dict(__lowercase						,				['''a.1''', '''a.2'''] )
        self.assertDictEqual(__lowercase						,				{'''a.1.a''': 0, '''a.2.a''': 2} )






"""simple docstring"""





import argparse

import torch

from transformers import LxmertConfig, LxmertForPreTraining, load_tf_weights_in_lxmert
from transformers.utils import logging


logging.set_verbosity_info()


def       lowerCamelCase_       (   UpperCAmelCase_  ,			UpperCAmelCase_  ,			UpperCAmelCase_  )						->Optional[int]:





  """simple docstring"""


  __UpperCAmelCase    :						Dict       =						LxmertConfig.from_json_file(UpperCAmelCase_  )
  print(f'''Building PyTorch model from configuration: {config}'''  )
  __UpperCAmelCase    :						Tuple       =						LxmertForPreTraining(UpperCAmelCase_  )

  # Load weights from tf checkpoint
  load_tf_weights_in_lxmert(UpperCAmelCase_  ,			UpperCAmelCase_  ,			UpperCAmelCase_  )

  # Save pytorch-model
  print(f'''Save PyTorch model to {pytorch_dump_path}'''  )
  torch.save(model.state_dict()  ,			UpperCAmelCase_  )


if __name__ == "__main__":
  lowercase__				:Optional[int]				   =       argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      '--tf_checkpoint_path', default=None, type=str, required=True, help='Path to the TensorFlow checkpoint path.'
  )
  parser.add_argument(
      '--config_file',
      default=None,
      type=str,
      required=True,
      help='The config json file corresponding to the pre-trained model. \nThis specifies the model architecture.',
  )
  parser.add_argument(
      '--pytorch_dump_path', default=None, type=str, required=True, help='Path to the output PyTorch model.'
  )
  lowercase__				:Union[str, Any]				   =       parser.parse_args()
  convert_tf_checkpoint_to_pytorch(args.tf_checkpoint_path, args.config_file, args.pytorch_dump_path)