





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

lowerCAmelCase_						   =    {
    "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",
}
lowerCAmelCase_						   =    [
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
    "projector",
    "classifier",
]


def     lowerCamelCase_	(      lowerCAmelCase:	List[str]				)->		List[str]:
  _snake_case   :				List[str]														=	{}
  with open(_A    ,							'r'				) as file:
    for line_number, line in enumerate(_A				):
      _snake_case   :				int														=	line.strip()
      if line:
        _snake_case   :				Any														=	line.split()
        _snake_case   :				Any														=	line_number
        _snake_case   :				List[Any]														=	words[0]
        _snake_case   :				int														=	value
  return result


def     lowerCamelCase_	(      lowerCAmelCase:	Tuple    ,							lowerCAmelCase:	Optional[Any]    ,							lowerCAmelCase:	Dict    ,							lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	Tuple				)->		int:




  for attribute in key.split('.'				):
    _snake_case   :				Any														=	getattr(_A    ,							_A				)

  _snake_case   :				str														=	None
  for param_key in PARAM_MAPPING.keys():
    if full_name.endswith(_A				):
      _snake_case   :				int														=	PARAM_MAPPING[full_name.split('.'				)[-1]]
      _snake_case   :				Tuple														=	"param"

  if weight_type is not None and weight_type != "param":
    _snake_case   :				Union[str, Any]														=	getattr(_A    ,							_A				).shape
  elif weight_type is not None and weight_type == "param":
    _snake_case   :				Tuple														=	hf_pointer
    for attribute in hf_param_name.split('.'				):
      _snake_case   :				Tuple														=	getattr(_A    ,							_A				)
    _snake_case   :				List[str]														=	shape_pointer.shape

    # let's reduce dimension
    _snake_case   :				List[str]														=	value[0]
  else:
    _snake_case   :				int														=	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":
    _snake_case   :				Dict														=	value
  elif weight_type == "weight_g":
    _snake_case   :				Union[str, Any]														=	value
  elif weight_type == "weight_v":
    _snake_case   :				Union[str, Any]														=	value
  elif weight_type == "bias":
    _snake_case   :				int														=	value
  elif weight_type == "param":
    for attribute in hf_param_name.split('.'				):
      _snake_case   :				Any														=	getattr(_A    ,							_A				)
    _snake_case   :				int														=	value
  else:
    _snake_case   :				Dict														=	value

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


def     lowerCamelCase_	(      lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	Any    ,							lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	Tuple				)->		str:
  _snake_case   :				Union[str, Any]														=	None
  for param_key in PARAM_MAPPING.keys():
    if full_name.endswith(_A				):
      _snake_case   :				List[Any]														=	PARAM_MAPPING[full_name.split('.'				)[-1]]
      _snake_case   :				int														=	"param"

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

  _snake_case   :				List[str]														=	value if "lm_head" in full_key else value[0]


