





import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

# Register SEW's fairseq modules
from sew_asapp import tasks  # noqa: F401

from transformers import (
    SEWConfig,
    SEWForCTC,
    SEWModel,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
SCREAMING_SNAKE_CASE__							:      Any         =    logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__							:      List[Any]         =    {
    """post_extract_proj""": """feature_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.upsample.0""": """encoder.upsample.projection""",
    """encoder.layer_norm""": """encoder.layer_norm""",
    """w2v_model.layer_norm""": """layer_norm""",
    """w2v_encoder.proj""": """lm_head""",
    """mask_emb""": """masked_spec_embed""",
}






def    _A	(   lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase ):
   for attribute in key.split("." ):
      a__						:    Optional[int]   		=	getattr(lowerCamelCase				,			lowerCamelCase )

   if weight_type is not None:
      a__						:    Any   		=	getattr(lowerCamelCase				,			lowerCamelCase ).shape
   else:
      a__						:    Optional[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":
      a__						:    Union[str, Any]   		=	value
   elif weight_type == "weight_g":
      a__						:    Union[str, Any]   		=	value
   elif weight_type == "weight_v":
      a__						:    List[str]   		=	value
   elif weight_type == "bias":
      a__						:    Optional[Any]   		=	value
   else:
      a__						:    List[str]   		=	value

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






def    _A	(   lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase ):
   a__						:    List[Any]   		=	[]
   a__						:    str   		=	fairseq_model.state_dict()

   a__						:    Optional[int]   		=	hf_model.sew.feature_extractor if is_finetuned else hf_model.feature_extractor

   for name, value in fairseq_dict.items():
      a__						:    int   		=	False
      if "conv_layers" in name:
         load_conv_layer(
             lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase				,			hf_model.config.feat_extract_norm == "group"				,			)
         a__						:    Any   		=	True
      else:
         for key, mapped_key in MAPPING.items():
            a__						:    Union[str, Any]   		=	"sew." + mapped_key if (is_finetuned and mapped_key != "lm_head") else mapped_key

            if key in name or key.split("w2v_model." )[-1] == name.split("." )[0]:
               a__						:    Optional[Any]   		=	True
               if "*" in mapped_key:
                  a__						:    Optional[int]   		=	name.split(lowerCamelCase )[0].split("." )[-2]
                  a__						:    List[Any]   		=	mapped_key.replace("*"				,			lowerCamelCase )
               if "weight_g" in name:
                  a__						:    Union[str, Any]   		=	"weight_g"
               elif "weight_v" in name:
                  a__						:    int   		=	"weight_v"
               elif "weight" in name:
                  a__						:    Dict   		=	"weight"
               elif "bias" in name:
                  a__						:    Optional[int]   		=	"bias"
               else:
                  a__						:    str   		=	None
               set_recursively(lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase )
            continue
      if not is_used:
         unused_weights.append(lowerCamelCase )

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






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






def    _A	(   lowerCamelCase				,			lowerCamelCase ):
   a__						:    Dict   		=	SEWConfig()
   if is_finetuned:
      a__						:    Optional[Any]   		=	model.wav_encoder.wav_model.cfg
   else:
      a__						:    int   		=	model.cfg

   a__						:    Any   		=	fs_config.conv_bias
   a__						:    List[str]   		=	eval(fs_config.conv_feature_layers )
   a__						:    Any   		=	[x[0] for x in conv_layers]
   a__						:    Union[str, Any]   		=	[x[1] for x in conv_layers]
   a__						:    Optional[Any]   		=	[x[2] for x in conv_layers]
   a__						:    Any   		=	"gelu"
   a__						:    Any   		=	"layer" if fs_config.extractor_mode == "layer_norm" else "group"
   a__						:    Tuple   		=	0.0
   a__						:    List[str]   		=	fs_config.activation_fn.name
   a__						:    List[str]   		=	fs_config.encoder_embed_dim
   a__						:    Tuple   		=	0.02
   a__						:    Optional[Any]   		=	fs_config.encoder_ffn_embed_dim
   a__						:    Optional[int]   		=	1E-5
   a__						:    Dict   		=	fs_config.encoder_layerdrop
   a__						:    Any   		=	fs_config.encoder_attention_heads
   a__						:    Any   		=	fs_config.conv_pos_groups
   a__						:    List[Any]   		=	fs_config.conv_pos
   a__						:    Tuple   		=	len(lowerCamelCase )
   a__						:    Optional[Any]   		=	fs_config.encoder_layers
   a__						:    str   		=	fs_config.squeeze_factor

   # take care of any params that are overridden by the Wav2VecCtc model
   if is_finetuned:
      a__						:    List[str]   		=	model.cfg
      a__						:    List[str]   		=	fs_config.final_dropout
      a__						:    int   		=	fs_config.layerdrop
   a__						:    int   		=	fs_config.activation_dropout
   a__						:    Tuple   		=	fs_config.mask_prob > 0 or fs_config.mask_channel_prob > 0
   a__						:    Tuple   		=	fs_config.attention_dropout
   a__						:    Tuple   		=	fs_config.dropout_input
   a__						:    int   		=	fs_config.dropout
   a__						:    int   		=	fs_config.mask_channel_length
   a__						:    int   		=	fs_config.mask_channel_prob
   a__						:    List[Any]   		=	fs_config.mask_length
   a__						:    Optional[Any]   		=	fs_config.mask_prob

   a__						:    Any   		=	"Wav2Vec2FeatureExtractor"
   a__						:    List[Any]   		=	"Wav2Vec2CTCTokenizer"

   return config






@torch.no_grad()
def    _A	(   lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase=None				,			lowerCamelCase=None				,			lowerCamelCase=True ):

   if is_finetuned:
      a__		,  a__		,  a__						:    Union[str, Any]   		=	fairseq.checkpoint_utils.load_model_ensemble_and_task(
          [checkpoint_path]				,			arg_overrides={"data": "/".join(dict_path.split("/" )[:-1] )} )
   else:
      a__		,  a__		,  a__						:    Tuple   		=	fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path] )

   if config_path is not None:
      a__						:    int   		=	SEWConfig.from_pretrained(lowerCamelCase )
   else:
      a__						:    List[str]   		=	convert_config(model[0]				,			lowerCamelCase )
   a__						:    str   		=	model[0].eval()

   a__						:    List[str]   		=	True if config.feat_extract_norm == "layer" else False
   a__						:    List[Any]   		=	WavaVecaFeatureExtractor(
       feature_size=1				,			sampling_rate=1_6000				,			padding_value=0				,			do_normalize=lowerCamelCase				,			return_attention_mask=lowerCamelCase				,			)

   if is_finetuned:
      if dict_path:
         a__						:    Optional[int]   		=	Dictionary.load(lowerCamelCase )

         # 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__						:    Union[str, Any]   		=	target_dict.bos_index
         a__						:    Any   		=	target_dict.pad_index
         a__						:    int   		=	target_dict.bos_index
         a__						:    str   		=	target_dict.eos_index
         a__						:    Union[str, Any]   		=	len(target_dict.symbols )
         a__						:    str   		=	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 )
         with open(lowerCamelCase				,			"w"				,			encoding="utf-8" ) as vocab_handle:
            json.dump(target_dict.indices				,			lowerCamelCase )
         a__						:    Union[str, Any]   		=	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				,			)
         a__						:    Dict   		=	WavaVecaProcessor(feature_extractor=lowerCamelCase				,			tokenizer=lowerCamelCase )
         processor.save_pretrained(lowerCamelCase )

      a__						:    Optional[int]   		=	SEWForCTC(lowerCamelCase )
   else:
      a__						:    str   		=	SEWModel(lowerCamelCase )
      feature_extractor.save_pretrained(lowerCamelCase )

   recursively_load_weights(lowerCamelCase				,			lowerCamelCase				,			lowerCamelCase )

   hf_model.save_pretrained(lowerCamelCase )


