





import argparse
import pathlib

import fairseq
import torch
from fairseq.models.roberta import RobertaModel as FairseqRobertaModel
from fairseq.modules import TransformerSentenceEncoderLayer
from packaging import version

from transformers import XLMRobertaConfig, XLMRobertaXLForMaskedLM, XLMRobertaXLForSequenceClassification
from transformers.models.bert.modeling_bert import (
    BertIntermediate,
    BertLayer,
    BertOutput,
    BertSelfAttention,
    BertSelfOutput,
)
from transformers.models.roberta.modeling_roberta import RobertaAttention
from transformers.utils import logging


if version.parse(fairseq.__version__) < version.parse("""1.0.0a"""):
      raise Exception("""requires fairseq >= 1.0.0a""")

logging.set_verbosity_info()
UpperCAmelCase_					:   List[str] 							=						logging.get_logger(__name__)

UpperCAmelCase_					:   Union[str, Any] 							=						"""Hello world! cécé herlolip"""






def 							_lowerCAmelCase (  _a   :					str					,   _a   :					str					,   _a   :					bool			)	->			int:
     lowerCAmelCase_					:       Dict    =		FairseqRobertaModel.from_pretrained(_a			)
     roberta.eval()  # disable dropout
     lowerCAmelCase_					:       Optional[Any]    =		roberta.model.encoder.sentence_encoder
     lowerCAmelCase_					:       int    =		XLMRobertaConfig(
         vocab_size=roberta_sent_encoder.embed_tokens.num_embeddings					,   hidden_size=roberta.cfg.model.encoder_embed_dim					,   num_hidden_layers=roberta.cfg.model.encoder_layers					,   num_attention_heads=roberta.cfg.model.encoder_attention_heads					,   intermediate_size=roberta.cfg.model.encoder_ffn_embed_dim					,   max_position_embeddings=5_14					,   type_vocab_size=1					,   layer_norm_eps=1E-5					,   )
     if classification_head:
          lowerCAmelCase_					:       Union[str, Any]    =		roberta.model.classification_heads["""mnli"""].out_proj.weight.shape[0]

     print("""Our RoBERTa config:"""					,   _a			)

     lowerCAmelCase_					:       Tuple    =		XLMRobertaXLForSequenceClassification(_a			) if classification_head else XLMRobertaXLForMaskedLM(_a			)
     model.eval()

     # Now let's copy all the weights.
     # Embeddings
     lowerCAmelCase_					:       int    =		roberta_sent_encoder.embed_tokens.weight
     lowerCAmelCase_					:       Union[str, Any]    =		roberta_sent_encoder.embed_positions.weight
     lowerCAmelCase_					:       List[str]    =		torch.zeros_like(
         model.roberta.embeddings.token_type_embeddings.weight			)  # just zero them out b/c RoBERTa doesn't use them.

     lowerCAmelCase_					:       str    =		roberta_sent_encoder.layer_norm.weight
     lowerCAmelCase_					:       Union[str, Any]    =		roberta_sent_encoder.layer_norm.bias

     for i in range(config.num_hidden_layers			):
          # Encoder: start of layer
          lowerCAmelCase_					:       BertLayer    =		model.roberta.encoder.layer[i]
          lowerCAmelCase_					:       TransformerSentenceEncoderLayer    =		roberta_sent_encoder.layers[i]

          lowerCAmelCase_					:       RobertaAttention    =		layer.attention
          lowerCAmelCase_					:       int    =		roberta_layer.self_attn_layer_norm.weight
          lowerCAmelCase_					:       Dict    =		roberta_layer.self_attn_layer_norm.bias

          # self attention
          lowerCAmelCase_					:       BertSelfAttention    =		layer.attention.self
          assert (
              roberta_layer.self_attn.k_proj.weight.data.shape
              == roberta_layer.self_attn.q_proj.weight.data.shape
              == roberta_layer.self_attn.v_proj.weight.data.shape
              == torch.Size((config.hidden_size, config.hidden_size)			)
          )

          lowerCAmelCase_					:       str    =		roberta_layer.self_attn.q_proj.weight
          lowerCAmelCase_					:       Dict    =		roberta_layer.self_attn.q_proj.bias
          lowerCAmelCase_					:       List[str]    =		roberta_layer.self_attn.k_proj.weight
          lowerCAmelCase_					:       Union[str, Any]    =		roberta_layer.self_attn.k_proj.bias
          lowerCAmelCase_					:       Optional[int]    =		roberta_layer.self_attn.v_proj.weight
          lowerCAmelCase_					:       int    =		roberta_layer.self_attn.v_proj.bias

          # self-attention output
          lowerCAmelCase_					:       BertSelfOutput    =		layer.attention.output
          assert self_output.dense.weight.shape == roberta_layer.self_attn.out_proj.weight.shape
          lowerCAmelCase_					:       List[Any]    =		roberta_layer.self_attn.out_proj.weight
          lowerCAmelCase_					:       Tuple    =		roberta_layer.self_attn.out_proj.bias

          # this one is final layer norm
          lowerCAmelCase_					:       List[Any]    =		roberta_layer.final_layer_norm.weight
          lowerCAmelCase_					:       List[str]    =		roberta_layer.final_layer_norm.bias

          # intermediate
          lowerCAmelCase_					:       BertIntermediate    =		layer.intermediate
          assert intermediate.dense.weight.shape == roberta_layer.fca.weight.shape
          lowerCAmelCase_					:       List[Any]    =		roberta_layer.fca.weight
          lowerCAmelCase_					:       Optional[Any]    =		roberta_layer.fca.bias

          # output
          lowerCAmelCase_					:       BertOutput    =		layer.output
          assert bert_output.dense.weight.shape == roberta_layer.fca.weight.shape
          lowerCAmelCase_					:       Tuple    =		roberta_layer.fca.weight
          lowerCAmelCase_					:       Any    =		roberta_layer.fca.bias
          # end of layer

     if classification_head:
          lowerCAmelCase_					:       str    =		roberta.model.classification_heads["""mnli"""].dense.weight
          lowerCAmelCase_					:       List[Any]    =		roberta.model.classification_heads["""mnli"""].dense.bias
          lowerCAmelCase_					:       Optional[Any]    =		roberta.model.classification_heads["""mnli"""].out_proj.weight
          lowerCAmelCase_					:       Tuple    =		roberta.model.classification_heads["""mnli"""].out_proj.bias
     else:
          # LM Head
          lowerCAmelCase_					:       Union[str, Any]    =		roberta.model.encoder.lm_head.dense.weight
          lowerCAmelCase_					:       Any    =		roberta.model.encoder.lm_head.dense.bias
          lowerCAmelCase_					:       List[Any]    =		roberta.model.encoder.lm_head.layer_norm.weight
          lowerCAmelCase_					:       List[str]    =		roberta.model.encoder.lm_head.layer_norm.bias
          lowerCAmelCase_					:       List[str]    =		roberta.model.encoder.lm_head.weight
          lowerCAmelCase_					:       Dict    =		roberta.model.encoder.lm_head.bias

     # Let's check that we get the same results.
     lowerCAmelCase_					:       torch.Tensor    =		roberta.encode(_a			).unsqueeze(0			)  # batch of size 1

     lowerCAmelCase_					:       Dict    =		model(_a			)[0]
     if classification_head:
          lowerCAmelCase_					:       int    =		roberta.model.classification_heads["""mnli"""](roberta.extract_features(_a			)			)
     else:
          lowerCAmelCase_					:       Dict    =		roberta.model(_a			)[0]
     print(our_output.shape					,   their_output.shape			)
     lowerCAmelCase_					:       Optional[Any]    =		torch.max(torch.abs(our_output - their_output			)			).item()
     print(F'max_absolute_diff = {max_absolute_diff}'			)  # ~ 1e-7
     lowerCAmelCase_					:       Tuple    =		torch.allclose(_a					,   _a					,   atol=1E-3			)
     print("""Do both models output the same tensors?"""					,   """🔥""" if success else """💩"""			)
     if not success:
          raise Exception("""Something went wRoNg"""			)

     pathlib.Path(_a			).mkdir(parents=_a					,   exist_ok=_a			)
     print(F'Saving model to {pytorch_dump_folder_path}'			)
     model.save_pretrained(_a			)


