





'''simple docstring'''




import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConfig,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaForCTC,
    WavaVecaForPreTraining,
    WavaVecaProcessor,
    logging,
)
from transformers.models.wavaveca.modeling_wavaveca import WavaVecaForSequenceClassification


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.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""",
    """adapter_layer""": """encoder.layers.*.adapter_layer""",
    """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""",
    """pooling_layer.linear""": """projector""",
    """pooling_layer.projection""": """classifier""",
}
__UpperCAmelCase            =       [
    """lm_head""",
    """quantizer.weight_proj""",
    """quantizer.codevectors""",
    """project_q""",
    """project_hid""",
    """projector""",
    """classifier""",
]



def 		__A		(  lowerCamelCase_	):


		"""simple docstring"""



		SCREAMING_SNAKE_CASE :				Any		=			{}
		with open(lowerCamelCase_			,			"""r"""	) as file:
				for line_number, line in enumerate(lowerCamelCase_	):
						SCREAMING_SNAKE_CASE :				List[Any]		=			line.strip()
						if line:
								SCREAMING_SNAKE_CASE :				List[Any]		=			line.split()
								SCREAMING_SNAKE_CASE :				Optional[Any]		=			line_number
								SCREAMING_SNAKE_CASE :				Any		=			words[0]
								SCREAMING_SNAKE_CASE :				Union[str, Any]		=			value
		return result



def 		__A		(  lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_	):


		"""simple docstring"""



		for attribute in key.split("""."""	):
				SCREAMING_SNAKE_CASE :				Optional[Any]		=			getattr(lowerCamelCase_			,			lowerCamelCase_	)

		SCREAMING_SNAKE_CASE :				Any		=			None
		for param_key in PARAM_MAPPING.keys():
				if full_name.endswith(lowerCamelCase_	):
						SCREAMING_SNAKE_CASE :				str		=			PARAM_MAPPING[full_name.split("""."""	)[-1]]
						SCREAMING_SNAKE_CASE :				int		=			"""param"""

		if weight_type is not None and weight_type != "param":
				SCREAMING_SNAKE_CASE :				Tuple		=			getattr(lowerCamelCase_			,			lowerCamelCase_	).shape
		elif weight_type is not None and weight_type == "param":
				SCREAMING_SNAKE_CASE :				List[Any]		=			hf_pointer
				for attribute in hf_param_name.split("""."""	):
						SCREAMING_SNAKE_CASE :				Optional[Any]		=			getattr(lowerCamelCase_			,			lowerCamelCase_	)
				SCREAMING_SNAKE_CASE :				Optional[int]		=			shape_pointer.shape

				# let's reduce dimension
				SCREAMING_SNAKE_CASE :				Tuple		=			value[0]
		else:
				SCREAMING_SNAKE_CASE :				Optional[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":
				SCREAMING_SNAKE_CASE :				Tuple		=			value
		elif weight_type == "weight_g":
				SCREAMING_SNAKE_CASE :				Optional[int]		=			value
		elif weight_type == "weight_v":
				SCREAMING_SNAKE_CASE :				str		=			value
		elif weight_type == "bias":
				SCREAMING_SNAKE_CASE :				int		=			value
		elif weight_type == "param":
				for attribute in hf_param_name.split("""."""	):
						SCREAMING_SNAKE_CASE :				str		=			getattr(lowerCamelCase_			,			lowerCamelCase_	)
				SCREAMING_SNAKE_CASE :				int		=			value
		else:
				SCREAMING_SNAKE_CASE :				Any		=			value

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



def 		__A		(  lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_	):


		"""simple docstring"""



		SCREAMING_SNAKE_CASE :				str		=			None
		for param_key in PARAM_MAPPING.keys():
				if full_name.endswith(lowerCamelCase_	):
						SCREAMING_SNAKE_CASE :				Dict		=			PARAM_MAPPING[full_name.split("""."""	)[-1]]
						SCREAMING_SNAKE_CASE :				Tuple		=			"""param"""

		if weight_type is not None and weight_type != "param":
				SCREAMING_SNAKE_CASE :				Any		=			""".""".join([key, weight_type]	)
		elif weight_type is not None and weight_type == "param":
				SCREAMING_SNAKE_CASE :				Optional[Any]		=			""".""".join([key, hf_param_name]	)
		else:
				SCREAMING_SNAKE_CASE :				Tuple		=			key

		SCREAMING_SNAKE_CASE :				int		=			value if """lm_head""" in full_key else value[0]


__UpperCAmelCase            =       {
    """W_a""": """linear_1.weight""",
    """W_b""": """linear_2.weight""",
    """b_a""": """linear_1.bias""",
    """b_b""": """linear_2.bias""",
    """ln_W""": """norm.weight""",
    """ln_b""": """norm.bias""",
}



def 		__A		(  lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_=None			,			lowerCamelCase_=None	):


		"""simple docstring"""



		SCREAMING_SNAKE_CASE :				Dict		=			False
		for key, mapped_key in MAPPING.items():
				SCREAMING_SNAKE_CASE :				Any		=			"""wav2vec2.""" + 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]:
						SCREAMING_SNAKE_CASE :				Union[str, Any]		=			True
						if "*" in mapped_key:
								SCREAMING_SNAKE_CASE :				str		=			name.split(lowerCamelCase_	)[0].split("""."""	)[-2]
								SCREAMING_SNAKE_CASE :				int		=			mapped_key.replace("""*"""			,			lowerCamelCase_	)
						if "weight_g" in name:
								SCREAMING_SNAKE_CASE :				Any		=			"""weight_g"""
						elif "weight_v" in name:
								SCREAMING_SNAKE_CASE :				Optional[int]		=			"""weight_v"""
						elif "bias" in name:
								SCREAMING_SNAKE_CASE :				Union[str, Any]		=			"""bias"""
						elif "weight" in name:
								# TODO: don't match quantizer.weight_proj
								SCREAMING_SNAKE_CASE :				List[Any]		=			"""weight"""
						else:
								SCREAMING_SNAKE_CASE :				Optional[int]		=			None
						if hf_dict is not None:
								rename_dict(lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_	)
						else:
								set_recursively(lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_	)
						return is_used
		return is_used



def 		__A		(  lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_	):


		"""simple docstring"""



		SCREAMING_SNAKE_CASE :				str		=			[]
		SCREAMING_SNAKE_CASE :				Dict		=			fairseq_model.state_dict()

		SCREAMING_SNAKE_CASE :				Dict		=			hf_model.wavaveca.feature_extractor

		for name, value in fairseq_dict.items():
				SCREAMING_SNAKE_CASE :				List[str]		=			False
				if "conv_layers" in name:
						load_conv_layer(
						    lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			hf_model.config.feat_extract_norm == """group"""			,			)
						SCREAMING_SNAKE_CASE :				Union[str, Any]		=			True
				else:
						SCREAMING_SNAKE_CASE :				Optional[int]		=			load_wavaveca_layer(lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_	)
				if not is_used:
						unused_weights.append(lowerCamelCase_	)

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



def 		__A		(  lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_	):


		"""simple docstring"""



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



@torch.no_grad()
def 		__A		(  lowerCamelCase_			,			lowerCamelCase_			,			lowerCamelCase_=None			,			lowerCamelCase_=None			,			lowerCamelCase_=True			,			lowerCamelCase_=False	):


		"""simple docstring"""



		if config_path is not None:
				SCREAMING_SNAKE_CASE :				List[Any]		=			WavaVecaConfig.from_pretrained(lowerCamelCase_	)
		else:
				SCREAMING_SNAKE_CASE :				Dict		=			WavaVecaConfig()

		if is_seq_class:
				SCREAMING_SNAKE_CASE :				Any		=			read_txt_into_dict(lowerCamelCase_	)
				SCREAMING_SNAKE_CASE :				Optional[Any]		=			idalabel
				SCREAMING_SNAKE_CASE :				List[str]		=			WavaVecaForSequenceClassification(lowerCamelCase_	)
				SCREAMING_SNAKE_CASE :				str		=			WavaVecaFeatureExtractor(
				    feature_size=1			,			sampling_rate=1_60_00			,			padding_value=0			,			do_normalize=lowerCamelCase_			,			return_attention_mask=lowerCamelCase_			,			)
				feature_extractor.save_pretrained(lowerCamelCase_	)

		elif is_finetuned:
				if dict_path:
						SCREAMING_SNAKE_CASE :				int		=			Dictionary.load(lowerCamelCase_	)

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

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

				SCREAMING_SNAKE_CASE :				Union[str, Any]		=			WavaVecaForCTC(lowerCamelCase_	)
		else:
				SCREAMING_SNAKE_CASE :				Union[str, Any]		=			WavaVecaForPreTraining(lowerCamelCase_	)

		if is_finetuned or is_seq_class:
				SCREAMING_SNAKE_CASE,							SCREAMING_SNAKE_CASE,							SCREAMING_SNAKE_CASE :				Any		=			fairseq.checkpoint_utils.load_model_ensemble_and_task(
				    [checkpoint_path]			,			arg_overrides={"""data""": """/""".join(dict_path.split("""/"""	)[:-1]	)}	)
		else:
				SCREAMING_SNAKE_CASE :				Optional[int]		=			argparse.Namespace(task="""audio_pretraining"""	)
				SCREAMING_SNAKE_CASE :				List[str]		=			fairseq.tasks.setup_task(lowerCamelCase_	)

				SCREAMING_SNAKE_CASE,							SCREAMING_SNAKE_CASE,							SCREAMING_SNAKE_CASE :				str		=			fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]			,			task=lowerCamelCase_	)

		SCREAMING_SNAKE_CASE :				Any		=			model[0].eval()

		recursively_load_weights(lowerCamelCase_			,			lowerCamelCase_			,			not is_finetuned	)

		hf_wavavec.save_pretrained(lowerCamelCase_	)


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"""
							)
							parser.add_argument(
							    """--is_seq_class""",
							    action="""store_true""",
							    help="""Whether the model to convert is a fine-tuned sequence classification model or not""",
							)
							__UpperCAmelCase            =       parser.parse_args()

							__UpperCAmelCase            =       not args.not_finetuned and not args.is_seq_class
							convert_wavaveca_checkpoint(
							    args.checkpoint_path,
							    args.pytorch_dump_folder_path,
							    args.config_path,
							    args.dict_path,
							    is_finetuned,
							    args.is_seq_class,
							)










