



'''simple docstring'''





import argparse
import os
from functools import reduce

import fairseq
import torch
from datasets import load_dataset

from transformers import WavaVecaProcessor, logging
from transformers.models.dataavec.configuration_dataavec_audio import DataaVecAudioConfig

# Copied from https://github.com/pytorch/fairseq/blob/main/examples/data2vec/models/data2vec_audio.py
from transformers.models.dataavec.dataavec_audio import DataaVecAudioModel as Dummy  # noqa: F401
from transformers.models.dataavec.modeling_dataavec_audio import DataaVecAudioForCTC, DataaVecAudioModel


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

A			:   Dict			=      {
    """post_extract_proj""": """feature_projection.projection""",
    """models.0.layer_norm""": """feature_projection.layer_norm""",
    """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""",
    """w2v_encoder.proj""": """lm_head""",
    """mask_emb""": """masked_spec_embed""",
}
A			:   Dict			=      [
    """lm_head""",
]
def 		_a   (  lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_							):
   for attribute in key.split('''.'''							):
      snake_case   :							Optional[int]						  =getattr(lowerCamelCase_					,						lowerCamelCase_							)

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

   logger.info(F'''{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}.'''							)
def 		_a   (  lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_							):
   snake_case   :							Optional[Any]						  =[]
   snake_case   :							Any						  =fairseq_model.state_dict()

   if not is_headless:
      snake_case   :							List[Any]						  =hf_model.dataavec_audio.feature_extractor
      snake_case   :							str						  =hf_model.dataavec_audio.encoder.pos_conv_embed

   else:
      snake_case   :							int						  =hf_model.feature_extractor
      snake_case   :							Tuple						  =hf_model.encoder.pos_conv_embed

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

   logger.warning(F'''Unused weights: {unused_weights}'''							)
def 		_a   (  lowerCamelCase_					,						lowerCamelCase_							):
   snake_case   :							Union[str, Any]						  =path.split('''.'''							)
   return reduce(lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_							)
def 		_a   (  lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_							):
   snake_case   :							Dict						  =access_by_string(lowerCamelCase_					,						lowerCamelCase_							)
   snake_case   :							Tuple						  =hf_weight.data

   if fsq_value.shape != hf_value.shape:
      raise ValueError(F'''{full_name} has size {fsq_value.shape}, but {hf_value.shape} was found.'''							)
   snake_case   :							int						  =fsq_value
   logger.info(F'''{full_name} was correctly initialized from {hf_weight_path}.'''							)
def 		_a   (  lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_							):
   snake_case   :							List[str]						  =full_name.split('''conv_layers.'''							)[-1]
   snake_case   :							int						  =name.split('''.'''							)
   snake_case   :							Optional[int]						  =int(items[0]							)
   snake_case   :							Optional[Any]						  =int(items[1]							)

   snake_case   :							List[str]						  =name.split('''.'''							)[-1]
   if type_id == 0:
      snake_case   :							str						  ='''conv'''
   elif type_id == 2:
      snake_case   :							str						  ='''layer_norm'''
   else:
      unused_weights.append(lowerCamelCase_							)
      return

   set_weights(lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_					,						F'''conv_layers.{layer_id}.{layer_type}.{weight_type}'''							)
def 		_a   (  lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_							):
   snake_case   :							str						  =full_name.split('''pos_conv.'''							)[-1]
   snake_case   :							int						  =name.split('''.'''							)
   snake_case   :							List[str]						  =int(items[0]							)
   snake_case   :							Tuple						  =int(items[1]							)

   snake_case   :							List[str]						  =name.split('''.'''							)[-1]
   if type_id != 0:
      unused_weights.append(lowerCamelCase_							)
      return
   else:
      snake_case   :							List[str]						  ='''conv'''

   set_weights(lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_					,						F'''layers.{layer_id}.{layer_type}.{weight_type}'''							)