if __name__ == "__main__":
  SCREAMING_SNAKE_CASE__							:      List[Any]         =    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(
      """--is_finetuned""", action="""store_true""", help="""Whether the model to convert is a fine-tuned model or not"""
  )
  SCREAMING_SNAKE_CASE__							:      Optional[Any]         =    parser.parse_args()
  convert_sew_checkpoint(
      args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, args.is_finetuned
  )








from collections import OrderedDict
from typing import TYPE_CHECKING, Any, Mapping, Optional, Union

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxSeqaSeqConfigWithPast
from ...utils import logging


if TYPE_CHECKING:
  from ...feature_extraction_utils import FeatureExtractionMixin
  from ...tokenization_utils_base import PreTrainedTokenizerBase
  from ...utils import TensorType

SCREAMING_SNAKE_CASE__							:      List[str]         =    logging.get_logger(__name__)

SCREAMING_SNAKE_CASE__							:      int         =    {
    """openai/whisper-base""": """https://huggingface.co/openai/whisper-base/resolve/main/config.json""",
}

# fmt: off
SCREAMING_SNAKE_CASE__							:      int         =    [
    1, 2, 7, 8, 9, 1_0, 1_4, 2_5,
    2_6, 2_7, 2_8, 2_9, 3_1, 5_8, 5_9, 6_0, 6_1, 6_2,
    6_3, 9_0, 9_1, 9_2, 9_3, 3_5_7, 3_6_6, 4_3_8, 5_3_2, 6_8_5,
    7_0_5, 7_9_6, 9_3_0, 1_0_5_8, 1_2_2_0, 1_2_6_7, 1_2_7_9, 1_3_0_3, 1_3_4_3, 1_3_7_7,
    1_3_9_1, 1_6_3_5, 1_7_8_2, 1_8_7_5, 2_1_6_2, 2_3_6_1, 2_4_8_8, 3_4_6_7, 4_0_0_8, 4_2_1_1,
    4_6_0_0, 4_8_0_8, 5_2_9_9, 5_8_5_5, 6_3_2_9, 7_2_0_3, 9_6_0_9, 9_9_5_9, 1_0_5_6_3, 1_0_7_8_6,
    1_1_4_2_0, 1_1_7_0_9, 1_1_9_0_7, 1_3_1_6_3, 1_3_6_9_7, 1_3_7_0_0, 1_4_8_0_8, 1_5_3_0_6, 1_6_4_1_0, 1_6_7_9_1,
    1_7_9_9_2, 1_9_2_0_3, 1_9_5_1_0, 2_0_7_2_4, 2_2_3_0_5, 2_2_9_3_5, 2_7_0_0_7, 3_0_1_0_9, 3_0_4_2_0, 3_3_4_0_9,
    3_4_9_4_9, 4_0_2_8_3, 4_0_4_9_3, 4_0_5_4_9, 4_7_2_8_2, 4_9_1_4_6, 5_0_2_5_7, 5_0_3_5_9, 5_0_3_6_0, 5_0_3_6_1
]
SCREAMING_SNAKE_CASE__							:      List[str]         =    [
    1, 2, 7, 8, 9, 1_0, 1_4, 2_5,
    2_6, 2_7, 2_8, 2_9, 3_1, 5_8, 5_9, 6_0, 6_1, 6_2,
    6_3, 9_0, 9_1, 9_2, 9_3, 3_5_9, 5_0_3, 5_2_2, 5_4_2, 8_7_3,
    8_9_3, 9_0_2, 9_1_8, 9_2_2, 9_3_1, 1_3_5_0, 1_8_5_3, 1_9_8_2, 2_4_6_0, 2_6_2_7,
    3_2_4_6, 3_2_5_3, 3_2_6_8, 3_5_3_6, 3_8_4_6, 3_9_6_1, 4_1_8_3, 4_6_6_7, 6_5_8_5, 6_6_4_7,
    7_2_7_3, 9_0_6_1, 9_3_8_3, 1_0_4_2_8, 1_0_9_2_9, 1_1_9_3_8, 1_2_0_3_3, 1_2_3_3_1, 1_2_5_6_2, 1_3_7_9_3,
    1_4_1_5_7, 1_4_6_3_5, 1_5_2_6_5, 1_5_6_1_8, 1_6_5_5_3, 1_6_6_0_4, 1_8_3_6_2, 1_8_9_5_6, 2_0_0_7_5, 2_1_6_7_5,
    2_2_5_2_0, 2_6_1_3_0, 2_6_1_6_1, 2_6_4_3_5, 2_8_2_7_9, 2_9_4_6_4, 3_1_6_5_0, 3_2_3_0_2, 3_2_4_7_0, 3_6_8_6_5,
    4_2_8_6_3, 4_7_4_2_5, 4_9_8_7_0, 5_0_2_5_4, 5_0_2_5_8, 5_0_3_6_0, 5_0_3_6_1, 5_0_3_6_2
]