if __name__ == "__main__":
      UpperCAmelCase_					:   Optional[Any] 							=						argparse.ArgumentParser()
      # Required parameters
      parser.add_argument(
          """--roberta_checkpoint_path""", default=None, type=str, required=True, help="""Path the official PyTorch dump."""
      )
      parser.add_argument(
          """--pytorch_dump_folder_path""", default=None, type=str, required=True, help="""Path to the output PyTorch model."""
      )
      parser.add_argument(
          """--classification_head""", action="""store_true""", help="""Whether to convert a final classification head."""
      )
      UpperCAmelCase_					:   Optional[Any] 							=						parser.parse_args()
      convert_xlm_roberta_xl_checkpoint_to_pytorch(
          args.roberta_checkpoint_path, args.pytorch_dump_folder_path, args.classification_head
      )







import json
import os
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from requests.exceptions import HTTPError

from transformers.utils import (
    CONFIG_NAME,
    FLAX_WEIGHTS_NAME,
    TF2_WEIGHTS_NAME,
    TRANSFORMERS_CACHE,
    WEIGHTS_NAME,
    cached_file,
    get_file_from_repo,
    has_file,
)


UpperCAmelCase_					:   List[str] 							=						"""hf-internal-testing/tiny-random-bert"""
UpperCAmelCase_					:   List[Any] 							=						os.path.join(TRANSFORMERS_CACHE, """models--hf-internal-testing--tiny-random-bert""")
UpperCAmelCase_					:   Tuple 							=						"""9b8c223d42b2188cb49d29af482996f9d0f3e5a6"""