lowerCAmelCase_						   =    {
    "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_	(      lowerCAmelCase:	Optional[int]    ,							lowerCAmelCase:	Any    ,							lowerCAmelCase:	Optional[Any]=None    ,							lowerCAmelCase:	Tuple=None				)->		List[Any]:
  _snake_case   :				int														=	False
  for key, mapped_key in MAPPING.items():
    _snake_case   :				Dict														=	"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]:
      _snake_case   :				Tuple														=	True
      if "*" in mapped_key:
        _snake_case   :				Any														=	name.split(_A				)[0].split('.'				)[-2]
        _snake_case   :				Optional[int]														=	mapped_key.replace('*'    ,							_A				)
      if "weight_g" in name:
        _snake_case   :				Union[str, Any]														=	"weight_g"
      elif "weight_v" in name:
        _snake_case   :				List[Any]														=	"weight_v"
      elif "bias" in name:
        _snake_case   :				List[Any]														=	"bias"
      elif "weight" in name:
        # TODO: don't match quantizer.weight_proj
        _snake_case   :				Optional[Any]														=	"weight"
      else:
        _snake_case   :				Tuple														=	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     lowerCamelCase_	(      lowerCAmelCase:	Dict    ,							lowerCAmelCase:	int    ,							lowerCAmelCase:	int				)->		Dict:
  _snake_case   :				List[Any]														=	[]
  _snake_case   :				Union[str, Any]														=	fairseq_model.state_dict()

  _snake_case   :				Optional[int]														=	hf_model.wavaveca.feature_extractor

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

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


def     lowerCamelCase_	(      lowerCAmelCase:	Union[str, Any]    ,							lowerCAmelCase:	Dict    ,							lowerCAmelCase:	Optional[Any]    ,							lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	Union[str, Any]				)->		Optional[Any]:
  _snake_case   :				int														=	full_name.split('conv_layers.'				)[-1]
  _snake_case   :				Optional[Any]														=	name.split('.'				)
  _snake_case   :				Any														=	int(items[0]				)
  _snake_case   :				int														=	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."""				)
      _snake_case   :				Tuple														=	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."""				)
      _snake_case   :				Tuple														=	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."""				)
      _snake_case   :				Union[str, Any]														=	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."""				)
      _snake_case   :				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     lowerCamelCase_	(      lowerCAmelCase:	int    ,							lowerCAmelCase:	Any    ,							lowerCAmelCase:	Any=None    ,							lowerCAmelCase:	List[Any]=None    ,							lowerCAmelCase:	Tuple=True    ,							lowerCAmelCase:	List[Any]=False				)->		List[str]:




  if config_path is not None:
    _snake_case   :				Optional[Any]														=	WavaVecaConfig.from_pretrained(_A				)
  else:
    _snake_case   :				str														=	WavaVecaConfig()

  if is_seq_class:
    _snake_case   :				str														=	read_txt_into_dict(_A				)
    _snake_case   :				List[Any]														=	idalabel
    _snake_case   :				List[Any]														=	WavaVecaForSequenceClassification(_A				)
    _snake_case   :				Optional[int]														=	WavaVecaFeatureExtractor(
        feature_size=1    ,							sampling_rate=1_60_00    ,							padding_value=0    ,							do_normalize=_A    ,							return_attention_mask=_A    ,							)
    feature_extractor.save_pretrained(_A				)

  elif is_finetuned:
    if dict_path:
      _snake_case   :				Dict														=	Dictionary.load(_A				)

      # important change bos & pad token id since CTC symbol is <pad> and
      # not <s> as in fairseq
      _snake_case   :				List[Any]														=	target_dict.pad_index
      _snake_case   :				str														=	target_dict.bos_index
      _snake_case   :				str														=	target_dict.eos_index
      _snake_case   :				Tuple														=	len(target_dict.symbols				)
      _snake_case   :				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				)
      _snake_case   :				List[str]														=	target_dict.indices

      # fairseq has the <pad> and <s> switched
      _snake_case   :				str														=	0
      _snake_case   :				Dict														=	1
      with open(_A    ,							'w'    ,							encoding='utf-8'				) as vocab_handle:
        json.dump(_A    ,							_A				)
      _snake_case   :				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    ,							)
      _snake_case   :				Dict														=	True if config.feat_extract_norm == "layer" else False
      _snake_case   :				List[Any]														=	WavaVecaFeatureExtractor(
          feature_size=1    ,							sampling_rate=1_60_00    ,							padding_value=0    ,							do_normalize=_A    ,							return_attention_mask=_A    ,							)
      _snake_case   :				Any														=	WavaVecaProcessor(feature_extractor=_A    ,							tokenizer=_A				)
      processor.save_pretrained(_A				)

    _snake_case   :				int														=	WavaVecaForCTC(_A				)
  else:
    _snake_case   :				Optional[Any]														=	WavaVecaForPreTraining(_A				)

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

    _snake_case   :				str														=	fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]    ,							task=_A				)

  _snake_case   :				List[str]														=	model[0].eval()

  recursively_load_weights(_A    ,							_A    ,							not is_finetuned				)

  hf_wavavec.save_pretrained(_A				)


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

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











import argparse
import os

import transformers

from .convert_slow_tokenizer import SLOW_TO_FAST_CONVERTERS
from .utils import logging


logging.set_verbosity_info()

lowerCAmelCase_						   =    logging.get_logger(__name__)


lowerCAmelCase_						   =    {name: getattr(transformers, name + """Fast""") for name in SLOW_TO_FAST_CONVERTERS}


