





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

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

def    __UpperCamelCase	(	a    : List[str]					,   a    : List[str]					,   a    : str					,   a    : Union[str, Any]					,   a    : Optional[Any]     ) ->Tuple:
	for attribute in key.split('''.'''     ):
		snake_case				=			getattr(a_					,   a_     )

	if weight_type is not None:
		snake_case				=			getattr(a_					,   a_     ).shape
	else:
		snake_case				=			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":
		snake_case				=			value
	elif weight_type == "weight_g":
		snake_case				=			value
	elif weight_type == "weight_v":
		snake_case				=			value
	elif weight_type == "bias":
		snake_case				=			value
	else:
		snake_case				=			value

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

def    __UpperCamelCase	(	a    : Optional[Any]					,   a    : int     ) ->Any:
	snake_case				=			[]
	snake_case				=			fairseq_model.state_dict()

	snake_case				=			hf_model.feature_extractor

	# if encoder has different dim to decoder -> use proj_weight
	snake_case				=			None

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

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

	return proj_weight

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

def    __UpperCamelCase	(	a    : Tuple     ) ->Union[str, Any]:
	snake_case				=			emb.weight.shape
	snake_case				=			nn.Linear(a_					,   a_					,   bias=a_     )
	snake_case				=			emb.weight.data
	return lin_layer

def    __UpperCamelCase	(	a    : Union[str, Any]     ) ->Optional[Any]:
	with open(a_					,   '''r'''					,   encoding='''utf-8'''     ) as f:
		snake_case				=			f.readlines()
		snake_case				=			[line.split(''' '''     )[0] for line in lines]

	snake_case				=			len(a_     )

	snake_case				=			{
	    '''<s>''': 0,
	    '''<pad>''': 1,
	    '''</s>''': 2,
	    '''<unk>''': 3,
	}

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

