



'''simple docstring'''





import argparse
import logging
from collections import namedtuple

import torch
from model_bertabs import BertAbsSummarizer
from models.model_builder import AbsSummarizer  # The authors' implementation

from transformers import BertTokenizer


logging.basicConfig(level=logging.INFO)
_lowerCAmelCase 		=      logging.getLogger(__name__)


_lowerCAmelCase 		=      '''Hello world! cécé herlolip'''


_lowerCAmelCase 		=      namedtuple(
    '''BertAbsConfig''',
    [
        '''temp_dir''',
        '''large''',
        '''use_bert_emb''',
        '''finetune_bert''',
        '''encoder''',
        '''share_emb''',
        '''max_pos''',
        '''enc_layers''',
        '''enc_hidden_size''',
        '''enc_heads''',
        '''enc_ff_size''',
        '''enc_dropout''',
        '''dec_layers''',
        '''dec_hidden_size''',
        '''dec_heads''',
        '''dec_ff_size''',
        '''dec_dropout''',
    ],
)
def   _SCREAMING_SNAKE_CASE    (			UpperCamelCase				,       UpperCamelCase     ):



    """simple docstring"""





    lowerCAmelCase__					:    int				 =   BertAbsConfig(
        temp_dir="""."""				,       finetune_bert=lowerCAmelCase_				,       large=lowerCAmelCase_				,       share_emb=lowerCAmelCase_				,       use_bert_emb=lowerCAmelCase_				,       encoder="""bert"""				,       max_pos=512				,       enc_layers=6				,       enc_hidden_size=512				,       enc_heads=8				,       enc_ff_size=512				,       enc_dropout=0.2				,       dec_layers=6				,       dec_hidden_size=768				,       dec_heads=8				,       dec_ff_size=2048				,       dec_dropout=0.2				,       )
    lowerCAmelCase__					:    int				 =   torch.load(lowerCAmelCase_				,       lambda UpperCamelCase				,       UpperCamelCase     : storage     )
    lowerCAmelCase__					:    Optional[Any]				 =   AbsSummarizer(lowerCAmelCase_				,       torch.device("""cpu"""     )				,       lowerCAmelCase_     )
    original.eval()

    lowerCAmelCase__					:    Optional[int]				 =   BertAbsSummarizer(lowerCAmelCase_				,       torch.device("""cpu"""     )     )
    new_model.eval()

    # -------------------
    # Convert the weights
    # -------------------

    logging.info("""convert the model"""     )
    new_model.bert.load_state_dict(original.bert.state_dict()     )
    new_model.decoder.load_state_dict(original.decoder.state_dict()     )
    new_model.generator.load_state_dict(original.generator.state_dict()     )

    # ----------------------------------
    # Make sure the outpus are identical
    # ----------------------------------

    logging.info("""Make sure that the models' outputs are identical"""     )
    lowerCAmelCase__					:    Union[str, Any]				 =   BertTokenizer.from_pretrained("""bert-base-uncased"""     )

    # prepare the model inputs
    lowerCAmelCase__					:    Union[str, Any]				 =   tokenizer.encode("""This is sample éàalj'-."""     )
    encoder_input_ids.extend([tokenizer.pad_token_id] * (512 - len(lowerCAmelCase_     ))     )
    lowerCAmelCase__					:    int				 =   torch.tensor(lowerCAmelCase_     ).unsqueeze(0     )
    lowerCAmelCase__					:    str				 =   tokenizer.encode("""This is sample 3 éàalj'-."""     )
    decoder_input_ids.extend([tokenizer.pad_token_id] * (512 - len(lowerCAmelCase_     ))     )
    lowerCAmelCase__					:    str				 =   torch.tensor(lowerCAmelCase_     ).unsqueeze(0     )

    # failsafe to make sure the weights reset does not affect the
    # loaded weights.
    assert torch.max(torch.abs(original.generator[0].weight - new_model.generator[0].weight     )     ) == 0

    # forward pass
    lowerCAmelCase__					:    str				 =   encoder_input_ids
    lowerCAmelCase__					:    List[Any]				 =   decoder_input_ids
    lowerCAmelCase__					:    Union[str, Any]				 =   None
    lowerCAmelCase__					:    Union[str, Any]				 =   None
    lowerCAmelCase__					:    Optional[Any]				 =   None
    lowerCAmelCase__					:    List[str]				 =   None
    lowerCAmelCase__					:    str				 =   None

    # The original model does not apply the geneator layer immediatly but rather in
    # the beam search (where it combines softmax + linear layer). Since we already
    # apply the softmax in our generation process we only apply the linear layer here.
    # We make sure that the outputs of the full stack are identical
    lowerCAmelCase__					:    int				 =   original(lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_     )[0]
    lowerCAmelCase__					:    List[str]				 =   original.generator(lowerCAmelCase_     )

    lowerCAmelCase__					:    int				 =   new_model(
        lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_				,       lowerCAmelCase_     )[0]
    lowerCAmelCase__					:    str				 =   new_model.generator(lowerCAmelCase_     )

    lowerCAmelCase__					:    Optional[int]				 =   torch.max(torch.abs(output_converted_model - output_original_model     )     ).item()
    print("""Maximum absolute difference beween weights: {:.2f}""".format(lowerCAmelCase_     )     )
    lowerCAmelCase__					:    int				 =   torch.max(torch.abs(output_converted_generator - output_original_generator     )     ).item()
    print("""Maximum absolute difference beween weights: {:.2f}""".format(lowerCAmelCase_     )     )

    lowerCAmelCase__					:    List[Any]				 =   torch.allclose(lowerCAmelCase_				,       lowerCAmelCase_				,       atol=1e-3     )
    if are_identical:
        logging.info("""all weights are equal up to 1e-3"""     )
    else:
        raise ValueError("""the weights are different. The new model is likely different from the original one."""     )

    # The model has been saved with torch.save(model) and this is bound to the exact
    # directory structure. We save the state_dict instead.
    logging.info("""saving the model's state dictionary"""     )
    torch.save(
        new_model.state_dict()				,       """./bertabs-finetuned-cnndm-extractive-abstractive-summarization/pytorch_model.bin"""     )


if __name__ == "__main__":
 _lowerCAmelCase 		=      argparse.ArgumentParser()
 parser.add_argument(
     '''--bertabs_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.''',
 )
 _lowerCAmelCase 		=      parser.parse_args()

 convert_bertabs_checkpoints(
     args.bertabs_checkpoint_path,
     args.pytorch_dump_folder_path,
 )






from __future__ import annotations



def       _lowerCAmelCase    (				lowerCAmelCase_				:int		,    lowerCAmelCase_				:int				)->list[str]:
   '''simple docstring'''






   if partitions <= 0:
      raise ValueError("partitions must be a positive number!"				)
   if partitions > number_of_bytes:
      raise ValueError("partitions can not > number_of_bytes!"				)
   snake_case_            =       number_of_bytes // partitions
   snake_case_            =       []
   for i in range(lowerCAmelCase_				):
      snake_case_            =       i * bytes_per_partition + 1
      snake_case_            =       (
          number_of_bytes if i == partitions - 1 else (i + 1) * bytes_per_partition
      )
      allocation_list.append(F'''{start_bytes}-{end_bytes}'''				)
   return allocation_list


if __name__ == "__main__":
   import doctest

   doctest.testmod()