class 	__lowerCAmelCase      (       _UpperCamelCase	):
   _UpperCamelCase			:    Any  		=					"""whisper"""
   _UpperCamelCase			:    Union[str, Any]  		=					["""past_key_values"""]
   _UpperCamelCase			:    Union[str, Any]  		=					{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}





   def __init__(			self       ,      snake_case=51_865       ,      snake_case=80       ,      snake_case=6       ,      snake_case=4       ,      snake_case=6       ,      snake_case=4       ,      snake_case=1_536       ,      snake_case=1_536       ,      snake_case=0.0       ,      snake_case=0.0       ,      snake_case=50_257       ,      snake_case=True       ,      snake_case=True       ,      snake_case="gelu"       ,      snake_case=256       ,      snake_case=0.0       ,      snake_case=0.0       ,      snake_case=0.0       ,      snake_case=0.02       ,      snake_case=False       ,      snake_case=1_500       ,      snake_case=448       ,      snake_case=50_256       ,      snake_case=50_256       ,      snake_case=50_256       ,      snake_case=None       ,      snake_case=[220, 50_256]       ,      snake_case=False       ,      snake_case=256       ,      snake_case=False       ,      snake_case=0.05       ,      snake_case=10       ,      snake_case=2       ,      snake_case=0.0       ,      snake_case=10       ,      snake_case=0       ,      snake_case=7       ,      **snake_case       ,      )					->	Dict:
      """simple docstring"""





      a__						:    Optional[Any]   		=	vocab_size
      a__						:    int   		=	num_mel_bins
      a__						:    Dict   		=	d_model
      a__						:    List[Any]   		=	encoder_layers
      a__						:    List[Any]   		=	encoder_attention_heads
      a__						:    Optional[int]   		=	decoder_layers
      a__						:    int   		=	decoder_attention_heads
      a__						:    Optional[Any]   		=	decoder_ffn_dim
      a__						:    List[Any]   		=	encoder_ffn_dim
      a__						:    int   		=	dropout
      a__						:    Optional[int]   		=	attention_dropout
      a__						:    Tuple   		=	activation_dropout
      a__						:    Optional[Any]   		=	activation_function
      a__						:    List[Any]   		=	init_std
      a__						:    List[Any]   		=	encoder_layerdrop
      a__						:    Dict   		=	decoder_layerdrop
      a__						:    List[Any]   		=	use_cache
      a__						:    Union[str, Any]   		=	encoder_layers
      a__						:    Union[str, Any]   		=	scale_embedding  # scale factor will be sqrt(d_model) if True
      a__						:    Tuple   		=	max_source_positions
      a__						:    Optional[Any]   		=	max_target_positions

      # Audio Classification-specific parameters. Feel free to ignore for other classes.
      a__						:    Optional[Any]   		=	classifier_proj_size
      a__						:    int   		=	use_weighted_layer_sum

      # fine-tuning config parameters for SpecAugment: https://arxiv.org/abs/1904.08779
      a__						:    Tuple   		=	apply_spec_augment
      a__						:    int   		=	mask_time_prob
      a__						:    Optional[Any]   		=	mask_time_length
      a__						:    List[str]   		=	mask_time_min_masks
      a__						:    List[str]   		=	mask_feature_prob
      a__						:    Dict   		=	mask_feature_length
      a__						:    Any   		=	mask_feature_min_masks

      a__						:    List[str]   		=	median_filter_width

      super().__init__(
          pad_token_id=snake_case       ,      bos_token_id=snake_case       ,      eos_token_id=snake_case       ,      is_encoder_decoder=snake_case       ,      decoder_start_token_id=snake_case       ,      suppress_tokens=snake_case       ,      begin_suppress_tokens=snake_case       ,      **snake_case       ,      )







