




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

_lowerCAmelCase       =   {
    '''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''',
}
_lowerCAmelCase       =   [
    '''lm_head''',
]

def      __lowerCAmelCase   (       snake_case__   , snake_case__   , snake_case__   , snake_case__   , snake_case__		):
			for attribute in key.split("."		):
						__UpperCamelCase :     Optional[Any]										=  getattr(__lowerCAmelCase   , __lowerCAmelCase		)

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

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

def      __lowerCAmelCase   (       snake_case__   , snake_case__   , snake_case__		):
			__UpperCamelCase :     int										=  []
			__UpperCamelCase :     int										=  fairseq_model.state_dict()

			if not is_headless:
						__UpperCamelCase :     int										=  hf_model.dataavec_audio.feature_extractor
						__UpperCamelCase :     Tuple										=  hf_model.dataavec_audio.encoder.pos_conv_embed

			else:
						__UpperCamelCase :     str										=  hf_model.feature_extractor
						__UpperCamelCase :     Optional[int]										=  hf_model.encoder.pos_conv_embed

			for name, value in fairseq_dict.items():
						__UpperCamelCase :     List[str]										=  False
						if "conv_layers" in name:
									load_conv_layer(
									    __lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase   , )
									__UpperCamelCase :     str										=  True
						elif "pos_conv" in name:
									load_pos_conv_layer(
									    __lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase   , )
									__UpperCamelCase :     Tuple										=  True
						else:
									for key, mapped_key in MAPPING.items():
												if not is_headless:
															__UpperCamelCase :     Optional[int]										=  "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 :     Tuple										=  True
															if "*" in mapped_key:
																		__UpperCamelCase :     int										=  name.split(__lowerCAmelCase		)[0].split("."		)[-2]
																		__UpperCamelCase :     str										=  mapped_key.replace("*"   , __lowerCAmelCase		)
															if "weight_g" in name:
																		__UpperCamelCase :     str										=  "weight_g"
															elif "weight_v" in name:
																		__UpperCamelCase :     Tuple										=  "weight_v"
															elif "bias" in name:
																		__UpperCamelCase :     Tuple										=  "bias"
															elif "weight" in name:
																		# TODO: don't match quantizer.weight_proj
																		__UpperCamelCase :     str										=  "weight"
															else:
																		__UpperCamelCase :     int										=  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      __lowerCAmelCase   (       snake_case__   , snake_case__		):
			__UpperCamelCase :     Any										=  path.split("."		)
			return reduce(__lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase		)

def      __lowerCAmelCase   (       snake_case__   , snake_case__   , snake_case__   , snake_case__		):
			__UpperCamelCase :     str										=  access_by_string(__lowerCAmelCase   , __lowerCAmelCase		)
			__UpperCamelCase :     List[Any]										=  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 :     List[str]										=  fsq_value
			logger.info(F"{full_name} was correctly initialized from {hf_weight_path}."		)

def      __lowerCAmelCase   (       snake_case__   , snake_case__   , snake_case__   , snake_case__		):
			__UpperCamelCase :     Tuple										=  full_name.split("conv_layers."		)[-1]
			__UpperCamelCase :     str										=  name.split("."		)
			__UpperCamelCase :     Union[str, Any]										=  int(items[0]		)
			__UpperCamelCase :     List[Any]										=  int(items[1]		)

			__UpperCamelCase :     Tuple										=  name.split("."		)[-1]
			if type_id == 0:
						__UpperCamelCase :     str										=  "conv"
			elif type_id == 2:
						__UpperCamelCase :     Optional[Any]										=  "layer_norm"
			else:
						unused_weights.append(__lowerCAmelCase		)
						return

			set_weights(__lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase   , F"conv_layers.{layer_id}.{layer_type}.{weight_type}"		)

def      __lowerCAmelCase   (       snake_case__   , snake_case__   , snake_case__   , snake_case__		):
			__UpperCamelCase :     int										=  full_name.split("pos_conv."		)[-1]
			__UpperCamelCase :     Union[str, Any]										=  name.split("."		)
			__UpperCamelCase :     Optional[Any]										=  int(items[0]		)
			__UpperCamelCase :     List[str]										=  int(items[1]		)

			__UpperCamelCase :     Optional[Any]										=  name.split("."		)[-1]
			if type_id != 0:
						unused_weights.append(__lowerCAmelCase		)
						return
			else:
						__UpperCamelCase :     str										=  "conv"

			set_weights(__lowerCAmelCase   , __lowerCAmelCase   , __lowerCAmelCase   , F"layers.{layer_id}.{layer_type}.{weight_type}"		)

