






import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConformerConfig,
    WavaVecaConformerForCTC,
    WavaVecaConformerForPreTraining,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
__UpperCAmelCase     							=						logging.get_logger(__name__)

__UpperCAmelCase     							=						{
    'post_extract_proj': 'feature_projection.projection',
    'encoder.pos_conv.0': 'encoder.pos_conv_embed.conv',
    'self_attn.linear_k': 'encoder.layers.*.self_attn.linear_k',
    'self_attn.linear_v': 'encoder.layers.*.self_attn.linear_v',
    'self_attn.linear_q': 'encoder.layers.*.self_attn.linear_q',
    'self_attn.pos_bias_u': 'encoder.layers.*.self_attn.pos_bias_u',
    'self_attn.pos_bias_v': 'encoder.layers.*.self_attn.pos_bias_v',
    'self_attn.linear_out': 'encoder.layers.*.self_attn.linear_out',
    'self_attn.linear_pos': 'encoder.layers.*.self_attn.linear_pos',
    'self_attn.rotary_emb': 'encoder.embed_positions',
    'self_attn_layer_norm': 'encoder.layers.*.self_attn_layer_norm',
    'conv_module.pointwise_conv1': 'encoder.layers.*.conv_module.pointwise_conv1',
    'conv_module.pointwise_conv2': 'encoder.layers.*.conv_module.pointwise_conv2',
    'conv_module.depthwise_conv': 'encoder.layers.*.conv_module.depthwise_conv',
    'conv_module.batch_norm': 'encoder.layers.*.conv_module.batch_norm',
    'conv_module.layer_norm': 'encoder.layers.*.conv_module.layer_norm',
    'ffn1.w_1': 'encoder.layers.*.ffn1.intermediate_dense',
    'ffn1.w_2': 'encoder.layers.*.ffn1.output_dense',
    'ffn1.layer_norm': 'encoder.layers.*.ffn1_layer_norm',
    'ffn2.w_1': 'encoder.layers.*.ffn2.intermediate_dense',
    'ffn2.w_2': 'encoder.layers.*.ffn2.output_dense',
    'ffn2.layer_norm': 'encoder.layers.*.ffn2_layer_norm',
    'final_layer_norm': 'encoder.layers.*.final_layer_norm',
    'encoder.layer_norm': 'encoder.layer_norm',
    'w2v_model.layer_norm': 'feature_projection.layer_norm',
    'quantizer.weight_proj': 'quantizer.weight_proj',
    'quantizer.vars': 'quantizer.codevectors',
    'project_q': 'project_q',
    'final_proj': 'project_hid',
    'w2v_encoder.proj': 'lm_head',
    'mask_emb': 'masked_spec_embed',
}
__UpperCAmelCase     							=						[
    'lm_head',
    'quantizer.weight_proj',
    'quantizer.codevectors',
    'project_q',
    'project_hid',
]




def 					lowercase__				(					__snake_case   :							List[Any]					,	__snake_case   :							int					,	__snake_case   :							str					,	__snake_case   :							Any					,	__snake_case   :							Any  ):


					'''simple docstring'''


					for attribute in key.split('.'  ):
										UpperCAmelCase_   :				Any						=						getattr(__snake_case					,	__snake_case  )

					if weight_type is not None:
										UpperCAmelCase_   :				List[Any]						=						getattr(__snake_case					,	__snake_case  ).shape
					else:
										UpperCAmelCase_   :				List[str]						=						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_   :				str						=						value
					elif weight_type == "weight_g":
										UpperCAmelCase_   :				Any						=						value
					elif weight_type == "weight_v":
										UpperCAmelCase_   :				List[str]						=						value
					elif weight_type == "bias":
										UpperCAmelCase_   :				List[Any]						=						value
					elif weight_type == "running_mean":
										UpperCAmelCase_   :				Any						=						value
					elif weight_type == "running_var":
										UpperCAmelCase_   :				Tuple						=						value
					elif weight_type == "num_batches_tracked":
										UpperCAmelCase_   :				Optional[int]						=						value
					elif weight_type == "inv_freq":
										UpperCAmelCase_   :				Tuple						=						value
					else:
										UpperCAmelCase_   :				List[str]						=						value

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




