




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


snake_case__ :						Optional[int]     =			collections.namedtuple(
    '''CheckpointState''', ['''model_dict''', '''optimizer_dict''', '''scheduler_dict''', '''offset''', '''epoch''', '''encoder_params''']
)


def 						lowercase		(     _lowerCAmelCase      ):
  print(F'''Reading saved model from {model_file}'''      )
  UpperCAmelCase__    						=   torch.load(_lowerCAmelCase   ,     map_location=lambda _lowerCAmelCase   ,     _lowerCAmelCase      : default_restore_location(_lowerCAmelCase   ,     """cpu"""      )      )
  return CheckpointState(**_lowerCAmelCase      )





class     snake_case  :


   '''simple docstring'''







   def __init__(					self     :	List[str]						,    lowerCamelCase_     :	Path			)					->List[Any]:
     '''simple docstring'''







     UpperCAmelCase__    						=   src_file




   def 	UpperCAmelCase       (					self     :	Optional[int]			)					->Any:
     '''simple docstring'''







     raise NotImplementedError







   @staticmethod
   def 	UpperCAmelCase       (					lowerCamelCase_     :	str						,    *lowerCamelCase_     :	Union[str, Any]						,    **lowerCamelCase_     :	List[str]			)					->"DPRState":
     '''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     snake_case  (	_snake_case							):


   '''simple docstring'''







   def 	UpperCAmelCase       (					self     :	Any			)					->Tuple:
     '''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(lowerCamelCase_			):
         UpperCAmelCase__    						=   key[len(lowerCamelCase_			) :]
         if not key.startswith("""encode_proj."""			):
           UpperCAmelCase__    						=   """bert_model.""" + key
         UpperCAmelCase__    						=   value
     encoder.load_state_dict(lowerCamelCase_			)
     return model





class     snake_case  (	_snake_case							):


   '''simple docstring'''







   def 	UpperCAmelCase       (					self     :	Dict			)					->List[str]:
     '''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(lowerCamelCase_			):
         UpperCAmelCase__    						=   key[len(lowerCamelCase_			) :]
         if not key.startswith("""encode_proj."""			):
           UpperCAmelCase__    						=   """bert_model.""" + key
         UpperCAmelCase__    						=   value
     encoder.load_state_dict(lowerCamelCase_			)
     return model





class     snake_case  (	_snake_case							):


   '''simple docstring'''







   def 	UpperCAmelCase       (					self     :	int			)					->Tuple:
     '''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(lowerCamelCase_			)
     return model



def 						lowercase		(     _lowerCAmelCase   ,     _lowerCAmelCase   ,     _lowerCAmelCase      ):
  UpperCAmelCase__    						=   Path(_lowerCAmelCase      )
  dest_dir.mkdir(exist_ok=_lowerCAmelCase      )

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


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

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










import shutil
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from transformers import SpeechaTextTokenizer, is_speech_available
from transformers.models.speech_to_text.tokenization_speech_to_text import VOCAB_FILES_NAMES, save_json
from transformers.testing_utils import get_tests_dir, require_sentencepiece, require_torch, require_torchaudio
from transformers.utils import FEATURE_EXTRACTOR_NAME

from .test_feature_extraction_speech_to_text import floats_list


if is_speech_available():
   from transformers import SpeechaTextFeatureExtractor, SpeechaTextProcessor


snake_case__ :						Union[str, Any]     =			get_tests_dir('''fixtures/test_sentencepiece.model''')





