




'''simple docstring'''







from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import center_crop, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
   import PIL


A			   =logging.get_logger(__name__)


class       _a  (     __a			):
       __a    :					Union[str, Any]           = ["""pixel_values"""]
       def __init__(     self						:     List[str]       ,    lowercase						:     bool = True       ,    lowercase						:     Dict[str, int] = None       ,    lowercase						:     PILImageResampling = PIL.Image.NEAREST       ,    lowercase						:     bool = False       ,    lowercase						:     Dict[str, int] = None       ,    lowercase						:     Union[int, float] = 1 / 255       ,    lowercase						:     bool = False       ,    lowercase						:     bool = True       ,    lowercase						:     bool = True       ,    lowercase						:     Optional[Union[float, List[float]]] = None       ,    lowercase						:     Optional[Union[float, List[float]]] = None       ,    lowercase						:     bool = True       ,    **lowercase						:     Optional[int]       ,    ):



           '''simple docstring'''

           super().__init__(**lowercase    )
           UpperCAmelCase      	=		size if size is not None else {'''height''': 346, '''width''': 346}
           UpperCAmelCase      	=		get_size_dict(lowercase    )
           UpperCAmelCase      	=		crop_size if crop_size is not None else {'''height''': 289, '''width''': 289}
           UpperCAmelCase      	=		get_size_dict(lowercase       ,    param_name='''crop_size'''    )

           UpperCAmelCase      	=		do_resize
           UpperCAmelCase      	=		size
           UpperCAmelCase      	=		resample
           UpperCAmelCase      	=		do_center_crop
           UpperCAmelCase      	=		crop_size
           UpperCAmelCase      	=		do_rescale
           UpperCAmelCase      	=		rescale_factor
           UpperCAmelCase      	=		rescale_offset
           UpperCAmelCase      	=		do_normalize
           UpperCAmelCase      	=		image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
           UpperCAmelCase      	=		image_std if image_std is not None else IMAGENET_STANDARD_STD
           UpperCAmelCase      	=		include_top
       def    A				(     self						:     List[str]       ,    lowercase						:     np.ndarray       ,    lowercase						:     Dict[str, int]       ,    lowercase						:     PILImageResampling = PIL.Image.NEAREST       ,    lowercase						:     Optional[Union[str, ChannelDimension]] = None       ,    **lowercase						:     Union[str, Any]       ,    ):



           '''simple docstring'''

           UpperCAmelCase      	=		get_size_dict(lowercase    )
           if "height" not in size or "width" not in size:
               raise ValueError(f"The size dictionary must have keys 'height' and 'width'. Got {size.keys()}"    )
           return resize(
               lowercase       ,    size=(size['''height'''], size['''width'''])       ,    resample=lowercase       ,    data_format=lowercase       ,    **lowercase    )
       def    A				(     self						:     str       ,    lowercase						:     np.ndarray       ,    lowercase						:     Dict[str, int]       ,    lowercase						:     Optional[Union[str, ChannelDimension]] = None       ,    **lowercase						:     str       ,    ):



           '''simple docstring'''

           UpperCAmelCase      	=		get_size_dict(lowercase    )
           if "height" not in size or "width" not in size:
               raise ValueError(f"The size dictionary must have keys 'height' and 'width'. Got {size.keys()}"    )
           return center_crop(lowercase       ,    size=(size['''height'''], size['''width'''])       ,    data_format=lowercase       ,    **lowercase    )
       def    A				(     self						:     Optional[Any]       ,    lowercase						:     np.ndarray       ,    lowercase						:     Union[int, float]       ,    lowercase						:     bool = True       ,    lowercase						:     Optional[Union[str, ChannelDimension]] = None       ,    **lowercase						:     int       ,    ):



           '''simple docstring'''

           if offset:
               UpperCAmelCase      	=		(image - 127.5) * scale
               if data_format is not None:
                   UpperCAmelCase      	=		to_channel_dimension_format(lowercase       ,    lowercase    )
               UpperCAmelCase      	=		rescaled_image.astype(np.floataa    )
           else:
               UpperCAmelCase      	=		rescale(lowercase       ,    scale=lowercase       ,    data_format=lowercase       ,    **lowercase    )
           return rescaled_image
       def    A				(     self						:     Union[str, Any]       ,    lowercase						:     np.ndarray       ,    lowercase						:     Union[float, List[float]]       ,    lowercase						:     Union[float, List[float]]       ,    lowercase						:     Optional[Union[str, ChannelDimension]] = None       ,    **lowercase						:     Optional[int]       ,    ):



           '''simple docstring'''

           return normalize(lowercase       ,    mean=lowercase       ,    std=lowercase       ,    data_format=lowercase       ,    **lowercase    )







       def    A				(     self						:     List[str]       ,    lowercase						:     ImageInput       ,    lowercase						:     bool = None       ,    lowercase						:     Dict[str, int] = None       ,    lowercase						:     Tuple=None       ,    lowercase						:     bool = None       ,    lowercase						:     Dict[str, int] = None       ,    lowercase						:     bool = None       ,    lowercase						:     float = None       ,    lowercase						:     bool = None       ,    lowercase						:     bool = None       ,    lowercase						:     Optional[Union[float, List[float]]] = None       ,    lowercase						:     Optional[Union[float, List[float]]] = None       ,    lowercase						:     bool = None       ,    lowercase						:     Optional[Union[str, TensorType]] = None       ,    lowercase						:     ChannelDimension = ChannelDimension.FIRST       ,    **lowercase						:     Tuple       ,    ):



           '''simple docstring'''

           UpperCAmelCase      	=		do_resize if do_resize is not None else self.do_resize
           UpperCAmelCase      	=		resample if resample is not None else self.resample
           UpperCAmelCase      	=		do_center_crop if do_center_crop is not None else self.do_center_crop
           UpperCAmelCase      	=		do_rescale if do_rescale is not None else self.do_rescale
           UpperCAmelCase      	=		rescale_factor if rescale_factor is not None else self.rescale_factor
           UpperCAmelCase      	=		rescale_offset if rescale_offset is not None else self.rescale_offset
           UpperCAmelCase      	=		do_normalize if do_normalize is not None else self.do_normalize
           UpperCAmelCase      	=		image_mean if image_mean is not None else self.image_mean
           UpperCAmelCase      	=		image_std if image_std is not None else self.image_std
           UpperCAmelCase      	=		include_top if include_top is not None else self.include_top

           UpperCAmelCase      	=		size if size is not None else self.size
           UpperCAmelCase      	=		get_size_dict(lowercase    )
           UpperCAmelCase      	=		crop_size if crop_size is not None else self.crop_size
           UpperCAmelCase      	=		get_size_dict(lowercase       ,    param_name='''crop_size'''    )

           UpperCAmelCase      	=		make_list_of_images(lowercase    )

           if not valid_images(lowercase    ):
               raise ValueError(
                   '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
                   '''torch.Tensor, tf.Tensor or jax.ndarray.'''    )

           if do_resize and size is None or resample is None:
               raise ValueError('''Size and resample must be specified if do_resize is True.'''    )

           if do_center_crop and crop_size is None:
               raise ValueError('''Crop size must be specified if do_center_crop is True.'''    )

           if do_rescale and rescale_factor is None:
               raise ValueError('''Rescale factor must be specified if do_rescale is True.'''    )

           if do_normalize and (image_mean is None or image_std is None):
               raise ValueError('''Image mean and std must be specified if do_normalize is True.'''    )

           # All transformations expect numpy arrays.
           UpperCAmelCase      	=		[to_numpy_array(lowercase    ) for image in images]

           if do_resize:
               UpperCAmelCase      	=		[self.resize(image=lowercase       ,    size=lowercase       ,    resample=lowercase    ) for image in images]

           if do_center_crop:
               UpperCAmelCase      	=		[self.center_crop(image=lowercase       ,    size=lowercase    ) for image in images]

           if do_rescale:
               UpperCAmelCase      	=		[self.rescale(image=lowercase       ,    scale=lowercase       ,    offset=lowercase    ) for image in images]

           if do_normalize:
               UpperCAmelCase      	=		[self.normalize(image=lowercase       ,    mean=lowercase       ,    std=lowercase    ) for image in images]

           if include_top:
               UpperCAmelCase      	=		[self.normalize(image=lowercase       ,    mean=[0, 0, 0]       ,    std=lowercase    ) for image in images]

           UpperCAmelCase      	=		[to_channel_dimension_format(lowercase       ,    lowercase    ) for image in images]

           UpperCAmelCase      	=		{'''pixel_values''': images}
           return BatchFeature(data=lowercase       ,    tensor_type=lowercase    )










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





