





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


lowerCAmelCase_        =			collections.namedtuple(
    """CheckpointState""", ["""model_dict""", """optimizer_dict""", """scheduler_dict""", """offset""", """epoch""", """encoder_params"""]
)


def 			lowerCamelCase_		(					lowerCAmelCase:		List[str] )->						CheckpointState:
  print(F"""Reading saved model from {model_file}""" )
  _snake_case      : int														=		torch.load(__UpperCamelCase			,							map_location=lambda lowerCAmelCase			,							lowerCAmelCase : default_restore_location(__UpperCamelCase			,							'cpu' ) )
  return CheckpointState(**__UpperCamelCase )



class 				_lowerCAmelCase			:


  '''simple docstring'''







  def __init__(					self					:    str					, UpperCamelCase					:    Dict							):


    '''simple docstring'''





    _snake_case      : Union[str, Any]														=		src_file







  def UpperCamelCase_       (					self					:    Tuple							):


    '''simple docstring'''





    raise NotImplementedError







  @staticmethod
  def UpperCamelCase_       (					UpperCamelCase					:    Union[str, Any]					, *UpperCamelCase					:    Union[str, Any]					, **UpperCamelCase					:    Optional[Any]							):


    '''simple docstring'''





    if comp_type.startswith('c'							):
      return DPRContextEncoderState(*__UpperCAmelCase					, **__UpperCAmelCase							)
    if comp_type.startswith('q'							):
      return DPRQuestionEncoderState(*__UpperCAmelCase					, **__UpperCAmelCase							)
    if comp_type.startswith('r'							):
      return DPRReaderState(*__UpperCAmelCase					, **__UpperCAmelCase							)
    else:
      raise ValueError('Component type must be either \'ctx_encoder\', \'question_encoder\' or \'reader\'.'							)




