


import argparse
import json
import os

import fairseq
import torch
from torch import nn

from transformers import (
    SpeechaTextaConfig,
    SpeechaTextaForCausalLM,
    SpeechaTextaTokenizer,
    SpeechEncoderDecoderConfig,
    SpeechEncoderDecoderModel,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaModel,
    logging,
)


logging.set_verbosity_info()
snake_case				:    Dict       				=     logging.get_logger(__name__)

snake_case				:    Dict       				=     {
    '''post_extract_proj''': '''feature_projection.projection''',
    '''encoder.pos_conv.0''': '''encoder.pos_conv_embed.conv''',
    '''self_attn.k_proj''': '''encoder.layers.*.attention.k_proj''',
    '''self_attn.v_proj''': '''encoder.layers.*.attention.v_proj''',
    '''self_attn.q_proj''': '''encoder.layers.*.attention.q_proj''',
    '''self_attn.out_proj''': '''encoder.layers.*.attention.out_proj''',
    '''self_attn_layer_norm''': '''encoder.layers.*.layer_norm''',
    '''fc1''': '''encoder.layers.*.feed_forward.intermediate_dense''',
    '''fc2''': '''encoder.layers.*.feed_forward.output_dense''',
    '''final_layer_norm''': '''encoder.layers.*.final_layer_norm''',
    '''encoder.layer_norm''': '''encoder.layer_norm''',
    '''w2v_model.layer_norm''': '''feature_projection.layer_norm''',
    '''quantizer.weight_proj''': '''quantizer.weight_proj''',
    '''quantizer.vars''': '''quantizer.codevectors''',
    '''project_q''': '''project_q''',
    '''final_proj''': '''project_hid''',
    '''w2v_encoder.proj''': '''lm_head''',
    '''mask_emb''': '''masked_spec_embed''',
}
snake_case				:    Union[str, Any]       				=     [
    '''lm_head''',
    '''quantizer.weight_proj''',
    '''quantizer.codevectors''',
    '''project_q''',
    '''project_hid''',
]
def 				__lowerCamelCase						( UpperCAmelCase_						:		Dict ,  UpperCAmelCase_						:		int ,  UpperCAmelCase_						:		Optional[Any] ,  UpperCAmelCase_						:		List[str] ,  UpperCAmelCase_						:		Any						):




      """simple docstring"""





      for attribute in key.split('''.'''						):
            a						:List[str] 							=       getattr(UpperCAmelCase_ ,  UpperCAmelCase_						)

      if weight_type is not None:
            a						:Dict 							=       getattr(UpperCAmelCase_ ,  UpperCAmelCase_						).shape
      else:
            a						:Dict 							=       hf_pointer.shape

      assert hf_shape == value.shape, (
          F'''Shape of hf {key + '.' + weight_type if weight_type is not None else ''} is {hf_shape}, but should be'''
          F''' {value.shape} for {full_name}'''
      )

      if weight_type == "weight":
            a						:Union[str, Any] 							=       value
      elif weight_type == "weight_g":
            a						:Dict 							=       value
      elif weight_type == "weight_v":
            a						:int 							=       value
      elif weight_type == "bias":
            a						:Optional[Any] 							=       value
      else:
            a						:Any 							=       value

      logger.info(F'''{key + '.' + weight_type if weight_type is not None else ''} was initialized from {full_name}.'''						)
