






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()
UpperCAmelCase__				   =		logging.get_logger(__name__)

UpperCAmelCase__				   =		{
    "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",
}
UpperCAmelCase__				   =		[
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
    "projector",
    "classifier",
]
def 						_a    (      a :List[str] )							->			int:
 a		  =      {}
 with open(UpperCamelCase__	,							'''r''' ) as file:
  for line_number, line in enumerate(UpperCamelCase__ ):
   a		  =      line.strip()
   if line:
    a		  =      line.split()
    a		  =      line_number
    a		  =      words[0]
    a		  =      value
 return result
def 						_a    (      a :Dict	,							a :int	,							a :Tuple	,							a :str	,							a :Union[str, Any] )							->			Tuple:


 for attribute in key.split('''.''' ):
  a		  =      getattr(UpperCamelCase__	,							UpperCamelCase__ )

 a		  =      None
 for param_key in PARAM_MAPPING.keys():
  if full_name.endswith(UpperCamelCase__ ):
   a		  =      PARAM_MAPPING[full_name.split('''.''' )[-1]]
   a		  =      '''param'''

 if weight_type is not None and weight_type != "param":
  a		  =      getattr(UpperCamelCase__	,							UpperCamelCase__ ).shape
 elif weight_type is not None and weight_type == "param":
  a		  =      hf_pointer
  for attribute in hf_param_name.split('''.''' ):
   a		  =      getattr(UpperCamelCase__	,							UpperCamelCase__ )
  a		  =      shape_pointer.shape

  # let's reduce dimension
  a		  =      value[0]
 else:
  a		  =      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		  =      value
 elif weight_type == "weight_g":
  a		  =      value
 elif weight_type == "weight_v":
  a		  =      value
 elif weight_type == "bias":
  a		  =      value
 elif weight_type == "param":
  for attribute in hf_param_name.split('''.''' ):
   a		  =      getattr(UpperCamelCase__	,							UpperCamelCase__ )
  a		  =      value
 else:
  a		  =      value

 logger.info(F"""{key + '.' + weight_type if weight_type is not None else ''} was initialized from {full_name}.""" )
def 						_a    (      a :List[Any]	,							a :List[Any]	,							a :List[str]	,							a :List[Any]	,							a :Any )							->			Union[str, Any]:
 a		  =      None
 for param_key in PARAM_MAPPING.keys():
  if full_name.endswith(UpperCamelCase__ ):
   a		  =      PARAM_MAPPING[full_name.split('''.''' )[-1]]
   a		  =      '''param'''

 if weight_type is not None and weight_type != "param":
  a		  =      '''.'''.join([key, weight_type] )
 elif weight_type is not None and weight_type == "param":
  a		  =      '''.'''.join([key, hf_param_name] )
 else:
  a		  =      key

 a		  =      value if '''lm_head''' in full_key else value[0]


