'''simple docstring'''





import argparse
import os
import re

import packaging.version


SCREAMING_SNAKE_CASE_:					List[str]  ='examples/'
SCREAMING_SNAKE_CASE_:					List[Any]  ={
    'examples': (re.compile(r'^check_min_version\("[^"]+"\)\s*$', re.MULTILINE), 'check_min_version("VERSION")\n'),
    'init': (re.compile(r'^__version__\s+=\s+"([^"]+)"\s*$', re.MULTILINE), '__version__ = "VERSION"\n'),
    'setup': (re.compile(r'^(\s*)version\s*=\s*"[^"]+",', re.MULTILINE), r'\1version="VERSION",'),
    'doc': (re.compile(r'^(\s*)release\s*=\s*"[^"]+"$', re.MULTILINE), 'release = "VERSION"\n'),
}
SCREAMING_SNAKE_CASE_:					str  ={
    'init': 'src/transformers/__init__.py',
    'setup': 'setup.py',
}
SCREAMING_SNAKE_CASE_:					str  ='README.md'





def      lowerCAmelCase_    ( snake_case_	:			List[str] ,   snake_case_	:			str ,   snake_case_	:			str      )   ->				List[str]:


     '''simple docstring'''




     with open(_A ,   "r" ,   encoding="utf-8" ,   newline="\n"      ) as f:
          UpperCAmelCase_	   =       f.read()
     UpperCAmelCase_	   =       REPLACE_PATTERNS[pattern]
     UpperCAmelCase_	   =       replace.replace("VERSION" ,   _A      )
     UpperCAmelCase_	   =       re_pattern.sub(_A ,   _A      )
     with open(_A ,   "w" ,   encoding="utf-8" ,   newline="\n"      ) as f:
          f.write(_A      )





def      lowerCAmelCase_    ( snake_case_	:			List[str]      )   ->				Any:


     '''simple docstring'''




     for folder, directories, fnames in os.walk(_A      ):
          # Removing some of the folders with non-actively maintained examples from the walk
          if "research_projects" in directories:
               directories.remove("research_projects"      )
          if "legacy" in directories:
               directories.remove("legacy"      )
          for fname in fnames:
               if fname.endswith(".py"      ):
                    update_version_in_file(os.path.join(_A ,   _A      ) ,   _A ,   pattern="examples"      )





def      lowerCAmelCase_    ( snake_case_	:			List[str] ,   snake_case_	:			Optional[int]=False      )   ->				Optional[int]:


     '''simple docstring'''




     for pattern, fname in REPLACE_FILES.items():
          update_version_in_file(_A ,   _A ,   _A      )
     if not patch:
          update_version_in_examples(_A      )





def      lowerCAmelCase_    ( )   ->				Optional[Any]:


     '''simple docstring'''




     UpperCAmelCase_	   =       '🤗 Transformers currently provides the following architectures'
     UpperCAmelCase_	   =       '1. Want to contribute a new model?'
     with open(_A ,   "r" ,   encoding="utf-8" ,   newline="\n"      ) as f:
          UpperCAmelCase_	   =       f.readlines()

     # Find the start of the list.
     UpperCAmelCase_	   =       0
     while not lines[start_index].startswith(_start_prompt      ):
          start_index += 1
     start_index += 1

     UpperCAmelCase_	   =       start_index
     # Update the lines in the model list.
     while not lines[index].startswith(_end_prompt      ):
          if lines[index].startswith("1."      ):
               UpperCAmelCase_	   =       lines[index].replace(
                   "https://huggingface.co/docs/transformers/main/model_doc" ,   "https://huggingface.co/docs/transformers/model_doc" ,   )
          index += 1

     with open(_A ,   "w" ,   encoding="utf-8" ,   newline="\n"      ) as f:
          f.writelines(_A      )





def      lowerCAmelCase_    ( )   ->				int:


     '''simple docstring'''




     with open(REPLACE_FILES["init"] ,   "r"      ) as f:
          UpperCAmelCase_	   =       f.read()
     UpperCAmelCase_	   =       REPLACE_PATTERNS['init'][0].search(_A      ).groups()[0]
     return packaging.version.parse(_A      )





