


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 			_a      (    __UpperCamelCase			):
     print(F'''Reading saved model from {model_file}'''			)
     a_     :  str				      =     torch.load(__UpperCamelCase	,	map_location=lambda __UpperCamelCase	,	__UpperCamelCase			: default_restore_location(__UpperCamelCase	,	"""cpu"""			)			)
     return CheckpointState(**__UpperCamelCase			)


class 	a__						:

       def __init__( self   :  int   ,				lowerCamelCase_   :  Path     ):
            a_     :  List[Any]				      =     src_file

       def     UpperCAmelCase( self   :  List[str]     ):
            raise NotImplementedError

       @staticmethod
       def     UpperCAmelCase( lowerCamelCase_   :  str   ,				*lowerCamelCase_   :  Optional[Any]   ,				**lowerCamelCase_   :  Any     ):
            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 	a__						(    lowerCAmelCase_     ):

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


class 	a__						(    lowerCAmelCase_     ):

       def     UpperCAmelCase( self   :  int     ):
            a_     :  List[str]				      =     DPRQuestionEncoder(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_      ,      a_     :  Any				      =     model.question_encoder, """question_model."""
            # Fix changes from https://github.com/huggingface/transformers/commit/614fef1691edb806de976756d4948ecbcd0c0ca3
            a_     :  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_     :  str				      =     key[len(lowerCamelCase_     ) :]
                      if not key.startswith("""encode_proj."""     ):
                           a_     :  List[str]				      =     """bert_model.""" + key
                      a_     :  Any				      =     value
            encoder.load_state_dict(lowerCamelCase_     )
            return model


class 	a__						(    lowerCAmelCase_     ):

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



def 			_a      (    __UpperCamelCase	,	__UpperCamelCase	,	__UpperCamelCase			):
     a_     :  Optional[Any]				      =     Path(__UpperCamelCase			)
     dest_dir.mkdir(exist_ok=__UpperCamelCase			)

     a_     :  str				      =     DPRState.from_type(__UpperCamelCase	,	src_file=__UpperCamelCase			)
     a_     :  Optional[Any]				      =     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)







import unittest

from transformers import SpeechaTextaConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
       import torch

       from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
           SpeechaTextaDecoder,
           SpeechaTextaForCausalLM,
       )


