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      lowerCAmelCase						(					nn.Module			):



     def __init__(					self       ):
            super().__init__()
            _A=	nn.Linear(3				,      4       )
            _A=	nn.BatchNormad(4       )
            _A=	nn.Linear(4				,      5       )





     def 			a__  (					self				,      lowerCAmelCase__       ):
            return self.lineara(self.batchnorm(self.lineara(lowerCAmelCase__       )       )       )





class      lowerCAmelCase						(					unittest.TestCase			):



     def 			a__  (					self       ):
            _A=	ModelForTest()
            with TemporaryDirectory() as tmp_dir:
                   offload_state_dict(lowerCAmelCase__				,      model.state_dict()       )
                   _A=	os.path.join(lowerCAmelCase__				,      'index.json'       )
                   self.assertTrue(os.path.isfile(lowerCAmelCase__       )       )
                   # TODO: add tests on what is inside the index

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



     def 			a__  (					self       ):
            _A=	[torch.floataa, torch.floataa, torch.bfloataa]

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

                          _A=	load_offloaded_weight(lowerCAmelCase__				,      index['weight']       )
                          self.assertTrue(torch.equal(lowerCAmelCase__				,      lowerCAmelCase__       )       )



     def 			a__  (					self       ):
            _A=	ModelForTest()
            _A=	model.state_dict()
            _A=	{k: v for k, v in state_dict.items() if 'linear2' not in k}
            _A=	{k: v for k, v in state_dict.items() if 'linear2' in k}

            with TemporaryDirectory() as tmp_dir:
                   offload_state_dict(lowerCAmelCase__				,      lowerCAmelCase__       )
                   _A=	OffloadedWeightsLoader(state_dict=lowerCAmelCase__				,      save_folder=lowerCAmelCase__       )

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

            _A=	{k: v for k, v in state_dict.items() if 'weight' in k}
            _A=	{k: v for k, v in state_dict.items() if 'weight' not in k}

            with TemporaryDirectory() as tmp_dir:
                   offload_state_dict(lowerCAmelCase__				,      lowerCAmelCase__       )
                   _A=	OffloadedWeightsLoader(state_dict=lowerCAmelCase__				,      save_folder=lowerCAmelCase__       )

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

            with TemporaryDirectory() as tmp_dir:
                   offload_state_dict(lowerCAmelCase__				,      lowerCAmelCase__       )
                   # Duplicates are removed
                   _A=	OffloadedWeightsLoader(state_dict=lowerCAmelCase__				,      save_folder=lowerCAmelCase__       )

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





     def 			a__  (					self       ):
            _A=	{'a.1': 0, 'a.10': 1, 'a.2': 2}
            _A=	extract_submodules_state_dict(lowerCAmelCase__				,      ['a.1', 'a.2']       )
            self.assertDictEqual(lowerCAmelCase__				,      {'a.1': 0, 'a.2': 2}       )

            _A=	{'a.1.a': 0, 'a.10.a': 1, 'a.2.a': 2}
            _A=	extract_submodules_state_dict(lowerCAmelCase__				,      ['a.1', 'a.2']       )
            self.assertDictEqual(lowerCAmelCase__				,      {'a.1.a': 0, 'a.2.a': 2}       )
import argparse

import torch

from transformers import BertForMaskedLM


if __name__ == "__main__":
      UpperCAmelCase_								=							argparse.ArgumentParser(
          description=(
              '''Extraction some layers of the full BertForMaskedLM or RObertaForMaskedLM for Transfer Learned'''
              ''' Distillation'''
          )
      )
      parser.add_argument('''--model_type''', default='''bert''', choices=['''bert'''])
      parser.add_argument('''--model_name''', default='''bert-base-uncased''', type=str)
      parser.add_argument('''--dump_checkpoint''', default='''serialization_dir/tf_bert-base-uncased_0247911.pth''', type=str)
      parser.add_argument('''--vocab_transform''', action='''store_true''')
      UpperCAmelCase_								=							parser.parse_args()

      if args.model_type == "bert":
            UpperCAmelCase_								=							BertForMaskedLM.from_pretrained(args.model_name)
            UpperCAmelCase_								=							'''bert'''
      else:
            raise ValueError('''args.model_type should be "bert".''')

      UpperCAmelCase_								=							model.state_dict()
      UpperCAmelCase_								=							{}

      for w in ["word_embeddings", "position_embeddings"]:
            UpperCAmelCase_								=							state_dict[F"{prefix}.embeddings.{w}.weight"]
      for w in ["weight", "bias"]:
            UpperCAmelCase_								=							state_dict[F"{prefix}.embeddings.LayerNorm.{w}"]

      UpperCAmelCase_								=							0
      for teacher_idx in [0, 2, 4, 7, 9, 11]:
            for w in ["weight", "bias"]:
                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.attention.self.query.{w}"
                  ]
                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.attention.self.key.{w}"
                  ]
                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.attention.self.value.{w}"
                  ]

                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.attention.output.dense.{w}"
                  ]
                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.attention.output.LayerNorm.{w}"
                  ]

                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.intermediate.dense.{w}"
                  ]
                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.output.dense.{w}"
                  ]
                  UpperCAmelCase_								=							state_dict[
                      F"{prefix}.encoder.layer.{teacher_idx}.output.LayerNorm.{w}"
                  ]
            std_idx += 1

      UpperCAmelCase_								=							state_dict['''cls.predictions.decoder.weight''']
      UpperCAmelCase_								=							state_dict['''cls.predictions.bias''']
      if args.vocab_transform:
            for w in ["weight", "bias"]:
                  UpperCAmelCase_								=							state_dict[F"cls.predictions.transform.dense.{w}"]
                  UpperCAmelCase_								=							state_dict[F"cls.predictions.transform.LayerNorm.{w}"]

      print(F"N layers selected for distillation: {std_idx}")
      print(F"Number of params transferred for distillation: {len(compressed_sd.keys())}")

      print(F"Save transferred checkpoint to {args.dump_checkpoint}.")
      torch.save(compressed_sd, args.dump_checkpoint)