





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


lowerCamelCase_	:		Dict         =   collections.namedtuple(
    'CheckpointState', ['model_dict', 'optimizer_dict', 'scheduler_dict', 'offset', 'epoch', 'encoder_params']
)





def 		UpperCAmelCase__						(	_UpperCAmelCase   ):






  """simple docstring"""





  print(f"""Reading saved model from {model_file}"""   )
  A_    :			Tuple									=    torch.load(a_					,   map_location=lambda _UpperCAmelCase					,   _UpperCAmelCase   : default_restore_location(a_					,   'cpu'   )   )
  return CheckpointState(**a_   )

class 					_UpperCAmelCase    :



      '''simple docstring'''

      def __init__(	self   , snake_case_    ):

        """simple docstring"""

        A_    :			int									=    src_file

      def    lowerCamelCase_     (	self    ):

        """simple docstring"""

        raise NotImplementedError

      @staticmethod
      def    lowerCamelCase_     (	snake_case_   , *snake_case_   , **snake_case_    ):

        """simple docstring"""

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


class 					_UpperCAmelCase    (  __A						):



      '''simple docstring'''

      def    lowerCamelCase_     (	self    ):

        """simple docstring"""

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


class 					_UpperCAmelCase    (  __A						):



      '''simple docstring'''

      def    lowerCamelCase_     (	self    ):

        """simple docstring"""

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


class 					_UpperCAmelCase    (  __A						):



      '''simple docstring'''

      def    lowerCamelCase_     (	self    ):

        """simple docstring"""

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





def 		UpperCAmelCase__						(	_UpperCAmelCase					,   _UpperCAmelCase					,   _UpperCAmelCase   ):






  """simple docstring"""





  A_    :			Tuple									=    Path(a_   )
  dest_dir.mkdir(exist_ok=a_   )

  A_    :			Optional[Any]									=    DPRState.from_type(a_					,   src_file=a_   )
  A_    :			Optional[Any]									=    dpr_state.load_dpr_model()
  model.save_pretrained(a_   )
  model.from_pretrained(a_   )  # sanity check


if __name__ == "__main__":
   lowerCamelCase_	:		int         =   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_	:		Dict         =   parser.parse_args()

   lowerCamelCase_	:		Dict         =   Path(args.src)
   lowerCamelCase_	:		Tuple         =   F"converted-{src_file.name}" if args.dest is None else args.dest
   lowerCamelCase_	:		Optional[Any]         =   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 unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
  import torch

if is_vision_available():
  from PIL import Image

  from transformers import ViltImageProcessor


