


'''simple docstring'''



import argparse
import collections
from pathlib import Path

import torch
from torch.serialization import default_restore_location

from transformers import BertConfig, DPRConfig, DPRContextEncoder, DPRQuestionEncoder, DPRReader


A__		: Optional[Any]												=  collections.namedtuple(
    """CheckpointState""", ["""model_dict""", """optimizer_dict""", """scheduler_dict""", """offset""", """epoch""", """encoder_params"""]
)



def       UpperCAmelCase__ (     UpperCAmelCase_   :    str							)					->		CheckpointState:
     print(F'Reading saved model from {model_file}'							)
     __lowerCamelCase       :     int        =		torch.load(UpperCAmelCase_				,   map_location=lambda UpperCAmelCase_				,   UpperCAmelCase_							: default_restore_location(UpperCAmelCase_				,   'cpu'							)							)
     return CheckpointState(**UpperCAmelCase_							)





class        UpperCAmelCase_  :





    """simple docstring"""






    def __init__(							self      ,					SCREAMING_SNAKE_CASE_     )    ->					Dict:
         __lowerCamelCase       :     List[Any]        =		src_file







    def 				lowercase_    (							self     )    ->					Union[str, Any]:
         raise NotImplementedError







    @staticmethod
    def 				lowercase_    (							SCREAMING_SNAKE_CASE_      ,					*SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     )    ->					"DPRState":
         if comp_type.startswith('c'     ):
              return DPRContextEncoderState(*SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     )
         if comp_type.startswith('q'     ):
              return DPRQuestionEncoderState(*SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     )
         if comp_type.startswith('r'     ):
              return DPRReaderState(*SCREAMING_SNAKE_CASE_      ,					**SCREAMING_SNAKE_CASE_     )
         else:
              raise ValueError('Component type must be either \'ctx_encoder\', \'question_encoder\' or \'reader\'.'     )







class        UpperCAmelCase_  (_UpperCAmelCase  ):





    """simple docstring"""






    def 				lowercase_    (							self     )    ->					List[Any]:
         __lowerCamelCase       :     Tuple        =		DPRContextEncoder(DPRConfig(**BertConfig.get_config_dict('bert-base-uncased'     )[0]     )     )
         print(f'Loading DPR biencoder from {self.src_file}'     )
         __lowerCamelCase       :     Tuple        =		load_states_from_checkpoint(self.src_file     )
         __lowerCamelCase    ,	__lowerCamelCase       :     Optional[int]        =		model.ctx_encoder, 'ctx_model.'
         # Fix changes from https://github.com/huggingface/transformers/commit/614fef1691edb806de976756d4948ecbcd0c0ca3
         __lowerCamelCase       :     Optional[Any]        =		{'bert_model.embeddings.position_ids': model.ctx_encoder.bert_model.embeddings.position_ids}
         for key, value in saved_state.model_dict.items():
              if key.startswith(SCREAMING_SNAKE_CASE_     ):
                   __lowerCamelCase       :     Any        =		key[len(SCREAMING_SNAKE_CASE_     ) :]
                   if not key.startswith('encode_proj.'     ):
                        __lowerCamelCase       :     str        =		'bert_model.' + key
                   __lowerCamelCase       :     List[str]        =		value
         encoder.load_state_dict(SCREAMING_SNAKE_CASE_     )
         return model