@torch.no_grad()
def      __lowerCAmelCase   (       snake_case__   , snake_case__   , snake_case__=None   , snake_case__=None   , snake_case__=True		):
			if config_path is not None:
						__UpperCamelCase :     Union[str, Any]										=  DataaVecAudioConfig.from_pretrained(__lowerCAmelCase		)
			else:
						__UpperCamelCase :     Optional[int]										=  DataaVecAudioConfig()

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

						__UpperCamelCase :     Optional[Any]										=  torch.load(__lowerCAmelCase		)
						__UpperCamelCase :     Optional[Any]										=  state_dict["model"].pop("final_proj.0.weight"		)
						__UpperCamelCase :     List[str]										=  state_dict["model"].pop("final_proj.0.bias"		)
						__UpperCamelCase :     Optional[Any]										=  os.path.join(__lowerCAmelCase   , "converted.pt"		)
						torch.save(__lowerCAmelCase   , __lowerCAmelCase		)
			else:
						__UpperCamelCase :     Dict										=  DataaVecAudioForCTC(__lowerCAmelCase		)
						__UpperCamelCase :     List[str]										=  checkpoint_path

			def load_dataavec(snake_case__		):
						__UpperCamelCase :     Optional[Any]										=  fairseq.checkpoint_utils.load_model_ensemble_and_task([path]		)
						return model[0].eval()

			__UpperCamelCase :     Optional[int]										=  load_dataavec(__lowerCAmelCase		)

			recursively_load_weights(__lowerCAmelCase   , __lowerCAmelCase   , not is_finetuned		)

			__UpperCamelCase :     Tuple										=  WavaVecaProcessor.from_pretrained("facebook/wav2vec2-large-lv60"		)

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

			__UpperCamelCase :     List[Any]										=  processor(__lowerCAmelCase   , return_tensors="pt"   , padding=__lowerCAmelCase		)

			__UpperCamelCase :     List[Any]										=  inputs.input_values
			__UpperCamelCase :     Union[str, Any]										=  inputs.attention_mask
			#    input_values = inputs.input_values[:, :-1]
			#    attention_mask = inputs.attention_mask[:, :-1]

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

						__UpperCamelCase :     Dict										=  torch.argmax(__lowerCAmelCase   , dim=-1		)
						__UpperCamelCase :     List[Any]										=  processor.batch_decode(__lowerCAmelCase		)

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

			print(our_output.shape   , their_output.shape		)
			__UpperCamelCase :     List[str]										=  torch.max(torch.abs(our_output - their_output		)		).item()
			print(F"max_absolute_diff = {max_absolute_diff}"		)  # ~ 1e-7
			__UpperCamelCase :     List[str]										=  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__":
				_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'''
				)
				_lowerCAmelCase       =   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 numpy as np
from numpy import floataa
from numpy.typing import NDArray






def 				__lowerCAmelCase					(					__lowerCAmelCase			:		NDArray[floataa]  ,	__lowerCAmelCase			:		NDArray[floataa]						)    ->       NDArray[floataa]:
 _UpperCamelCase      ,	_UpperCamelCase      :	Optional[Any]					      =  np.shape(__lowerCAmelCase						)

 _UpperCamelCase      :	NDArray[floataa]					      =  np.zeros((rows, 1)  ,	dtype=__lowerCAmelCase						)
 for row in reversed(range(__lowerCAmelCase						)						):
  _UpperCamelCase      :	List[str]					      =  0
  for col in range(row + 1  ,	__lowerCAmelCase						):
   total += coefficients[row, col] * x[col]

  _UpperCamelCase      :	Dict					      =  (vector[row] - total) / coefficients[row, row]

 return x






def 				__lowerCAmelCase					(					__lowerCAmelCase			:		NDArray[floataa]  ,	__lowerCAmelCase			:		NDArray[floataa]						)    ->       NDArray[floataa]:
 _UpperCamelCase      ,	_UpperCamelCase      :	Optional[Any]					      =  np.shape(__lowerCAmelCase						)
 if rows != columns:
  return np.array(()  ,	dtype=__lowerCAmelCase						)

 # augmented matrix
 _UpperCamelCase      :	NDArray[floataa]					      =  np.concatenate((coefficients, vector)  ,	axis=1						)
 _UpperCamelCase      :	List[str]					      =  augmented_mat.astype("float64"						)

 # scale the matrix leaving it triangular
 for row in range(rows - 1						):
  _UpperCamelCase      :	Optional[Any]					      =  augmented_mat[row, row]
  for col in range(row + 1  ,	__lowerCAmelCase						):
   _UpperCamelCase      :	Union[str, Any]					      =  augmented_mat[col, row] / pivot
   augmented_mat[col, :] -= factor * augmented_mat[row, :]

 _UpperCamelCase      :	Any					      =  retroactive_resolution(
     augmented_mat[:, 0:columns]  ,	augmented_mat[:, columns : columns + 1]						)

 return x


if __name__ == "__main__":
    import doctest

    doctest.testmod()



