


import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    UniSpeechConfig,
    UniSpeechForCTC,
    UniSpeechForPreTraining,
    WavaVecaFeatureExtractor,
    WavaVecaPhonemeCTCTokenizer,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
snake_case							:       Optional[int]										=	logging.get_logger(__name__)

snake_case							:       Optional[Any]										=	{
    '''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''': '''ctc_proj''',
    '''mask_emb''': '''masked_spec_embed''',
}
snake_case							:       Dict										=	[
    '''ctc_proj''',
    '''quantizer.weight_proj''',
    '''quantizer.codevectors''',
    '''project_q''',
    '''project_hid''',
]
def 							__lowerCamelCase		(					UpperCAmelCase_		:						List[str]					,    UpperCAmelCase_		:						Tuple					,    UpperCAmelCase_		:						List[str]					,    UpperCAmelCase_		:						Union[str, Any]					,    UpperCAmelCase_		:						Optional[Any]					,    UpperCAmelCase_		:						int		):




      """simple docstring"""





      for attribute in key.split('''.'''		):
            if is_finetuned:
                  if attribute in ["quantizer", "project_q", "project_hid"]:
                        # those layers are only relevant for pretraining and should be dropped
                        return

                  if attribute == "ctc_proj":
                        # we should rename `ctc_proj` to `lm_head` for fine-tuned phoneme models
                        a		:Union[str, Any]									=			'''lm_head'''

            a		:Union[str, Any]									=			getattr(UpperCAmelCase_					,    UpperCAmelCase_		)

      if weight_type is not None:
            a		:Optional[int]									=			getattr(UpperCAmelCase_					,    UpperCAmelCase_		).shape
      else:
            a		:Tuple									=			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":
            a		:List[str]									=			value
      elif weight_type == "weight_g":
            a		:Tuple									=			value
      elif weight_type == "weight_v":
            a		:Tuple									=			value
      elif weight_type == "bias":
            a		:Dict									=			value
      else:
            a		:str									=			value

      logger.info(F'''{key + '.' + weight_type if weight_type is not None else ''} was initialized from {full_name}.'''		)
def 							__lowerCamelCase		(					UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						Any		):




      """simple docstring"""





      a		:Dict									=			[]
      a		:int									=			fairseq_model.state_dict()

      a		:Tuple									=			hf_model.unispeech.feature_extractor

      for name, value in fairseq_dict.items():
            a		:Dict									=			False
            if "conv_layers" in name:
                  load_conv_layer(
                      UpperCAmelCase_					,    UpperCAmelCase_					,    UpperCAmelCase_					,    UpperCAmelCase_					,    hf_model.config.feat_extract_norm == '''group'''					,    )
                  a		:List[str]									=			True
            else:
                  for key, mapped_key in MAPPING.items():
                        a		:List[str]									=			'''unispeech.''' + 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		:List[Any]									=			True
                              if "*" in mapped_key:
                                    a		:int									=			name.split(UpperCAmelCase_		)[0].split('''.'''		)[-2]
                                    a		:List[str]									=			mapped_key.replace('''*'''					,    UpperCAmelCase_		)
                              if "weight_g" in name:
                                    a		:List[str]									=			'''weight_g'''
                              elif "weight_v" in name:
                                    a		:Optional[int]									=			'''weight_v'''
                              elif "bias" in name:
                                    a		:Union[str, Any]									=			'''bias'''
                              elif "weight" in name:
                                    # TODO: don't match quantizer.weight_proj
                                    a		:int									=			'''weight'''
                              else:
                                    a		:List[Any]									=			None
                              set_recursively(UpperCAmelCase_					,    UpperCAmelCase_					,    UpperCAmelCase_					,    UpperCAmelCase_					,    UpperCAmelCase_					,    UpperCAmelCase_		)
                        continue
            if not is_used:
                  unused_weights.append(UpperCAmelCase_		)

      logger.warning(F'''Unused weights: {unused_weights}'''		)
def 							__lowerCamelCase		(					UpperCAmelCase_		:						str					,    UpperCAmelCase_		:						Dict					,    UpperCAmelCase_		:						Union[str, Any]					,    UpperCAmelCase_		:						List[str]					,    UpperCAmelCase_		:						str		):




      """simple docstring"""





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




      """simple docstring"""





      if config_path is not None:
            a		:Union[str, Any]									=			UniSpeechConfig.from_pretrained(UpperCAmelCase_		)
      else:
            a		:List[Any]									=			UniSpeechConfig()

      if is_finetuned:
            if dict_path:
                  a		:Union[str, Any]									=			Dictionary.load_from_json(UpperCAmelCase_		)

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

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

            a		:Dict									=			UniSpeechForCTC(UpperCAmelCase_		)
      else:
            a		:Union[str, Any]									=			UniSpeechForPreTraining(UpperCAmelCase_		)

      if is_finetuned:
            a				,							a				,							a		:Any									=			fairseq.checkpoint_utils.load_model_ensemble_and_task(
                [checkpoint_path]					,    arg_overrides={'''data''': '''/'''.join(dict_path.split('''/'''		)[:-1]		), '''w2v_path''': checkpoint_path}		)
      else:
            a				,							a				,							a		:Union[str, Any]									=			fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]		)

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

      recursively_load_weights(UpperCAmelCase_					,    UpperCAmelCase_					,    UpperCAmelCase_		)

      hf_unispeech.save_pretrained(UpperCAmelCase_		)