def     lowerCamelCase_	(      lowerCAmelCase:	List[Any]    ,							lowerCAmelCase:	Optional[Any]    ,							lowerCAmelCase:	Dict    ,							lowerCAmelCase:	Union[str, Any]				)->		Optional[int]:
  if tokenizer_name is not None and tokenizer_name not in TOKENIZER_CLASSES:
    raise ValueError(F"""Unrecognized tokenizer name, should be one of {list(TOKENIZER_CLASSES.keys()				)}."""				)

  if tokenizer_name is None:
    _snake_case   :				Tuple														=	TOKENIZER_CLASSES
  else:
    _snake_case   :				Union[str, Any]														=	{tokenizer_name: getattr(lowerCAmelCase    ,							tokenizer_name + 'Fast'				)}

  logger.info(F"""Loading tokenizer classes: {tokenizer_names}"""				)

  for tokenizer_name in tokenizer_names:
    _snake_case   :				Dict														=	TOKENIZER_CLASSES[tokenizer_name]

    _snake_case   :				Optional[Any]														=	True
    if checkpoint_name is None:
      _snake_case   :				Union[str, Any]														=	list(tokenizer_class.max_model_input_sizes.keys()				)
    else:
      _snake_case   :				Optional[int]														=	[checkpoint_name]

    logger.info(F"""For tokenizer {tokenizer_class.__class__.__name__} loading checkpoints: {checkpoint_names}"""				)

    for checkpoint in checkpoint_names:
      logger.info(F"""Loading {tokenizer_class.__class__.__name__} {checkpoint}"""				)

      # Load tokenizer
      _snake_case   :				str														=	tokenizer_class.from_pretrained(lowerCAmelCase    ,							force_download=lowerCAmelCase				)

      # Save fast tokenizer
      logger.info(F"""Save fast tokenizer to {dump_path} with prefix {checkpoint} add_prefix {add_prefix}"""				)

      # For organization names we create sub-directories
      if "/" in checkpoint:
        _snake_case							,					_snake_case   :				Tuple														=	checkpoint.split('/'				)
        _snake_case   :				int														=	os.path.join(lowerCAmelCase    ,							lowerCAmelCase				)
      elif add_prefix:
        _snake_case   :				Dict														=	checkpoint
        _snake_case   :				Optional[Any]														=	dump_path
      else:
        _snake_case   :				str														=	None
        _snake_case   :				Union[str, Any]														=	dump_path

      logger.info(F"""=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}"""				)

      if checkpoint in list(tokenizer.pretrained_vocab_files_map.values()				)[0]:
        _snake_case   :				Optional[Any]														=	list(tokenizer.pretrained_vocab_files_map.values()				)[0][checkpoint]
        _snake_case   :				Optional[int]														=	file_path.split(lowerCAmelCase				)[-1][0]
        if next_char == "/":
          _snake_case   :				Union[str, Any]														=	os.path.join(lowerCAmelCase    ,							lowerCAmelCase				)
          _snake_case   :				str														=	None

        logger.info(F"""=> {dump_path_full} with prefix {checkpoint_prefix_name}, add_prefix {add_prefix}"""				)

      _snake_case   :				Optional[int]														=	tokenizer.save_pretrained(
          lowerCAmelCase    ,							legacy_format=lowerCAmelCase    ,							filename_prefix=lowerCAmelCase				)
      logger.info(F"""=> File names {file_names}"""				)

      for file_name in file_names:
        if not file_name.endswith('tokenizer.json'				):
          os.remove(lowerCAmelCase				)
          logger.info(F"""=> removing {file_name}"""				)


if __name__ == "__main__":
      lowerCAmelCase_						   =    argparse.ArgumentParser()
      # Required parameters
      parser.add_argument(
          """--dump_path""", default=None, type=str, required=True, help="""Path to output generated fast tokenizer files."""
      )
      parser.add_argument(
          """--tokenizer_name""",
          default=None,
          type=str,
          help=(
              F"""Optional tokenizer type selected in the list of {list(TOKENIZER_CLASSES.keys())}. If not given, will """
              """download and convert all the checkpoints from AWS."""
          ),
      )
      parser.add_argument(
          """--checkpoint_name""",
          default=None,
          type=str,
          help="""Optional checkpoint name. If not given, will download and convert the canonical checkpoints from AWS.""",
      )
      parser.add_argument(
          """--force_download""",
          action="""store_true""",
          help="""Re-download checkpoints.""",
      )
      lowerCAmelCase_						   =    parser.parse_args()

      convert_slow_checkpoint_to_fast(args.tokenizer_name, args.checkpoint_name, args.dump_path, args.force_download)




