






import argparse
import json
import os

import torch

from transformers import LukeConfig, LukeModel, LukeTokenizer, RobertaTokenizer
from transformers.tokenization_utils_base import AddedToken


@torch.no_grad()
def 					snake_case__				(				UpperCAmelCase				:	Tuple     ,      UpperCAmelCase				:	int     ,      UpperCAmelCase				:	Dict     ,      UpperCAmelCase				:	List[Any]     ,      UpperCAmelCase				:	Tuple							):
    # Load configuration defined in the metadata file
    with open(UpperCAmelCase							) as metadata_file:
        lowerCAmelCase__ :Optional[int]   				=					json.load(UpperCAmelCase							)
    lowerCAmelCase__ :Optional[int]   				=					LukeConfig(use_entity_aware_attention=UpperCAmelCase     ,      **metadata["model_config"]							)

    # Load in the weights from the checkpoint_path
    lowerCAmelCase__ :Dict   				=					torch.load(UpperCAmelCase     ,      map_location="cpu"							)

    # Load the entity vocab file
    lowerCAmelCase__ :List[str]   				=					load_entity_vocab(UpperCAmelCase							)

    lowerCAmelCase__ :Union[str, Any]   				=					RobertaTokenizer.from_pretrained(metadata["model_config"]["bert_model_name"]							)

    # Add special tokens to the token vocabulary for downstream tasks
    lowerCAmelCase__ :Any   				=					AddedToken("<ent>"     ,      lstrip=UpperCAmelCase     ,      rstrip=UpperCAmelCase							)
    lowerCAmelCase__ :Tuple   				=					AddedToken("<ent2>"     ,      lstrip=UpperCAmelCase     ,      rstrip=UpperCAmelCase							)
    tokenizer.add_special_tokens({"additional_special_tokens": [entity_token_a, entity_token_a]}							)
    config.vocab_size += 2

    print(F'''Saving tokenizer to {pytorch_dump_folder_path}'''							)
    tokenizer.save_pretrained(UpperCAmelCase							)
    with open(os.path.join(UpperCAmelCase     ,      LukeTokenizer.vocab_files_names["entity_vocab_file"]							)     ,      "w"							) as f:
        json.dump(UpperCAmelCase     ,      UpperCAmelCase							)

    lowerCAmelCase__ :int   				=					LukeTokenizer.from_pretrained(UpperCAmelCase							)

    # Initialize the embeddings of the special tokens
    lowerCAmelCase__ :Any   				=					state_dict["""embeddings.word_embeddings.weight"""]
    lowerCAmelCase__ :Any   				=					word_emb[tokenizer.convert_tokens_to_ids(["@"]							)[0]].unsqueeze(0							)
    lowerCAmelCase__ :List[Any]   				=					word_emb[tokenizer.convert_tokens_to_ids(["#"]							)[0]].unsqueeze(0							)
    lowerCAmelCase__ :List[Any]   				=					torch.cat([word_emb, ent_emb, enta_emb]							)

    # Initialize the query layers of the entity-aware self-attention mechanism
    for layer_index in range(config.num_hidden_layers							):
        for matrix_name in ["query.weight", "query.bias"]:
            lowerCAmelCase__ :Any   				=					F'''encoder.layer.{layer_index}.attention.self.'''
            lowerCAmelCase__ :List[Any]   				=					state_dict[prefix + matrix_name]
            lowerCAmelCase__ :Dict   				=					state_dict[prefix + matrix_name]
            lowerCAmelCase__ :Optional[int]   				=					state_dict[prefix + matrix_name]

    # Initialize the embedding of the [MASK2] entity using that of the [MASK] entity for downstream tasks
    lowerCAmelCase__ :List[str]   				=					state_dict["""entity_embeddings.entity_embeddings.weight"""]
    lowerCAmelCase__ :int   				=					entity_emb[entity_vocab["""[MASK]"""]]

    lowerCAmelCase__ :Tuple   				=					LukeModel(config=UpperCAmelCase							).eval()

    lowerCAmelCase__ :Optional[Any]   				=					model.load_state_dict(UpperCAmelCase     ,      strict=UpperCAmelCase							)
    if not (len(UpperCAmelCase							) == 1 and missing_keys[0] == "embeddings.position_ids"):
        raise ValueError(F'''Missing keys {", ".join(UpperCAmelCase							)}. Expected only missing embeddings.position_ids'''							)
    if not (all(key.startswith("entity_predictions"							) or key.startswith("lm_head"							) for key in unexpected_keys							)):
        raise ValueError(
            "Unexpected keys"
            F''' {", ".join([key for key in unexpected_keys if not (key.startswith("entity_predictions"							) or key.startswith("lm_head"							))]							)}'''							)

    # Check outputs
    lowerCAmelCase__ :Optional[Any]   				=					LukeTokenizer.from_pretrained(UpperCAmelCase     ,      task="entity_classification"							)

    lowerCAmelCase__ :Dict   				=					(
        """Top seed Ana Ivanovic said on Thursday she could hardly believe her luck as a fortuitous netcord helped the"""
        """ new world number one avoid a humiliating second- round exit at Wimbledon ."""
    )
    lowerCAmelCase__ :List[Any]   				=					(3_9, 4_2)
    lowerCAmelCase__ :str   				=					tokenizer(UpperCAmelCase     ,      entity_spans=[span]     ,      add_prefix_space=UpperCAmelCase     ,      return_tensors="pt"							)

    lowerCAmelCase__ :List[str]   				=					model(**UpperCAmelCase							)

    # Verify word hidden states
    if model_size == "large":
        lowerCAmelCase__ :Optional[Any]   				=					torch.Size((1, 4_2, 1_0_2_4)							)
        lowerCAmelCase__ :Union[str, Any]   				=					torch.tensor(
            [[0.0133, 0.0865, 0.0095], [0.3093, -0.2576, -0.7418], [-0.1720, -0.2117, -0.2869]]							)
    else:  # base
        lowerCAmelCase__ :Optional[int]   				=					torch.Size((1, 4_2, 7_6_8)							)
        lowerCAmelCase__ :Optional[Any]   				=					torch.tensor([[0.0037, 0.1368, -0.0091], [0.1099, 0.3329, -0.1095], [0.0765, 0.5335, 0.1179]]							)

    if not (outputs.last_hidden_state.shape == expected_shape):
        raise ValueError(
            F'''Outputs.last_hidden_state.shape is {outputs.last_hidden_state.shape}, Expected shape is {expected_shape}'''							)
    if not torch.allclose(outputs.last_hidden_state[0, :3, :3]     ,      UpperCAmelCase     ,      atol=1E-4							):
        raise ValueError

    # Verify entity hidden states
    if model_size == "large":
        lowerCAmelCase__ :Union[str, Any]   				=					torch.Size((1, 1, 1_0_2_4)							)
        lowerCAmelCase__ :str   				=					torch.tensor([[0.0466, -0.0106, -0.0179]]							)
    else:  # base
        lowerCAmelCase__ :Tuple   				=					torch.Size((1, 1, 7_6_8)							)
        lowerCAmelCase__ :List[str]   				=					torch.tensor([[0.1457, 0.1044, 0.0174]]							)

    if not (outputs.entity_last_hidden_state.shape != expected_shape):
        raise ValueError(
            F'''Outputs.entity_last_hidden_state.shape is {outputs.entity_last_hidden_state.shape}, Expected shape is'''
            F''' {expected_shape}'''							)
    if not torch.allclose(outputs.entity_last_hidden_state[0, :3, :3]     ,      UpperCAmelCase     ,      atol=1E-4							):
        raise ValueError

    # Finally, save our PyTorch model and tokenizer
    print("Saving PyTorch model to {}".format(UpperCAmelCase							)							)
    model.save_pretrained(UpperCAmelCase							)


