






"""simple docstring"""




from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


UpperCamelCase							       =    {
    '''configuration_bigbird_pegasus''': [
        '''BIGBIRD_PEGASUS_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''BigBirdPegasusConfig''',
        '''BigBirdPegasusOnnxConfig''',
    ],
}

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       UpperCamelCase							       =    [
           '''BIGBIRD_PEGASUS_PRETRAINED_MODEL_ARCHIVE_LIST''',
           '''BigBirdPegasusForCausalLM''',
           '''BigBirdPegasusForConditionalGeneration''',
           '''BigBirdPegasusForQuestionAnswering''',
           '''BigBirdPegasusForSequenceClassification''',
           '''BigBirdPegasusModel''',
           '''BigBirdPegasusPreTrainedModel''',
       ]


if TYPE_CHECKING:
       from .configuration_bigbird_pegasus import (
           BIGBIRD_PEGASUS_PRETRAINED_CONFIG_ARCHIVE_MAP,
           BigBirdPegasusConfig,
           BigBirdPegasusOnnxConfig,
       )

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_bigbird_pegasus import (
                  BIGBIRD_PEGASUS_PRETRAINED_MODEL_ARCHIVE_LIST,
                  BigBirdPegasusForCausalLM,
                  BigBirdPegasusForConditionalGeneration,
                  BigBirdPegasusForQuestionAnswering,
                  BigBirdPegasusForSequenceClassification,
                  BigBirdPegasusModel,
                  BigBirdPegasusPreTrainedModel,
              )


else:
       import sys

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










"""simple docstring"""




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        lowerCAmelCase (      UpperCamelCase_:							List[str]					)       ->							List[Any]:







  '''simple docstring'''







  _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        lowerCAmelCase (      UpperCamelCase_:							List[str]     , UpperCamelCase_:							Dict     , UpperCamelCase_:							List[str]     , UpperCamelCase_:							Union[str, Any]     , UpperCamelCase_:							Union[str, Any]					)       ->							Union[str, Any]:







  '''simple docstring'''







  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        lowerCAmelCase (      UpperCamelCase_:							Union[str, Any]     , UpperCamelCase_:							Dict     , UpperCamelCase_:							Tuple     , UpperCamelCase_:							Any     , UpperCamelCase_:							Union[str, Any]					)       ->							str:







  '''simple docstring'''







  _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        lowerCAmelCase (      UpperCamelCase_:							str     , UpperCamelCase_:							Union[str, Any]     , UpperCamelCase_:							int=None     , UpperCamelCase_:							Optional[int]=None					)       ->							Tuple:







  '''simple docstring'''







  _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        lowerCAmelCase (      UpperCamelCase_:							Dict     , UpperCamelCase_:							Dict     , UpperCamelCase_:							Dict					)       ->							Dict:







  '''simple docstring'''







  _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        lowerCAmelCase (      UpperCamelCase_:							Union[str, Any]     , UpperCamelCase_:							List[Any]     , UpperCamelCase_:							int     , UpperCamelCase_:							Optional[int]     , UpperCamelCase_:							Optional[int]					)       ->							Dict:







  '''simple docstring'''







  _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        lowerCAmelCase (      UpperCamelCase_:							Tuple     , UpperCamelCase_:							List[str]     , UpperCamelCase_:							int=None     , UpperCamelCase_:							int=None     , UpperCamelCase_:							Optional[Any]=True     , UpperCamelCase_:							Any=False					)       ->							Optional[Any]:







  '''simple docstring'''







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