def 					lowercase__				(					__snake_case   :							Dict					,	__snake_case   :							List[str]					,	__snake_case   :							Any  ):


					'''simple docstring'''


					UpperCAmelCase_   :				List[str]						=						[]
					UpperCAmelCase_   :				Tuple						=						fairseq_model.state_dict()

					UpperCAmelCase_   :				str						=						hf_model.wavaveca_conformer.feature_extractor

					for name, value in fairseq_dict.items():
										UpperCAmelCase_   :				List[str]						=						False
										if "conv_layers" in name:
															load_conv_layer(
															    __snake_case					,	__snake_case					,	__snake_case					,	__snake_case					,	hf_model.config.feat_extract_norm == 'group'					,	)
															UpperCAmelCase_   :				Union[str, Any]						=						True
										else:
															for key, mapped_key in MAPPING.items():
																				UpperCAmelCase_   :				str						=						'wav2vec2_conformer.' + 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_   :				Union[str, Any]						=						name.split(__snake_case  )[0].split('.'  )[-2]
																														UpperCAmelCase_   :				Any						=						mapped_key.replace('*'					,	__snake_case  )
																									if "pos_bias_u" in name:
																														UpperCAmelCase_   :				str						=						None
																									elif "pos_bias_v" in name:
																														UpperCAmelCase_   :				Dict						=						None
																									elif "weight_g" in name:
																														UpperCAmelCase_   :				Dict						=						'weight_g'
																									elif "weight_v" in name:
																														UpperCAmelCase_   :				Tuple						=						'weight_v'
																									elif "bias" in name:
																														UpperCAmelCase_   :				List[str]						=						'bias'
																									elif "weight" in name:
																														# TODO: don't match quantizer.weight_proj
																														UpperCAmelCase_   :				Optional[Any]						=						'weight'
																									elif "running_mean" in name:
																														UpperCAmelCase_   :				Optional[Any]						=						'running_mean'
																									elif "inv_freq" in name:
																														UpperCAmelCase_   :				int						=						'inv_freq'
																									elif "running_var" in name:
																														UpperCAmelCase_   :				Tuple						=						'running_var'
																									elif "num_batches_tracked" in name:
																														UpperCAmelCase_   :				str						=						'num_batches_tracked'
																									else:
																														UpperCAmelCase_   :				int						=						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 					lowercase__				(					__snake_case   :							Union[str, Any]					,	__snake_case   :							str					,	__snake_case   :							Optional[int]					,	__snake_case   :							Any					,	__snake_case   :							int  ):


					'''simple docstring'''


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

					if type_id == 0:
										if "bias" in name:
															if value.shape != feature_extractor.conv_layers[layer_id].conv.bias.data.shape:
																				raise ValueError(
																				    F"{full_name} has size {value.shape}, but"
																				    F" {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found."  )
															UpperCAmelCase_   :				List[str]						=						value
															logger.info(F"Feat extract conv layer {layer_id} was initialized from {full_name}."  )
										elif "weight" in name:
															if value.shape != feature_extractor.conv_layers[layer_id].conv.weight.data.shape:
																				raise ValueError(
																				    F"{full_name} has size {value.shape}, but"
																				    F" {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found."  )
															UpperCAmelCase_   :				Tuple						=						value
															logger.info(F"Feat extract conv layer {layer_id} was initialized from {full_name}."  )
					elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
										if "bias" in name:
															if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape:
																				raise ValueError(
																				    F"{full_name} has size {value.shape}, but"
																				    F" {feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape} was found."  )
															UpperCAmelCase_   :				str						=						value
															logger.info(F"Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."  )
										elif "weight" in name:
															if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape:
																				raise ValueError(
																				    F"{full_name} has size {value.shape}, but"
																				    F" {feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape} was found."  )
															UpperCAmelCase_   :				Optional[Any]						=						value
															logger.info(F"Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."  )
					else:
										unused_weights.append(__snake_case  )