UpperCAmelCase__				   =		{
    "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 :List[Any]	,							a :str	,							a :Union[str, Any]=None	,							a :Optional[int]=None )							->			Union[str, Any]:
 a		  =      False
 for key, mapped_key in MAPPING.items():
  a		  =      '''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		  =      True
   if "*" in mapped_key:
    a		  =      name.split(UpperCamelCase__ )[0].split('''.''' )[-2]
    a		  =      mapped_key.replace('''*'''	,							UpperCamelCase__ )
   if "weight_g" in name:
    a		  =      '''weight_g'''
   elif "weight_v" in name:
    a		  =      '''weight_v'''
   elif "bias" in name:
    a		  =      '''bias'''
   elif "weight" in name:
    # TODO: don't match quantizer.weight_proj
    a		  =      '''weight'''
   else:
    a		  =      None
   if hf_dict is not None:
    rename_dict(UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__ )
   else:
    set_recursively(UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__ )
   return is_used
 return is_used
def 						_a    (      a :List[str]	,							a :List[Any]	,							a :Union[str, Any] )							->			Optional[Any]:
 a		  =      []
 a		  =      fairseq_model.state_dict()

 a		  =      hf_model.wavaveca.feature_extractor

 for name, value in fairseq_dict.items():
  a		  =      False
  if "conv_layers" in name:
   load_conv_layer(
       UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__	,							hf_model.config.feat_extract_norm == '''group'''	,							)
   a		  =      True
  else:
   a		  =      load_wavaveca_layer(UpperCamelCase__	,							UpperCamelCase__	,							UpperCamelCase__ )
  if not is_used:
   unused_weights.append(UpperCamelCase__ )

 logger.warning(F"""Unused weights: {unused_weights}""" )
def 						_a    (      a :Dict	,							a :Optional[Any]	,							a :Dict	,							a :str	,							a :int )							->			Any:
 a		  =      full_name.split('''conv_layers.''' )[-1]
 a		  =      name.split('''.''' )
 a		  =      int(items[0] )
 a		  =      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		  =      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		  =      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		  =      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		  =      value
   logger.info(F"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.""" )
 else:
  unused_weights.append(UpperCamelCase__ )
@torch.no_grad()
def 						_a    (      a :Any	,							a :int	,							a :Any=None	,							a :str=None	,							a :int=True	,							a :str=False )							->			str:


 if config_path is not None:
  a		  =      WavaVecaConfig.from_pretrained(UpperCamelCase__ )
 else:
  a		  =      WavaVecaConfig()

 if is_seq_class:
  a		  =      read_txt_into_dict(UpperCamelCase__ )
  a		  =      idalabel
  a		  =      WavaVecaForSequenceClassification(UpperCamelCase__ )
  a		  =      WavaVecaFeatureExtractor(
      feature_size=1	,							sampling_rate=16_000	,							padding_value=0	,							do_normalize=UpperCamelCase__	,							return_attention_mask=UpperCamelCase__	,							)
  feature_extractor.save_pretrained(UpperCamelCase__ )

 elif is_finetuned:
  if dict_path:
   a		  =      Dictionary.load(UpperCamelCase__ )

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

   # fairseq has the <pad> and <s> switched
   a		  =      0
   a		  =      1
   with open(UpperCamelCase__	,							'''w'''	,							encoding='''utf-8''' ) as vocab_handle:
    json.dump(UpperCamelCase__	,							UpperCamelCase__ )
   a		  =      WavaVecaCTCTokenizer(
       UpperCamelCase__	,							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=UpperCamelCase__	,							)
   a		  =      True if config.feat_extract_norm == '''layer''' else False
   a		  =      WavaVecaFeatureExtractor(
       feature_size=1	,							sampling_rate=16_000	,							padding_value=0	,							do_normalize=UpperCamelCase__	,							return_attention_mask=UpperCamelCase__	,							)
   a		  =      WavaVecaProcessor(feature_extractor=UpperCamelCase__	,							tokenizer=UpperCamelCase__ )
   processor.save_pretrained(UpperCamelCase__ )

  a		  =      WavaVecaForCTC(UpperCamelCase__ )
 else:
  a		  =      WavaVecaForPreTraining(UpperCamelCase__ )

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

  a  ,     a  ,     a		  =      fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]	,							task=UpperCamelCase__ )

 a		  =      model[0].eval()

 recursively_load_weights(UpperCamelCase__	,							UpperCamelCase__	,							not is_finetuned )

 hf_wavavec.save_pretrained(UpperCamelCase__ )


if __name__ == "__main__":
     UpperCAmelCase__				   =		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",
     )
     UpperCAmelCase__				   =		parser.parse_args()

     UpperCAmelCase__				   =		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,
     )








from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_speech_available, is_torch_available


__A        =    {
    "configuration_audio_spectrogram_transformer": [
        "AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "ASTConfig",
    ]
}

try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					__A        =    [
					    "AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST",
					    "ASTForAudioClassification",
					    "ASTModel",
					    "ASTPreTrainedModel",
					]

try:
					if not is_speech_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					__A        =    ["ASTFeatureExtractor"]

if TYPE_CHECKING:
					from .configuration_audio_spectrogram_transformer import (
					    AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP,
					    ASTConfig,
					)

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_audio_spectrogram_transformer import (
										    AUDIO_SPECTROGRAM_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
										    ASTForAudioClassification,
										    ASTModel,
										    ASTPreTrainedModel,
										)

					try:
										if not is_speech_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .feature_extraction_audio_spectrogram_transformer import ASTFeatureExtractor


else:
					import sys

					__A        =    _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)