class      lowercase__    (							unittest.TestCase  ):







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						_lowercase						)
       # Should have downloaded the file in here
       self.assertTrue(os.path.isdir(_lowercase						)						)
       # Cache should contain at least those three subfolders:
       for subfolder in ["blobs", "refs", "snapshots"]:
            self.assertTrue(os.path.isdir(os.path.join(_lowercase    ,						_lowercase						)						)						)
       with open(os.path.join(_lowercase    ,						"""refs"""    ,						"""main"""						)						) as f:
            lowerCAmelCase_					:       Any    =		f.read()
       self.assertEqual(_lowercase    ,						os.path.join(_lowercase    ,						"""snapshots"""    ,						_lowercase    ,						_lowercase						)						)
       self.assertTrue(os.path.isfile(_lowercase						)						)

       # File is cached at the same place the second time.
       lowerCAmelCase_					:       Dict    =		cached_file(_lowercase    ,						_lowercase						)
       self.assertEqual(_lowercase    ,						_lowercase						)

       # Using a specific revision to test the full commit hash.
       lowerCAmelCase_					:       Optional[int]    =		cached_file(_lowercase    ,						_lowercase    ,						revision="""9b8c223"""						)
       self.assertEqual(_lowercase    ,						os.path.join(_lowercase    ,						"""snapshots"""    ,						_lowercase    ,						_lowercase						)						)







  def 	UpperCAmelCase__		(	self						):
       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid model identifier"""						):
            lowerCAmelCase_					:       Dict    =		cached_file("""tiny-random-bert"""    ,						_lowercase						)

       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid git identifier"""						):
            lowerCAmelCase_					:       Tuple    =		cached_file(_lowercase    ,						_lowercase    ,						revision="""aaaa"""						)

       with self.assertRaisesRegex(_lowercase    ,						"""does not appear to have a file named"""						):
            lowerCAmelCase_					:       List[Any]    =		cached_file(_lowercase    ,						"""conf"""						)







  def 	UpperCAmelCase__		(	self						):
       with self.assertRaisesRegex(_lowercase    ,						"""does not appear to have a file named"""						):
            lowerCAmelCase_					:       Any    =		cached_file(_lowercase    ,						"""conf"""						)

       with open(os.path.join(_lowercase    ,						"""refs"""    ,						"""main"""						)						) as f:
            lowerCAmelCase_					:       str    =		f.read()
       self.assertTrue(os.path.isfile(os.path.join(_lowercase    ,						""".no_exist"""    ,						_lowercase    ,						"""conf"""						)						)						)

       lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						"""conf"""    ,						_raise_exceptions_for_missing_entries=_lowercase						)
       self.assertIsNone(_lowercase						)

       lowerCAmelCase_					:       Any    =		cached_file(_lowercase    ,						"""conf"""    ,						local_files_only=_lowercase    ,						_raise_exceptions_for_missing_entries=_lowercase						)
       self.assertIsNone(_lowercase						)

       lowerCAmelCase_					:       Any    =		mock.Mock()
       lowerCAmelCase_					:       Optional[Any]    =		500
       lowerCAmelCase_					:       Any    =		{}
       lowerCAmelCase_					:       List[str]    =		HTTPError
       lowerCAmelCase_					:       List[str]    =		{}

       # Under the mock environment we get a 500 error when trying to reach the tokenizer.
       with mock.patch("""requests.Session.request"""    ,						return_value=_lowercase						) as mock_head:
            lowerCAmelCase_					:       Union[str, Any]    =		cached_file(_lowercase    ,						"""conf"""    ,						_raise_exceptions_for_connection_errors=_lowercase						)
            self.assertIsNone(_lowercase						)
            # This check we did call the fake head request
            mock_head.assert_called()







  def 	UpperCAmelCase__		(	self						):
       self.assertTrue(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)
       self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)
       self.assertFalse(has_file("""hf-internal-testing/tiny-bert-pt-only"""    ,						_lowercase						)						)







  def 	UpperCAmelCase__		(	self						):
       # `get_file_from_repo` returns None if the file does not exist
       self.assertIsNone(get_file_from_repo("""bert-base-cased"""    ,						"""ahah.txt"""						)						)

       # The function raises if the repository does not exist.
       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid model identifier"""						):
            get_file_from_repo("""bert-base-case"""    ,						_lowercase						)

       # The function raises if the revision does not exist.
       with self.assertRaisesRegex(_lowercase    ,						"""is not a valid git identifier"""						):
            get_file_from_repo("""bert-base-cased"""    ,						_lowercase    ,						revision="""ahaha"""						)

       lowerCAmelCase_					:       Union[str, Any]    =		get_file_from_repo("""bert-base-cased"""    ,						_lowercase						)
       # The name is the cached name which is not very easy to test, so instead we load the content.
       lowerCAmelCase_					:       Any    =		json.loads(open(_lowercase    ,						"""r"""						).read()						)
       self.assertEqual(config["""hidden_size"""]    ,						768						)







  def 	UpperCAmelCase__		(	self						):
       with tempfile.TemporaryDirectory() as tmp_dir:
            lowerCAmelCase_					:       Optional[Any]    =		Path(_lowercase						) / """a.txt"""
            filename.touch()
            self.assertEqual(get_file_from_repo(_lowercase    ,						"""a.txt"""						)    ,						str(_lowercase						)						)

            self.assertIsNone(get_file_from_repo(_lowercase    ,						"""b.txt"""						)						)
