




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()
lowercase_			:       int						       =				logging.get_logger(__name__)

lowercase_			:       Union[str, 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',
}
lowercase_			:       Dict						       =				[
    'lm_head',
]


def 							__SCREAMING_SNAKE_CASE						(    snake_case_			,					snake_case_			,					snake_case_			,					snake_case_			,					snake_case_		):

						'''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 							__SCREAMING_SNAKE_CASE						(    snake_case_			,					snake_case_			,					snake_case_		):

						'''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 							__SCREAMING_SNAKE_CASE						(    snake_case_			,					snake_case_		):

						'''simple docstring'''


						_UpperCAmelCase									=  path.split("."		)
						return reduce(UpperCamelCase__			,					UpperCamelCase__			,					UpperCamelCase__		)


def 							__SCREAMING_SNAKE_CASE						(    snake_case_			,					snake_case_			,					snake_case_			,					snake_case_		):

						'''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 							__SCREAMING_SNAKE_CASE						(    snake_case_			,					snake_case_			,					snake_case_			,					snake_case_		):

						'''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 							__SCREAMING_SNAKE_CASE						(    snake_case_			,					snake_case_			,					snake_case_			,					snake_case_		):

						'''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 							__SCREAMING_SNAKE_CASE						(    snake_case_			,					snake_case_			,					snake_case_=None			,					snake_case_=None			,					snake_case_=True		):

						'''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(snake_case_		):
												_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__":
	lowercase_			:       Union[str, 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'
	)
	lowercase_			:       Optional[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
	)










def 							__SCREAMING_SNAKE_CASE						(    snake_case_ = 99_8001		):

						'''simple docstring'''


						for number in range(n - 1			,					9999			,					-1		):
												_UpperCAmelCase									=  str(snake_case_		)

												# checks whether 'str_number' is a palindrome.
												if str_number == str_number[::-1]:
																		_UpperCAmelCase									=  999

																		# if 'number' is a product of two 3-digit numbers
																		# then number is the answer otherwise fetch next number.
																		while divisor != 99:
																								if (number % divisor == 0) and (len(str(number // divisor		)		) == 3.0):
																														return number
																								divisor -= 1
						raise ValueError("That number is larger than our acceptable range."		)


if __name__ == "__main__":
				print(f"""{solution() = }""")