class        UpperCAmelCase_  (_UpperCAmelCase  ):





    """simple docstring"""






    def 				lowercase_    (							self     )    ->					Dict:
         __lowerCamelCase       :     int        =		DPRQuestionEncoder(DPRConfig(**BertConfig.get_config_dict('bert-base-uncased'     )[0]     )     )
         print(f'Loading DPR biencoder from {self.src_file}'     )
         __lowerCamelCase       :     int        =		load_states_from_checkpoint(self.src_file     )
         __lowerCamelCase    ,	__lowerCamelCase       :     Union[str, Any]        =		model.question_encoder, 'question_model.'
         # Fix changes from https://github.com/huggingface/transformers/commit/614fef1691edb806de976756d4948ecbcd0c0ca3
         __lowerCamelCase       :     Optional[int]        =		{'bert_model.embeddings.position_ids': model.question_encoder.bert_model.embeddings.position_ids}
         for key, value in saved_state.model_dict.items():
              if key.startswith(SCREAMING_SNAKE_CASE_     ):
                   __lowerCamelCase       :     str        =		key[len(SCREAMING_SNAKE_CASE_     ) :]
                   if not key.startswith('encode_proj.'     ):
                        __lowerCamelCase       :     Dict        =		'bert_model.' + key
                   __lowerCamelCase       :     Tuple        =		value
         encoder.load_state_dict(SCREAMING_SNAKE_CASE_     )
         return model







class        UpperCAmelCase_  (_UpperCAmelCase  ):





    """simple docstring"""






    def 				lowercase_    (							self     )    ->					List[Any]:
         __lowerCamelCase       :     List[str]        =		DPRReader(DPRConfig(**BertConfig.get_config_dict('bert-base-uncased'     )[0]     )     )
         print(f'Loading DPR reader from {self.src_file}'     )
         __lowerCamelCase       :     List[str]        =		load_states_from_checkpoint(self.src_file     )
         # Fix changes from https://github.com/huggingface/transformers/commit/614fef1691edb806de976756d4948ecbcd0c0ca3
         __lowerCamelCase       :     Optional[int]        =		{
             'encoder.bert_model.embeddings.position_ids': model.span_predictor.encoder.bert_model.embeddings.position_ids
         }
         for key, value in saved_state.model_dict.items():
              if key.startswith('encoder.'     ) and not key.startswith('encoder.encode_proj'     ):
                   __lowerCamelCase       :     Optional[int]        =		'encoder.bert_model.' + key[len('encoder.'     ) :]
              __lowerCamelCase       :     List[Any]        =		value
         model.span_predictor.load_state_dict(SCREAMING_SNAKE_CASE_     )
         return model







def       UpperCAmelCase__ (     UpperCAmelCase_   :    str				,   UpperCAmelCase_   :    Path				,   UpperCAmelCase_   :    Path							)					->		Optional[Any]:
     __lowerCamelCase       :     Optional[Any]        =		Path(UpperCAmelCase_							)
     dest_dir.mkdir(exist_ok=UpperCAmelCase_							)

     __lowerCamelCase       :     List[Any]        =		DPRState.from_type(UpperCAmelCase_				,   src_file=UpperCAmelCase_							)
     __lowerCamelCase       :     Union[str, Any]        =		dpr_state.load_dpr_model()
     model.save_pretrained(UpperCAmelCase_							)
     model.from_pretrained(UpperCAmelCase_							)  # sanity check


if __name__ == "__main__":
      A__		: Union[str, Any]												=  argparse.ArgumentParser()
      # Required parameters
      parser.add_argument(
          """--type""", type=str, help="""Type of the component to convert: 'ctx_encoder', 'question_encoder' or 'reader'."""
      )
      parser.add_argument(
          """--src""",
          type=str,
          help=(
              """Path to the dpr checkpoint file. They can be downloaded from the official DPR repo"""
              """ https://github.com/facebookresearch/DPR. Note that in the official repo, both encoders are stored in the"""
              """ 'retriever' checkpoints."""
          ),
      )
      parser.add_argument("""--dest""", type=str, default=None, help="""Path to the output PyTorch model directory.""")
      A__		: List[str]												=  parser.parse_args()

      A__		: Optional[Any]												=  Path(args.src)
      A__		: List[Any]												=  f'''converted-{src_file.name}''' if args.dest is None else args.dest
      A__		: Tuple												=  Path(dest_dir)
      assert src_file.exists()
      assert (
          args.type is not None
      ), "Please specify the component type of the DPR model to convert: 'ctx_encoder', 'question_encoder' or 'reader'."
      convert(args.type, src_file, dest_dir)