@torch.no_grad()
def    __UpperCamelCase	(	a    : List[Any]					,   a    : Dict					,   a    : Optional[int]					,   a    : str					,   a    : Tuple					,   a    : Any					,   a    : List[str]					,   ) ->Union[str, Any]:
	snake_case				=			WavaVecaConfig.from_pretrained(a_     )
	snake_case				=			SpeechaTextaConfig.from_pretrained(
	    a_					,   vocab_size=a_					,   decoder_layers=a_					,   do_stable_layer_norm=a_     )

	snake_case				=			WavaVecaFeatureExtractor(
	    feature_size=1					,   sampling_rate=1_6000					,   padding_value=0					,   do_normalize=a_					,   return_attention_mask=a_					,   )

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

	# set weights for wav2vec2 encoder
	snake_case				=			WavaVecaModel(a_     )
	snake_case				=			recursively_load_weights_wavaveca(model.encoder					,   a_     )

	snake_case				=			SpeechaTextaForCausalLM(a_     )
	snake_case				=			hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict()					,   strict=a_     )

	# set output linear layer
	unexpected_keys.remove('''embed_out'''     )
	snake_case				=			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}"""     )

	snake_case				=			SpeechEncoderDecoderModel(encoder=a_					,   decoder=a_     )
	snake_case				=			False

	# add projection layer
	snake_case				=			nn.Parameter(projection_layer.weight     )
	snake_case				=			nn.Parameter(projection_layer.bias     )

	snake_case				=			create_vocab_dict(a_     )

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

	snake_case				=			SpeechaTextaTokenizer(os.path.join(a_					,   '''vocab.json'''     )     )
	tokenizer.save_pretrained(a_     )

	snake_case				=			hf_wavavec.config.to_dict()
	snake_case				=			tokenizer.pad_token_id
	snake_case				=			tokenizer.bos_token_id
	snake_case				=			tokenizer.eos_token_id
	snake_case				=			'''speech_to_text_2'''
	snake_case				=			'''wav2vec2'''

	snake_case				=			SpeechEncoderDecoderConfig.from_dict(a_     )

	hf_wavavec.save_pretrained(a_     )
	feature_extractor.save_pretrained(a_     )


if __name__ == "__main__":
			_lowercase       						=			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=10_224, type=int, help='Vocab size of decoder')
			parser.add_argument('--num_decoder_layers', default=7, type=int, help='Number of decoder layers')

			_lowercase       						=			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'''







from typing import Dict, List, Optional, Union

import numpy as np

from .feature_extraction_utils import BatchFeature, FeatureExtractionMixin
from .utils import PaddingStrategy, TensorType, is_tf_tensor, is_torch_tensor, logging, to_numpy


_lowercase       						=			logging.get_logger(__name__)
class 			_lowercase						(       __a					):







     def __init__( self    ,   A__    ,   A__    ,   A__    ,   **A__    ) ->	Union[str, Any]:
      snake_case				=			feature_size
      snake_case				=			sampling_rate
      snake_case				=			padding_value

      snake_case				=			kwargs.pop('''padding_side'''    ,   '''right'''    )
      snake_case				=			kwargs.pop('''return_attention_mask'''    ,   A__    )

      super().__init__(**A__    )







     def     UpperCamelCase			( self    ,   A__    ,   A__ = True    ,   A__ = None    ,   A__ = False    ,   A__ = None    ,   A__ = None    ,   A__ = None    ,   ) ->	BatchFeature:
      # If we have a list of dicts, let's convert it in a dict of lists
      # We do this to allow using this method as a collate_fn function in PyTorch Dataloader
      if isinstance(A__    ,   (list, tuple)    ) and isinstance(processed_features[0]    ,   (dict, BatchFeature)    ):
       snake_case				=			{
           key: [example[key] for example in processed_features] for key in processed_features[0].keys()
       }

      # The model's main input name, usually `input_values`, has be passed for padding
      if self.model_input_names[0] not in processed_features:
       raise ValueError(
           '''You should supply an instance of `transformers.BatchFeature` or list of `transformers.BatchFeature`'''
           F""" to this method that includes {self.model_input_names[0]}, but you provided"""
           F""" {list(processed_features.keys()    )}"""    )

      snake_case				=			processed_features[self.model_input_names[0]]
      snake_case				=			(
          return_attention_mask if return_attention_mask is not None else self.return_attention_mask
      )

      if len(A__    ) == 0:
       if return_attention_mask:
        snake_case				=			[]
       return processed_features

      # If we have PyTorch/TF tensors or lists as inputs, we cast them as Numpy arrays
      # and rebuild them afterwards if no return_tensors is specified
      # Note that we lose the specific device the tensor may be on for PyTorch

      snake_case				=			required_input[0]
      if isinstance(A__    ,   (list, tuple)    ):
       # first_element might be an empty list/tuple in some edge cases so we grab the first non empty element.
       snake_case				=			0
       while len(required_input[index]    ) == 0:
        index += 1
       if index < len(A__    ):
        snake_case				=			required_input[index][0]

      if return_tensors is None:
       if is_tf_tensor(A__    ):
        snake_case				=			'''tf'''
       elif is_torch_tensor(A__    ):
        snake_case				=			'''pt'''
       elif isinstance(A__    ,   (int, float, list, tuple, np.ndarray)    ):
        snake_case				=			'''np'''
       else:
        raise ValueError(
            F"""type of {first_element} unknown: {type(A__    )}. """
            '''Should be one of a python, numpy, pytorch or tensorflow object.'''    )

      for key, value in processed_features.items():
       if isinstance(value[0]    ,   (int, float)    ):
        snake_case				=			to_numpy(A__    )
       else:
        snake_case				=			[to_numpy(A__    ) for v in value]

        # Convert padding_strategy in PaddingStrategy
      snake_case				=			self._get_padding_strategies(padding=A__    ,   max_length=A__    )

      snake_case				=			processed_features[self.model_input_names[0]]

      snake_case				=			len(A__    )
      if not all(len(A__    ) == batch_size for v in processed_features.values()    ):
       raise ValueError('''Some items in the output dictionary have a different batch size than others.'''    )

      snake_case				=			[]
      for i in range(A__    ):
       snake_case				=			{k: v[i] for k, v in processed_features.items()}
       # truncation
       snake_case				=			self._truncate(
           A__    ,   max_length=A__    ,   pad_to_multiple_of=A__    ,   truncation=A__    ,   )
       truncated_inputs.append(A__    )

      if padding_strategy == PaddingStrategy.LONGEST:
       # make sure that `max_length` cannot be longer than the longest truncated length
       snake_case				=			max(len(input_slice[self.model_input_names[0]]    ) for input_slice in truncated_inputs    )
       snake_case				=			PaddingStrategy.MAX_LENGTH

      snake_case				=			{}
      for i in range(A__    ):
       # padding
       snake_case				=			self._pad(
           truncated_inputs[i]    ,   max_length=A__    ,   padding_strategy=A__    ,   pad_to_multiple_of=A__    ,   return_attention_mask=A__    ,   )

       for key, value in outputs.items():
        if key not in batch_outputs:
         snake_case				=			[]
        if value.dtype is np.dtype(np.floataa    ):
         snake_case				=			value.astype(np.floataa    )
        batch_outputs[key].append(A__    )

      return BatchFeature(A__    ,   tensor_type=A__    )







     def     UpperCamelCase			( self    ,   A__    ,   A__ = None    ,   A__ = PaddingStrategy.DO_NOT_PAD    ,   A__ = None    ,   A__ = None    ,   ) ->	dict:
      snake_case				=			processed_features[self.model_input_names[0]]

      if padding_strategy == PaddingStrategy.LONGEST:
       snake_case				=			len(A__    )

      if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
       snake_case				=			((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

      snake_case				=			padding_strategy != PaddingStrategy.DO_NOT_PAD and len(A__    ) < max_length

      if return_attention_mask and "attention_mask" not in processed_features:
       snake_case				=			np.ones(len(A__    )    ,   dtype=np.intaa    )

      if needs_to_be_padded:
       snake_case				=			max_length - len(A__    )
       if self.padding_side == "right":
        if return_attention_mask:
         snake_case				=			np.pad(
             processed_features['''attention_mask''']    ,   (0, difference)    )
        snake_case				=			((0, difference), (0, 0)) if self.feature_size > 1 else (0, difference)
        snake_case				=			np.pad(
            A__    ,   A__    ,   '''constant'''    ,   constant_values=self.padding_value    )
       elif self.padding_side == "left":
        if return_attention_mask:
         snake_case				=			np.pad(
             processed_features['''attention_mask''']    ,   (difference, 0)    )
        snake_case				=			((difference, 0), (0, 0)) if self.feature_size > 1 else (difference, 0)
        snake_case				=			np.pad(
            A__    ,   A__    ,   '''constant'''    ,   constant_values=self.padding_value    )
       else:
        raise ValueError('''Invalid padding strategy:''' + str(self.padding_side    )    )

      return processed_features







     def     UpperCamelCase			( self    ,   A__    ,   A__ = None    ,   A__ = None    ,   A__ = None    ,   ) ->	Union[str, Any]:
      if not truncation:
       return processed_features
      elif truncation and max_length is None:
       raise ValueError('''When setting ``truncation=True``, make sure that ``max_length`` is defined.'''    )

      snake_case				=			processed_features[self.model_input_names[0]]

      # find `max_length` that fits `pad_to_multiple_of`
      if max_length is not None and pad_to_multiple_of is not None and (max_length % pad_to_multiple_of != 0):
       snake_case				=			((max_length // pad_to_multiple_of) + 1) * pad_to_multiple_of

      snake_case				=			len(A__    ) > max_length

      if needs_to_be_truncated:
       snake_case				=			processed_features[self.model_input_names[0]][:max_length]
       if "attention_mask" in processed_features:
        snake_case				=			processed_features['''attention_mask'''][:max_length]

      return processed_features







     def     UpperCamelCase			( self    ,   A__=False    ,   A__=None    ) ->	Union[str, Any]:

      # Get padding strategy
      if padding is not False:
       if padding is True:
        snake_case				=			PaddingStrategy.LONGEST  # Default to pad to the longest sequence in the batch
       elif not isinstance(A__    ,   A__    ):
        snake_case				=			PaddingStrategy(A__    )
       elif isinstance(A__    ,   A__    ):
        snake_case				=			padding
      else:
       snake_case				=			PaddingStrategy.DO_NOT_PAD

      # Set max length if needed
      if max_length is None:
       if padding_strategy == PaddingStrategy.MAX_LENGTH:
        raise ValueError(
            F"""When setting ``padding={PaddingStrategy.MAX_LENGTH}``, make sure that max_length is defined"""    )

        # Test if we have a padding value
      if padding_strategy != PaddingStrategy.DO_NOT_PAD and (self.padding_value is None):
       raise ValueError(
           '''Asking to pad but the feature_extractor does not have a padding value. Please select a value to use'''
           ''' as `padding_value`. For example: `feature_extractor.padding_value = 0.0`.'''    )

      return padding_strategy