@require_torch
class 	a__						:

       def __init__( self   :  Dict   ,				lowerCamelCase_   :  Any   ,				lowerCamelCase_   :  Dict=9_9   ,				lowerCamelCase_   :  List[Any]=1_3   ,				lowerCamelCase_   :  Dict=1_6   ,				lowerCamelCase_   :  Any=7   ,				lowerCamelCase_   :  Dict=True   ,				lowerCamelCase_   :  List[str]=True   ,				lowerCamelCase_   :  Tuple=True   ,				lowerCamelCase_   :  Any=False   ,				lowerCamelCase_   :  Union[str, Any]=True   ,				lowerCamelCase_   :  int=2   ,				lowerCamelCase_   :  Tuple=3_2   ,				lowerCamelCase_   :  int=4   ,				lowerCamelCase_   :  Dict=4   ,				lowerCamelCase_   :  str=3_0   ,				lowerCamelCase_   :  Optional[int]=0   ,				lowerCamelCase_   :  Optional[Any]=1   ,				lowerCamelCase_   :  Any=2   ,				lowerCamelCase_   :  Optional[int]=None   ,				):
            a_     :  Optional[Any]				      =     parent
            a_     :  Union[str, Any]				      =     batch_size
            a_     :  Tuple				      =     decoder_seq_length
            # For common tests
            a_     :  Union[str, Any]				      =     self.decoder_seq_length
            a_     :  List[str]				      =     is_training
            a_     :  int				      =     use_attention_mask
            a_     :  Dict				      =     use_labels

            a_     :  List[str]				      =     vocab_size
            a_     :  Optional[int]				      =     d_model
            a_     :  str				      =     d_model
            a_     :  Any				      =     decoder_layers
            a_     :  Tuple				      =     decoder_layers
            a_     :  Dict				      =     decoder_ffn_dim
            a_     :  List[str]				      =     decoder_attention_heads
            a_     :  List[Any]				      =     decoder_attention_heads
            a_     :  List[str]				      =     eos_token_id
            a_     :  Any				      =     bos_token_id
            a_     :  List[Any]				      =     pad_token_id
            a_     :  Dict				      =     decoder_start_token_id
            a_     :  Tuple				      =     use_cache
            a_     :  List[Any]				      =     max_position_embeddings

            a_     :  List[str]				      =     None
            a_     :  Any				      =     decoder_seq_length
            a_     :  List[Any]				      =     2
            a_     :  Optional[Any]				      =     1

       def     UpperCAmelCase( self   :  str     ):
            a_     :  Any				      =     ids_tensor([self.batch_size, self.decoder_seq_length]   ,				self.vocab_size     )

            a_     :  Any				      =     None
            if self.use_attention_mask:
                 a_     :  Optional[Any]				      =     ids_tensor([self.batch_size, self.decoder_seq_length]   ,				vocab_size=2     )

            a_     :  Tuple				      =     None
            if self.use_labels:
                 a_     :  int				      =     ids_tensor([self.batch_size, self.decoder_seq_length]   ,				self.vocab_size     )

            a_     :  int				      =     SpeechaTextaConfig(
                vocab_size=self.vocab_size   ,				d_model=self.d_model   ,				decoder_layers=self.decoder_layers   ,				decoder_ffn_dim=self.decoder_ffn_dim   ,				decoder_attention_heads=self.decoder_attention_heads   ,				eos_token_id=self.eos_token_id   ,				bos_token_id=self.bos_token_id   ,				use_cache=self.use_cache   ,				pad_token_id=self.pad_token_id   ,				decoder_start_token_id=self.decoder_start_token_id   ,				max_position_embeddings=self.max_position_embeddings   ,				)

            return (
                config,
                input_ids,
                attention_mask,
                lm_labels,
            )

       def     UpperCAmelCase( self   :  Tuple   ,				lowerCamelCase_   :  List[str]   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  Optional[int]   ,				lowerCamelCase_   :  List[str]   ,				):
            a_     :  List[Any]				      =     True
            a_     :  Optional[int]				      =     SpeechaTextaDecoder(config=lowerCamelCase_     ).to(lowerCamelCase_     ).eval()
            a_     :  Optional[int]				      =     input_ids[:2]

            input_ids[input_ids == 0] += 1
            # first forward pass
            a_     :  Union[str, Any]				      =     model(lowerCamelCase_   ,				use_cache=lowerCamelCase_     )
            a_     :  Tuple				      =     model(lowerCamelCase_     )
            a_     :  str				      =     model(lowerCamelCase_   ,				use_cache=lowerCamelCase_     )

            self.parent.assertTrue(len(lowerCamelCase_     ) == len(lowerCamelCase_     )     )
            self.parent.assertTrue(len(lowerCamelCase_     ) == len(lowerCamelCase_     ) + 1     )

            a_     :  Tuple				      =     outputs["""past_key_values"""]

            # create hypothetical next token and extent to next_input_ids
            a_     :  Any				      =     ids_tensor((2, 1)   ,				config.vocab_size - 1     ) + 1

            # append to next input_ids and
            a_     :  str				      =     torch.cat([input_ids, next_tokens]   ,				dim=-1     )

            print(lowerCamelCase_     )

            a_     :  Any				      =     model(lowerCamelCase_     )["""last_hidden_state"""]
            a_     :  int				      =     model(lowerCamelCase_   ,				past_key_values=lowerCamelCase_     )["""last_hidden_state"""]

            # select random slice
            a_     :  Tuple				      =     ids_tensor((1,)   ,				output_from_past.shape[-1]     ).item()
            a_     :  List[str]				      =     output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
            a_     :  List[Any]				      =     output_from_past[:, 0, random_slice_idx].detach()

            # test that outputs are equal for slice
            assert torch.allclose(lowerCamelCase_   ,				lowerCamelCase_   ,				atol=1E-3     )

       def     UpperCAmelCase( self   :  Any     ):
            a_     :  List[Any]				      =     self.prepare_config_and_inputs()
            (
                (
                a_
            )      ,      (
                a_
            )      ,      (
                a_
            )      ,      (
                a_
            )      ,      
            )     :  Optional[int]				      =     config_and_inputs

            a_     :  Tuple				      =     {
                """input_ids""": input_ids,
                """attention_mask""": attention_mask,
            }
            return config, inputs_dict


@require_torch
class 	a__						(    lowerCAmelCase_    ,							lowerCAmelCase_    ,							lowerCAmelCase_    ,							unittest.TestCase     ):
       lowerCamelCase__:  int				     =  (SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
       lowerCamelCase__:  int				     =  (SpeechaTextaForCausalLM,) if is_torch_available() else ()
       lowerCamelCase__:  int				     =  {"""text-generation""": SpeechaTextaForCausalLM} if is_torch_available() else {}
       lowerCamelCase__:  Dict				     =  True
       lowerCamelCase__:  List[Any]				     =  False

       def     UpperCAmelCase( self   :  List[str]   ,				):
            a_     :  Optional[int]				      =     SpeechaTextaStandaloneDecoderModelTester(self   ,				is_training=lowerCamelCase_     )
            a_     :  str				      =     ConfigTester(self   ,				config_class=lowerCamelCase_     )

       def     UpperCAmelCase( self   :  Optional[Any]     ):
            pass

       def     UpperCAmelCase( self   :  Optional[Any]     ):
            pass

       def     UpperCAmelCase( self   :  Optional[Any]     ):
            pass

       def     UpperCAmelCase( self   :  str     ):
            self.config_tester.run_common_tests()

       def     UpperCAmelCase( self   :  Dict     ):
            a_     :  int				      =     self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_decoder_model_past(*lowerCamelCase_     )

       def     UpperCAmelCase( self   :  int     ):
            return