'''simple docstring'''




import os
import sys
import unittest


__UpperCAmelCase            =       os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, """utils"""))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
__UpperCAmelCase            =       os.path.join(git_repo_path, """src""", """transformers""")

__UpperCAmelCase            =       """
{0} = None
"""

__UpperCAmelCase            =       """
class {0}(metaclass=DummyObject):
    _backends = {1}

    def __init__(self, *args, **kwargs):
        requires_backends(self, {1})
"""


__UpperCAmelCase            =       """
def {0}(*args, **kwargs):
    requires_backends({0}, {1})
"""




class      UpperCamelCase__					(					unittest.TestCase			):




					"""simple docstring"""


					def lowerCamelCase_			(						self				:			List[str]						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Dict		=			find_backend("""    _import_structure[\"models.albert\"].append(\"AlbertTokenizerFast\")"""						)
							self.assertIsNone(lowerCamelCase_						)

							SCREAMING_SNAKE_CASE :				str		=			find_backend("""    if not is_tokenizers_available():"""						)
							self.assertEqual(lowerCamelCase_      ,						"""tokenizers"""						)

							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			find_backend("""    if not is_tensorflow_text_available():"""						)
							self.assertEqual(lowerCamelCase_      ,						"""tensorflow_text"""						)

							SCREAMING_SNAKE_CASE :				List[str]		=			find_backend("""    if not (is_sentencepiece_available() and is_tokenizers_available()):"""						)
							self.assertEqual(lowerCamelCase_      ,						"""sentencepiece_and_tokenizers"""						)

							SCREAMING_SNAKE_CASE :				Tuple		=			find_backend(
							    """    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"""						)
							self.assertEqual(lowerCamelCase_      ,						"""sentencepiece_and_tensorflow_text"""						)

							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			find_backend(
							    """    if not (is_sentencepiece_available() and is_tokenizers_available() and is_vision_available()):"""						)
							self.assertEqual(lowerCamelCase_      ,						"""sentencepiece_and_tokenizers_and_vision"""						)


					def lowerCamelCase_			(						self				:			Any						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Optional[Any]		=			read_init()
							# We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
							self.assertIn("""torch"""      ,						lowerCamelCase_						)
							self.assertIn("""tensorflow_text"""      ,						lowerCamelCase_						)
							self.assertIn("""sentencepiece_and_tokenizers"""      ,						lowerCamelCase_						)

							# Likewise, we can't assert on the exact content of a key
							self.assertIn("""BertModel"""      ,						objects["""torch"""]						)
							self.assertIn("""TFBertModel"""      ,						objects["""tf"""]						)
							self.assertIn("""FlaxBertModel"""      ,						objects["""flax"""]						)
							self.assertIn("""BertModel"""      ,						objects["""torch"""]						)
							self.assertIn("""TFBertTokenizer"""      ,						objects["""tensorflow_text"""]						)
							self.assertIn("""convert_slow_tokenizer"""      ,						objects["""sentencepiece_and_tokenizers"""]						)


					def lowerCamelCase_			(						self				:			int						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				str		=			create_dummy_object("""CONSTANT"""      ,						"""'torch'"""						)
							self.assertEqual(lowerCamelCase_      ,						"""\nCONSTANT = None\n"""						)

							SCREAMING_SNAKE_CASE :				List[Any]		=			create_dummy_object("""function"""      ,						"""'torch'"""						)
							self.assertEqual(
							    lowerCamelCase_      ,						"""\ndef function(*args, **kwargs):\n    requires_backends(function, 'torch')\n"""						)

							SCREAMING_SNAKE_CASE :				str		=			"""
class FakeClass(metaclass=DummyObject):
    _backends = 'torch'

    def __init__(self, *args, **kwargs):
        requires_backends(self, 'torch')
"""
							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			create_dummy_object("""FakeClass"""      ,						"""'torch'"""						)
							self.assertEqual(lowerCamelCase_      ,						lowerCamelCase_						)



					def lowerCamelCase_			(						self				:			int						):

							'''simple docstring'''



							SCREAMING_SNAKE_CASE :				Union[str, Any]		=			"""# This file is autogenerated by the command `make fix-copies`, do not edit.
from ..utils import DummyObject, requires_backends


CONSTANT = None


def function(*args, **kwargs):
    requires_backends(function, [\"torch\"])


class FakeClass(metaclass=DummyObject):
    _backends = [\"torch\"]

    def __init__(self, *args, **kwargs):
        requires_backends(self, [\"torch\"])
"""
							SCREAMING_SNAKE_CASE :				List[Any]		=			create_dummy_files({"""torch""": ["""CONSTANT""", """function""", """FakeClass"""]}						)
							self.assertEqual(dummy_files["""torch"""]      ,						lowerCamelCase_						)