class 							_lowerCAmelCase	(  unittest.TestCase							):


 """simple docstring"""

 def __init__(      self					,    _lowerCamelCase					,    _lowerCamelCase=7					,    _lowerCamelCase=3					,    _lowerCamelCase=18					,    _lowerCamelCase=30					,    _lowerCamelCase=400					,    _lowerCamelCase=True					,    _lowerCamelCase=None					,    _lowerCamelCase=2					,    _lowerCamelCase=True					,    _lowerCamelCase=[0.5, 0.5, 0.5]					,    _lowerCamelCase=[0.5, 0.5, 0.5]					,    )    ->    Tuple:
    A_  :		Dict						=					size if size is not None else {"""shortest_edge""": 30}
    A_  :		List[Any]						=					parent
    A_  :		Optional[Any]						=					batch_size
    A_  :		Tuple						=					num_channels
    A_  :		Any						=					image_size
    A_  :		str						=					min_resolution
    A_  :		str						=					max_resolution
    A_  :		Any						=					do_resize
    A_  :		List[Any]						=					size
    A_  :		Union[str, Any]						=					size_divisor
    A_  :		Dict						=					do_normalize
    A_  :		Dict						=					image_mean
    A_  :		Optional[Any]						=					image_std







 def       UpperCAmelCase_       (      self  )    ->    Optional[Any]:
    return {
        "image_mean": self.image_mean,
        "image_std": self.image_std,
        "do_normalize": self.do_normalize,
        "do_resize": self.do_resize,
        "size": self.size,
        "size_divisor": self.size_divisor,
    }







 def       UpperCAmelCase_       (      self					,    _lowerCamelCase					,    _lowerCamelCase=False  )    ->    Union[str, Any]:
    if not batched:
       A_  :		List[str]						=					self.size["""shortest_edge"""]
       A_  :		str						=					image_inputs[0]
       if isinstance(_lowerCamelCase					,    Image.Image  ):
          A_						,  A_  :		Dict						=					image.size
       else:
          A_						,  A_  :		Dict						=					image.shape[1], image.shape[2]
       A_  :		Dict						=					size / min(_lowerCamelCase					,    _lowerCamelCase  )
       if h < w:
          A_						,  A_  :		Any						=					size, scale * w
       else:
          A_						,  A_  :		List[str]						=					scale * h, size

       A_  :		Dict						=					int((1333 / 800) * size  )
       if max(_lowerCamelCase					,    _lowerCamelCase  ) > max_size:
          A_  :		Dict						=					max_size / max(_lowerCamelCase					,    _lowerCamelCase  )
          A_  :		str						=					newh * scale
          A_  :		Tuple						=					neww * scale

       A_						,  A_  :		str						=					int(newh + 0.5  ), int(neww + 0.5  )
       A_						,  A_  :		Union[str, Any]						=					(
           newh // self.size_divisor * self.size_divisor,
           neww // self.size_divisor * self.size_divisor,
       )

    else:
       A_  :		Union[str, Any]						=					[]
       for image in image_inputs:
          A_						,  A_  :		Any						=					self.get_expected_values([image]  )
          expected_values.append((expected_height, expected_width)  )
       A_  :		List[str]						=					max(_lowerCamelCase					,    key=lambda _lowerCamelCase  : item[0]  )[0]
       A_  :		List[str]						=					max(_lowerCamelCase					,    key=lambda _lowerCamelCase  : item[1]  )[1]

    return expected_height, expected_width