def      lowerCAmelCase_    ( snake_case_	:			List[Any]=False      )   ->				List[str]:


     '''simple docstring'''




     UpperCAmelCase_	   =       get_version()
     if patch and default_version.is_devrelease:
          raise ValueError("Can\'t create a patch version from the dev branch, checkout a released version!"      )
     if default_version.is_devrelease:
          UpperCAmelCase_	   =       default_version.base_version
     elif patch:
          UpperCAmelCase_	   =       f"""{default_version.major}.{default_version.minor}.{default_version.micro + 1}"""
     else:
          UpperCAmelCase_	   =       f"""{default_version.major}.{default_version.minor + 1}.0"""

     # Now let's ask nicely if that's the right one.
     UpperCAmelCase_	   =       input(f"""Which version are you releasing? [{default_version}]"""      )
     if len(_A      ) == 0:
          UpperCAmelCase_	   =       default_version

     print(f"""Updating version to {version}."""      )
     global_version_update(_A ,   patch=_A      )
     if not patch:
          print("Cleaning main README, don\'t forget to run `make fix-copies`."      )
          clean_main_ref_in_model_list()





def      lowerCAmelCase_    ( )   ->				Union[str, Any]:


     '''simple docstring'''




     UpperCAmelCase_	   =       get_version()
     UpperCAmelCase_	   =       f"""{current_version.major}.{current_version.minor + 1}.0.dev0"""
     UpperCAmelCase_	   =       current_version.base_version

     # Check with the user we got that right.
     UpperCAmelCase_	   =       input(f"""Which version are we developing now? [{dev_version}]"""      )
     if len(_A      ) == 0:
          UpperCAmelCase_	   =       dev_version

     print(f"""Updating version to {version}."""      )
     global_version_update(_A      )
     print("Cleaning main README, don\'t forget to run `make fix-copies`."      )
     clean_main_ref_in_model_list()


if __name__ == "__main__":
    SCREAMING_SNAKE_CASE_:					Tuple  =argparse.ArgumentParser()
    parser.add_argument('--post_release', action='store_true', help='Whether this is pre or post release.')
    parser.add_argument('--patch', action='store_true', help='Whether or not this is a patch release.')
    SCREAMING_SNAKE_CASE_:					Optional[Any]  =parser.parse_args()
    if not args.post_release:
        pre_release_work(patch=args.patch)
    elif args.patch:
        print('Nothing to do after a patch :-)')
    else:
        post_release_work()





'''simple docstring'''





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()
SCREAMING_SNAKE_CASE_:					int  =logging.get_logger(__name__)

SCREAMING_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',
}
SCREAMING_SNAKE_CASE_:					List[str]  =[
    'lm_head',
    'quantizer.weight_proj',
    'quantizer.codevectors',
    'project_q',
    'project_hid',
]





def      lowerCAmelCase_    ( snake_case_	:			str ,   snake_case_	:			List[Any] ,   snake_case_	:			Union[str, Any] ,   snake_case_	:			int ,   snake_case_	:			Any      )   ->				Optional[int]:


     '''simple docstring'''




     for attribute in key.split("."      ):
          UpperCAmelCase_	   =       getattr(snake_case_ ,   snake_case_      )

     if weight_type is not None:
          UpperCAmelCase_	   =       getattr(snake_case_ ,   snake_case_      ).shape
     else:
          UpperCAmelCase_	   =       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":
          UpperCAmelCase_	   =       value
     elif weight_type == "weight_g":
          UpperCAmelCase_	   =       value
     elif weight_type == "weight_v":
          UpperCAmelCase_	   =       value
     elif weight_type == "bias":
          UpperCAmelCase_	   =       value
     else:
          UpperCAmelCase_	   =       value

     logger.info(f"""{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}."""      )





def      lowerCAmelCase_    ( snake_case_	:			List[Any] ,   snake_case_	:			List[str]      )   ->				int:


     '''simple docstring'''




     UpperCAmelCase_	   =       []
     UpperCAmelCase_	   =       fairseq_model.state_dict()

     UpperCAmelCase_	   =       hf_model.feature_extractor

     # if encoder has different dim to decoder -> use proj_weight
     UpperCAmelCase_	   =       None

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

     logger.warning(f"""Unused weights: {unused_weights}"""      )

     return proj_weight