@require_torch
@require_torchaudio
@require_sentencepiece
class     snake_case  (	unittest.TestCase							):


   '''simple docstring'''







   def 	UpperCAmelCase       (					self     :	str			)					->str:
     '''simple docstring'''







     UpperCAmelCase__    						=   tempfile.mkdtemp()

     UpperCAmelCase__    						=   ["""<s>""", """<pad>""", """</s>""", """<unk>""", """▁This""", """▁is""", """▁a""", """▁t""", """est"""]
     UpperCAmelCase__    						=   dict(zip(lowerCamelCase_						,    range(len(lowerCamelCase_			)			)			)			)
     UpperCAmelCase__    						=   Path(self.tmpdirname			)
     save_json(lowerCamelCase_						,    save_dir / VOCAB_FILES_NAMES["""vocab_file"""]			)
     if not (save_dir / VOCAB_FILES_NAMES["spm_file"]).exists():
       copyfile(lowerCamelCase_						,    save_dir / VOCAB_FILES_NAMES["""spm_file"""]			)

     UpperCAmelCase__    						=   SpeechaTextTokenizer.from_pretrained(self.tmpdirname			)
     tokenizer.save_pretrained(self.tmpdirname			)

     UpperCAmelCase__    						=   {
         """feature_size""": 24,
         """num_mel_bins""": 24,
         """padding_value""": 0.0,
         """sampling_rate""": 1_6000,
         """return_attention_mask""": False,
         """do_normalize""": True,
     }
     save_json(lowerCamelCase_						,    save_dir / FEATURE_EXTRACTOR_NAME			)




   def 	UpperCAmelCase       (					self     :	Union[str, Any]						,    **lowerCamelCase_     :	Tuple			)					->Any:
     '''simple docstring'''







     return SpeechaTextTokenizer.from_pretrained(self.tmpdirname						,    **lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	Dict						,    **lowerCamelCase_     :	str			)					->Optional[Any]:
     '''simple docstring'''







     return SpeechaTextFeatureExtractor.from_pretrained(self.tmpdirname						,    **lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	List[str]			)					->Dict:
     '''simple docstring'''







     shutil.rmtree(self.tmpdirname			)




   def 	UpperCAmelCase       (					self     :	str			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_tokenizer()
     UpperCAmelCase__    						=   self.get_feature_extractor()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     processor.save_pretrained(self.tmpdirname			)
     UpperCAmelCase__    						=   SpeechaTextProcessor.from_pretrained(self.tmpdirname			)

     self.assertEqual(processor.tokenizer.get_vocab()						,    tokenizer.get_vocab()			)
     self.assertIsInstance(processor.tokenizer						,    lowerCamelCase_			)

     self.assertEqual(processor.feature_extractor.to_json_string()						,    feature_extractor.to_json_string()			)
     self.assertIsInstance(processor.feature_extractor						,    lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	List[str]			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   SpeechaTextProcessor(
         tokenizer=self.get_tokenizer()						,    feature_extractor=self.get_feature_extractor()			)
     processor.save_pretrained(self.tmpdirname			)

     UpperCAmelCase__    						=   self.get_tokenizer(bos_token="""(BOS)"""						,    eos_token="""(EOS)"""			)
     UpperCAmelCase__    						=   self.get_feature_extractor(do_normalize=lowerCamelCase_						,    padding_value=1.0			)

     UpperCAmelCase__    						=   SpeechaTextProcessor.from_pretrained(
         self.tmpdirname						,    bos_token="""(BOS)"""						,    eos_token="""(EOS)"""						,    do_normalize=lowerCamelCase_						,    padding_value=1.0			)

     self.assertEqual(processor.tokenizer.get_vocab()						,    tokenizer_add_kwargs.get_vocab()			)
     self.assertIsInstance(processor.tokenizer						,    lowerCamelCase_			)

     self.assertEqual(processor.feature_extractor.to_json_string()						,    feature_extractor_add_kwargs.to_json_string()			)
     self.assertIsInstance(processor.feature_extractor						,    lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	Union[str, Any]			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     UpperCAmelCase__    						=   floats_list((3, 1000)			)

     UpperCAmelCase__    						=   feature_extractor(lowerCamelCase_						,    return_tensors="""np"""			)
     UpperCAmelCase__    						=   processor(lowerCamelCase_						,    return_tensors="""np"""			)

     for key in input_feat_extract.keys():
       self.assertAlmostEqual(input_feat_extract[key].sum()						,    input_processor[key].sum()						,    delta=1E-2			)




   def 	UpperCAmelCase       (					self     :	str			)					->Union[str, Any]:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     UpperCAmelCase__    						=   """This is a test string"""

     UpperCAmelCase__    						=   processor(text=lowerCamelCase_			)

     UpperCAmelCase__    						=   tokenizer(lowerCamelCase_			)

     for key in encoded_tok.keys():
       self.assertListEqual(encoded_tok[key]						,    encoded_processor[key]			)




   def 	UpperCAmelCase       (					self     :	Optional[Any]			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     UpperCAmelCase__    						=   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

     UpperCAmelCase__    						=   processor.batch_decode(lowerCamelCase_			)
     UpperCAmelCase__    						=   tokenizer.batch_decode(lowerCamelCase_			)

     self.assertListEqual(lowerCamelCase_						,    lowerCamelCase_			)







   def 	UpperCAmelCase       (					self     :	Dict			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     self.assertListEqual(
         processor.model_input_names						,    feature_extractor.model_input_names						,    msg="""`processor` and `feature_extractor` model input names do not match"""						,    )




