




'''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


lowercase						:   List[str]      						=							collections.namedtuple(
    '''CheckpointState''', ['''model_dict''', '''optimizer_dict''', '''scheduler_dict''', '''offset''', '''epoch''', '''encoder_params''']
)




def 		a_(			lowerCamelCase   )					->CheckpointState:






       '''simple docstring'''




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

class 	SCREAMING_SNAKE_CASE_		:



 def __init__(	self ,		A			):






        '''simple docstring'''




        lowercase	:		Union[str, Any]    =   src_file



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        raise NotImplementedError





 @staticmethod
 def  SCREAMING_SNAKE_CASE(	A ,		*A ,		**A			):






        '''simple docstring'''




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






class 	SCREAMING_SNAKE_CASE_		(							lowerCamelCase_					):



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




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






class 	SCREAMING_SNAKE_CASE_		(							lowerCamelCase_					):



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




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






class 	SCREAMING_SNAKE_CASE_		(							lowerCamelCase_					):



 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




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






def 		a_(			lowerCamelCase ,					lowerCamelCase ,					lowerCamelCase   )					->str:






       '''simple docstring'''




       lowercase	:		List[str]    =   Path(lowerCamelCase   )
       dest_dir.mkdir(exist_ok=lowerCamelCase   )

       lowercase	:		int    =   DPRState.from_type(lowerCamelCase ,					src_file=lowerCamelCase   )
       lowercase	:		int    =   dpr_state.load_dpr_model()
       model.save_pretrained(lowerCamelCase   )
       model.from_pretrained(lowerCamelCase   )  # sanity check


if __name__ == "__main__":
      lowercase						:   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.''')
      lowercase						:   List[Any]      						=							parser.parse_args()

      lowercase						:   Any      						=							Path(args.src)
      lowercase						:   str      						=							f'converted-{src_file.name}' if args.dest is None else args.dest
      lowercase						:   Dict      						=							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 os

from packaging import version

from .. import __version__
from .accelerate_utils import apply_forward_hook
from .constants import (
    CONFIG_NAME,
    DEPRECATED_REVISION_ARGS,
    DIFFUSERS_CACHE,
    DIFFUSERS_DYNAMIC_MODULE_NAME,
    FLAX_WEIGHTS_NAME,
    HF_MODULES_CACHE,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    ONNX_EXTERNAL_WEIGHTS_NAME,
    ONNX_WEIGHTS_NAME,
    SAFETENSORS_WEIGHTS_NAME,
    TEXT_ENCODER_ATTN_MODULE,
    WEIGHTS_NAME,
)
from .deprecation_utils import deprecate
from .doc_utils import replace_example_docstring
from .dynamic_modules_utils import get_class_from_dynamic_module
from .hub_utils import (
    HF_HUB_OFFLINE,
    _add_variant,
    _get_model_file,
    extract_commit_hash,
    http_user_agent,
)
from .import_utils import (
    BACKENDS_MAPPING,
    ENV_VARS_TRUE_AND_AUTO_VALUES,
    ENV_VARS_TRUE_VALUES,
    USE_JAX,
    USE_TF,
    USE_TORCH,
    DummyObject,
    OptionalDependencyNotAvailable,
    is_accelerate_available,
    is_accelerate_version,
    is_bsa_available,
    is_flax_available,
    is_ftfy_available,
    is_inflect_available,
    is_invisible_watermark_available,
    is_k_diffusion_available,
    is_k_diffusion_version,
    is_librosa_available,
    is_note_seq_available,
    is_omegaconf_available,
    is_onnx_available,
    is_safetensors_available,
    is_scipy_available,
    is_tensorboard_available,
    is_tf_available,
    is_torch_available,
    is_torch_version,
    is_torchsde_available,
    is_transformers_available,
    is_transformers_version,
    is_unidecode_available,
    is_wandb_available,
    is_xformers_available,
    requires_backends,
)
from .logging import get_logger
from .outputs import BaseOutput
from .pil_utils import PIL_INTERPOLATION, numpy_to_pil, pt_to_pil
from .torch_utils import is_compiled_module, randn_tensor


if is_torch_available():
      from .testing_utils import (
          floats_tensor,
          load_hf_numpy,
          load_image,
          load_numpy,
          load_pt,
          nightly,
          parse_flag_from_env,
          print_tensor_test,
          require_torch_a,
          require_torch_gpu,
          skip_mps,
          slow,
          torch_all_close,
          torch_device,
      )
      from .torch_utils import maybe_allow_in_graph

from .testing_utils import export_to_gif, export_to_video


lowercase						:   Tuple      						=							get_logger(__name__)




def 		a_(			lowerCamelCase   )					->Any:






       '''simple docstring'''




       if version.parse(lowerCamelCase   ) < version.parse(lowerCamelCase   ):
              if "dev" in min_version:
                     lowercase	:		str    =   (
                         '''This example requires a source install from HuggingFace diffusers (see '''
                         '''`https://huggingface.co/docs/diffusers/installation#install-from-source`),'''
                     )
              else:
                     lowercase	:		List[str]    =   f"""This example requires a minimum version of {min_version},"""
              error_message += f""" but the version found is {__version__}.\n"""
              raise ImportError(lowerCamelCase   )




