


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

lowerCAmelCase   :int								=    {
    '''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   :Any								=    [
    '''lm_head''',
]





def lowerCamelCase    (    lowerCAmelCase       :			Optional[int]	,	lowerCAmelCase       :			Any	,	lowerCAmelCase       :			Tuple	,	lowerCAmelCase       :			Tuple	,	lowerCAmelCase       :			int	):





	"""simple docstring"""







	for attribute in key.split('.'	):
		__magic_name__ :					Dict  			=  getattr(lowerCAmelCase	,	lowerCAmelCase	)

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

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





def lowerCamelCase    (    lowerCAmelCase       :			Optional[int]	,	lowerCAmelCase       :			Optional[Any]	,	lowerCAmelCase       :			List[str]	):





	"""simple docstring"""







	__magic_name__ :					List[str]  			=  []
	__magic_name__ :					Dict  			=  fairseq_model.state_dict()

	if not is_headless:
		__magic_name__ :					int  			=  hf_model.dataavec_audio.feature_extractor
		__magic_name__ :					Dict  			=  hf_model.dataavec_audio.encoder.pos_conv_embed

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

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





	"""simple docstring"""







	__magic_name__ :					Tuple  			=  path.split('.'	)
	return reduce(lowerCAmelCase	,	lowerCAmelCase	,	lowerCAmelCase	)





def lowerCamelCase    (    lowerCAmelCase       :			List[Any]	,	lowerCAmelCase       :			int	,	lowerCAmelCase       :			str	,	lowerCAmelCase       :			str	):





	"""simple docstring"""







	__magic_name__ :					Dict  			=  access_by_string(lowerCAmelCase	,	lowerCAmelCase	)
	__magic_name__ :					Tuple  			=  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.'	)
	__magic_name__ :					int  			=  fsq_value
	logger.info(f'{full_name} was correctly initialized from {hf_weight_path}.'	)





def lowerCamelCase    (    lowerCAmelCase       :			str	,	lowerCAmelCase       :			str	,	lowerCAmelCase       :			str	,	lowerCAmelCase       :			List[Any]	):





	"""simple docstring"""







	__magic_name__ :					Any  			=  full_name.split('conv_layers.'	)[-1]
	__magic_name__ :					Union[str, Any]  			=  name.split('.'	)
	__magic_name__ :					List[str]  			=  int(items[0]	)
	__magic_name__ :					int  			=  int(items[1]	)

	__magic_name__ :					Tuple  			=  name.split('.'	)[-1]
	if type_id == 0:
		__magic_name__ :					Tuple  			=  'conv'
	elif type_id == 2:
		__magic_name__ :					str  			=  'layer_norm'
	else:
		unused_weights.append(lowerCAmelCase	)
		return

	set_weights(lowerCAmelCase	,	lowerCAmelCase	,	lowerCAmelCase	,	f'conv_layers.{layer_id}.{layer_type}.{weight_type}'	)





def lowerCamelCase    (    lowerCAmelCase       :			Union[str, Any]	,	lowerCAmelCase       :			Dict	,	lowerCAmelCase       :			str	,	lowerCAmelCase       :			Union[str, Any]	):





	"""simple docstring"""







	__magic_name__ :					str  			=  full_name.split('pos_conv.'	)[-1]
	__magic_name__ :					Dict  			=  name.split('.'	)
	__magic_name__ :					Dict  			=  int(items[0]	)
	__magic_name__ :					str  			=  int(items[1]	)

	__magic_name__ :					Optional[int]  			=  name.split('.'	)[-1]
	if type_id != 0:
		unused_weights.append(lowerCAmelCase	)
		return
	else:
		__magic_name__ :					Union[str, Any]  			=  'conv'

	set_weights(lowerCAmelCase	,	lowerCAmelCase	,	lowerCAmelCase	,	f'layers.{layer_id}.{layer_type}.{weight_type}'	)





@torch.no_grad()
def lowerCamelCase    (    lowerCAmelCase       :			List[Any]	,	lowerCAmelCase       :			Union[str, Any]	,	lowerCAmelCase       :			Optional[Any]=None	,	lowerCAmelCase       :			List[str]=None	,	lowerCAmelCase       :			str=True	):





	"""simple docstring"""







	if config_path is not None:
		__magic_name__ :					List[str]  			=  DataaVecAudioConfig.from_pretrained(lowerCAmelCase	)
	else:
		__magic_name__ :					List[str]  			=  DataaVecAudioConfig()

	if not is_finetuned:
		# Modify final_proj layer name
		__magic_name__ :					Union[str, Any]  			=  DataaVecAudioModel(lowerCAmelCase	)
		__magic_name__ :					Optional[Any]  			=  os.path.dirname(lowerCAmelCase	)

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

	def load_dataavec(lowerCAmelCase       :			str	):
		__magic_name__			,       __magic_name__			,       __magic_name__ :					Tuple  			=  fairseq.checkpoint_utils.load_model_ensemble_and_task([path]	)
		return model[0].eval()

	__magic_name__ :					Tuple  			=  load_dataavec(lowerCAmelCase	)

	recursively_load_weights(lowerCAmelCase	,	lowerCAmelCase	,	not is_finetuned	)

	__magic_name__ :					Optional[Any]  			=  WavaVecaProcessor.from_pretrained('facebook/wav2vec2-large-lv60'	)

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

	__magic_name__ :					List[Any]  			=  processor(lowerCAmelCase	,	return_tensors='pt'	,	padding=lowerCAmelCase	)

	__magic_name__ :					Optional[Any]  			=  inputs.input_values
	__magic_name__ :					Optional[Any]  			=  inputs.attention_mask
	#    input_values = inputs.input_values[:, :-1]
	#    attention_mask = inputs.attention_mask[:, :-1]

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

		__magic_name__ :					Any  			=  torch.argmax(lowerCAmelCase	,	dim=-1	)
		__magic_name__ :					Dict  			=  processor.batch_decode(lowerCAmelCase	)

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

	print(our_output.shape	,	their_output.shape	)
	__magic_name__ :					Optional[int]  			=  torch.max(torch.abs(our_output - their_output	)	).item()
	print(f'max_absolute_diff = {max_absolute_diff}'	)  # ~ 1e-7
	__magic_name__ :					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__":
							lowerCAmelCase   :Dict								=    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   :List[str]								=    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 math





def lowerCamelCase    (    lowerCAmelCase       :			int	):





	"""simple docstring"""







	if not isinstance(lowerCAmelCase	,	lowerCAmelCase	):
		__magic_name__ :					str  			=  f'Input value of [number={number}] must be an integer'
		raise TypeError(lowerCAmelCase	)
	if number < 1:
		__magic_name__ :					Tuple  			=  f'Input value of [number={number}] must be a positive integer'
		raise ValueError(lowerCAmelCase	)
	__magic_name__ :					Tuple  			=  [number]
	while number != 1:
		if number % 2 == 0:
			__magic_name__ :					Optional[int]  			=  math.floor(math.sqrt(lowerCAmelCase	)	)
		else:
			__magic_name__ :					Union[str, Any]  			=  math.floor(
			    math.sqrt(lowerCAmelCase	) * math.sqrt(lowerCAmelCase	) * math.sqrt(lowerCAmelCase	)	)
		sequence.append(lowerCAmelCase	)
	return sequence


if __name__ == "__main__":
							import doctest

							doctest.testmod()