



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

UpperCamelCase__:       Tuple							     =      {
    "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",
}
UpperCamelCase__:       Union[str, Any]							     =      [
    "lm_head",
]


def 			snake_case_					( _lowerCAmelCase  :						Tuple       ,				_lowerCAmelCase  :						int       ,				_lowerCAmelCase  :						List[str]       ,				_lowerCAmelCase  :						Any       ,				_lowerCAmelCase  :						Any  )				->       Union[str, Any]:
							for attribute in key.split('''.'''  ):
														UpperCAmelCase			:	Tuple										=		getattr(_lowerCAmelCase       ,				_lowerCAmelCase  )

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

							if not is_headless:
														UpperCAmelCase			:	Optional[int]										=		hf_model.dataavec_audio.feature_extractor
														UpperCAmelCase			:	Optional[int]										=		hf_model.dataavec_audio.encoder.pos_conv_embed

							else:
														UpperCAmelCase			:	Optional[Any]										=		hf_model.feature_extractor
														UpperCAmelCase			:	List[str]										=		hf_model.encoder.pos_conv_embed

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


def 			snake_case_					( _lowerCAmelCase  :						Dict       ,				_lowerCAmelCase  :						Union[str, Any]       ,				_lowerCAmelCase  :						Union[str, Any]       ,				_lowerCAmelCase  :						str  )				->       Dict:
							UpperCAmelCase			:	str										=		access_by_string(_lowerCAmelCase       ,				_lowerCAmelCase  )
							UpperCAmelCase			:	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			:	Optional[int]										=		fsq_value
							logger.info(f"""{full_name} was correctly initialized from {hf_weight_path}."""  )


def 			snake_case_					( _lowerCAmelCase  :						Tuple       ,				_lowerCAmelCase  :						Optional[Any]       ,				_lowerCAmelCase  :						Dict       ,				_lowerCAmelCase  :						Dict  )				->       Optional[int]:
							UpperCAmelCase			:	Any										=		full_name.split('''conv_layers.'''  )[-1]
							UpperCAmelCase			:	int										=		name.split('''.'''  )
							UpperCAmelCase			:	Dict										=		int(items[0]  )
							UpperCAmelCase			:	Tuple										=		int(items[1]  )

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

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


def 			snake_case_					( _lowerCAmelCase  :						List[str]       ,				_lowerCAmelCase  :						Optional[Any]       ,				_lowerCAmelCase  :						Tuple       ,				_lowerCAmelCase  :						Tuple  )				->       Tuple:
							UpperCAmelCase			:	str										=		full_name.split('''pos_conv.'''  )[-1]
							UpperCAmelCase			:	List[str]										=		name.split('''.'''  )
							UpperCAmelCase			:	List[str]										=		int(items[0]  )
							UpperCAmelCase			:	Dict										=		int(items[1]  )

							UpperCAmelCase			:	str										=		name.split('''.'''  )[-1]
							if type_id != 0:
														unused_weights.append(_lowerCAmelCase  )
														return
							else:
														UpperCAmelCase			:	Tuple										=		'''conv'''

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


@torch.no_grad()
def 			snake_case_					( _lowerCAmelCase  :						str       ,				_lowerCAmelCase  :						str       ,				_lowerCAmelCase  :						List[str]=None       ,				_lowerCAmelCase  :						Union[str, Any]=None       ,				_lowerCAmelCase  :						List[Any]=True  )				->       str:
							if config_path is not None:
														UpperCAmelCase			:	Dict										=		DataaVecAudioConfig.from_pretrained(_lowerCAmelCase  )
							else:
														UpperCAmelCase			:	Any										=		DataaVecAudioConfig()

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

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

							def load_dataavec(_lowerCAmelCase  :						str  ):
														UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase			:	int										=		fairseq.checkpoint_utils.load_model_ensemble_and_task([path]  )
														return model[0].eval()

							UpperCAmelCase			:	Union[str, Any]										=		load_dataavec(_lowerCAmelCase  )

							recursively_load_weights(_lowerCAmelCase       ,				_lowerCAmelCase       ,				not is_finetuned  )

							UpperCAmelCase			:	Dict										=		WavaVecaProcessor.from_pretrained('''facebook/wav2vec2-large-lv60'''  )

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

							UpperCAmelCase			:	int										=		processor(_lowerCAmelCase       ,				return_tensors='''pt'''       ,				padding=_lowerCAmelCase  )

							UpperCAmelCase			:	Any										=		inputs.input_values
							UpperCAmelCase			:	Dict										=		inputs.attention_mask
							#    input_values = inputs.input_values[:, :-1]
							#    attention_mask = inputs.attention_mask[:, :-1]

							hf_wavavec.eval()
							model.eval()
							if is_finetuned:
														UpperCAmelCase			:	Optional[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			:	Union[str, Any]										=		torch.argmax(_lowerCAmelCase       ,				dim=-1  )
														UpperCAmelCase			:	List[str]										=		processor.batch_decode(_lowerCAmelCase  )

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

							print(our_output.shape       ,				their_output.shape  )
							UpperCAmelCase			:	Optional[int]										=		torch.max(torch.abs(our_output - their_output  )  ).item()
							print(f"""max_absolute_diff = {max_absolute_diff}"""  )  # ~ 1e-7
							UpperCAmelCase			:	Optional[Any]										=		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__":
						UpperCamelCase__:       Any							     =      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"
						)
						UpperCamelCase__:       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 re


def 			snake_case_					( _lowerCAmelCase  :						List[str]  )				->       Dict:
							def docstring_decorator(_lowerCAmelCase  :						List[str]  ):
														UpperCAmelCase			:	str										=		fn.__doc__
														UpperCAmelCase			:	List[str]										=		func_doc.split('''\n'''  )
														UpperCAmelCase			:	Optional[Any]										=		0
														while i < len(_lowerCAmelCase  ) and re.search(R'''^\s*Examples?:\s*$'''       ,				lines[i]  ) is None:
																					i += 1
														if i < len(_lowerCAmelCase  ):
																					UpperCAmelCase			:	Union[str, Any]										=		example_docstring
																					UpperCAmelCase			:	Tuple										=		'''\n'''.join(_lowerCAmelCase  )
														else:
																					raise ValueError(
																					    f"""The function {fn} should have an empty 'Examples:' in its docstring as placeholder, """
																					    f"""current docstring is:\n{func_doc}"""  )
														UpperCAmelCase			:	Any										=		func_doc
														return fn

							return docstring_decorator