@torch.no_grad()
def 		_a   (  lowerCamelCase_					,						lowerCamelCase_					,						lowerCamelCase_=None					,						lowerCamelCase_=None					,						lowerCamelCase_=True							):
   if config_path is not None:
      snake_case   :							str						  =DataaVecAudioConfig.from_pretrained(lowerCamelCase_							)
   else:
      snake_case   :							Union[str, Any]						  =DataaVecAudioConfig()

   if not is_finetuned:
      # Modify final_proj layer name
      snake_case   :							Union[str, Any]						  =DataaVecAudioModel(lowerCamelCase_							)
      snake_case   :							Optional[int]						  =os.path.dirname(lowerCamelCase_							)

      snake_case   :							str						  =torch.load(lowerCamelCase_							)
      snake_case   :							str						  =state_dict['''model'''].pop('''final_proj.0.weight'''							)
      snake_case   :							Union[str, Any]						  =state_dict['''model'''].pop('''final_proj.0.bias'''							)
      snake_case   :							Optional[int]						  =os.path.join(lowerCamelCase_					,						'''converted.pt'''							)
      torch.save(lowerCamelCase_					,						lowerCamelCase_							)
   else:
      snake_case   :							Union[str, Any]						  =DataaVecAudioForCTC(lowerCamelCase_							)
      snake_case   :							Optional[int]						  =checkpoint_path

   def load_dataavec(lowerCamelCase_							):
      snake_case  ,    snake_case  ,    snake_case   :							List[Any]						  =fairseq.checkpoint_utils.load_model_ensemble_and_task([path]							)
      return model[0].eval()

   snake_case   :							Tuple						  =load_dataavec(lowerCamelCase_							)

   recursively_load_weights(lowerCamelCase_					,						lowerCamelCase_					,						not is_finetuned							)

   snake_case   :							List[str]						  =WavaVecaProcessor.from_pretrained('''facebook/wav2vec2-large-lv60'''							)

   snake_case   :							Tuple						  =load_dataset('''patrickvonplaten/librispeech_asr_dummy'''					,						'''clean'''					,						split='''validation'''							)
   snake_case   :							Dict						  =[x['''array'''] for x in ds[:4]['''audio''']]

   snake_case   :							Tuple						  =processor(lowerCamelCase_					,						return_tensors='''pt'''					,						padding=lowerCamelCase_							)

   snake_case   :							Optional[int]						  =inputs.input_values
   snake_case   :							List[str]						  =inputs.attention_mask
   #    input_values = inputs.input_values[:, :-1]
   #    attention_mask = inputs.attention_mask[:, :-1]

   hf_wavavec.eval()
   model.eval()
   if is_finetuned:
      snake_case   :							Dict						  =model(source=lowerCamelCase_					,						padding_mask=(1 - attention_mask)					,						mask=lowerCamelCase_					,						features_only=lowerCamelCase_							)[
          '''encoder_out'''
      ].transpose(0					,						1							)
      snake_case   :							Dict						  =hf_wavavec(lowerCamelCase_					,						attention_mask=lowerCamelCase_							)['''logits''']

      snake_case   :							List[str]						  =torch.argmax(lowerCamelCase_					,						dim=-1							)
      snake_case   :							Tuple						  =processor.batch_decode(lowerCamelCase_							)

      print(F'''Expected Output: {ds[:4]["text"]}, Pred: {output_string}'''							)
   else:
      snake_case   :							List[Any]						  =model(source=lowerCamelCase_					,						padding_mask=(1 - attention_mask)					,						mask=lowerCamelCase_					,						features_only=lowerCamelCase_							)[
          '''layer_results'''
      ][-1][0].transpose(0					,						1							)
      snake_case   :							Optional[int]						  =hf_wavavec(lowerCamelCase_					,						attention_mask=lowerCamelCase_							)['''last_hidden_state''']

   print(our_output.shape					,						their_output.shape							)
   snake_case   :							Optional[Any]						  =torch.max(torch.abs(our_output - their_output							)							).item()
   print(F'''max_absolute_diff = {max_absolute_diff}'''							)  # ~ 1e-7
   snake_case   :							Dict						  =torch.allclose(lowerCamelCase_					,						lowerCamelCase_					,						atol=1e-3							)
   print('''Do both models output the same tensors?'''					,						'''🔥''' if success else '''💩'''							)
   if not success:
      raise Exception('''Something went wRoNg'''							)

   hf_wavavec.save_pretrained(lowerCamelCase_							)

   if is_finetuned:
      processor.save_pretrained(lowerCamelCase_							)
   else:
      processor.feature_extractor.save_pretrained(lowerCamelCase_							)


if __name__ == "__main__":
      A			:   int			=      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"""
      )
      A			:   Optional[int]			=      parser.parse_args()
      convert_wavaveca_checkpoint(
          args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
      )










'''simple docstring'''





import os
import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow, torch_device


if is_torch_available():
      import torch

      from transformers import GPTaLMHeadModel





@require_torch
@require_sentencepiece
@require_tokenizers
class 							lowerCAmelCase_		(	unittest.TestCase						):





    @slow
    @unittest.skip('''Model is not available.'''     )
    def 					__snake_case       (    self    :    List[Any]     ):



       '''simple docstring'''







       snake_case   :							Any						  ='''nvidia/megatron-gpt2-345m/'''
       if "MYDIR" in os.environ:
          snake_case   :							Any						  =os.path.join(os.environ['''MYDIR'''],    _snake_case     )
       snake_case   :							Optional[Any]						  =GPTaLMHeadModel.from_pretrained(_snake_case     )
       model.to(_snake_case     )
       model.half()

       snake_case   :							int						  =torch.tensor(
           [[101, 7_110, 1_005, 1_056, 2_023, 11_333, 17_413, 1_029, 102]],    device=_snake_case,    dtype=torch.long,    )

       with torch.no_grad():
          snake_case   :							Dict						  =model(_snake_case     ).logits

       snake_case   :							Dict						  =torch.Size((1, 9, 50_257)     )
       self.assertEqual(output.shape,    _snake_case     )

       snake_case   :							str						  =torch.tensor(
           [
               4.9414,
               -0.2920,
               -1.2148,
               -4.0273,
               -0.5161,
               -5.2109,
               -1.2412,
               -1.8301,
               -1.7734,
               -4.7148,
               -0.2317,
               -1.0811,
               -2.1777,
               0.4141,
               -3.7969,
               -4.0586,
               -2.5332,
               -3.3809,
               4.3867,
           ],    device=_snake_case,    dtype=torch.half,    )

       for i in range(19     ):
          snake_case  ,    snake_case   :							Tuple						  =8 * i // 17, 2_792 * i  # along the diagonal
          snake_case  ,    snake_case   :							List[str]						  =output[0, r, c], expected_diag[i]
          snake_case   :							Optional[int]						  =f'''row={r} col={c} computed={computed} expected={expected}'''
          self.assertAlmostEqual(_snake_case,    _snake_case,    delta=1E-4,    msg=_snake_case     )





