





"""simple docstring"""






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()
A	     =					logging.get_logger(__name__)

A	     =					{
    "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",
}
A	     =					[
    "lm_head",
    "quantizer.weight_proj",
    "quantizer.codevectors",
    "project_q",
    "project_hid",
]

def       __A (     a_  :Union[str, Any]				,    a_  :str				,    a_  :Tuple				,    a_  :Any				,    a_  :Union[str, Any])						->						List[Any]:





				for attribute in key.split('''.'''):
								__a	:  Optional[int]         =					getattr(A__				,    A__)

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

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

def       __A (     a_  :Optional[Any]				,    a_  :Any				,    a_  :List[Any])						->						List[str]:
				__a	:  List[Any]         =					[]
				__a	:  Tuple         =					fairseq_model.state_dict()

				__a	:  Tuple         =					hf_model.wavaveca_conformer.feature_extractor

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

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

def       __A (     a_  :Tuple				,    a_  :Any				,    a_  :List[Any]				,    a_  :str				,    a_  :Tuple)						->						List[Any]:
				__a	:  List[Any]         =					full_name.split('''conv_layers.''')[-1]
				__a	:  int         =					name.split('''.''')
				__a	:  Dict         =					int(items[0])
				__a	:  Dict         =					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.""")
												__a	:  Optional[Any]         =					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.""")
												__a	:  Union[str, Any]         =					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.""")
												__a	:  Tuple         =					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.""")
												__a	:  List[Any]         =					value
												logger.info(F"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.""")
				else:
								unused_weights.append(A__)

@torch.no_grad()
def       __A (     a_  :Dict				,    a_  :Dict				,    a_  :Dict=None				,    a_  :List[Any]=None				,    a_  :Any=True)						->						List[str]:





				if config_path is not None:
								__a	:  List[str]         =					WavaVecaConformerConfig.from_pretrained(A__				,    hidden_act='''swish''')
				else:
								__a	:  int         =					WavaVecaConformerConfig()

				if "rope" in checkpoint_path:
								__a	:  str         =					"""rotary"""

				if is_finetuned:
								if dict_path:
												__a	:  Optional[Any]         =					Dictionary.load(A__)

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

												# fairseq has the <pad> and <s> switched
												__a	:  str         =					0
												__a	:  List[str]         =					1
												with open(A__				,    '''w'''				,    encoding='''utf-8''') as vocab_handle:
																json.dump(A__				,    A__)
												__a	:  List[str]         =					WavaVecaCTCTokenizer(
												    A__				,    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=A__				,    )
												__a	:  Optional[Any]         =					True if config.feat_extract_norm == """layer""" else False
												__a	:  str         =					WavaVecaFeatureExtractor(
												    feature_size=1				,    sampling_rate=1_60_00				,    padding_value=0				,    do_normalize=A__				,    return_attention_mask=A__				,    )
												__a	:  str         =					WavaVecaProcessor(feature_extractor=A__				,    tokenizer=A__)
												processor.save_pretrained(A__)

								__a	:  Dict         =					WavaVecaConformerForCTC(A__)
				else:
								__a	:  List[Any]         =					WavaVecaConformerForPreTraining(A__)

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

								__a	:  int         =					fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]				,    task=A__)

				__a	:  int         =					model[0].eval()

				recursively_load_weights(A__				,    A__				,    not is_finetuned)

				hf_wavavec.save_pretrained(A__)


if __name__ == "__main__":
		A	     =					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'''
		)
		A	     =					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
		)






"""simple docstring"""






import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


A	     =					logging.get_logger(__name__)
A	     =					{'''vocab_file''': '''spiece.model'''}

A	     =					{
    '''vocab_file''': {
        '''albert-base-v1''': '''https://huggingface.co/albert-base-v1/resolve/main/spiece.model''',
        '''albert-large-v1''': '''https://huggingface.co/albert-large-v1/resolve/main/spiece.model''',
        '''albert-xlarge-v1''': '''https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model''',
        '''albert-xxlarge-v1''': '''https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model''',
        '''albert-base-v2''': '''https://huggingface.co/albert-base-v2/resolve/main/spiece.model''',
        '''albert-large-v2''': '''https://huggingface.co/albert-large-v2/resolve/main/spiece.model''',
        '''albert-xlarge-v2''': '''https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model''',
        '''albert-xxlarge-v2''': '''https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model''',
    }
}

