






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

UpperCamelCase__						:      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""",
}
UpperCamelCase__						:      Dict								=					[
    """lm_head""",
]





def   SCREAMING_SNAKE_CASE__						(   snake_case_,  snake_case_,  snake_case_,  snake_case_,  snake_case_  )   ->  Optional[int]:


	"""simple docstring"""



	for attribute in key.split('''.'''  ):
		a		=			getattr(snake_case_,  snake_case_  )

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


	"""simple docstring"""



	a		=			[]
	a		=			fairseq_model.state_dict()

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

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

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

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





def   SCREAMING_SNAKE_CASE__						(   snake_case_,  snake_case_  )   ->  Tuple:


	"""simple docstring"""



	a		=			path.split('''.'''  )
	return reduce(snake_case_,  snake_case_,  snake_case_  )





def   SCREAMING_SNAKE_CASE__						(   snake_case_,  snake_case_,  snake_case_,  snake_case_  )   ->  Tuple:


	"""simple docstring"""



	a		=			access_by_string(snake_case_,  snake_case_  )
	a		=			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."""  )
	a		=			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_  )   ->  Optional[int]:


	"""simple docstring"""



	a		=			full_name.split('''conv_layers.'''  )[-1]
	a		=			name.split('''.'''  )
	a		=			int(items[0]  )
	a		=			int(items[1]  )

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

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





def   SCREAMING_SNAKE_CASE__						(   snake_case_,  snake_case_,  snake_case_,  snake_case_  )   ->  Union[str, Any]:


	"""simple docstring"""



	a		=			full_name.split('''pos_conv.'''  )[-1]
	a		=			name.split('''.'''  )
	a		=			int(items[0]  )
	a		=			int(items[1]  )

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

	set_weights(snake_case_,  snake_case_,  snake_case_,  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  )   ->  Union[str, Any]:


	"""simple docstring"""



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

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

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

	def load_dataavec(snake_case_  ):
		a	, a	, a		=			fairseq.checkpoint_utils.load_model_ensemble_and_task([path]  )
		return model[0].eval()

	a		=			load_dataavec(snake_case_  )

	recursively_load_weights(snake_case_,  snake_case_,  not is_finetuned  )

	a		=			WavaVecaProcessor.from_pretrained('''facebook/wav2vec2-large-lv60'''  )

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

	a		=			processor(snake_case_,  return_tensors='''pt''',  padding=snake_case_  )

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

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

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

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

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

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


if __name__ == "__main__":
				UpperCamelCase__						:      Optional[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__						:      Dict								=					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_  )   ->  int:


	"""simple docstring"""



	return n * (2 * n - 1)





def   SCREAMING_SNAKE_CASE__						(   snake_case_  )   ->  bool:


	"""simple docstring"""



	a		=			(1 + 2_4 * n) ** 0.5
	return ((1 + root) / 6) % 1 == 0





def   SCREAMING_SNAKE_CASE__						(   snake_case_ = 1_4_4  )   ->  int:


	"""simple docstring"""



	a		=			start
	a		=			hexagonal_num(snake_case_  )
	while not is_pentagonal(snake_case_  ):
		n += 1
		a		=			hexagonal_num(snake_case_  )
	return num


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





