





'''simple docstring'''



snake_case_   :  List[str]								=      [0.4_85, 0.4_56, 0.4_06]
snake_case_   :  Dict								=      [0.2_29, 0.2_24, 0.2_25]
snake_case_   :  List[Any]								=      [0.5, 0.5, 0.5]
snake_case_   :  Tuple								=      [0.5, 0.5, 0.5]
snake_case_   :  List[str]								=      [0.48_14_54_66, 0.4_57_82_75, 0.40_82_10_73]
snake_case_   :  List[str]								=      [0.26_86_29_54, 0.26_13_02_58, 0.27_57_77_11]










'''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()
snake_case_   :  Optional[Any]								=      logging.get_logger(__name__)

snake_case_   :  Optional[Any]								=      {
    '''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''',
}
snake_case_   :  Optional[Any]								=      [
    '''lm_head''',
]



def  lowercase__(		_UpperCamelCase  : List[Any] ,   _UpperCamelCase  : Optional[Any] ,   _UpperCamelCase  : Tuple ,   _UpperCamelCase  : Tuple ,   _UpperCamelCase  : Optional[Any]				)->   Tuple:






					"""simple docstring"""

					for attribute in key.split("."				):
										_UpperCamelCase											=   getattr(_UpperCamelCase ,   _UpperCamelCase				)

					if weight_type is not None:
										_UpperCamelCase											=   getattr(_UpperCamelCase ,   _UpperCamelCase				).shape
					else:
										_UpperCamelCase											=   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":
										_UpperCamelCase											=   value
					elif weight_type == "weight_g":
										_UpperCamelCase											=   value
					elif weight_type == "weight_v":
										_UpperCamelCase											=   value
					elif weight_type == "bias":
										_UpperCamelCase											=   value
					else:
										_UpperCamelCase											=   value

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



def  lowercase__(		_UpperCamelCase  : Tuple ,   _UpperCamelCase  : Optional[Any] ,   _UpperCamelCase  : List[str]				)->   List[str]:






					"""simple docstring"""

					_UpperCamelCase											=   []
					_UpperCamelCase											=   fairseq_model.state_dict()

					if not is_headless:
										_UpperCamelCase											=   hf_model.dataavec_audio.feature_extractor
										_UpperCamelCase											=   hf_model.dataavec_audio.encoder.pos_conv_embed

					else:
										_UpperCamelCase											=   hf_model.feature_extractor
										_UpperCamelCase											=   hf_model.encoder.pos_conv_embed

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

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



def  lowercase__(		_UpperCamelCase  : List[Any] ,   _UpperCamelCase  : Tuple				)->   int:






					"""simple docstring"""

					_UpperCamelCase											=   path.split("."				)
					return reduce(_UpperCamelCase ,   _UpperCamelCase ,   _UpperCamelCase				)



def  lowercase__(		_UpperCamelCase  : str ,   _UpperCamelCase  : List[Any] ,   _UpperCamelCase  : Dict ,   _UpperCamelCase  : List[str]				)->   List[str]:






					"""simple docstring"""

					_UpperCamelCase											=   access_by_string(_UpperCamelCase ,   _UpperCamelCase				)
					_UpperCamelCase											=   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."				)
					_UpperCamelCase											=   fsq_value
					logger.info(f"{full_name} was correctly initialized from {hf_weight_path}."				)



def  lowercase__(		_UpperCamelCase  : Tuple ,   _UpperCamelCase  : Union[str, Any] ,   _UpperCamelCase  : int ,   _UpperCamelCase  : str				)->   Dict:






					"""simple docstring"""

					_UpperCamelCase											=   full_name.split("conv_layers."				)[-1]
					_UpperCamelCase											=   name.split("."				)
					_UpperCamelCase											=   int(items[0]				)
					_UpperCamelCase											=   int(items[1]				)

					_UpperCamelCase											=   name.split("."				)[-1]
					if type_id == 0:
										_UpperCamelCase											=   "conv"
					elif type_id == 2:
										_UpperCamelCase											=   "layer_norm"
					else:
										unused_weights.append(_UpperCamelCase				)
										return

					set_weights(_UpperCamelCase ,   _UpperCamelCase ,   _UpperCamelCase ,   f"conv_layers.{layer_id}.{layer_type}.{weight_type}"				)



def  lowercase__(		_UpperCamelCase  : Union[str, Any] ,   _UpperCamelCase  : List[Any] ,   _UpperCamelCase  : Tuple ,   _UpperCamelCase  : int				)->   Dict:






					"""simple docstring"""

					_UpperCamelCase											=   full_name.split("pos_conv."				)[-1]
					_UpperCamelCase											=   name.split("."				)
					_UpperCamelCase											=   int(items[0]				)
					_UpperCamelCase											=   int(items[1]				)

					_UpperCamelCase											=   name.split("."				)[-1]
					if type_id != 0:
										unused_weights.append(_UpperCamelCase				)
										return
					else:
										_UpperCamelCase											=   "conv"

					set_weights(_UpperCamelCase ,   _UpperCamelCase ,   _UpperCamelCase ,   f"layers.{layer_id}.{layer_type}.{weight_type}"				)



@torch.no_grad()
def  lowercase__(		_UpperCamelCase  : List[Any] ,   _UpperCamelCase  : Tuple ,   _UpperCamelCase  : Optional[Any]=None ,   _UpperCamelCase  : List[Any]=None ,   _UpperCamelCase  : Optional[Any]=True				)->   Optional[int]:






					"""simple docstring"""

					if config_path is not None:
										_UpperCamelCase											=   DataaVecAudioConfig.from_pretrained(_UpperCamelCase				)
					else:
										_UpperCamelCase											=   DataaVecAudioConfig()

					if not is_finetuned:
										# Modify final_proj layer name
										_UpperCamelCase											=   DataaVecAudioModel(_UpperCamelCase				)
										_UpperCamelCase											=   os.path.dirname(_UpperCamelCase				)

										_UpperCamelCase											=   torch.load(_UpperCamelCase				)
										_UpperCamelCase											=   state_dict["model"].pop("final_proj.0.weight"				)
										_UpperCamelCase											=   state_dict["model"].pop("final_proj.0.bias"				)
										_UpperCamelCase											=   os.path.join(_UpperCamelCase ,   "converted.pt"				)
										torch.save(_UpperCamelCase ,   _UpperCamelCase				)
					else:
										_UpperCamelCase											=   DataaVecAudioForCTC(_UpperCamelCase				)
										_UpperCamelCase											=   checkpoint_path

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

					_UpperCamelCase											=   load_dataavec(_UpperCamelCase				)

					recursively_load_weights(_UpperCamelCase ,   _UpperCamelCase ,   not is_finetuned				)

					_UpperCamelCase											=   WavaVecaProcessor.from_pretrained("facebook/wav2vec2-large-lv60"				)

					_UpperCamelCase											=   load_dataset("patrickvonplaten/librispeech_asr_dummy" ,   "clean" ,   split="validation"				)
					_UpperCamelCase											=   [x["array"] for x in ds[:4]["audio"]]

					_UpperCamelCase											=   processor(_UpperCamelCase ,   return_tensors="pt" ,   padding=_UpperCamelCase				)

					_UpperCamelCase											=   inputs.input_values
					_UpperCamelCase											=   inputs.attention_mask
					#    input_values = inputs.input_values[:, :-1]
					#    attention_mask = inputs.attention_mask[:, :-1]

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

										_UpperCamelCase											=   torch.argmax(_UpperCamelCase ,   dim=-1				)
										_UpperCamelCase											=   processor.batch_decode(_UpperCamelCase				)

										print(f"Expected Output: {ds[:4]['text']}, Pred: {output_string}"				)
					else:
										_UpperCamelCase											=   model(source=_UpperCamelCase ,   padding_mask=(1 - attention_mask) ,   mask=_UpperCamelCase ,   features_only=_UpperCamelCase				)[
										    "layer_results"
										][-1][0].transpose(0 ,   1				)
										_UpperCamelCase											=   hf_wavavec(_UpperCamelCase ,   attention_mask=_UpperCamelCase				)["last_hidden_state"]

					print(our_output.shape ,   their_output.shape				)
					_UpperCamelCase											=   torch.max(torch.abs(our_output - their_output				)				).item()
					print(f"max_absolute_diff = {max_absolute_diff}"				)  # ~ 1e-7
					_UpperCamelCase											=   torch.allclose(_UpperCamelCase ,   _UpperCamelCase ,   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(_UpperCamelCase				)

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


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