class 	__lowerCAmelCase      (       _UpperCamelCase	):





   @property
   def 						_snake_case			(			self   )					->	Mapping[str, Mapping[int, str]]:
      """simple docstring"""





      a__						:    Dict   		=	OrderedDict(
          [
              ("input_features", {0: "batch", 1: "feature_size", 2: "encoder_sequence"}),
          ]   )
      if self.use_past:
         a__						:    Tuple   		=	{0: "batch"}
      else:
         a__						:    Union[str, Any]   		=	{0: "batch", 1: "decoder_sequence"}

      if self.use_past:
         self.fill_with_past_key_values_(snake_case       ,      direction="inputs"   )

      return common_inputs





   def 						_snake_case			(			self       ,      snake_case       ,      snake_case = -1       ,      snake_case = -1       ,      snake_case = False       ,      snake_case = None       ,      snake_case = 22_050       ,      snake_case = 5.0       ,      snake_case = 220       ,      )					->	Mapping[str, Any]:
      """simple docstring"""





      a__						:    int   		=	OrderedDict()
      a__						:    List[str]   		=	OnnxConfig.generate_dummy_inputs(
          self       ,      preprocessor=preprocessor.feature_extractor       ,      batch_size=snake_case       ,      framework=snake_case       ,      sampling_rate=snake_case       ,      time_duration=snake_case       ,      frequency=snake_case       ,      )
      a__						:    Optional[int]   		=	encoder_inputs["input_features"].shape[2]
      a__						:    str   		=	encoder_sequence_length // 2 if self.use_past else seq_length

      a__						:    Optional[int]   		=	super().generate_dummy_inputs(
          preprocessor.tokenizer       ,      snake_case       ,      snake_case       ,      snake_case       ,      snake_case   )

      a__						:    Any   		=	encoder_inputs.pop("input_features"   )
      a__						:    Dict   		=	decoder_inputs.pop("decoder_input_ids"   )

      if "past_key_values" in decoder_inputs:
         a__						:    Tuple   		=	decoder_inputs.pop("past_key_values"   )

      return dummy_inputs





   @property
   def 						_snake_case			(			self   )					->	float:
      """simple docstring"""





      return 1E-3

