

"""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__			:						int							=			collections.namedtuple(
    """CheckpointState""", ["""model_dict""", """optimizer_dict""", """scheduler_dict""", """offset""", """epoch""", """encoder_params"""]
)





def 					__UpperCamelCase (     a__:			str      )					->      CheckpointState:




					"""simple docstring"""



					print(f'''Reading saved model from {model_file}'''      )
					__lowercase        =      torch.load(a__			,						map_location=lambda a__			,						a__      : default_restore_location(a__			,						'''cpu'''      )      )
					return CheckpointState(**a__      )


class       _snake_case  :







						def __init__(self  ,				UpperCAmelCase_				)      ->    Union[str, Any]:
											__lowercase        =      src_file







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







						@staticmethod
						def 							_UpperCamelCase     (UpperCAmelCase_  ,				*UpperCAmelCase_  ,				**UpperCAmelCase_				)      ->    "DPRState":
											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       _snake_case  (    UpperCamelCase							):







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



class       _snake_case  (    UpperCamelCase							):







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



class       _snake_case  (    UpperCamelCase							):







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





def 					__UpperCamelCase (     a__:			str			,						a__:			Path			,						a__:			Path      )					->      List[str]:




					"""simple docstring"""



					__lowercase        =      Path(a__      )
					dest_dir.mkdir(exist_ok=a__      )

					__lowercase        =      DPRState.from_type(a__			,						src_file=a__      )
					__lowercase        =      dpr_state.load_dpr_model()
					model.save_pretrained(a__      )
					model.from_pretrained(a__      )  # sanity check


if __name__ == "__main__":
							a__			:						Tuple							=			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__			:						Union[str, Any]							=			Path(args.src)
							a__			:						int							=			f"""converted-{src_file.name}""" if args.dest is None else args.dest
							a__			:						Optional[int]							=			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__			:						Optional[int]							=			set()
a__			:						str							=			get_logger(__name__)





def 					__UpperCamelCase (     a__:			Optional[str] = None      )					->      Dict:




					"""simple docstring"""



					def decorator(a__:			Callable      ):
										global _emitted_deprecation_warnings

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

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

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

										__lowercase        =      '''deprecated'''

										if inspect.isclass(a__      ):
															__lowercase        =      wrapper
															return deprecated_class_or_function
										else:
															return wrapper

					return decorator


class       _snake_case  (    enum.EnumMeta							):







						def __getattribute__(cls  ,				UpperCAmelCase_				)      ->    int:
											__lowercase        =      super().__getattribute__(UpperCAmelCase_				)
											if isinstance(UpperCAmelCase_  ,				enum.Enum				) and obj._on_access:
																obj._on_access()
											return obj







						def __getitem__(cls  ,				UpperCAmelCase_				)      ->    Union[str, Any]:
											__lowercase        =      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				)      ->    List[Any]:
											__lowercase        =      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       _snake_case  (    enum.Enum    ,      metaclass=UpperCamelCase							):







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







						@property
						def 							_UpperCamelCase     (self				)      ->    Dict:
											return ""







						def 							_UpperCamelCase     (self				)      ->    Optional[Any]:
											__lowercase        =      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  ,				)


