





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

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



def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     )			->       List[Any]:


      '''simple docstring'''


      for attribute in key.split("""."""     ):
            SCREAMING_SNAKE_CASE		=						getattr(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     )

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

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



def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     )			->       Tuple:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						[]
      SCREAMING_SNAKE_CASE		=						fairseq_model.state_dict()

      SCREAMING_SNAKE_CASE		=						hf_model.feature_extractor

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

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

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

      return proj_weight



def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     )			->       List[str]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						full_name.split("""conv_layers."""     )[-1]
      SCREAMING_SNAKE_CASE		=						name.split("""."""     )
      SCREAMING_SNAKE_CASE		=						int(items[0]     )
      SCREAMING_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."""
                  )
                  SCREAMING_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."""
                  )
                  SCREAMING_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."
                  )
                  SCREAMING_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."""
                  )
                  SCREAMING_SNAKE_CASE		=						value
                  logger.info(F"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""     )
      else:
            unused_weights.append(_SCREAMING_SNAKE_CASE     )



def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       Optional[int]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE, SCREAMING_SNAKE_CASE		=						emb.weight.shape
      SCREAMING_SNAKE_CASE		=						nn.Linear(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						bias=_SCREAMING_SNAKE_CASE     )
      SCREAMING_SNAKE_CASE		=						emb.weight.data
      return lin_layer



def        __lowercase       (				_SCREAMING_SNAKE_CASE     )			->       Any:


      '''simple docstring'''


      with open(_SCREAMING_SNAKE_CASE			,						"""r"""			,						encoding="""utf-8"""     ) as f:
            SCREAMING_SNAKE_CASE		=						f.readlines()
            SCREAMING_SNAKE_CASE		=						[line.split(""" """     )[0] for line in lines]

      SCREAMING_SNAKE_CASE		=						len(_SCREAMING_SNAKE_CASE     )

      SCREAMING_SNAKE_CASE		=						{
          """<s>""": 0,
          """<pad>""": 1,
          """</s>""": 2,
          """<unk>""": 3,
      }

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



@torch.no_grad()
def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						)			->       List[str]:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE		=						WavaVecaConfig.from_pretrained(_SCREAMING_SNAKE_CASE     )
      SCREAMING_SNAKE_CASE		=						SpeechaTextaConfig.from_pretrained(
          _SCREAMING_SNAKE_CASE			,						vocab_size=_SCREAMING_SNAKE_CASE			,						decoder_layers=_SCREAMING_SNAKE_CASE			,						do_stable_layer_norm=_SCREAMING_SNAKE_CASE     )

      SCREAMING_SNAKE_CASE		=						WavaVecaFeatureExtractor(
          feature_size=1			,						sampling_rate=1_60_00			,						padding_value=0			,						do_normalize=_SCREAMING_SNAKE_CASE			,						return_attention_mask=_SCREAMING_SNAKE_CASE			,						)

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

      # set weights for wav2vec2 encoder
      SCREAMING_SNAKE_CASE		=						WavaVecaModel(_SCREAMING_SNAKE_CASE     )
      SCREAMING_SNAKE_CASE		=						recursively_load_weights_wavaveca(model.encoder			,						_SCREAMING_SNAKE_CASE     )

      SCREAMING_SNAKE_CASE		=						SpeechaTextaForCausalLM(_SCREAMING_SNAKE_CASE     )
      SCREAMING_SNAKE_CASE, SCREAMING_SNAKE_CASE		=						hf_decoder.model.decoder.load_state_dict(model.decoder.state_dict()			,						strict=_SCREAMING_SNAKE_CASE     )

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

      SCREAMING_SNAKE_CASE		=						SpeechEncoderDecoderModel(encoder=_SCREAMING_SNAKE_CASE			,						decoder=_SCREAMING_SNAKE_CASE     )
      SCREAMING_SNAKE_CASE		=						False

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

      SCREAMING_SNAKE_CASE		=						create_vocab_dict(_SCREAMING_SNAKE_CASE     )

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

      SCREAMING_SNAKE_CASE		=						SpeechaTextaTokenizer(os.path.join(_SCREAMING_SNAKE_CASE			,						"""vocab.json"""     )     )
      tokenizer.save_pretrained(_SCREAMING_SNAKE_CASE     )

      SCREAMING_SNAKE_CASE		=						hf_wavavec.config.to_dict()
      SCREAMING_SNAKE_CASE		=						tokenizer.pad_token_id
      SCREAMING_SNAKE_CASE		=						tokenizer.bos_token_id
      SCREAMING_SNAKE_CASE		=						tokenizer.eos_token_id
      SCREAMING_SNAKE_CASE		=						"""speech_to_text_2"""
      SCREAMING_SNAKE_CASE		=						"""wav2vec2"""

      SCREAMING_SNAKE_CASE		=						SpeechEncoderDecoderConfig.from_dict(_SCREAMING_SNAKE_CASE     )

      hf_wavavec.save_pretrained(_SCREAMING_SNAKE_CASE     )
      feature_extractor.save_pretrained(_SCREAMING_SNAKE_CASE     )


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

      SCREAMING_SNAKE_CASE_						       =	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,
      )









from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


SCREAMING_SNAKE_CASE_						       =	logging.get_logger(__name__)


SCREAMING_SNAKE_CASE_						       =	TypeVar("""DatasetType""", Dataset, IterableDataset)



def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = "first_exhausted"			,						)			->       DatasetType:


      '''simple docstring'''


      from .arrow_dataset import Dataset
      from .iterable_dataset import IterableDataset

      if not datasets:
            raise ValueError("""Unable to interleave an empty list of datasets."""     )
      for i, dataset in enumerate(_SCREAMING_SNAKE_CASE     ):
            if not isinstance(_SCREAMING_SNAKE_CASE			,						(Dataset, IterableDataset)     ):
                  if isinstance(_SCREAMING_SNAKE_CASE			,						(DatasetDict, IterableDatasetDict)     ):
                        if not dataset:
                              raise ValueError(
                                  F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
                                  """is an empty dataset dictionary."""     )
                        raise ValueError(
                            F"""Dataset at position {i} has at least one split: {list(_SCREAMING_SNAKE_CASE     )}\n"""
                            F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(_SCREAMING_SNAKE_CASE     )     )}']"""     )
                  raise ValueError(
                      F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(_SCREAMING_SNAKE_CASE     ).__name__}."""     )
            if i == 0:
                  SCREAMING_SNAKE_CASE, SCREAMING_SNAKE_CASE		=						(
                      (Dataset, IterableDataset) if isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ):
                  raise ValueError(
                      F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""     )
      if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
            raise ValueError(F"""{stopping_strategy} is not supported. Please enter a valid stopping_strategy."""     )
      if dataset_type is Dataset:
            return _interleave_map_style_datasets(
                _SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						stopping_strategy=_SCREAMING_SNAKE_CASE     )
      else:
            return _interleave_iterable_datasets(
                _SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						stopping_strategy=_SCREAMING_SNAKE_CASE     )



def        __lowercase       (				_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = None			,						_SCREAMING_SNAKE_CASE = 0			,						)			->       DatasetType:


      '''simple docstring'''


      if not dsets:
            raise ValueError("""Unable to concatenate an empty list of datasets."""     )
      for i, dataset in enumerate(_SCREAMING_SNAKE_CASE     ):
            if not isinstance(_SCREAMING_SNAKE_CASE			,						(Dataset, IterableDataset)     ):
                  if isinstance(_SCREAMING_SNAKE_CASE			,						(DatasetDict, IterableDatasetDict)     ):
                        if not dataset:
                              raise ValueError(
                                  F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
                                  """is an empty dataset dictionary."""     )
                        raise ValueError(
                            F"""Dataset at position {i} has at least one split: {list(_SCREAMING_SNAKE_CASE     )}\n"""
                            F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(_SCREAMING_SNAKE_CASE     )     )}']"""     )
                  raise ValueError(
                      F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(_SCREAMING_SNAKE_CASE     ).__name__}."""     )
            if i == 0:
                  SCREAMING_SNAKE_CASE, SCREAMING_SNAKE_CASE		=						(
                      (Dataset, IterableDataset) if isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ) else (IterableDataset, Dataset)
                  )
            elif not isinstance(_SCREAMING_SNAKE_CASE			,						_SCREAMING_SNAKE_CASE     ):
                  raise ValueError(
                      F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""     )
      if dataset_type is Dataset:
            return _concatenate_map_style_datasets(_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						axis=_SCREAMING_SNAKE_CASE     )
      else:
            return _concatenate_iterable_datasets(_SCREAMING_SNAKE_CASE			,						info=_SCREAMING_SNAKE_CASE			,						split=_SCREAMING_SNAKE_CASE			,						axis=_SCREAMING_SNAKE_CASE     )