if __name__ == "__main__":
     snake_case							:       str										=	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'''
     )
     snake_case							:       Dict										=	parser.parse_args()
     convert_unispeech_checkpoint(
         args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
     )









import warnings

from ...utils import is_sklearn_available, requires_backends


if is_sklearn_available():
     from scipy.stats import pearsonr, spearmanr
     from sklearn.metrics import fa_score, matthews_corrcoef


snake_case							:       List[str]										=	(
    '''This metric will be removed from the library soon, metrics should be handled with the 🤗 Evaluate '''
    '''library. You can have a look at this example script for pointers: '''
    '''https://github.com/huggingface/transformers/blob/main/examples/pytorch/text-classification/run_glue.py'''
)
def 							__lowerCamelCase		(					UpperCAmelCase_		:						List[str]					,    UpperCAmelCase_		:						Optional[Any]		):




      """simple docstring"""





      warnings.warn(UpperCAmelCase_					,    UpperCAmelCase_		)
      requires_backends(UpperCAmelCase_					,    '''sklearn'''		)
      return (preds == labels).mean()
def 							__lowerCamelCase		(					UpperCAmelCase_		:						str					,    UpperCAmelCase_		:						int		):




      """simple docstring"""





      warnings.warn(UpperCAmelCase_					,    UpperCAmelCase_		)
      requires_backends(UpperCAmelCase_					,    '''sklearn'''		)
      a		:Dict									=			simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)
      a		:Any									=			fa_score(y_true=UpperCAmelCase_					,    y_pred=UpperCAmelCase_		)
      return {
          "acc": acc,
          "f1": fa,
          "acc_and_f1": (acc + fa) / 2,
      }
def 							__lowerCamelCase		(					UpperCAmelCase_		:						Dict					,    UpperCAmelCase_		:						List[Any]		):




      """simple docstring"""





      warnings.warn(UpperCAmelCase_					,    UpperCAmelCase_		)
      requires_backends(UpperCAmelCase_					,    '''sklearn'''		)
      a		:str									=			pearsonr(UpperCAmelCase_					,    UpperCAmelCase_		)[0]
      a		:Union[str, Any]									=			spearmanr(UpperCAmelCase_					,    UpperCAmelCase_		)[0]
      return {
          "pearson": pearson_corr,
          "spearmanr": spearman_corr,
          "corr": (pearson_corr + spearman_corr) / 2,
      }
def 							__lowerCamelCase		(					UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						Dict		):




      """simple docstring"""





      warnings.warn(UpperCAmelCase_					,    UpperCAmelCase_		)
      requires_backends(UpperCAmelCase_					,    '''sklearn'''		)
      assert len(UpperCAmelCase_		) == len(UpperCAmelCase_		), F'''Predictions and labels have mismatched lengths {len(UpperCAmelCase_		)} and {len(UpperCAmelCase_		)}'''
      if task_name == "cola":
            return {"mcc": matthews_corrcoef(UpperCAmelCase_					,    UpperCAmelCase_		)}
      elif task_name == "sst-2":
            return {"acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      elif task_name == "mrpc":
            return acc_and_fa(UpperCAmelCase_					,    UpperCAmelCase_		)
      elif task_name == "sts-b":
            return pearson_and_spearman(UpperCAmelCase_					,    UpperCAmelCase_		)
      elif task_name == "qqp":
            return acc_and_fa(UpperCAmelCase_					,    UpperCAmelCase_		)
      elif task_name == "mnli":
            return {"mnli/acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      elif task_name == "mnli-mm":
            return {"mnli-mm/acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      elif task_name == "qnli":
            return {"acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      elif task_name == "rte":
            return {"acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      elif task_name == "wnli":
            return {"acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      elif task_name == "hans":
            return {"acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      else:
            raise KeyError(UpperCAmelCase_		)






def 							__lowerCamelCase		(					UpperCAmelCase_		:						List[str]					,    UpperCAmelCase_		:						int					,    UpperCAmelCase_		:						List[Any]		):




      """simple docstring"""





      warnings.warn(UpperCAmelCase_					,    UpperCAmelCase_		)
      requires_backends(UpperCAmelCase_					,    '''sklearn'''		)
      if len(UpperCAmelCase_		) != len(UpperCAmelCase_		):
            raise ValueError(F'''Predictions and labels have mismatched lengths {len(UpperCAmelCase_		)} and {len(UpperCAmelCase_		)}'''		)
      if task_name == "xnli":
            return {"acc": simple_accuracy(UpperCAmelCase_					,    UpperCAmelCase_		)}
      else:
            raise KeyError(UpperCAmelCase_		)





