





'''simple docstring'''




import argparse
import os

import gluonnlp as nlp
import mxnet as mx
import numpy as np
import torch
from gluonnlp.base import get_home_dir
from gluonnlp.model.bert import BERTEncoder
from gluonnlp.model.utils import _load_vocab
from gluonnlp.vocab import Vocab
from packaging import version
from torch import nn

from transformers import BertConfig, BertForMaskedLM, BertModel, RobertaTokenizer
from transformers.models.bert.modeling_bert import (
    BertIntermediate,
    BertLayer,
    BertOutput,
    BertSelfAttention,
    BertSelfOutput,
)
from transformers.utils import logging


if version.parse(nlp.__version__) != version.parse('''0.8.3'''):
  raise Exception('''requires gluonnlp == 0.8.3''')

if version.parse(mx.__version__) != version.parse('''1.5.0'''):
  raise Exception('''requires mxnet == 1.5.0''')

logging.set_verbosity_info()
lowercase__       : Tuple         =       logging.get_logger(__name__)

lowercase__       : Optional[int]         =       '''The Nymphenburg Palace is a beautiful palace in Munich!'''






def 			_lowerCAmelCase     (    __snake_case    :					int					,				__snake_case    :					Optional[int]     )	->	Any:
     __A		:					Tuple						=			{
         "attention_cell": "multi_head",
         "num_layers": 4,
         "units": 10_24,
         "hidden_size": 7_68,
         "max_length": 5_12,
         "num_heads": 8,
         "scaled": True,
         "dropout": 0.1,
         "use_residual": True,
         "embed_size": 10_24,
         "embed_dropout": 0.1,
         "word_embed": None,
         "layer_norm_eps": 1e-5,
         "token_type_vocab_size": 2,
     }

     __A		:					Optional[int]						=			bort_4_8_768_1024_hparams

     # Let's construct the original Bort model here
     # Taken from official BERT implementation, see:
     # https://github.com/alexa/bort/blob/master/bort/bort.py
     __A		:					List[str]						=			BERTEncoder(
         attention_cell=predefined_args['attention_cell']					,				num_layers=predefined_args['num_layers']					,				units=predefined_args['units']					,				hidden_size=predefined_args['hidden_size']					,				max_length=predefined_args['max_length']					,				num_heads=predefined_args['num_heads']					,				scaled=predefined_args['scaled']					,				dropout=predefined_args['dropout']					,				output_attention=lowercase_					,				output_all_encodings=lowercase_					,				use_residual=predefined_args['use_residual']					,				activation=predefined_args.get('activation'					,				'gelu'     )					,				layer_norm_eps=predefined_args.get('layer_norm_eps'					,				lowercase_     )					,				)

     # Vocab information needs to be fetched first
     # It's the same as RoBERTa, so RobertaTokenizer can be used later
     __A		:					Tuple						=			"openwebtext_ccnews_stories_books_cased"

     # Specify download folder to Gluonnlp's vocab
     __A		:					Union[str, Any]						=			os.path.join(get_home_dir()					,				'models'     )
     __A		:					Any						=			_load_vocab(lowercase_					,				lowercase_					,				lowercase_					,				cls=lowercase_     )

     __A		:					List[Any]						=			nlp.model.BERTModel(
         lowercase_					,				len(lowercase_     )					,				units=predefined_args['units']					,				embed_size=predefined_args['embed_size']					,				embed_dropout=predefined_args['embed_dropout']					,				word_embed=predefined_args['word_embed']					,				use_pooler=lowercase_					,				use_token_type_embed=lowercase_					,				token_type_vocab_size=predefined_args['token_type_vocab_size']					,				use_classifier=lowercase_					,				use_decoder=lowercase_					,				)

     original_bort.load_parameters(lowercase_					,				cast_dtype=lowercase_					,				ignore_extra=lowercase_     )
     __A		:					Union[str, Any]						=			original_bort._collect_params_with_prefix()

     # Build our config 🤗
     __A		:					Tuple						=			{
         "architectures": ["BertForMaskedLM"],
         "attention_probs_dropout_prob": predefined_args["dropout"],
         "hidden_act": "gelu",
         "hidden_dropout_prob": predefined_args["dropout"],
         "hidden_size": predefined_args["embed_size"],
         "initializer_range": 0.02,
         "intermediate_size": predefined_args["hidden_size"],
         "layer_norm_eps": predefined_args["layer_norm_eps"],
         "max_position_embeddings": predefined_args["max_length"],
         "model_type": "bort",
         "num_attention_heads": predefined_args["num_heads"],
         "num_hidden_layers": predefined_args["num_layers"],
         "pad_token_id": 1,  # 2 = BERT, 1 = RoBERTa
         "type_vocab_size": 1,  # 2 = BERT, 1 = RoBERTa
         "vocab_size": len(lowercase_     ),
     }

     __A		:					Tuple						=			BertConfig.from_dict(lowercase_     )
     __A		:					str						=			BertForMaskedLM(lowercase_     )
     hf_bort_model.eval()

     # Parameter mapping table (Gluonnlp to Transformers)
     # * denotes layer index
     #
     # | Gluon Parameter                                                | Transformers Parameter
     # | -------------------------------------------------------------- | ----------------------
     # | `encoder.layer_norm.beta`                                      | `bert.embeddings.LayerNorm.bias`
     # | `encoder.layer_norm.gamma`                                     | `bert.embeddings.LayerNorm.weight`
     # | `encoder.position_weight`                                      | `bert.embeddings.position_embeddings.weight`
     # | `word_embed.0.weight`                                          | `bert.embeddings.word_embeddings.weight`
     # | `encoder.transformer_cells.*.attention_cell.proj_key.bias`     | `bert.encoder.layer.*.attention.self.key.bias`
     # | `encoder.transformer_cells.*.attention_cell.proj_key.weight`   | `bert.encoder.layer.*.attention.self.key.weight`
     # | `encoder.transformer_cells.*.attention_cell.proj_query.bias`   | `bert.encoder.layer.*.attention.self.query.bias`
     # | `encoder.transformer_cells.*.attention_cell.proj_query.weight` | `bert.encoder.layer.*.attention.self.query.weight`
     # | `encoder.transformer_cells.*.attention_cell.proj_value.bias`   | `bert.encoder.layer.*.attention.self.value.bias`
     # | `encoder.transformer_cells.*.attention_cell.proj_value.weight` | `bert.encoder.layer.*.attention.self.value.weight`
     # | `encoder.transformer_cells.*.ffn.ffn_2.bias`                   | `bert.encoder.layer.*.attention.output.dense.bias`
     # | `encoder.transformer_cells.*.ffn.ffn_2.weight`                 | `bert.encoder.layer.*.attention.output.dense.weight`
     # | `encoder.transformer_cells.*.layer_norm.beta`                  | `bert.encoder.layer.*.attention.output.LayerNorm.bias`
     # | `encoder.transformer_cells.*.layer_norm.gamma`                 | `bert.encoder.layer.*.attention.output.LayerNorm.weight`
     # | `encoder.transformer_cells.*.ffn.ffn_1.bias`                   | `bert.encoder.layer.*.intermediate.dense.bias`
     # | `encoder.transformer_cells.*.ffn.ffn_1.weight`                 | `bert.encoder.layer.*.intermediate.dense.weight`
     # | `encoder.transformer_cells.*.ffn.layer_norm.beta`              | `bert.encoder.layer.*.output.LayerNorm.bias`
     # | `encoder.transformer_cells.*.ffn.layer_norm.gamma`             | `bert.encoder.layer.*.output.LayerNorm.weight`
     # | `encoder.transformer_cells.*.proj.bias`                        | `bert.encoder.layer.*.output.dense.bias`
     # | `encoder.transformer_cells.*.proj.weight`                      | `bert.encoder.layer.*.output.dense.weight`

     # Helper function to convert MXNET Arrays to PyTorch
     def to_torch(__snake_case    :					List[Any]     ) -> nn.Parameter:
          return nn.Parameter(torch.FloatTensor(mx_array.data().asnumpy()     )     )

     # Check param shapes and map new HF param back
     def check_and_map_params(__snake_case    :					int					,				__snake_case    :					Optional[int]     ):
          __A		:					Dict						=			hf_param.shape

          __A		:					List[str]						=			to_torch(params[gluon_param]     )
          __A		:					Union[str, Any]						=			gluon_param.shape

          assert (
              shape_hf == shape_gluon
          ), f'The gluon parameter {gluon_param} has shape {shape_gluon}, but expects shape {shape_hf} for Transformers'

          return gluon_param

     __A		:					Dict						=			check_and_map_params(
         hf_bort_model.bert.embeddings.word_embeddings.weight					,				'word_embed.0.weight'     )
     __A		:					int						=			check_and_map_params(
         hf_bort_model.bert.embeddings.position_embeddings.weight					,				'encoder.position_weight'     )
     __A		:					Optional[Any]						=			check_and_map_params(
         hf_bort_model.bert.embeddings.LayerNorm.bias					,				'encoder.layer_norm.beta'     )
     __A		:					str						=			check_and_map_params(
         hf_bort_model.bert.embeddings.LayerNorm.weight					,				'encoder.layer_norm.gamma'     )

     # Inspired by RoBERTa conversion script, we just zero them out (Bort does not use them)
     __A		:					str						=			torch.zeros_like(
         hf_bort_model.bert.embeddings.token_type_embeddings.weight.data     )

     for i in range(hf_bort_config.num_hidden_layers     ):
          __A		:					BertLayer						=			hf_bort_model.bert.encoder.layer[i]

          # self attention
          __A		:					BertSelfAttention						=			layer.attention.self

          __A		:					str						=			check_and_map_params(
              self_attn.key.bias.data					,				f'encoder.transformer_cells.{i}.attention_cell.proj_key.bias'     )

          __A		:					Optional[int]						=			check_and_map_params(
              self_attn.key.weight.data					,				f'encoder.transformer_cells.{i}.attention_cell.proj_key.weight'     )
          __A		:					Optional[Any]						=			check_and_map_params(
              self_attn.query.bias.data					,				f'encoder.transformer_cells.{i}.attention_cell.proj_query.bias'     )
          __A		:					Any						=			check_and_map_params(
              self_attn.query.weight.data					,				f'encoder.transformer_cells.{i}.attention_cell.proj_query.weight'     )
          __A		:					List[Any]						=			check_and_map_params(
              self_attn.value.bias.data					,				f'encoder.transformer_cells.{i}.attention_cell.proj_value.bias'     )
          __A		:					Optional[int]						=			check_and_map_params(
              self_attn.value.weight.data					,				f'encoder.transformer_cells.{i}.attention_cell.proj_value.weight'     )

          # self attention output
          __A		:					BertSelfOutput						=			layer.attention.output

          __A		:					Optional[Any]						=			check_and_map_params(
              self_output.dense.bias					,				f'encoder.transformer_cells.{i}.proj.bias'     )
          __A		:					List[Any]						=			check_and_map_params(
              self_output.dense.weight					,				f'encoder.transformer_cells.{i}.proj.weight'     )
          __A		:					Union[str, Any]						=			check_and_map_params(
              self_output.LayerNorm.bias					,				f'encoder.transformer_cells.{i}.layer_norm.beta'     )
          __A		:					Tuple						=			check_and_map_params(
              self_output.LayerNorm.weight					,				f'encoder.transformer_cells.{i}.layer_norm.gamma'     )

          # intermediate
          __A		:					BertIntermediate						=			layer.intermediate

          __A		:					int						=			check_and_map_params(
              intermediate.dense.bias					,				f'encoder.transformer_cells.{i}.ffn.ffn_1.bias'     )
          __A		:					int						=			check_and_map_params(
              intermediate.dense.weight					,				f'encoder.transformer_cells.{i}.ffn.ffn_1.weight'     )

          # output
          __A		:					BertOutput						=			layer.output

          __A		:					List[Any]						=			check_and_map_params(
              bert_output.dense.bias					,				f'encoder.transformer_cells.{i}.ffn.ffn_2.bias'     )
          __A		:					List[str]						=			check_and_map_params(
              bert_output.dense.weight					,				f'encoder.transformer_cells.{i}.ffn.ffn_2.weight'     )
          __A		:					Union[str, Any]						=			check_and_map_params(
              bert_output.LayerNorm.bias					,				f'encoder.transformer_cells.{i}.ffn.layer_norm.beta'     )
          __A		:					Any						=			check_and_map_params(
              bert_output.LayerNorm.weight					,				f'encoder.transformer_cells.{i}.ffn.layer_norm.gamma'     )

     # Save space and energy 🎄
     hf_bort_model.half()

     # Compare output of both models
     __A		:					Optional[Any]						=			RobertaTokenizer.from_pretrained('roberta-base'     )

     __A		:					Optional[Any]						=			tokenizer.encode_plus(lowercase_     )["input_ids"]

     # Get gluon output
     __A		:					str						=			mx.nd.array([input_ids]     )
     __A		:					Union[str, Any]						=			original_bort(inputs=lowercase_					,				token_types=[]     )

     # Get Transformer output (save and reload model again)
     hf_bort_model.save_pretrained(lowercase_     )
     __A		:					int						=			BertModel.from_pretrained(lowercase_     )
     hf_bort_model.eval()

     __A		:					Optional[Any]						=			tokenizer.encode_plus(lowercase_					,				return_tensors='pt'     )
     __A		:					Optional[int]						=			hf_bort_model(**lowercase_     )[0]

     __A		:					List[Any]						=			output_gluon[0].asnumpy()
     __A		:					int						=			output_hf[0].detach().numpy()

     __A		:					Dict						=			np.max(np.abs(hf_layer - gluon_layer     )     ).item()
     __A		:					List[str]						=			np.allclose(lowercase_					,				lowercase_					,				atol=1e-3     )

     if success:
          print('✔️ Both model do output the same tensors'     )
     else:
          print('❌ Both model do **NOT** output the same tensors'     )
          print('Absolute difference is:'					,				lowercase_     )