def 					snake_case__				(				UpperCAmelCase				:	Optional[int]							):
    lowerCAmelCase__ :List[str]   				=					{}
    with open(UpperCAmelCase     ,      "r"     ,      encoding="utf-8"							) as f:
        for index, line in enumerate(UpperCAmelCase							):
            lowerCAmelCase__ :Tuple   				=					line.rstrip().split("\t"							)
            lowerCAmelCase__ :int   				=					index

    return entity_vocab


if __name__ == "__main__":
 _a     : List[Any]   	=      argparse.ArgumentParser()
 # Required parameters
 parser.add_argument("""--checkpoint_path""", type=str, help="""Path to a pytorch_model.bin file.""")
 parser.add_argument(
     """--metadata_path""", default=None, type=str, help="""Path to a metadata.json file, defining the configuration."""
 )
 parser.add_argument(
     """--entity_vocab_path""",
     default=None,
     type=str,
     help="""Path to an entity_vocab.tsv file, containing the entity vocabulary.""",
 )
 parser.add_argument(
     """--pytorch_dump_folder_path""", default=None, type=str, help="""Path to where to dump the output PyTorch model."""
 )
 parser.add_argument(
     """--model_size""", default="""base""", type=str, choices=["""base""", """large"""], help="""Size of the model to be converted."""
 )
 _a     : Any   	=      parser.parse_args()
 convert_luke_checkpoint(
     args.checkpoint_path,
     args.metadata_path,
     args.entity_vocab_path,
     args.pytorch_dump_folder_path,
     args.model_size,
 )