A	     =					{
    '''albert-base-v1''': 512,
    '''albert-large-v1''': 512,
    '''albert-xlarge-v1''': 512,
    '''albert-xxlarge-v1''': 512,
    '''albert-base-v2''': 512,
    '''albert-large-v2''': 512,
    '''albert-xlarge-v2''': 512,
    '''albert-xxlarge-v2''': 512,
}

A	     =					'''▁'''




class 			__lowercase     (     _UpperCamelCase							):







					'''simple docstring'''




					__lowerCAmelCase					=					VOCAB_FILES_NAMES
					__lowerCAmelCase					=					PRETRAINED_VOCAB_FILES_MAP
					__lowerCAmelCase					=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES







					def __init__(  self		,     _UpperCAmelCase		,     _UpperCAmelCase=True		,     _UpperCAmelCase=True		,     _UpperCAmelCase=False		,     _UpperCAmelCase="[CLS]"		,     _UpperCAmelCase="[SEP]"		,     _UpperCAmelCase="<unk>"		,     _UpperCAmelCase="[SEP]"		,     _UpperCAmelCase="<pad>"		,     _UpperCAmelCase="[CLS]"		,     _UpperCAmelCase="[MASK]"		,     _UpperCAmelCase = None		,     **_UpperCAmelCase		,     ):
									# Mask token behave like a normal word, i.e. include the space before it and
									# is included in the raw text, there should be a match in a non-normalized sentence.
									__a	:  int         =					(
									    AddedToken(_UpperCAmelCase		,     lstrip=_UpperCAmelCase		,     rstrip=_UpperCAmelCase		,     normalized=_UpperCAmelCase       )
									    if isinstance(_UpperCAmelCase		,     _UpperCAmelCase       )
									    else mask_token
									)

									__a	:  Any         =					{} if sp_model_kwargs is None else sp_model_kwargs

									super().__init__(
									    do_lower_case=_UpperCAmelCase		,     remove_space=_UpperCAmelCase		,     keep_accents=_UpperCAmelCase		,     bos_token=_UpperCAmelCase		,     eos_token=_UpperCAmelCase		,     unk_token=_UpperCAmelCase		,     sep_token=_UpperCAmelCase		,     pad_token=_UpperCAmelCase		,     cls_token=_UpperCAmelCase		,     mask_token=_UpperCAmelCase		,     sp_model_kwargs=self.sp_model_kwargs		,     **_UpperCAmelCase		,     )

									__a	:  Tuple         =					do_lower_case
									__a	:  Optional[Any]         =					remove_space
									__a	:  Optional[Any]         =					keep_accents
									__a	:  Union[str, Any]         =					vocab_file

									__a	:  List[str]         =					spm.SentencePieceProcessor(**self.sp_model_kwargs       )
									self.sp_model.Load(_UpperCAmelCase       )







					@property
					def 	_lowerCamelCase						(  self       ):
									return len(self.sp_model       )







					def 	_lowerCamelCase						(  self       ):
									__a	:  Any         =					{self.convert_ids_to_tokens(_UpperCAmelCase       ): i for i in range(self.vocab_size       )}
									vocab.update(self.added_tokens_encoder       )
									return vocab







					def __getstate__(  self       ):
									__a	:  str         =					self.__dict__.copy()
									__a	:  Tuple         =					None
									return state







					def __setstate__(  self		,     _UpperCAmelCase       ):
									__a	:  Any         =					d

									# for backward compatibility
									if not hasattr(self		,     '''sp_model_kwargs'''       ):
													__a	:  Optional[Any]         =					{}

									__a	:  Dict         =					spm.SentencePieceProcessor(**self.sp_model_kwargs       )
									self.sp_model.Load(self.vocab_file       )







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									if self.remove_space:
													__a	:  Any         =					''' '''.join(inputs.strip().split()       )
									else:
													__a	:  Tuple         =					inputs
									__a	:  Union[str, Any]         =					outputs.replace('''``'''		,     '''"'''       ).replace('''\'\''''		,     '''"'''       )

									if not self.keep_accents:
													__a	:  List[str]         =					unicodedata.normalize('''NFKD'''		,     _UpperCAmelCase       )
													__a	:  Optional[int]         =					''''''.join([c for c in outputs if not unicodedata.combining(_UpperCAmelCase       )]       )
									if self.do_lower_case:
													__a	:  Optional[Any]         =					outputs.lower()

									return outputs







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									__a	:  int         =					self.preprocess_text(_UpperCAmelCase       )
									__a	:  Tuple         =					self.sp_model.encode(_UpperCAmelCase		,     out_type=_UpperCAmelCase       )
									__a	:  int         =					[]
									for piece in pieces:
													if len(_UpperCAmelCase       ) > 1 and piece[-1] == str(''','''       ) and piece[-2].isdigit():
																	__a	:  List[str]         =					self.sp_model.EncodeAsPieces(piece[:-1].replace(_UpperCAmelCase		,     ''''''       )       )
																	if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																					if len(cur_pieces[0]       ) == 1:
																									__a	:  Tuple         =					cur_pieces[1:]
																					else:
																									__a	:  Optional[Any]         =					cur_pieces[0][1:]
																	cur_pieces.append(piece[-1]       )
																	new_pieces.extend(_UpperCAmelCase       )
													else:
																	new_pieces.append(_UpperCAmelCase       )

									return new_pieces







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									return self.sp_model.PieceToId(_UpperCAmelCase       )







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									return self.sp_model.IdToPiece(_UpperCAmelCase       )







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase       ):
									__a	:  List[str]         =					[]
									__a	:  str         =					''''''
									__a	:  Any         =					False
									for token in tokens:
													# make sure that special tokens are not decoded using sentencepiece model
													if token in self.all_special_tokens:
																	if not prev_is_special:
																					out_string += " "
																	out_string += self.sp_model.decode(_UpperCAmelCase       ) + token
																	__a	:  Tuple         =					True
																	__a	:  Tuple         =					[]
													else:
																	current_sub_tokens.append(_UpperCAmelCase       )
																	__a	:  Optional[int]         =					False
									out_string += self.sp_model.decode(_UpperCAmelCase       )
									return out_string.strip()







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase		,     _UpperCAmelCase = None       ):
									__a	:  int         =					[self.sep_token_id]
									__a	:  Union[str, Any]         =					[self.cls_token_id]
									if token_ids_a is None:
													return cls + token_ids_a + sep
									return cls + token_ids_a + sep + token_ids_a + sep







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase		,     _UpperCAmelCase = None		,     _UpperCAmelCase = False       ):

									if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=_UpperCAmelCase		,     token_ids_a=_UpperCAmelCase		,     already_has_special_tokens=_UpperCAmelCase       )

									if token_ids_a is not None:
													return [1] + ([0] * len(_UpperCAmelCase       )) + [1] + ([0] * len(_UpperCAmelCase       )) + [1]
									return [1] + ([0] * len(_UpperCAmelCase       )) + [1]







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase		,     _UpperCAmelCase = None       ):
									__a	:  Union[str, Any]         =					[self.sep_token_id]
									__a	:  Tuple         =					[self.cls_token_id]

									if token_ids_a is None:
													return len(cls + token_ids_a + sep       ) * [0]
									return len(cls + token_ids_a + sep       ) * [0] + len(token_ids_a + sep       ) * [1]







					def 	_lowerCamelCase						(  self		,     _UpperCAmelCase		,     _UpperCAmelCase = None       ):
									if not os.path.isdir(_UpperCAmelCase       ):
													logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""       )
													return
									__a	:  List[str]         =					os.path.join(
									    _UpperCAmelCase		,     (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']       )

									if os.path.abspath(self.vocab_file       ) != os.path.abspath(_UpperCAmelCase       ) and os.path.isfile(self.vocab_file       ):
													copyfile(self.vocab_file		,     _UpperCAmelCase       )
									elif not os.path.isfile(self.vocab_file       ):
													with open(_UpperCAmelCase		,     '''wb'''       ) as fi:
																	__a	:  Any         =					self.sp_model.serialized_model_proto()
																	fi.write(_UpperCAmelCase       )

									return (out_vocab_file,)