def 						snake_case_							(_a     :							str					):
    print(F"Reading saved model from {model_file}"					)
    UpperCAmelCase      	=		torch.load(_a   ,       map_location=lambda _a   ,       _a					: default_restore_location(_a   ,       '''cpu'''					)					)
    return CheckpointState(**_a					)


class       _a  :
       def __init__(     self						:     List[Any]       ,    lowercase						:     Path    ):



           '''simple docstring'''

           UpperCAmelCase      	=		src_file
       def    A				(     self						:     str    ):



           '''simple docstring'''

           raise NotImplementedError







       @staticmethod
       def    A				(     lowercase						:     str       ,    *lowercase						:     Tuple       ,    **lowercase						:     Union[str, Any]    ):



           '''simple docstring'''

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






class       _a  (     __a			):
       def    A				(     self						:     int    ):



           '''simple docstring'''

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






class       _a  (     __a			):
       def    A				(     self						:     int    ):



           '''simple docstring'''

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






class       _a  (     __a			):
       def    A				(     self						:     int    ):



           '''simple docstring'''

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






def 						snake_case_							(_a     :							str   ,       _a     :							Path   ,       _a     :							Path					):
    UpperCAmelCase      	=		Path(_a					)
    dest_dir.mkdir(exist_ok=_a					)

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


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

   A			   =Path(args.src)
   A			   =f"""converted-{src_file.name}""" if args.dest is None else args.dest
   A			   =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)