'''simple docstring'''



import torch

from transformers import AutoTokenizer





class        UpperCAmelCase_  (_UpperCAmelCase  ):





    """simple docstring"""






    def __init__(							self      ,					SCREAMING_SNAKE_CASE_     )    ->					int:
         __lowerCamelCase       :     Dict        =		AutoTokenizer.from_pretrained(SCREAMING_SNAKE_CASE_     )







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_     )    ->					int:

         if isinstance(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     ) and all(isinstance(_x      ,					SCREAMING_SNAKE_CASE_     ) for _x in x     ):
              __lowerCamelCase       :     Union[str, Any]        =		None
              for l in x:
                   __lowerCamelCase       :     Tuple        =		self.tokenizer(
                       SCREAMING_SNAKE_CASE_      ,					padding='max_length'      ,					max_length=3_84      ,					truncation=SCREAMING_SNAKE_CASE_      ,					return_tensors='pt'      ,					)
                   __lowerCamelCase       :     str        =		torch.tensor([len(SCREAMING_SNAKE_CASE_     )]     )
                   if d is not None:
                        for k in d.keys():
                             __lowerCamelCase       :     List[Any]        =		torch.cat((d[k], t[k])      ,					0     )
                   else:
                        __lowerCamelCase       :     str        =		t

              __lowerCamelCase       :     List[Any]        =		torch.tensor(self.tokenizer.convert_tokens_to_ids('[E]'     )     )
              __lowerCamelCase       :     Tuple        =		torch.tensor(self.tokenizer.convert_tokens_to_ids('[/E]'     )     )

         elif isinstance(SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_     ) and all(isinstance(_x      ,					SCREAMING_SNAKE_CASE_     ) for _x in x     ):
              __lowerCamelCase       :     Optional[Any]        =		self.tokenizer(
                  SCREAMING_SNAKE_CASE_      ,					padding='max_length'      ,					max_length=3_84      ,					truncation=SCREAMING_SNAKE_CASE_      ,					return_tensors='pt'      ,					)

         else:
              raise Exception(
                  'Type of parameter x was not recognized! Only `list of strings` for query or `list of lists of'
                  ' strings` for supports are supported.'     )

         return d







    def 				lowercase_    (							self      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_      ,					SCREAMING_SNAKE_CASE_=0.7_0     )    ->					str:
         __lowerCamelCase       :     str        =		[]
         for idx in range(len(W_query['input_ids']     )     ):
              __lowerCamelCase       :     Optional[Any]        =		range(p_start.shape[1]     )

              __lowerCamelCase       :     List[Any]        =		[]
              for start_id in start_indexes:
                   for end_id in end_indexes:
                        if start_id < end_id:
                             output.append(
                                 (
                                     start_id,
                                     end_id,
                                     p_start[idx][start_id].item(),
                                     p_end[idx][end_id].item(),
                                 )     )

              output.sort(key=lambda SCREAMING_SNAKE_CASE_     : (tup[2] * tup[3])      ,					reverse=SCREAMING_SNAKE_CASE_     )
              __lowerCamelCase       :     str        =		[]
              for k in range(len(SCREAMING_SNAKE_CASE_     )     ):
                   if output[k][2] * output[k][3] >= thresh:
                        __lowerCamelCase    ,	__lowerCamelCase       :     Union[str, Any]        =		output[k][0], output[k][1]
                        __lowerCamelCase       :     List[str]        =		self.tokenizer.decode(W_query['input_ids'][idx][c_start_pos:c_end_pos]     )
                        temp.append((decoded, output[k][2] * output[k][3])     )

              final_outputs.append(SCREAMING_SNAKE_CASE_     )

         return final_outputs