@require_torch
@require_vision
class 							_lowerCAmelCase	(  __A,							unittest.TestCase							):


 """simple docstring"""

 lowerCamelCase									=     ViltImageProcessor if is_vision_available() else None







 def       UpperCAmelCase_       (      self  )    ->    int:
    A_  :		Optional[int]						=					ViltImageProcessingTester(self  )







 @property
 def       UpperCAmelCase_       (      self  )    ->    Dict:
    return self.image_processor_tester.prepare_image_processor_dict()







 def       UpperCAmelCase_       (      self  )    ->    Optional[Any]:
    A_  :		List[str]						=					self.image_processing_class(**self.image_processor_dict  )
    self.assertTrue(hasattr(_lowerCamelCase					,    """image_mean"""  )  )
    self.assertTrue(hasattr(_lowerCamelCase					,    """image_std"""  )  )
    self.assertTrue(hasattr(_lowerCamelCase					,    """do_normalize"""  )  )
    self.assertTrue(hasattr(_lowerCamelCase					,    """do_resize"""  )  )
    self.assertTrue(hasattr(_lowerCamelCase					,    """size"""  )  )
    self.assertTrue(hasattr(_lowerCamelCase					,    """size_divisor"""  )  )







 def       UpperCAmelCase_       (      self  )    ->    List[str]:
    A_  :		Optional[int]						=					self.image_processing_class.from_dict(self.image_processor_dict  )
    self.assertEqual(image_processor.size					,    {"""shortest_edge""": 30}  )

    A_  :		Union[str, Any]						=					self.image_processing_class.from_dict(self.image_processor_dict					,    size=42  )
    self.assertEqual(image_processor.size					,    {"""shortest_edge""": 42}  )







 def       UpperCAmelCase_       (      self  )    ->    Any:
    pass







 def       UpperCAmelCase_       (      self  )    ->    List[Any]:
    # Initialize image_processing
    A_  :		Any						=					self.image_processing_class(**self.image_processor_dict  )
    # create random PIL images
    A_  :		Any						=					prepare_image_inputs(self.image_processor_tester					,    equal_resolution=_lowerCamelCase  )
    for image in image_inputs:
       self.assertIsInstance(_lowerCamelCase					,    Image.Image  )

    # Test not batched input
    A_  :		Tuple						=					image_processing(image_inputs[0]					,    return_tensors="""pt"""  ).pixel_values

    A_						,  A_  :		Union[str, Any]						=					self.image_processor_tester.get_expected_values(_lowerCamelCase  )
    self.assertEqual(
        encoded_images.shape					,    (1, self.image_processor_tester.num_channels, expected_height, expected_width)					,    )

    # Test batched
    A_  :		List[Any]						=					image_processing(_lowerCamelCase					,    return_tensors="""pt"""  ).pixel_values

    A_						,  A_  :		Union[str, Any]						=					self.image_processor_tester.get_expected_values(_lowerCamelCase					,    batched=_lowerCamelCase  )
    self.assertEqual(
        encoded_images.shape					,    (
            self.image_processor_tester.batch_size,
            self.image_processor_tester.num_channels,
            expected_height,
            expected_width,
        )					,    )







 def       UpperCAmelCase_       (      self  )    ->    str:
    # Initialize image_processing
    A_  :		Tuple						=					self.image_processing_class(**self.image_processor_dict  )
    # create random numpy tensors
    A_  :		List[str]						=					prepare_image_inputs(self.image_processor_tester					,    equal_resolution=_lowerCamelCase					,    numpify=_lowerCamelCase  )
    for image in image_inputs:
       self.assertIsInstance(_lowerCamelCase					,    np.ndarray  )

    # Test not batched input
    A_  :		Any						=					image_processing(image_inputs[0]					,    return_tensors="""pt"""  ).pixel_values

    A_						,  A_  :		int						=					self.image_processor_tester.get_expected_values(_lowerCamelCase  )
    self.assertEqual(
        encoded_images.shape					,    (1, self.image_processor_tester.num_channels, expected_height, expected_width)					,    )

    # Test batched
    A_  :		List[str]						=					image_processing(_lowerCamelCase					,    return_tensors="""pt"""  ).pixel_values

    A_						,  A_  :		Any						=					self.image_processor_tester.get_expected_values(_lowerCamelCase					,    batched=_lowerCamelCase  )
    self.assertEqual(
        encoded_images.shape					,    (
            self.image_processor_tester.batch_size,
            self.image_processor_tester.num_channels,
            expected_height,
            expected_width,
        )					,    )







 def       UpperCAmelCase_       (      self  )    ->    Optional[Any]:
    # Initialize image_processing
    A_  :		List[str]						=					self.image_processing_class(**self.image_processor_dict  )
    # create random PyTorch tensors
    A_  :		str						=					prepare_image_inputs(self.image_processor_tester					,    equal_resolution=_lowerCamelCase					,    torchify=_lowerCamelCase  )
    for image in image_inputs:
       self.assertIsInstance(_lowerCamelCase					,    torch.Tensor  )

    # Test not batched input
    A_  :		Optional[int]						=					image_processing(image_inputs[0]					,    return_tensors="""pt"""  ).pixel_values

    A_						,  A_  :		List[str]						=					self.image_processor_tester.get_expected_values(_lowerCamelCase  )
    self.assertEqual(
        encoded_images.shape					,    (1, self.image_processor_tester.num_channels, expected_height, expected_width)					,    )

    # Test batched
    A_  :		Any						=					image_processing(_lowerCamelCase					,    return_tensors="""pt"""  ).pixel_values

    A_						,  A_  :		Any						=					self.image_processor_tester.get_expected_values(_lowerCamelCase					,    batched=_lowerCamelCase  )
    self.assertEqual(
        encoded_images.shape					,    (
            self.image_processor_tester.batch_size,
            self.image_processor_tester.num_channels,
            expected_height,
            expected_width,
        )					,    )