if __name__ == "__main__":
  lowercase__       : Optional[Any]         =       argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      '''--bort_checkpoint_path''', default=None, type=str, required=True, help='''Path the official Bort params file.'''
  )
  parser.add_argument(
      '''--pytorch_dump_folder_path''', default=None, type=str, required=True, help='''Path to the output PyTorch model.'''
  )
  lowercase__       : Any         =       parser.parse_args()
  convert_bort_checkpoint_to_pytorch(args.bort_checkpoint_path, args.pytorch_dump_folder_path)






"""simple docstring"""


import os
import tempfile
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch


if is_torch_available():
  import torch
  from torch import nn

  from transformers import (
      Adafactor,
      AdamW,
      get_constant_schedule,
      get_constant_schedule_with_warmup,
      get_cosine_schedule_with_warmup,
      get_cosine_with_hard_restarts_schedule_with_warmup,
      get_inverse_sqrt_schedule,
      get_linear_schedule_with_warmup,
      get_polynomial_decay_schedule_with_warmup,
  )




def        lowercase__     (       lowercase_		,lowercase_=10      )       ->		Tuple:




      """simple docstring"""

      _UpperCamelCase   :						str    =							[]
      for _ in range(lowercase_      ):
            lrs.append(scheduler.get_lr()[0]      )
            scheduler.step()
      return lrs