class 				_lowerCAmelCase			(	UpperCAmelCase_							):


  '''simple docstring'''







  def UpperCamelCase_       (					self					:    Any							):


    '''simple docstring'''





    _snake_case      : Optional[Any]														=		DPRContextEncoder(DPRConfig(**BertConfig.get_config_dict('bert-base-uncased'							)[0]							)							)
    print(f"""Loading DPR biencoder from {self.src_file}"""							)
    _snake_case      : Optional[int]														=		load_states_from_checkpoint(self.src_file							)
    _snake_case							,    _snake_case      : Optional[Any]														=		model.ctx_encoder, 'ctx_model.'
    # Fix changes from https://github.com/huggingface/transformers/commit/614fef1691edb806de976756d4948ecbcd0c0ca3
    _snake_case      : List[str]														=		{'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(__UpperCAmelCase							):
        _snake_case      : int														=		key[len(__UpperCAmelCase							) :]
        if not key.startswith('encode_proj.'							):
          _snake_case      : Optional[int]														=		'bert_model.' + key
        _snake_case      : Dict														=		value
    encoder.load_state_dict(__UpperCAmelCase							)
    return model




class 				_lowerCAmelCase			(	UpperCAmelCase_							):


  '''simple docstring'''







  def UpperCamelCase_       (					self					:    Optional[Any]							):


    '''simple docstring'''





    _snake_case      : List[Any]														=		DPRQuestionEncoder(DPRConfig(**BertConfig.get_config_dict('bert-base-uncased'							)[0]							)							)
    print(f"""Loading DPR biencoder from {self.src_file}"""							)
    _snake_case      : List[Any]														=		load_states_from_checkpoint(self.src_file							)
    _snake_case							,    _snake_case      : List[str]														=		model.question_encoder, 'question_model.'
    # Fix changes from https://github.com/huggingface/transformers/commit/614fef1691edb806de976756d4948ecbcd0c0ca3
    _snake_case      : Any														=		{'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(__UpperCAmelCase							):
        _snake_case      : Optional[int]														=		key[len(__UpperCAmelCase							) :]
        if not key.startswith('encode_proj.'							):
          _snake_case      : Optional[Any]														=		'bert_model.' + key
        _snake_case      : Union[str, Any]														=		value
    encoder.load_state_dict(__UpperCAmelCase							)
    return model




class 				_lowerCAmelCase			(	UpperCAmelCase_							):


  '''simple docstring'''







  def UpperCamelCase_       (					self					:    List[str]							):


    '''simple docstring'''





    _snake_case      : Any														=		DPRReader(DPRConfig(**BertConfig.get_config_dict('bert-base-uncased'							)[0]							)							)
    print(f"""Loading DPR reader from {self.src_file}"""							)
    _snake_case      : Union[str, Any]														=		load_states_from_checkpoint(self.src_file							)
    # Fix changes from https://github.com/huggingface/transformers/commit/614fef1691edb806de976756d4948ecbcd0c0ca3
    _snake_case      : Union[str, Any]														=		{
        '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'							):
        _snake_case      : List[str]														=		'encoder.bert_model.' + key[len('encoder.'							) :]
      _snake_case      : Any														=		value
    model.span_predictor.load_state_dict(__UpperCAmelCase							)
    return model




def 			lowerCamelCase_		(					lowerCAmelCase:		int			,							lowerCAmelCase:		List[str]			,							lowerCAmelCase:		str )->						Union[str, Any]:
  _snake_case      : int														=		Path(__UpperCamelCase )
  dest_dir.mkdir(exist_ok=__UpperCamelCase )

  _snake_case      : List[Any]														=		DPRState.from_type(__UpperCamelCase			,							src_file=__UpperCamelCase )
  _snake_case      : Tuple														=		dpr_state.load_dpr_model()
  model.save_pretrained(__UpperCamelCase )
  model.from_pretrained(__UpperCamelCase )  # sanity check


if __name__ == "__main__":
      lowerCAmelCase_        =			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.""")
      lowerCAmelCase_        =			parser.parse_args()

      lowerCAmelCase_        =			Path(args.src)
      lowerCAmelCase_        =			F"""converted-{src_file.name}""" if args.dest is None else args.dest
      lowerCAmelCase_        =			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 enum
import inspect
import warnings
from functools import wraps
from typing import Callable, Optional

from .logging import get_logger


__A       						=       set()
__A       						=       get_logger(__name__)



def     _lowerCamelCase(__UpperCamelCase = None       )				-> Optional[Any]:

	def decorator(__UpperCamelCase       ):
		global _emitted_deprecation_warnings

		if inspect.isclass(__UpperCamelCase       ):
			_lowerCAmelCase								=deprecated_class_or_function.__init__
			_lowerCAmelCase								=deprecated_class_or_function.__name__
		else:
			_lowerCAmelCase								=deprecated_class_or_function
			_lowerCAmelCase								=deprecated_function.__name__
			# Support deprecating __init__ class method: class name instead
			_lowerCAmelCase								=name if name != """__init__""" else deprecated_function.__qualname__.split("""."""       )[-2]

		_lowerCAmelCase								=(
		    F'''{name} is deprecated and will be removed in the next major version of datasets.''' + F''' {help_message}'''
		    if help_message
		    else """"""
		)

		@wraps(__UpperCamelCase       )
		def wrapper(*__UpperCamelCase    , **__UpperCamelCase       ):
			_lowerCAmelCase								=hash(__UpperCamelCase       )
			if func_hash not in _emitted_deprecation_warnings:
				warnings.warn(__UpperCamelCase    , category=__UpperCamelCase    , stacklevel=2       )
				_emitted_deprecation_warnings.add(__UpperCamelCase       )
			return deprecated_function(*__UpperCamelCase    , **__UpperCamelCase       )

		_lowerCAmelCase								="""deprecated"""

		if inspect.isclass(__UpperCamelCase       ):
			_lowerCAmelCase								=wrapper
			return deprecated_class_or_function
		else:
			return wrapper

	return decorator







class 		lowerCamelCase__					(					enum.EnumMeta				):





			'''simple docstring'''



			def __getattribute__(							cls     ,     __UpperCAmelCase  ) ->     Optional[int]:
				_lowerCAmelCase								=super().__getattribute__(__UpperCAmelCase  )
				if isinstance(__UpperCAmelCase     ,     enum.Enum  ) and obj._on_access:
					obj._on_access()
				return obj



			def __getitem__(							cls     ,     __UpperCAmelCase  ) ->     List[str]:
				_lowerCAmelCase								=super().__getitem__(__UpperCAmelCase  )
				if member._on_access:
					member._on_access()
				return member



			def __call__(							cls     ,     __UpperCAmelCase     ,     __UpperCAmelCase=None     ,     *, __UpperCAmelCase=None     ,     __UpperCAmelCase=None     ,     __UpperCAmelCase=None     ,     __UpperCAmelCase=1  ) ->     Optional[Any]:
				_lowerCAmelCase								=super().__call__(__UpperCAmelCase     ,     __UpperCAmelCase     ,     module=__UpperCAmelCase     ,     qualname=__UpperCAmelCase     ,     type=__UpperCAmelCase     ,     start=__UpperCAmelCase  )
				if isinstance(__UpperCAmelCase     ,     enum.Enum  ) and obj._on_access:
					obj._on_access()
				return obj







class 		lowerCamelCase__					(					enum.Enum ,	metaclass=__magic_name__				):





			'''simple docstring'''



			def __new__(							cls     ,     __UpperCAmelCase  ) ->     Optional[int]:
				_lowerCAmelCase								=object.__new__(cls  )
				_lowerCAmelCase								=value
				_lowerCAmelCase								=member.deprecate
				return member



			@property
			def 					_lowerCAmelCase		(							self  ) ->     List[Any]:
				return ""



			def 					_lowerCAmelCase		(							self  ) ->     Tuple:
				_lowerCAmelCase								=f''' {self.help_message}''' if self.help_message else """"""
				warnings.warn(
				    f'''\'{self.__objclass__.__name__}\' is deprecated and will be removed in the next major version of datasets.'''
				    + help_message     ,     __UpperCAmelCase     ,     stacklevel=3     ,     )



