

'''simple docstring'''





import argparse
import json
import os

import fairseq
import torch
from torch import nn

from transformers import (
    SpeechaTextaConfig,
    SpeechaTextaForCausalLM,
    SpeechaTextaTokenizer,
    SpeechEncoderDecoderConfig,
    SpeechEncoderDecoderModel,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaModel,
    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.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""",
    """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 				_lowerCAmelCase			(   __magic_name__						:     List[str]   , __magic_name__						:     Optional[Any]   , __magic_name__						:     List[Any]   , __magic_name__						:     str   , __magic_name__						:     Tuple						)				->  List[Any]:
						for attribute in key.split('''.'''						):
												lowercase :							List[str]    		=getattr(__magic_name__   , __magic_name__						)

						if weight_type is not None:
												lowercase :							Union[str, Any]    		=getattr(__magic_name__   , __magic_name__						).shape
						else:
												lowercase :							Union[str, Any]    		=hf_pointer.shape

						assert hf_shape == value.shape, (
						    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":
												lowercase :							Tuple    		=value
						elif weight_type == "weight_g":
												lowercase :							Union[str, Any]    		=value
						elif weight_type == "weight_v":
												lowercase :							Any    		=value
						elif weight_type == "bias":
												lowercase :							Dict    		=value
						else:
												lowercase :							Union[str, Any]    		=value

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






def 				_lowerCAmelCase			(   __magic_name__						:     Optional[Any]   , __magic_name__						:     Union[str, Any]						)				->  Union[str, Any]:
						lowercase :							Optional[int]    		=[]
						lowercase :							Union[str, Any]    		=fairseq_model.state_dict()

						lowercase :							Tuple    		=hf_model.feature_extractor

						# if encoder has different dim to decoder -> use proj_weight
						lowercase :							Union[str, Any]    		=None

						for name, value in fairseq_dict.items():
												lowercase :							Any    		=False
												if "conv_layers" in name:
																		load_conv_layer(
																		    __magic_name__   , __magic_name__   , __magic_name__   , __magic_name__   , hf_model.config.feat_extract_norm == '''group'''   , )
																		lowercase :							Tuple    		=True
												elif name.split('''.'''						)[0] == "proj":
																		lowercase :							List[Any]    		=fairseq_model.proj
																		lowercase :							Union[str, Any]    		=True
												else:
																		for key, mapped_key in MAPPING.items():
																								if key in name or key.split('''w2v_model.'''						)[-1] == name.split('''.'''						)[0]:
																														lowercase :							Tuple    		=True
																														if "*" in mapped_key:
																																				lowercase :							Any    		=name.split(__magic_name__						)[0].split('''.'''						)[-2]
																																				lowercase :							Dict    		=mapped_key.replace('''*'''   , __magic_name__						)
																														if "weight_g" in name:
																																				lowercase :							Optional[int]    		='''weight_g'''
																														elif "weight_v" in name:
																																				lowercase :							Dict    		='''weight_v'''
																														elif "bias" in name:
																																				lowercase :							Optional[Any]    		='''bias'''
																														elif "weight" in name:
																																				lowercase :							Any    		='''weight'''
																														else:
																																				lowercase :							Dict    		=None
																														set_recursively(__magic_name__   , __magic_name__   , __magic_name__   , __magic_name__   , __magic_name__						)
																								continue
												if not is_used:
																		unused_weights.append(__magic_name__						)

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

						return proj_weight






def 				_lowerCAmelCase			(   __magic_name__						:     List[str]   , __magic_name__						:     Optional[int]   , __magic_name__						:     Optional[int]   , __magic_name__						:     Optional[int]   , __magic_name__						:     Any						)				->  str:
						lowercase :							str    		=full_name.split('''conv_layers.'''						)[-1]
						lowercase :							Optional[Any]    		=name.split('''.'''						)
						lowercase :							List[Any]    		=int(items[0]						)
						lowercase :							Optional[Any]    		=int(items[1]						)

						if type_id == 0:
												if "bias" in name:
																		assert value.shape == feature_extractor.conv_layers[layer_id].conv.bias.data.shape, (
																		    f'''{full_name} has size {value.shape}, but'''
																		    f''' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.'''
																		)
																		lowercase :							Any    		=value
																		logger.info(f'''Feat extract conv layer {layer_id} was initialized from {full_name}.'''						)
												elif "weight" in name:
																		assert value.shape == feature_extractor.conv_layers[layer_id].conv.weight.data.shape, (
																		    f'''{full_name} has size {value.shape}, but'''
																		    f''' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.'''
																		)
																		lowercase :							Optional[int]    		=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:
																		assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape, (
																		    f'''{full_name} has size {value.shape}, but {feature_extractor[layer_id].layer_norm.bias.data.shape} was'''
																		    " found."
																		)
																		lowercase :							Dict    		=value
																		logger.info(f'''Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'''						)
												elif "weight" in name:
																		assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape, (
																		    f'''{full_name} has size {value.shape}, but'''
																		    f''' {feature_extractor[layer_id].layer_norm.weight.data.shape} was found.'''
																		)
																		lowercase :							List[str]    		=value
																		logger.info(f'''Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'''						)
						else:
												unused_weights.append(__magic_name__						)






def 				_lowerCAmelCase			(   __magic_name__						:     str						)				->  Optional[int]:
						lowercase		,	lowercase :							List[Any]    		=emb.weight.shape
						lowercase :							Optional[Any]    		=nn.Linear(__magic_name__   , __magic_name__   , bias=__magic_name__						)
						lowercase :							Any    		=emb.weight.data
						return lin_layer






def 				_lowerCAmelCase			(   __magic_name__						:     int						)				->  Optional[int]:
						with open(__magic_name__   , '''r'''   , encoding='''utf-8'''						) as f:
												lowercase :							Dict    		=f.readlines()
												lowercase :							Optional[int]    		=[line.split(''' '''						)[0] for line in lines]

						lowercase :							List[Any]    		=len(__magic_name__						)

						lowercase :							List[str]    		={
						    '''<s>''': 0,
						    '''<pad>''': 1,
						    '''</s>''': 2,
						    '''<unk>''': 3,
						}

						vocab_dict.update(dict(zip(__magic_name__   , range(4   , num_words + 4						)						)						)						)
						return vocab_dict






@torch.no_grad()
def 				_lowerCAmelCase			(   __magic_name__						:     Union[str, Any]   , __magic_name__						:     Dict   , __magic_name__						:     Union[str, Any]   , __magic_name__						:     Optional[int]   , __magic_name__						:     List[Any]   , __magic_name__						:     List[Any]   , __magic_name__						:     int   , )				->  List[str]:
						lowercase :							List[str]    		=WavaVecaConfig.from_pretrained(__magic_name__						)
						lowercase :							int    		=SpeechaTextaConfig.from_pretrained(
						    __magic_name__   , vocab_size=__magic_name__   , decoder_layers=__magic_name__   , do_stable_layer_norm=__magic_name__						)

						lowercase :							Optional[int]    		=WavaVecaFeatureExtractor(
						    feature_size=1   , sampling_rate=16000   , padding_value=0   , do_normalize=__magic_name__   , return_attention_mask=__magic_name__   , )

						lowercase		,	lowercase		,	lowercase :							Dict    		=fairseq.checkpoint_utils.load_model_ensemble_and_task(
						    [checkpoint_path]   , arg_overrides={'''data''': '''/'''.join(dict_path.split('''/'''						)[:-1]						)}						)
						lowercase :							Union[str, Any]    		=model[0].eval()

						# set weights for wav2vec2 encoder
						lowercase :							Union[str, Any]    		=WavaVecaModel(__magic_name__						)
						lowercase :							Union[str, Any]    		=recursively_load_weights_wavaveca(model.encoder   , __magic_name__						)

						lowercase :							Optional[int]    		=SpeechaTextaForCausalLM(__magic_name__						)
						lowercase		,	lowercase :							Union[str, Any]    		=hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict()   , strict=__magic_name__						)

						# set output linear layer
						unexpected_keys.remove('''embed_out'''						)
						lowercase :							List[Any]    		=nn.Parameter(model.decoder.embed_out.detach()						)

						# layer norm is init to identity matrix so leaving it is fine
						logger.warning(f'''The following keys are missing when loading the decoder weights: {missing_keys}'''						)
						logger.warning(f'''The following keys are unexpected when loading the decoder weights: {unexpected_keys}'''						)

						lowercase :							Optional[Any]    		=SpeechEncoderDecoderModel(encoder=__magic_name__   , decoder=__magic_name__						)
						lowercase :							Tuple    		=False

						# add projection layer
						lowercase :							Union[str, Any]    		=nn.Parameter(projection_layer.weight						)
						lowercase :							Dict    		=nn.Parameter(projection_layer.bias						)

						lowercase :							str    		=create_vocab_dict(__magic_name__						)

						with open(os.path.join(__magic_name__   , '''vocab.json'''						)   , '''w'''						) as fp:
												json.dump(__magic_name__   , __magic_name__						)

						lowercase :							Dict    		=SpeechaTextaTokenizer(os.path.join(__magic_name__   , '''vocab.json'''						)						)
						tokenizer.save_pretrained(__magic_name__						)

						lowercase :							Optional[Any]    		=hf_wavavec.config.to_dict()
						lowercase :							List[Any]    		=tokenizer.pad_token_id
						lowercase :							Any    		=tokenizer.bos_token_id
						lowercase :							Any    		=tokenizer.eos_token_id
						lowercase :							int    		='''speech_to_text_2'''
						lowercase :							int    		='''wav2vec2'''

						lowercase :							Optional[Any]    		=SpeechEncoderDecoderConfig.from_dict(__magic_name__						)

						hf_wavavec.save_pretrained(__magic_name__						)
						feature_extractor.save_pretrained(__magic_name__						)


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(
							    """--encoder_config_path""",
							    default="""facebook/wav2vec2-large-lv60""",
							    type=str,
							    help="""Path to hf encoder wav2vec2 checkpoint config""",
							)
							parser.add_argument(
							    """--decoder_config_path""",
							    default="""facebook/s2t-small-mustc-en-fr-st""",
							    type=str,
							    help="""Path to hf decoder s2t checkpoint config""",
							)
							parser.add_argument("""--vocab_size""", default=10224, type=int, help="""Vocab size of decoder""")
							parser.add_argument("""--num_decoder_layers""", default=7, type=int, help="""Number of decoder layers""")

							UpperCamelCase_    						=       parser.parse_args()
							convert_wavaveca_checkpoint(
							    args.checkpoint_path,
							    args.pytorch_dump_folder_path,
							    args.dict_path,
							    encoder_config_path=args.encoder_config_path,
							    decoder_config_path=args.decoder_config_path,
							    vocab_size=args.vocab_size,
							    num_decoder_layers=args.num_decoder_layers,
							)








'''simple docstring'''





# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import doctest
import sys
import warnings
from os.path import abspath, dirname, join

import _pytest

from transformers.testing_utils import HfDoctestModule, HfDocTestParser


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
UpperCamelCase_    						=       abspath(join(dirname(__file__), """src"""))
sys.path.insert(1, git_repo_path)

# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action="""ignore""", category=FutureWarning)






def 				_lowerCAmelCase			(   __magic_name__						:     int						)				->  Tuple:
						config.addinivalue_line(
						    '''markers'''   , '''is_pt_tf_cross_test: mark test to run only when PT and TF interactions are tested'''						)
						config.addinivalue_line(
						    '''markers'''   , '''is_pt_flax_cross_test: mark test to run only when PT and FLAX interactions are tested'''						)
						config.addinivalue_line('''markers'''   , '''is_pipeline_test: mark test to run only when pipelines are tested'''						)
						config.addinivalue_line('''markers'''   , '''is_staging_test: mark test to run only in the staging environment'''						)
						config.addinivalue_line('''markers'''   , '''accelerate_tests: mark test that require accelerate'''						)
						config.addinivalue_line('''markers'''   , '''tool_tests: mark the tool tests that are run on their specific schedule'''						)






def 				_lowerCAmelCase			(   __magic_name__						:     int						)				->  Any:
						from transformers.testing_utils import pytest_addoption_shared

						pytest_addoption_shared(__magic_name__						)






def 				_lowerCAmelCase			(   __magic_name__						:     Any						)				->  Any:
						from transformers.testing_utils import pytest_terminal_summary_main

						lowercase :							Optional[Any]    		=terminalreporter.config.getoption('''--make-reports'''						)
						if make_reports:
												pytest_terminal_summary_main(__magic_name__   , id=__magic_name__						)






def 				_lowerCAmelCase			(   __magic_name__						:     Tuple   , __magic_name__						:     Union[str, Any]						)				->  List[str]:
						# If no tests are collected, pytest exists with code 5, which makes the CI fail.
						if exitstatus == 5:
												lowercase :							Optional[int]    		=0


# Doctest custom flag to ignore output.
UpperCamelCase_    						=       doctest.register_optionflag("""IGNORE_RESULT""")

UpperCamelCase_    						=       doctest.OutputChecker





class 		__SCREAMING_SNAKE_CASE							(			lowercase__      ):







				def  lowerCamelCase_	(       self     :		Optional[Any]		,					UpperCAmelCase__     :		Union[str, Any]		,					UpperCAmelCase__     :		List[Any]		,					UpperCAmelCase__     :		List[Any]   ):
										'''simple docstring'''
										if IGNORE_RESULT & optionflags:
																return True
										return OutputChecker.check_output(self		,					UpperCAmelCase__		,					UpperCAmelCase__		,					UpperCAmelCase__   )



UpperCamelCase_    						=       CustomOutputChecker
UpperCamelCase_    						=       HfDoctestModule







UpperCamelCase_    						=       HfDocTestParser