def 				__lowerCamelCase						( UpperCAmelCase_						:		List[Any] ,  UpperCAmelCase_						:		Union[str, Any]						):




      """simple docstring"""





      a						:Dict 							=       []
      a						:Optional[int] 							=       fairseq_model.state_dict()

      a						:Optional[Any] 							=       hf_model.feature_extractor

      # if encoder has different dim to decoder -> use proj_weight
      a						:Dict 							=       None

      for name, value in fairseq_dict.items():
            a						:List[Any] 							=       False
            if "conv_layers" in name:
                  load_conv_layer(
                      UpperCAmelCase_ ,  UpperCAmelCase_ ,  UpperCAmelCase_ ,  UpperCAmelCase_ ,  hf_model.config.feat_extract_norm == '''group''' ,  )
                  a						:Optional[int] 							=       True
            elif name.split('''.'''						)[0] == "proj":
                  a						:Union[str, Any] 							=       fairseq_model.proj
                  a						:Optional[Any] 							=       True
            else:
                  for key, mapped_key in MAPPING.items():
                        if key in name or key.split('''w2v_model.'''						)[-1] == name.split('''.'''						)[0]:
                              a						:int 							=       True
                              if "*" in mapped_key:
                                    a						:Any 							=       name.split(UpperCAmelCase_						)[0].split('''.'''						)[-2]
                                    a						:int 							=       mapped_key.replace('''*''' ,  UpperCAmelCase_						)
                              if "weight_g" in name:
                                    a						:int 							=       '''weight_g'''
                              elif "weight_v" in name:
                                    a						:Any 							=       '''weight_v'''
                              elif "bias" in name:
                                    a						:Union[str, Any] 							=       '''bias'''
                              elif "weight" in name:
                                    a						:Any 							=       '''weight'''
                              else:
                                    a						:Any 							=       None
                              set_recursively(UpperCAmelCase_ ,  UpperCAmelCase_ ,  UpperCAmelCase_ ,  UpperCAmelCase_ ,  UpperCAmelCase_						)
                        continue
            if not is_used:
                  unused_weights.append(UpperCAmelCase_						)

      logger.warning(F'''Unused weights: {unused_weights}'''						)

      return proj_weight
def 				__lowerCamelCase						( UpperCAmelCase_						:		str ,  UpperCAmelCase_						:		int ,  UpperCAmelCase_						:		Union[str, Any] ,  UpperCAmelCase_						:		Tuple ,  UpperCAmelCase_						:		int						):




      """simple docstring"""





      a						:List[Any] 							=       full_name.split('''conv_layers.'''						)[-1]
      a						:int 							=       name.split('''.'''						)
      a						:List[str] 							=       int(items[0]						)
      a						:Optional[int] 							=       int(items[1]						)

      if type_id == 0:
            if "bias" in name:
                  assert value.shape == feature_extractor.conv_layers[layer_id].conv.bias.data.shape, (
                      F'''{full_name} has size {value.shape}, but'''
                      F''' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.'''
                  )
                  a						:str 							=       value
                  logger.info(F'''Feat extract conv layer {layer_id} was initialized from {full_name}.'''						)
            elif "weight" in name:
                  assert value.shape == feature_extractor.conv_layers[layer_id].conv.weight.data.shape, (
                      F'''{full_name} has size {value.shape}, but'''
                      F''' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.'''
                  )
                  a						:Any 							=       value
                  logger.info(F'''Feat extract conv layer {layer_id} was initialized from {full_name}.'''						)
      elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
            if "bias" in name:
                  assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape, (
                      F'''{full_name} has size {value.shape}, but {feature_extractor[layer_id].layer_norm.bias.data.shape} was'''
                      " found."
                  )
                  a						:Optional[int] 							=       value
                  logger.info(F'''Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'''						)
            elif "weight" in name:
                  assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape, (
                      F'''{full_name} has size {value.shape}, but'''
                      F''' {feature_extractor[layer_id].layer_norm.weight.data.shape} was found.'''
                  )
                  a						:Dict 							=       value
                  logger.info(F'''Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'''						)
      else:
            unused_weights.append(UpperCAmelCase_						)
def 				__lowerCamelCase						( UpperCAmelCase_						:		List[str]						):




      """simple docstring"""





      a							,				a						:Any 							=       emb.weight.shape
      a						:Optional[int] 							=       nn.Linear(UpperCAmelCase_ ,  UpperCAmelCase_ ,  bias=UpperCAmelCase_						)
      a						:Dict 							=       emb.weight.data
      return lin_layer
def 				__lowerCamelCase						( UpperCAmelCase_						:		Dict						):




      """simple docstring"""





      with open(UpperCAmelCase_ ,  '''r''' ,  encoding='''utf-8'''						) as f:
            a						:Tuple 							=       f.readlines()
            a						:List[str] 							=       [line.split(''' '''						)[0] for line in lines]

      a						:str 							=       len(UpperCAmelCase_						)

      a						:Union[str, Any] 							=       {
          '''<s>''': 0,
          '''<pad>''': 1,
          '''</s>''': 2,
          '''<unk>''': 3,
      }

      vocab_dict.update(dict(zip(UpperCAmelCase_ ,  range(4 ,  num_words + 4						)						)						)						)
      return vocab_dict
