





import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConfig,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaForCTC,
    WavaVecaForPreTraining,
    WavaVecaProcessor,
    logging,
)
from transformers.models.wavaveca.modeling_wavaveca import WavaVecaForSequenceClassification


logging.set_verbosity_info()
lowercase				:	Optional[Any]      						=  logging.get_logger(__name__)

lowercase				:	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""",
    """adapter_layer""": """encoder.layers.*.adapter_layer""",
    """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""",
    """pooling_layer.linear""": """projector""",
    """pooling_layer.projection""": """classifier""",
}
lowercase				:	Any      						=  [
    """lm_head""",
    """quantizer.weight_proj""",
    """quantizer.codevectors""",
    """project_q""",
    """project_hid""",
    """projector""",
    """classifier""",
]



def A_     (		A__    )  ->		int:
       a__						:     int			=    {}
       with open(A__		,				'r'    ) as file:
              for line_number, line in enumerate(A__    ):
                     a__						:     Any			=    line.strip()
                     if line:
                            a__						:     Optional[int]			=    line.split()
                            a__						:     str			=    line_number
                            a__						:     Tuple			=    words[0]
                            a__						:     Tuple			=    value
       return result



def A_     (		A__		,				A__		,				A__		,				A__		,				A__    )  ->		Union[str, Any]:
       for attribute in key.split('.'    ):
              a__						:     Any			=    getattr(A__		,				A__    )

       a__						:     Union[str, Any]			=    None
       for param_key in PARAM_MAPPING.keys():
              if full_name.endswith(A__    ):
                     a__						:     Dict			=    PARAM_MAPPING[full_name.split('.'    )[-1]]
                     a__						:     str			=    'param'

       if weight_type is not None and weight_type != "param":
              a__						:     Dict			=    getattr(A__		,				A__    ).shape
       elif weight_type is not None and weight_type == "param":
              a__						:     Dict			=    hf_pointer
              for attribute in hf_param_name.split('.'    ):
                     a__						:     Tuple			=    getattr(A__		,				A__    )
              a__						:     str			=    shape_pointer.shape

              # let's reduce dimension
              a__						:     str			=    value[0]
       else:
              a__						:     Dict			=    hf_pointer.shape

       if hf_shape != value.shape:
              raise ValueError(
                  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__						:     Dict			=    value
       elif weight_type == "weight_g":
              a__						:     List[str]			=    value
       elif weight_type == "weight_v":
              a__						:     Dict			=    value
       elif weight_type == "bias":
              a__						:     Dict			=    value
       elif weight_type == "param":
              for attribute in hf_param_name.split('.'    ):
                     a__						:     Optional[int]			=    getattr(A__		,				A__    )
              a__						:     Tuple			=    value
       else:
              a__						:     Union[str, Any]			=    value

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



def A_     (		A__		,				A__		,				A__		,				A__		,				A__    )  ->		Optional[int]:
       a__						:     List[str]			=    None
       for param_key in PARAM_MAPPING.keys():
              if full_name.endswith(A__    ):
                     a__						:     Any			=    PARAM_MAPPING[full_name.split('.'    )[-1]]
                     a__						:     Any			=    'param'

       if weight_type is not None and weight_type != "param":
              a__						:     Dict			=    '.'.join([key, weight_type]    )
       elif weight_type is not None and weight_type == "param":
              a__						:     str			=    '.'.join([key, hf_param_name]    )
       else:
              a__						:     List[Any]			=    key

       a__						:     Optional[Any]			=    value if 'lm_head' in full_key else value[0]


lowercase				:	Optional[Any]      						=  {
    """W_a""": """linear_1.weight""",
    """W_b""": """linear_2.weight""",
    """b_a""": """linear_1.bias""",
    """b_b""": """linear_2.bias""",
    """ln_W""": """norm.weight""",
    """ln_b""": """norm.bias""",
}



def A_     (		A__		,				A__		,				A__=None		,				A__=None    )  ->		Optional[Any]:
       a__						:     Tuple			=    False
       for key, mapped_key in MAPPING.items():
              a__						:     Optional[int]			=    'wav2vec2.' + mapped_key if mapped_key not in TOP_LEVEL_KEYS else mapped_key
              if key in name or key.split('w2v_model.'    )[-1] == name.split('.'    )[0]:
                     a__						:     Dict			=    True
                     if "*" in mapped_key:
                            a__						:     Any			=    name.split(A__    )[0].split('.'    )[-2]
                            a__						:     Any			=    mapped_key.replace('*'		,				A__    )
                     if "weight_g" in name:
                            a__						:     List[str]			=    'weight_g'
                     elif "weight_v" in name:
                            a__						:     Optional[Any]			=    'weight_v'
                     elif "bias" in name:
                            a__						:     Optional[Any]			=    'bias'
                     elif "weight" in name:
                            # TODO: don't match quantizer.weight_proj
                            a__						:     Any			=    'weight'
                     else:
                            a__						:     List[str]			=    None
                     if hf_dict is not None:
                            rename_dict(A__		,				A__		,				A__		,				A__		,				A__    )
                     else:
                            set_recursively(A__		,				A__		,				A__		,				A__		,				A__    )
                     return is_used
       return is_used



def A_     (		A__		,				A__		,				A__    )  ->		Tuple:
       a__						:     Dict			=    []
       a__						:     Tuple			=    fairseq_model.state_dict()

       a__						:     List[str]			=    hf_model.wavaveca.feature_extractor

       for name, value in fairseq_dict.items():
              a__						:     Tuple			=    False
              if "conv_layers" in name:
                     load_conv_layer(
                         A__		,				A__		,				A__		,				A__		,				hf_model.config.feat_extract_norm == 'group'		,				)
                     a__						:     Optional[Any]			=    True
              else:
                     a__						:     Union[str, Any]			=    load_wavaveca_layer(A__		,				A__		,				A__    )
              if not is_used:
                     unused_weights.append(A__    )

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



def A_     (		A__		,				A__		,				A__		,				A__		,				A__    )  ->		int:
       a__						:     List[Any]			=    full_name.split('conv_layers.'    )[-1]
       a__						:     Optional[Any]			=    name.split('.'    )
       a__						:     Union[str, Any]			=    int(items[0]    )
       a__						:     Union[str, Any]			=    int(items[1]    )

       if type_id == 0:
              if "bias" in name:
                     if value.shape != feature_extractor.conv_layers[layer_id].conv.bias.data.shape:
                            raise ValueError(
                                F'{full_name} has size {value.shape}, but'
                                F' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.'    )
                     a__						:     List[Any]			=    value
                     logger.info(F'Feat extract conv layer {layer_id} was initialized from {full_name}.'    )
              elif "weight" in name:
                     if value.shape != feature_extractor.conv_layers[layer_id].conv.weight.data.shape:
                            raise ValueError(
                                F'{full_name} has size {value.shape}, but'
                                F' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.'    )
                     a__						:     Optional[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:
                     if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape:
                            raise ValueError(
                                F'{full_name} has size {value.shape}, but'
                                F' {feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape} was found.'    )
                     a__						:     int			=    value
                     logger.info(F'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'    )
              elif "weight" in name:
                     if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape:
                            raise ValueError(
                                F'{full_name} has size {value.shape}, but'
                                F' {feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape} was found.'    )
                     a__						:     List[Any]			=    value
                     logger.info(F'Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'    )
       else:
              unused_weights.append(A__    )



@torch.no_grad()
def A_     (		A__		,				A__		,				A__=None		,				A__=None		,				A__=True		,				A__=False    )  ->		Dict:
       if config_path is not None:
              a__						:     Union[str, Any]			=    WavaVecaConfig.from_pretrained(A__    )
       else:
              a__						:     str			=    WavaVecaConfig()

       if is_seq_class:
              a__						:     Optional[int]			=    read_txt_into_dict(A__    )
              a__						:     Dict			=    idalabel
              a__						:     Tuple			=    WavaVecaForSequenceClassification(A__    )
              a__						:     List[str]			=    WavaVecaFeatureExtractor(
                  feature_size=1		,				sampling_rate=1_6000		,				padding_value=0		,				do_normalize=A__		,				return_attention_mask=A__		,				)
              feature_extractor.save_pretrained(A__    )

       elif is_finetuned:
              if dict_path:
                     a__						:     Any			=    Dictionary.load(A__    )

                     # important change bos & pad token id since CTC symbol is <pad> and
                     # not <s> as in fairseq
                     a__						:     Any			=    target_dict.pad_index
                     a__						:     Tuple			=    target_dict.bos_index
                     a__						:     Any			=    target_dict.eos_index
                     a__						:     Any			=    len(target_dict.symbols    )
                     a__						:     Optional[Any]			=    os.path.join(A__		,				'vocab.json'    )
                     if not os.path.isdir(A__    ):
                            logger.error('--pytorch_dump_folder_path ({}) should be a directory'.format(A__    )    )
                            return
                     os.makedirs(A__		,				exist_ok=A__    )
                     a__						:     Union[str, Any]			=    target_dict.indices

                     # fairseq has the <pad> and <s> switched
                     a__						:     str			=    0
                     a__						:     Dict			=    1
                     with open(A__		,				'w'		,				encoding='utf-8'    ) as vocab_handle:
                            json.dump(A__		,				A__    )
                     a__						:     Optional[Any]			=    WavaVecaCTCTokenizer(
                         A__		,				unk_token=target_dict.unk_word		,				pad_token=target_dict.pad_word		,				bos_token=target_dict.bos_word		,				eos_token=target_dict.eos_word		,				word_delimiter_token='|'		,				do_lower_case=A__		,				)
                     a__						:     Union[str, Any]			=    True if config.feat_extract_norm == 'layer' else False
                     a__						:     int			=    WavaVecaFeatureExtractor(
                         feature_size=1		,				sampling_rate=1_6000		,				padding_value=0		,				do_normalize=A__		,				return_attention_mask=A__		,				)
                     a__						:     Union[str, Any]			=    WavaVecaProcessor(feature_extractor=A__		,				tokenizer=A__    )
                     processor.save_pretrained(A__    )

              a__						:     Any			=    WavaVecaForCTC(A__    )
       else:
              a__						:     Union[str, Any]			=    WavaVecaForPreTraining(A__    )

       if is_finetuned or is_seq_class:
              a__			,							a__			,							a__						:     int			=    fairseq.checkpoint_utils.load_model_ensemble_and_task(
                  [checkpoint_path]		,				arg_overrides={'data': '/'.join(dict_path.split('/'    )[:-1]    )}    )
       else:
              a__						:     Optional[int]			=    argparse.Namespace(task='audio_pretraining'    )
              a__						:     Optional[int]			=    fairseq.tasks.setup_task(A__    )

              a__			,							a__			,							a__						:     int			=    fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]		,				task=A__    )

       a__						:     Any			=    model[0].eval()

       recursively_load_weights(A__		,				A__		,				not is_finetuned    )

       hf_wavavec.save_pretrained(A__    )


if __name__ == "__main__":
    lowercase				:	Tuple      						=  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("""--config_path""", default=None, type=str, help="""Path to hf config.json of model to convert""")
    parser.add_argument(
        """--not_finetuned""", action="""store_true""", help="""Whether the model to convert is a fine-tuned model or not"""
    )
    parser.add_argument(
        """--is_seq_class""",
        action="""store_true""",
        help="""Whether the model to convert is a fine-tuned sequence classification model or not""",
    )
    lowercase				:	List[str]      						=  parser.parse_args()

    lowercase				:	Tuple      						=  not args.not_finetuned and not args.is_seq_class
    convert_wavaveca_checkpoint(
        args.checkpoint_path,
        args.pytorch_dump_folder_path,
        args.config_path,
        args.dict_path,
        is_finetuned,
        args.is_seq_class,
    )









def A_     (		A__		,				A__		,				A__    )  ->		float:
       if principal <= 0:
              raise Exception('Principal borrowed must be > 0'    )
       if rate_per_annum < 0:
              raise Exception('Rate of interest must be >= 0'    )
       if years_to_repay <= 0 or not isinstance(A__		,				A__    ):
              raise Exception('Years to repay must be an integer > 0'    )

       # Yearly rate is divided by 12 to get monthly rate
       a__						:     str			=    rate_per_annum / 12

       # Years to repay is multiplied by 12 to get number of payments as payment is monthly
       a__						:     List[Any]			=    years_to_repay * 12

       return (
           principal
           * rate_per_month
           * (1 + rate_per_month) ** number_of_payments
           / ((1 + rate_per_month) ** number_of_payments - 1)
       )


if __name__ == "__main__":
    import doctest

    doctest.testmod()