"""simple docstring"""
import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


_A							=  logging.getLogger(__name__)

if is_torch_tpu_available(check_device=False):
    import torch_xla.core.xla_model as xm
    import torch_xla.debug.metrics as met

class       lowerCamelCase	(       lowerCAmelCase__   ):


  '''simple docstring'''







  def __init__(self							,       *_lowerCamelCase							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       **_lowerCamelCase	):


        """simple docstring"""

        super().__init__(*_lowerCamelCase							,       **_lowerCamelCase	)
        UpperCAmelCase__			:							Any							=    eval_examples
        UpperCAmelCase__			:							Union[str, Any]							=    post_process_function
        UpperCAmelCase__			:							int							=    quant_trainer_args
        UpperCAmelCase__			:							Dict							=    128  # default number of calibration samples



  def  _a      (self							,       _lowerCamelCase=None	):


        """simple docstring"""

        if calib_dataset is None and self.calib_dataset is None:
              raise ValueError("""Trainer: calibration requires an calib_dataset."""	)
        UpperCAmelCase__			:							List[str]							=    calib_dataset if calib_dataset is not None else self.calib_dataset

        UpperCAmelCase__			:							Optional[int]							=    self._remove_unused_columns(_lowerCamelCase							,       description="""Calibration"""	)

        return DataLoader(
            _lowerCamelCase							,       batch_size=self.args.eval_batch_size							,       collate_fn=self.data_collator							,       drop_last=self.args.dataloader_drop_last							,       num_workers=self.args.dataloader_num_workers							,       pin_memory=self.args.dataloader_pin_memory							,       shuffle=_lowerCamelCase							,       )



  def  _a      (self							,       _lowerCamelCase=None	):


        """simple docstring"""

        UpperCAmelCase__			:							int							=    self.train_dataset if calib_dataset is None else calib_dataset
        UpperCAmelCase__			:							List[str]							=    self.get_calib_dataloader(_lowerCamelCase	)

        UpperCAmelCase__			:							List[str]							=    self.model
        quant_trainer.configure_model(_lowerCamelCase							,       self.quant_trainer_args							,       calib=_lowerCamelCase	)
        model.eval()
        quant_trainer.enable_calibration(_lowerCamelCase	)

        logger.info("""***** Running calibration *****"""	)
        logger.info(F"""  Num examples = {self.calib_num}"""	)
        logger.info(F"""  Batch size = {calib_dataloader.batch_size}"""	)

        for step, inputs in enumerate(_lowerCamelCase	):
              # Prediction step
              UpperCAmelCase__					,		UpperCAmelCase__					,		UpperCAmelCase__			:							List[str]							=    self.prediction_step(_lowerCamelCase							,       _lowerCamelCase							,       prediction_loss_only=_lowerCamelCase	)
              if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
                    break

        quant_trainer.finish_calibration(_lowerCamelCase							,       self.quant_trainer_args	)
        UpperCAmelCase__			:							Tuple							=    model



  def  _a      (self							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       _lowerCamelCase=None							,       _lowerCamelCase = "eval"	):


        """simple docstring"""

        UpperCAmelCase__			:							Tuple							=    self.eval_dataset if eval_dataset is None else eval_dataset
        UpperCAmelCase__			:							List[Any]							=    self.get_eval_dataloader(_lowerCamelCase	)
        UpperCAmelCase__			:							Tuple							=    self.eval_examples if eval_examples is None else eval_examples

        # Temporarily disable metric computation, we will do it in the loop here.
        UpperCAmelCase__			:							Optional[Any]							=    self.compute_metrics
        UpperCAmelCase__			:							int							=    None
        UpperCAmelCase__			:							int							=    self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
        try:
              UpperCAmelCase__			:							str							=    eval_loop(
                  _lowerCamelCase							,       description="""Evaluation"""							,       prediction_loss_only=True if compute_metrics is None else None							,       ignore_keys=_lowerCamelCase							,       )
        finally:
              UpperCAmelCase__			:							List[str]							=    compute_metrics

        if self.post_process_function is not None and self.compute_metrics is not None:
              UpperCAmelCase__			:							int							=    self.post_process_function(_lowerCamelCase							,       _lowerCamelCase							,       output.predictions	)
              UpperCAmelCase__			:							List[Any]							=    self.compute_metrics(_lowerCamelCase	)

              # Prefix all keys with metric_key_prefix + '_'
              for key in list(metrics.keys()	):
                    if not key.startswith(F"""{metric_key_prefix}_"""	):
                          UpperCAmelCase__			:							Optional[int]							=    metrics.pop(_lowerCamelCase	)

              self.log(_lowerCamelCase	)
        else:
              UpperCAmelCase__			:							List[Any]							=    {}

        if self.args.tpu_metrics_debug or self.args.debug:
              # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
              xm.master_print(met.metrics_report()	)

        UpperCAmelCase__			:							Optional[int]							=    self.callback_handler.on_evaluate(self.args							,       self.state							,       self.control							,       _lowerCamelCase	)
        return metrics



  def  _a      (self							,       _lowerCamelCase							,       _lowerCamelCase							,       _lowerCamelCase=None							,       _lowerCamelCase = "test"	):


        """simple docstring"""

        UpperCAmelCase__			:							Union[str, Any]							=    self.get_test_dataloader(_lowerCamelCase	)

        # Temporarily disable metric computation, we will do it in the loop here.
        UpperCAmelCase__			:							Union[str, Any]							=    self.compute_metrics
        UpperCAmelCase__			:							Optional[Any]							=    None
        UpperCAmelCase__			:							Any							=    self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
        try:
              UpperCAmelCase__			:							Optional[Any]							=    eval_loop(
                  _lowerCamelCase							,       description="""Prediction"""							,       prediction_loss_only=True if compute_metrics is None else None							,       ignore_keys=_lowerCamelCase							,       )
        finally:
              UpperCAmelCase__			:							Dict							=    compute_metrics

        if self.post_process_function is None or self.compute_metrics is None:
              return output

        UpperCAmelCase__			:							Dict							=    self.post_process_function(_lowerCamelCase							,       _lowerCamelCase							,       output.predictions							,       """predict"""	)
        UpperCAmelCase__			:							Any							=    self.compute_metrics(_lowerCamelCase	)

        # Prefix all keys with metric_key_prefix + '_'
        for key in list(metrics.keys()	):
              if not key.startswith(F"""{metric_key_prefix}_"""	):
                    UpperCAmelCase__			:							Any							=    metrics.pop(_lowerCamelCase	)

        return PredictionOutput(predictions=predictions.predictions							,       label_ids=predictions.label_ids							,       metrics=_lowerCamelCase	)




  def  _a      (self							,       _lowerCamelCase="./"	):


        """simple docstring"""

        UpperCAmelCase__			:							Tuple							=    self.eval_dataset
        UpperCAmelCase__			:							int							=    self.get_eval_dataloader(_lowerCamelCase	)

        UpperCAmelCase__			:							Union[str, Any]							=    next(iter(_lowerCamelCase	)	)

        # saving device - to make it consistent
        UpperCAmelCase__			:							int							=    torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""	)

        # convert to tuple
        UpperCAmelCase__			:							List[Any]							=    tuple(v.to(_lowerCamelCase	) for k, v in batch.items()	)

        logger.info("""Converting model to be onnx compatible"""	)
        from pytorch_quantization.nn import TensorQuantizer

        UpperCAmelCase__			:							Optional[Any]							=    True

        UpperCAmelCase__			:							Tuple							=    self.model.to(_lowerCamelCase	)

        model.eval()
        model.float()

        UpperCAmelCase__			:							str							=    model.module if hasattr(_lowerCamelCase							,       """module"""	) else model
        quant_trainer.configure_model(_lowerCamelCase							,       self.quant_trainer_args	)

        UpperCAmelCase__			:							List[str]							=    os.path.join(_lowerCamelCase							,       """model.onnx"""	)
        logger.info(F"""exporting model to {output_model_file}"""	)

        UpperCAmelCase__			:							Optional[Any]							=    {0: """batch_size""", 1: """seq_len"""}

        torch.onnx.export(
            _lowerCamelCase							,       _lowerCamelCase							,       _lowerCamelCase							,       export_params=_lowerCamelCase							,       opset_version=13							,       do_constant_folding=_lowerCamelCase							,       input_names=["""input_ids""", """attention_mask""", """token_type_ids"""]							,       output_names=["""output_start_logits""", """output_end_logits"""]							,       dynamic_axes={
                """input_ids""": axes,
                """attention_mask""": axes,
                """token_type_ids""": axes,
                """output_start_logits""": axes,
                """output_end_logits""": axes,
            }							,       verbose=_lowerCamelCase							,       )
        logger.info("""onnx export finished"""	)