def      lowerCAmelCase_    ( snake_case_	:			Dict ,   snake_case_	:			List[Any] ,   snake_case_	:			str ,   snake_case_	:			Any ,   snake_case_	:			int      )   ->				Dict:


     '''simple docstring'''




     UpperCAmelCase_	   =       full_name.split("conv_layers."      )[-1]
     UpperCAmelCase_	   =       name.split("."      )
     UpperCAmelCase_	   =       int(items[0]      )
     UpperCAmelCase_	   =       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."""
               )
               UpperCAmelCase_	   =       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."""
               )
               UpperCAmelCase_	   =       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."
               )
               UpperCAmelCase_	   =       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."""
               )
               UpperCAmelCase_	   =       value
               logger.info(f"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""      )
     else:
          unused_weights.append(snake_case_      )





def      lowerCAmelCase_    ( snake_case_	:			Any      )   ->				Any:


     '''simple docstring'''




     UpperCAmelCase_   ,			UpperCAmelCase_	   =       emb.weight.shape
     UpperCAmelCase_	   =       nn.Linear(snake_case_ ,   snake_case_ ,   bias=snake_case_      )
     UpperCAmelCase_	   =       emb.weight.data
     return lin_layer





def      lowerCAmelCase_    ( snake_case_	:			int      )   ->				List[str]:


     '''simple docstring'''




     with open(snake_case_ ,   "r" ,   encoding="utf-8"      ) as f:
          UpperCAmelCase_	   =       f.readlines()
          UpperCAmelCase_	   =       [line.split(" "      )[0] for line in lines]

     UpperCAmelCase_	   =       len(snake_case_      )

     UpperCAmelCase_	   =       {
         "<s>": 0,
         "<pad>": 1,
         "</s>": 2,
         "<unk>": 3,
     }

     vocab_dict.update(dict(zip(snake_case_ ,   range(4 ,   num_words + 4      )      )      )      )
     return vocab_dict





@torch.no_grad()
def      lowerCAmelCase_    ( snake_case_	:			Dict ,   snake_case_	:			Tuple ,   snake_case_	:			List[str] ,   snake_case_	:			Dict ,   snake_case_	:			Union[str, Any] ,   snake_case_	:			Tuple ,   snake_case_	:			Any ,   )   ->				Dict:


     '''simple docstring'''




     UpperCAmelCase_	   =       WavaVecaConfig.from_pretrained(snake_case_      )
     UpperCAmelCase_	   =       SpeechaTextaConfig.from_pretrained(
         snake_case_ ,   vocab_size=snake_case_ ,   decoder_layers=snake_case_ ,   do_stable_layer_norm=snake_case_      )

     UpperCAmelCase_	   =       WavaVecaFeatureExtractor(
         feature_size=1 ,   sampling_rate=1_60_00 ,   padding_value=0 ,   do_normalize=snake_case_ ,   return_attention_mask=snake_case_ ,   )

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

     # set weights for wav2vec2 encoder
     UpperCAmelCase_	   =       WavaVecaModel(snake_case_      )
     UpperCAmelCase_	   =       recursively_load_weights_wavaveca(model.encoder ,   snake_case_      )

     UpperCAmelCase_	   =       SpeechaTextaForCausalLM(snake_case_      )
     UpperCAmelCase_   ,			UpperCAmelCase_	   =       hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict() ,   strict=snake_case_      )

     # set output linear layer
     unexpected_keys.remove("embed_out"      )
     UpperCAmelCase_	   =       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}"""      )

     UpperCAmelCase_	   =       SpeechEncoderDecoderModel(encoder=snake_case_ ,   decoder=snake_case_      )
     UpperCAmelCase_	   =       False

     # add projection layer
     UpperCAmelCase_	   =       nn.Parameter(projection_layer.weight      )
     UpperCAmelCase_	   =       nn.Parameter(projection_layer.bias      )

     UpperCAmelCase_	   =       create_vocab_dict(snake_case_      )

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

     UpperCAmelCase_	   =       SpeechaTextaTokenizer(os.path.join(snake_case_ ,   "vocab.json"      )      )
     tokenizer.save_pretrained(snake_case_      )

     UpperCAmelCase_	   =       hf_wavavec.config.to_dict()
     UpperCAmelCase_	   =       tokenizer.pad_token_id
     UpperCAmelCase_	   =       tokenizer.bos_token_id
     UpperCAmelCase_	   =       tokenizer.eos_token_id
     UpperCAmelCase_	   =       "speech_to_text_2"
     UpperCAmelCase_	   =       "wav2vec2"

     UpperCAmelCase_	   =       SpeechEncoderDecoderConfig.from_dict(snake_case_      )

     hf_wavavec.save_pretrained(snake_case_      )
     feature_extractor.save_pretrained(snake_case_      )


if __name__ == "__main__":
    SCREAMING_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')

    SCREAMING_SNAKE_CASE_:					Dict  =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,
    )