@torch.no_grad()
def 					lowercase__				(					__snake_case   :							Any					,	__snake_case   :							int					,	__snake_case   :							Optional[Any]=None					,	__snake_case   :							Dict=None					,	__snake_case   :							List[str]=True  ):


					'''simple docstring'''


					if config_path is not None:
										UpperCAmelCase_   :				Optional[Any]						=						WavaVecaConformerConfig.from_pretrained(__snake_case					,	hidden_act='swish'  )
					else:
										UpperCAmelCase_   :				Dict						=						WavaVecaConformerConfig()

					if "rope" in checkpoint_path:
										UpperCAmelCase_   :				List[str]						=						'rotary'

					if is_finetuned:
										if dict_path:
															UpperCAmelCase_   :				Optional[int]						=						Dictionary.load(__snake_case  )

															# important change bos & pad token id since CTC symbol is <pad> and
															# not <s> as in fairseq
															UpperCAmelCase_   :				Dict						=						target_dict.pad_index
															UpperCAmelCase_   :				List[str]						=						target_dict.bos_index
															UpperCAmelCase_   :				Optional[int]						=						target_dict.eos_index
															UpperCAmelCase_   :				Any						=						len(target_dict.symbols  )
															UpperCAmelCase_   :				List[Any]						=						os.path.join(__snake_case					,	'vocab.json'  )
															if not os.path.isdir(__snake_case  ):
																				logger.error('--pytorch_dump_folder_path ({}) should be a directory'.format(__snake_case  )  )
																				return
															os.makedirs(__snake_case					,	exist_ok=__snake_case  )
															UpperCAmelCase_   :				Optional[int]						=						target_dict.indices

															# fairseq has the <pad> and <s> switched
															UpperCAmelCase_   :				Union[str, Any]						=						0
															UpperCAmelCase_   :				Union[str, Any]						=						1
															with open(__snake_case					,	'w'					,	encoding='utf-8'  ) as vocab_handle:
																				json.dump(__snake_case					,	__snake_case  )
															UpperCAmelCase_   :				str						=						WavaVecaCTCTokenizer(
															    __snake_case					,	unk_token=target_dict.unk_word					,	pad_token=target_dict.pad_word					,	bos_token=target_dict.bos_word					,	eos_token=target_dict.eos_word					,	word_delimiter_token='|'					,	do_lower_case=__snake_case					,	)
															UpperCAmelCase_   :				int						=						True if config.feat_extract_norm == 'layer' else False
															UpperCAmelCase_   :				Optional[int]						=						WavaVecaFeatureExtractor(
															    feature_size=1					,	sampling_rate=16_000					,	padding_value=0					,	do_normalize=__snake_case					,	return_attention_mask=__snake_case					,	)
															UpperCAmelCase_   :				Any						=						WavaVecaProcessor(feature_extractor=__snake_case					,	tokenizer=__snake_case  )
															processor.save_pretrained(__snake_case  )

										UpperCAmelCase_   :				List[str]						=						WavaVecaConformerForCTC(__snake_case  )
					else:
										UpperCAmelCase_   :				int						=						WavaVecaConformerForPreTraining(__snake_case  )

					if is_finetuned:
										UpperCAmelCase_   :				int						=						fairseq.checkpoint_utils.load_model_ensemble_and_task(
										    [checkpoint_path]					,	arg_overrides={'data': '/'.join(dict_path.split('/'  )[:-1]  )}  )
					else:
										UpperCAmelCase_   :				Optional[Any]						=						argparse.Namespace(task='audio_pretraining'  )
										UpperCAmelCase_   :				str						=						fairseq.tasks.setup_task(__snake_case  )

										UpperCAmelCase_   :				str						=						fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]					,	task=__snake_case  )

					UpperCAmelCase_   :				Dict						=						model[0].eval()

					recursively_load_weights(__snake_case					,	__snake_case					,	not is_finetuned  )

					hf_wavavec.save_pretrained(__snake_case  )


if __name__ == "__main__":
		__UpperCAmelCase     							=						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     							=						parser.parse_args()
		convert_wavaveca_conformer_checkpoint(
		    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
		)














def 					lowercase__				(					__snake_case   :							str					,	__snake_case   :							int					,	__snake_case   :							List[str]  ):


					'''simple docstring'''


					if n == 0:
										return 1

					elif n % 2 == 1:
										return (binary_exponentiation(__snake_case					,	n - 1					,	__snake_case  ) * a) % mod

					else:
										UpperCAmelCase_   :				Optional[int]						=						binary_exponentiation(__snake_case					,	n / 2					,	__snake_case  )
										return (b * b) % mod


# a prime number
__UpperCAmelCase     							=						701

__UpperCAmelCase     							=						1000000000
__UpperCAmelCase     							=						10

# using binary exponentiation function, O(log(p)):
print((a / b) % p == (a * binary_exponentiation(b, p - 2, p)) % p)






print((a / b) % p == (a * b ** (p - 2)) % p)