@torch.no_grad()
def 				__lowerCamelCase						( UpperCAmelCase_						:		List[Any] ,  UpperCAmelCase_						:		Any ,  UpperCAmelCase_						:		List[Any] ,  UpperCAmelCase_						:		List[str] ,  UpperCAmelCase_						:		int ,  UpperCAmelCase_						:		Dict ,  UpperCAmelCase_						:		Optional[int] ,  ):




      """simple docstring"""





      a						:int 							=       WavaVecaConfig.from_pretrained(UpperCAmelCase_						)
      a						:Union[str, Any] 							=       SpeechaTextaConfig.from_pretrained(
          UpperCAmelCase_ ,  vocab_size=UpperCAmelCase_ ,  decoder_layers=UpperCAmelCase_ ,  do_stable_layer_norm=UpperCAmelCase_						)

      a						:str 							=       WavaVecaFeatureExtractor(
          feature_size=1 ,  sampling_rate=1_6000 ,  padding_value=0 ,  do_normalize=UpperCAmelCase_ ,  return_attention_mask=UpperCAmelCase_ ,  )

      a							,				a							,				a						:int 							=       fairseq.checkpoint_utils.load_model_ensemble_and_task(
          [checkpoint_path] ,  arg_overrides={'''data''': '''/'''.join(dict_path.split('''/'''						)[:-1]						)}						)
      a						:Optional[int] 							=       model[0].eval()

      # set weights for wav2vec2 encoder
      a						:List[Any] 							=       WavaVecaModel(UpperCAmelCase_						)
      a						:Optional[int] 							=       recursively_load_weights_wavaveca(model.encoder ,  UpperCAmelCase_						)

      a						:Union[str, Any] 							=       SpeechaTextaForCausalLM(UpperCAmelCase_						)
      a							,				a						:str 							=       hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict() ,  strict=UpperCAmelCase_						)

      # set output linear layer
      unexpected_keys.remove('''embed_out'''						)
      a						:List[str] 							=       nn.Parameter(model.decoder.embed_out.detach()						)

      # layer norm is init to identity matrix so leaving it is fine
      logger.warning(F'''The following keys are missing when loading the decoder weights: {missing_keys}'''						)
      logger.warning(F'''The following keys are unexpected when loading the decoder weights: {unexpected_keys}'''						)

      a						:Optional[int] 							=       SpeechEncoderDecoderModel(encoder=UpperCAmelCase_ ,  decoder=UpperCAmelCase_						)
      a						:List[str] 							=       False

      # add projection layer
      a						:int 							=       nn.Parameter(projection_layer.weight						)
      a						:List[str] 							=       nn.Parameter(projection_layer.bias						)

      a						:Optional[Any] 							=       create_vocab_dict(UpperCAmelCase_						)

      with open(os.path.join(UpperCAmelCase_ ,  '''vocab.json'''						) ,  '''w'''						) as fp:
            json.dump(UpperCAmelCase_ ,  UpperCAmelCase_						)

      a						:Tuple 							=       SpeechaTextaTokenizer(os.path.join(UpperCAmelCase_ ,  '''vocab.json'''						)						)
      tokenizer.save_pretrained(UpperCAmelCase_						)

      a						:Any 							=       hf_wavavec.config.to_dict()
      a						:Optional[int] 							=       tokenizer.pad_token_id
      a						:List[str] 							=       tokenizer.bos_token_id
      a						:List[str] 							=       tokenizer.eos_token_id
      a						:Dict 							=       '''speech_to_text_2'''
      a						:Union[str, Any] 							=       '''wav2vec2'''

      a						:List[str] 							=       SpeechEncoderDecoderConfig.from_dict(UpperCAmelCase_						)

      hf_wavavec.save_pretrained(UpperCAmelCase_						)
      feature_extractor.save_pretrained(UpperCAmelCase_						)