def        lowercase__     (       lowercase_		,lowercase_=10      )       ->		str:




      """simple docstring"""

      _UpperCamelCase   :						Optional[int]    =							[]
      for step in range(lowercase_      ):
            lrs.append(scheduler.get_lr()[0]      )
            scheduler.step()
            if step == num_steps // 2:
                  with tempfile.TemporaryDirectory() as tmpdirname:
                        _UpperCamelCase   :						List[str]    =							os.path.join(lowercase_		,"schedule.bin"      )
                        torch.save(scheduler.state_dict()		,lowercase_      )

                        _UpperCamelCase   :						Optional[int]    =							torch.load(lowercase_      )
                        scheduler.load_state_dict(lowercase_      )
      return lrs



@require_torch
class   __SCREAMING_SNAKE_CASE     (    unittest.TestCase ):







  '''simple docstring'''


  def  __SCREAMING_SNAKE_CASE					(		self :							Dict       , __a :							Dict       , __a :							Any       , __a :							List[str]						)		->					List[str]:
        self.assertEqual(len(__a						)       , len(__a						)						)
        for a, b in zip(__a       , __a						):
              self.assertAlmostEqual(__a       , __a       , delta=__a						)

  def  __SCREAMING_SNAKE_CASE					(		self :							str						)		->					str:
        _UpperCamelCase   :						Optional[int]    =							torch.tensor([0.1, -0.2, -0.1]       , requires_grad=__a						)
        _UpperCamelCase   :						List[Any]    =							torch.tensor([0.4, 0.2, -0.5]						)
        _UpperCamelCase   :						Any    =							nn.MSELoss()
        # No warmup, constant schedule, no gradient clipping
        _UpperCamelCase   :						Dict    =							AdamW(params=[w]       , lr=2e-1       , weight_decay=0.0						)
        for _ in range(100						):
              _UpperCamelCase   :						List[Any]    =							criterion(__a       , __a						)
              loss.backward()
              optimizer.step()
              w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
              w.grad.zero_()
        self.assertListAlmostEqual(w.tolist()       , [0.4, 0.2, -0.5]       , tol=1e-2						)


  def  __SCREAMING_SNAKE_CASE					(		self :							List[str]						)		->					Tuple:
        _UpperCamelCase   :						List[Any]    =							torch.tensor([0.1, -0.2, -0.1]       , requires_grad=__a						)
        _UpperCamelCase   :						Any    =							torch.tensor([0.4, 0.2, -0.5]						)
        _UpperCamelCase   :						List[str]    =							nn.MSELoss()
        # No warmup, constant schedule, no gradient clipping
        _UpperCamelCase   :						Dict    =							Adafactor(
            params=[w]       , lr=1e-2       , eps=(1e-3_0, 1e-3)       , clip_threshold=1.0       , decay_rate=-0.8       , betaa=__a       , weight_decay=0.0       , relative_step=__a       , scale_parameter=__a       , warmup_init=__a       , )
        for _ in range(1000						):
              _UpperCamelCase   :						Optional[Any]    =							criterion(__a       , __a						)
              loss.backward()
              optimizer.step()
              w.grad.detach_()  # No zero_grad() function on simple tensors. we do it ourselves.
              w.grad.zero_()
        self.assertListAlmostEqual(w.tolist()       , [0.4, 0.2, -0.5]       , tol=1e-2						)