if __name__ == "__main__":
     snake_case				:    Optional[int]       				=     argparse.ArgumentParser()
     parser.add_argument('''--pytorch_dump_folder_path''', default=None, type=str, help='''Path to the output PyTorch model.''')
     parser.add_argument('''--checkpoint_path''', default=None, type=str, help='''Path to fairseq checkpoint''')
     parser.add_argument('''--dict_path''', default=None, type=str, help='''Path to dict of fine-tuned model''')
     parser.add_argument(
         '''--encoder_config_path''',
         default='''facebook/wav2vec2-large-lv60''',
         type=str,
         help='''Path to hf encoder wav2vec2 checkpoint config''',
     )
     parser.add_argument(
         '''--decoder_config_path''',
         default='''facebook/s2t-small-mustc-en-fr-st''',
         type=str,
         help='''Path to hf decoder s2t checkpoint config''',
     )
     parser.add_argument('''--vocab_size''', default=1_02_24, type=int, help='''Vocab size of decoder''')
     parser.add_argument('''--num_decoder_layers''', default=7, type=int, help='''Number of decoder layers''')

     snake_case				:    List[Any]       				=     parser.parse_args()
     convert_wavaveca_checkpoint(
         args.checkpoint_path,
         args.pytorch_dump_folder_path,
         args.dict_path,
         encoder_config_path=args.encoder_config_path,
         decoder_config_path=args.decoder_config_path,
         vocab_size=args.vocab_size,
         num_decoder_layers=args.num_decoder_layers,
     )









from pathlib import PurePosixPath
from typing import Optional

import fsspec
from fsspec import AbstractFileSystem
from huggingface_hub.hf_api import DatasetInfo

from ..utils.file_utils import get_authentication_headers_for_url
from ..utils.hub import hf_hub_url







class 		_snake_case  (      _snake_case							):
 SCREAMING_SNAKE_CASE__      =       ''
 SCREAMING_SNAKE_CASE__      =       'hf-legacy'  # "hf://"" is reserved for hffs




 def __init__(		self					,  _lowerCamelCase = None					,  _lowerCamelCase = None					,  **_lowerCamelCase					,  ):
       super().__init__(self					,  **_lowerCamelCase    )
       a						:Union[str, Any] 							=       repo_info
       a						:int 							=       token
       a						:int 							=       None




 def        SCREAMING_SNAKE_CASE__				(		self    ):
       if self.dir_cache is None:
             a						:Dict 							=       {}
             for hf_file in self.repo_info.siblings:
                   # TODO(QL): add sizes
                   a						:List[Any] 							=       {
                       '''name''': hf_file.rfilename,
                       '''size''': None,
                       '''type''': '''file''',
                   }
                   self.dir_cache.update(
                       {
                           str(_lowerCamelCase    ): {'''name''': str(_lowerCamelCase    ), '''size''': None, '''type''': '''directory'''}
                           for d in list(PurePosixPath(hf_file.rfilename    ).parents    )[:-1]
                       }    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = "rb"					,  **_lowerCamelCase					,  ):
       if not isinstance(self.repo_info					,  _lowerCamelCase    ):
             raise NotImplementedError(F'''Open is only implemented for dataset repositories, but got {self.repo_info}'''    )
       a						:Optional[int] 							=       hf_hub_url(self.repo_info.id					,  _lowerCamelCase					,  revision=self.repo_info.sha    )
       return fsspec.open(
           _lowerCamelCase					,  mode=_lowerCamelCase					,  headers=get_authentication_headers_for_url(_lowerCamelCase					,  use_auth_token=self.token    )					,  client_kwargs={'''trust_env''': True}					,  ).open()




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  **_lowerCamelCase    ):
       self._get_dirs()
       a						:Union[str, Any] 							=       self._strip_protocol(_lowerCamelCase    )
       if path in self.dir_cache:
             return self.dir_cache[path]
       else:
             raise FileNotFoundError(_lowerCamelCase    )




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase=False					,  **_lowerCamelCase    ):
       self._get_dirs()
       a						:str 							=       PurePosixPath(path.strip('''/'''    )    )
       a						:Tuple 							=       {}
       for p, f in self.dir_cache.items():
             a						:Optional[int] 							=       PurePosixPath(p.strip('''/'''    )    )
             a						:str 							=       p.parent
             if root == path:
                   a						:List[str] 							=       f
       a						:Any 							=       list(paths.values()    )
       if detail:
             return out
       else:
             return sorted(f['''name'''] for f in out    )