@require_torch
class   __SCREAMING_SNAKE_CASE     (    unittest.TestCase ):







  '''simple docstring'''


  SCREAMING_SNAKE_CASE__				:Optional[Any]						  =       nn.Linear(50		,	50 ) if is_torch_available() else None
  SCREAMING_SNAKE_CASE__				:List[Any]						  =       AdamW(m.parameters()		,	lr=10.0 ) if is_torch_available() else None
  SCREAMING_SNAKE_CASE__				:List[str]						  =       10

  def  __SCREAMING_SNAKE_CASE					(		self :							List[Any]       , __a :							Tuple       , __a :							List[Any]       , __a :							Dict       , __a :							str=None						)		->					List[Any]:
        self.assertEqual(len(__a						)       , len(__a						)						)
        for a, b in zip(__a       , __a						):
              self.assertAlmostEqual(__a       , __a       , delta=__a       , msg=__a						)


  def  __SCREAMING_SNAKE_CASE					(		self :							Optional[Any]						)		->					List[Any]:
        _UpperCamelCase   :						Union[str, Any]    =							{"num_warmup_steps": 2, "num_training_steps": 10}
        # schedulers doct format
        # function: (sched_args_dict, expected_learning_rates)
        _UpperCamelCase   :						Optional[Any]    =							{
            get_constant_schedule: ({}, [10.0] * self.num_steps),
            get_constant_schedule_with_warmup: (
                {"num_warmup_steps": 4},
                [0.0, 2.5, 5.0, 7.5, 10.0, 10.0, 10.0, 10.0, 10.0, 10.0],
            ),
            get_linear_schedule_with_warmup: (
                {**common_kwargs},
                [0.0, 5.0, 10.0, 8.75, 7.5, 6.25, 5.0, 3.75, 2.5, 1.25],
            ),
            get_cosine_schedule_with_warmup: (
                {**common_kwargs},
                [0.0, 5.0, 10.0, 9.61, 8.53, 6.91, 5.0, 3.08, 1.46, 0.38],
            ),
            get_cosine_with_hard_restarts_schedule_with_warmup: (
                {**common_kwargs, "num_cycles": 2},
                [0.0, 5.0, 10.0, 8.53, 5.0, 1.46, 10.0, 8.53, 5.0, 1.46],
            ),
            get_polynomial_decay_schedule_with_warmup: (
                {**common_kwargs, "power": 2.0, "lr_end": 1e-7},
                [0.0, 5.0, 10.0, 7.6_56, 5.6_25, 3.9_06, 2.5, 1.4_06, 0.6_25, 0.1_56],
            ),
            get_inverse_sqrt_schedule: (
                {"num_warmup_steps": 2},
                [0.0, 5.0, 10.0, 8.1_65, 7.0_71, 6.3_25, 5.7_74, 5.3_45, 5.0, 4.7_14],
            ),
        }

        for scheduler_func, data in scheds.items():
              _UpperCamelCase,	_UpperCamelCase   :						List[str]    =							data

              _UpperCamelCase   :						List[str]    =							scheduler_func(self.optimizer       , **__a						)
              self.assertEqual(len([scheduler.get_lr()[0]]						)       , 1						)
              _UpperCamelCase   :						Optional[int]    =							unwrap_schedule(__a       , self.num_steps						)
              self.assertListAlmostEqual(
                  __a       , __a       , tol=1e-2       , msg=F'''failed for {scheduler_func} in normal scheduler'''       , )

              _UpperCamelCase   :						List[str]    =							scheduler_func(self.optimizer       , **__a						)
              if scheduler_func.__name__ != "get_constant_schedule":
                    LambdaScheduleWrapper.wrap_scheduler(__a						)  # wrap to test picklability of the schedule
              _UpperCamelCase   :						int    =							unwrap_and_save_reload_schedule(__a       , self.num_steps						)
              self.assertListEqual(__a       , __a       , msg=F'''failed for {scheduler_func} in save and reload'''						)





class   __SCREAMING_SNAKE_CASE     :







  '''simple docstring'''


  def __init__(		self :							List[Any]       , __a :							Optional[int]						)		->					Tuple:
        _UpperCamelCase   :						Any    =							fn

  def __call__(		self :							Tuple       , *__a :							Optional[Any]       , **__a :							str						)		->					Any:
        return self.fn(*__a       , **__a						)


  @classmethod
  def  __SCREAMING_SNAKE_CASE					(		self :							Any       , __a :							Any						)		->					Tuple:
        _UpperCamelCase   :						Any    =							list(map(self       , scheduler.lr_lambdas						)						)
