


import logging
import os
from dataclasses import dataclass
from enum import Enum
from typing import List, Optional, Union

from filelock import FileLock

from transformers import PreTrainedTokenizer, is_tf_available, is_torch_available


__UpperCamelCase    :						Any     =							logging.getLogger(__name__)



@dataclass
class 							_UpperCamelCase :

   '''simple docstring'''







   a_				:							str
   a_				:							List[str]
   a_				:							Optional[List[str]]



@dataclass
class 							_UpperCamelCase :

   '''simple docstring'''







   a_				:							List[int]
   a_				:							List[int]
   a_				:							Optional[List[int]]    = None
   a_				:							Optional[List[int]]    = None



class 							_UpperCamelCase (						A						):

   '''simple docstring'''







   a_				:							Optional[Any]    = "train"
   a_				:							Optional[int]    = "dev"
   a_				:							List[str]    = "test"



class 							_UpperCamelCase :

   '''simple docstring'''







   @staticmethod
   def   _snake_case    (  _lowerCamelCase  : Any       ,  _lowerCamelCase  : Union[Split, str]							):






         '''simple docstring'''



         raise NotImplementedError

   @staticmethod
   def   _snake_case    (  _lowerCamelCase  : str							):






         '''simple docstring'''



         raise NotImplementedError


   @staticmethod
   def   _snake_case    (  _lowerCamelCase  : List[InputExample]       ,  _lowerCamelCase  : List[str]       ,  _lowerCamelCase  : int       ,  _lowerCamelCase  : PreTrainedTokenizer       ,  _lowerCamelCase  : List[str]=False       ,  _lowerCamelCase  : Any="[CLS]"       ,  _lowerCamelCase  : Tuple=1       ,  _lowerCamelCase  : Any="[SEP]"       ,  _lowerCamelCase  : Optional[int]=False       ,  _lowerCamelCase  : Optional[int]=False       ,  _lowerCamelCase  : str=0       ,  _lowerCamelCase  : List[Any]=0       ,  _lowerCamelCase  : int=-1_0_0       ,  _lowerCamelCase  : Any=0       ,  _lowerCamelCase  : Optional[Any]=True       ,  ):






         '''simple docstring'''



         __lowerCamelCase     :      int       						=						{label: i for i, label in enumerate(_lowerCamelCase							)}

         __lowerCamelCase     :      Dict       						=						[]
         for ex_index, example in enumerate(_lowerCamelCase							):
               if ex_index % 1_0_0_0_0 == 0:
                     logger.info("""Writing example %d of %d"""       ,  _lowerCamelCase       ,  len(_lowerCamelCase							)							)

               __lowerCamelCase     :      List[Any]       						=						[]
               __lowerCamelCase     :      Dict       						=						[]
               for word, label in zip(example.words       ,  example.labels							):
                     __lowerCamelCase     :      List[Any]       						=						tokenizer.tokenize(_lowerCamelCase							)

                     # bert-base-multilingual-cased sometimes output "nothing ([]) when calling tokenize with just a space.
                     if len(_lowerCamelCase							) > 0:
                           tokens.extend(_lowerCamelCase							)
                           # Use the real label id for the first token of the word, and padding ids for the remaining tokens
                           label_ids.extend([label_map[label]] + [pad_token_label_id] * (len(_lowerCamelCase							) - 1)							)

            # Account for [CLS] and [SEP] with "- 2" and with "- 3" for RoBERTa.
               __lowerCamelCase     :      str       						=						tokenizer.num_special_tokens_to_add()
               if len(_lowerCamelCase							) > max_seq_length - special_tokens_count:
                     __lowerCamelCase     :      Optional[Any]       						=						tokens[: (max_seq_length - special_tokens_count)]
                     __lowerCamelCase     :      str       						=						label_ids[: (max_seq_length - special_tokens_count)]

               # The convention in BERT is:
               # (a) For sequence pairs:
               #  tokens:   [CLS] is this jack ##son ##ville ? [SEP] no it is not . [SEP]
               #  type_ids:   0   0  0    0    0     0       0   0   1  1  1  1   1   1
               # (b) For single sequences:
               #  tokens:   [CLS] the dog is hairy . [SEP]
               #  type_ids:   0   0   0   0  0     0   0
               #
               # Where "type_ids" are used to indicate whether this is the first
               # sequence or the second sequence. The embedding vectors for `type=0` and
               # `type=1` were learned during pre-training and are added to the wordpiece
               # embedding vector (and position vector). This is not *strictly* necessary
               # since the [SEP] token unambiguously separates the sequences, but it makes
               # it easier for the model to learn the concept of sequences.
               #
               # For classification tasks, the first vector (corresponding to [CLS]) is
               # used as the "sentence vector". Note that this only makes sense because
               # the entire model is fine-tuned.
               tokens += [sep_token]
               label_ids += [pad_token_label_id]
               if sep_token_extra:
                     # roberta uses an extra separator b/w pairs of sentences
                     tokens += [sep_token]
                     label_ids += [pad_token_label_id]
               __lowerCamelCase     :      Any       						=						[sequence_a_segment_id] * len(_lowerCamelCase							)

               if cls_token_at_end:
                     tokens += [cls_token]
                     label_ids += [pad_token_label_id]
                     segment_ids += [cls_token_segment_id]
               else:
                     __lowerCamelCase     :      int       						=						[cls_token] + tokens
                     __lowerCamelCase     :      int       						=						[pad_token_label_id] + label_ids
                     __lowerCamelCase     :      Dict       						=						[cls_token_segment_id] + segment_ids

               __lowerCamelCase     :      Dict       						=						tokenizer.convert_tokens_to_ids(_lowerCamelCase							)

               # The mask has 1 for real tokens and 0 for padding tokens. Only real
               # tokens are attended to.
               __lowerCamelCase     :      Tuple       						=						[1 if mask_padding_with_zero else 0] * len(_lowerCamelCase							)

               # Zero-pad up to the sequence length.
               __lowerCamelCase     :      List[Any]       						=						max_seq_length - len(_lowerCamelCase							)
               if pad_on_left:
                     __lowerCamelCase     :      Optional[int]       						=						([pad_token] * padding_length) + input_ids
                     __lowerCamelCase     :      Optional[Any]       						=						([0 if mask_padding_with_zero else 1] * padding_length) + input_mask
                     __lowerCamelCase     :      Tuple       						=						([pad_token_segment_id] * padding_length) + segment_ids
                     __lowerCamelCase     :      Tuple       						=						([pad_token_label_id] * padding_length) + label_ids
               else:
                     input_ids += [pad_token] * padding_length
                     input_mask += [0 if mask_padding_with_zero else 1] * padding_length
                     segment_ids += [pad_token_segment_id] * padding_length
                     label_ids += [pad_token_label_id] * padding_length

               assert len(_lowerCamelCase							) == max_seq_length
               assert len(_lowerCamelCase							) == max_seq_length
               assert len(_lowerCamelCase							) == max_seq_length
               assert len(_lowerCamelCase							) == max_seq_length

               if ex_index < 5:
                     logger.info("""*** Example ***"""							)
                     logger.info("""guid: %s"""       ,  example.guid							)
                     logger.info("""tokens: %s"""       ,  """ """.join([str(_lowerCamelCase							) for x in tokens]							)							)
                     logger.info("""input_ids: %s"""       ,  """ """.join([str(_lowerCamelCase							) for x in input_ids]							)							)
                     logger.info("""input_mask: %s"""       ,  """ """.join([str(_lowerCamelCase							) for x in input_mask]							)							)
                     logger.info("""segment_ids: %s"""       ,  """ """.join([str(_lowerCamelCase							) for x in segment_ids]							)							)
                     logger.info("""label_ids: %s"""       ,  """ """.join([str(_lowerCamelCase							) for x in label_ids]							)							)

               if "token_type_ids" not in tokenizer.model_input_names:
                     __lowerCamelCase     :      str       						=						None

               features.append(
                   InputFeatures(
                       input_ids=_lowerCamelCase       ,  attention_mask=_lowerCamelCase       ,  token_type_ids=_lowerCamelCase       ,  label_ids=_lowerCamelCase							)							)
         return features


if is_torch_available():
    import torch
    from torch import nn
    from torch.utils.data import Dataset



    class 							_UpperCamelCase (						A						):

       '''simple docstring'''







       a_				:							List[InputFeatures]
       a_				:							int    = nn.CrossEntropyLoss().ignore_index

       def __init__(  self  : Tuple       ,  _lowerCamelCase  : TokenClassificationTask       ,  _lowerCamelCase  : str       ,  _lowerCamelCase  : PreTrainedTokenizer       ,  _lowerCamelCase  : List[str]       ,  _lowerCamelCase  : str       ,  _lowerCamelCase  : Optional[int] = None       ,  _lowerCamelCase  : Optional[int]=False       ,  _lowerCamelCase  : Split = Split.train       ,  ):






             '''simple docstring'''



             __lowerCamelCase     :      List[Any]       						=						os.path.join(
                 _lowerCamelCase       ,  """cached_{}_{}_{}""".format(mode.value       ,  tokenizer.__class__.__name__       ,  str(_lowerCamelCase							)							)       ,  )

             # Make sure only the first process in distributed training processes the dataset,
             # and the others will use the cache.
             __lowerCamelCase     :      List[Any]       						=						cached_features_file + """.lock"""
             with FileLock(_lowerCamelCase							):
                   if os.path.exists(_lowerCamelCase							) and not overwrite_cache:
                         logger.info(F"""Loading features from cached file {cached_features_file}"""							)
                         __lowerCamelCase     :      Tuple       						=						torch.load(_lowerCamelCase							)
                   else:
                         logger.info(F"""Creating features from dataset file at {data_dir}"""							)
                         __lowerCamelCase     :      List[str]       						=						token_classification_task.read_examples_from_file(_lowerCamelCase       ,  _lowerCamelCase							)
                         # TODO clean up all this to leverage built-in features of tokenizers
                         __lowerCamelCase     :      List[str]       						=						token_classification_task.convert_examples_to_features(
                             _lowerCamelCase       ,  _lowerCamelCase       ,  _lowerCamelCase       ,  _lowerCamelCase       ,  cls_token_at_end=bool(model_type in ["""xlnet"""]							)       ,  cls_token=tokenizer.cls_token       ,  cls_token_segment_id=2 if model_type in ["""xlnet"""] else 0       ,  sep_token=tokenizer.sep_token       ,  sep_token_extra=_lowerCamelCase       ,  pad_on_left=bool(tokenizer.padding_side == """left"""							)       ,  pad_token=tokenizer.pad_token_id       ,  pad_token_segment_id=tokenizer.pad_token_type_id       ,  pad_token_label_id=self.pad_token_label_id       ,  )
                         logger.info(F"""Saving features into cached file {cached_features_file}"""							)
                         torch.save(self.features       ,  _lowerCamelCase							)

       def __len__(  self  : Optional[int]							):






             '''simple docstring'''



             return len(self.features							)


       def __getitem__(  self  : str       ,  _lowerCamelCase  : Union[str, Any]							):






             '''simple docstring'''



             return self.features[i]


if is_tf_available():
    import tensorflow as tf



    class 							_UpperCamelCase :

       '''simple docstring'''







       a_				:							List[InputFeatures]
       a_				:							int    = -100

       def __init__(  self  : Optional[Any]       ,  _lowerCamelCase  : TokenClassificationTask       ,  _lowerCamelCase  : str       ,  _lowerCamelCase  : PreTrainedTokenizer       ,  _lowerCamelCase  : List[str]       ,  _lowerCamelCase  : str       ,  _lowerCamelCase  : Optional[int] = None       ,  _lowerCamelCase  : int=False       ,  _lowerCamelCase  : Split = Split.train       ,  ):






             '''simple docstring'''



             __lowerCamelCase     :      Tuple       						=						token_classification_task.read_examples_from_file(_lowerCamelCase       ,  _lowerCamelCase							)
             # TODO clean up all this to leverage built-in features of tokenizers
             __lowerCamelCase     :      int       						=						token_classification_task.convert_examples_to_features(
                 _lowerCamelCase       ,  _lowerCamelCase       ,  _lowerCamelCase       ,  _lowerCamelCase       ,  cls_token_at_end=bool(model_type in ["""xlnet"""]							)       ,  cls_token=tokenizer.cls_token       ,  cls_token_segment_id=2 if model_type in ["""xlnet"""] else 0       ,  sep_token=tokenizer.sep_token       ,  sep_token_extra=_lowerCamelCase       ,  pad_on_left=bool(tokenizer.padding_side == """left"""							)       ,  pad_token=tokenizer.pad_token_id       ,  pad_token_segment_id=tokenizer.pad_token_type_id       ,  pad_token_label_id=self.pad_token_label_id       ,  )

             def gen():
                   for ex in self.features:
                         if ex.token_type_ids is None:
                               yield (
                                   {"input_ids": ex.input_ids, "attention_mask": ex.attention_mask},
                                   ex.label_ids,
                               )
                         else:
                               yield (
                                   {
                                       "input_ids": ex.input_ids,
                                       "attention_mask": ex.attention_mask,
                                       "token_type_ids": ex.token_type_ids,
                                   },
                                   ex.label_ids,
                               )

             if "token_type_ids" not in tokenizer.model_input_names:
                   __lowerCamelCase     :      Any       						=						tf.data.Dataset.from_generator(
                       _lowerCamelCase       ,  ({"""input_ids""": tf.intaa, """attention_mask""": tf.intaa}, tf.intaa)       ,  (
                           {"""input_ids""": tf.TensorShape([None]							), """attention_mask""": tf.TensorShape([None]							)},
                           tf.TensorShape([None]							),
                       )       ,  )
             else:
                   __lowerCamelCase     :      int       						=						tf.data.Dataset.from_generator(
                       _lowerCamelCase       ,  ({"""input_ids""": tf.intaa, """attention_mask""": tf.intaa, """token_type_ids""": tf.intaa}, tf.intaa)       ,  (
                           {
                               """input_ids""": tf.TensorShape([None]							),
                               """attention_mask""": tf.TensorShape([None]							),
                               """token_type_ids""": tf.TensorShape([None]							),
                           },
                           tf.TensorShape([None]							),
                       )       ,  )

       def   _snake_case    (  self  : str							):






             '''simple docstring'''



             __lowerCamelCase     :      Union[str, Any]       						=						self.dataset.apply(tf.data.experimental.assert_cardinality(len(self.features							)							)							)

             return self.dataset

       def __len__(  self  : str							):






             '''simple docstring'''



             return len(self.features							)


       def __getitem__(  self  : List[Any]       ,  _lowerCamelCase  : Optional[Any]							):






             '''simple docstring'''



             return self.features[i]







import argparse
import hashlib
import os
import urllib
import warnings

import torch
from torch import nn
from tqdm import tqdm

from transformers import WhisperConfig, WhisperForConditionalGeneration


__UpperCamelCase    :						Tuple     =							{
    'tiny.en': 'https://openaipublic.azureedge.net/main/whisper/models/d3dd57d32accea0b295c96e26691aa14d8822fac7d9d27d5dc00b4ca2826dd03/tiny.en.pt',
    'tiny': 'https://openaipublic.azureedge.net/main/whisper/models/65147644a518d12f04e32d6f3b26facc3f8dd46e5390956a9424a650c0ce22b9/tiny.pt',
    'base.en': 'https://openaipublic.azureedge.net/main/whisper/models/25a8566e1d0c1e2231d1c762132cd20e0f96a85d16145c3a00adf5d1ac670ead/base.en.pt',
    'base': 'https://openaipublic.azureedge.net/main/whisper/models/ed3a0b6b1c0edf879ad9b11b1af5a0e6ab5db9205f891f668f8b0e6c6326e34e/base.pt',
    'small.en': 'https://openaipublic.azureedge.net/main/whisper/models/f953ad0fd29cacd07d5a9eda5624af0f6bcf2258be67c92b79389873d91e0872/small.en.pt',
    'small': 'https://openaipublic.azureedge.net/main/whisper/models/9ecf779972d90ba49c06d968637d720dd632c55bbf19d441fb42bf17a411e794/small.pt',
    'medium.en': 'https://openaipublic.azureedge.net/main/whisper/models/d7440d1dc186f76616474e0ff0b3b6b879abc9d1a4926b7adfa41db2d497ab4f/medium.en.pt',
    'medium': 'https://openaipublic.azureedge.net/main/whisper/models/345ae4da62f9b3d59415adc60127b97c714f32e89e936602e85993674d08dcb1/medium.pt',
    'large': 'https://openaipublic.azureedge.net/main/whisper/models/e4b87e7e0bf463eb8e6956e646f1e277e901512310def2c24bf0e11bd3c28e9a/large.pt',
    'large-v2': 'https://openaipublic.azureedge.net/main/whisper/models/81f7c96c852ee8fc832187b0132e569d6c3065a3252ed18e56effd0b6a73e524/large-v2.pt',
}





def   _UpperCAmelCase       (     UpperCAmelCase      :			Dict      ):
      """simple docstring"""




      __lowerCamelCase     :      int       						=						["""layers""", """blocks"""]
      for k in ignore_keys:
            state_dict.pop(UpperCAmelCase						,	UpperCAmelCase      )


__UpperCamelCase    :						str     =							{
    'blocks': 'layers',
    'mlp.0': 'fc1',
    'mlp.2': 'fc2',
    'mlp_ln': 'final_layer_norm',
    '.attn.query': '.self_attn.q_proj',
    '.attn.key': '.self_attn.k_proj',
    '.attn.value': '.self_attn.v_proj',
    '.attn_ln': '.self_attn_layer_norm',
    '.attn.out': '.self_attn.out_proj',
    '.cross_attn.query': '.encoder_attn.q_proj',
    '.cross_attn.key': '.encoder_attn.k_proj',
    '.cross_attn.value': '.encoder_attn.v_proj',
    '.cross_attn_ln': '.encoder_attn_layer_norm',
    '.cross_attn.out': '.encoder_attn.out_proj',
    'decoder.ln.': 'decoder.layer_norm.',
    'encoder.ln.': 'encoder.layer_norm.',
    'token_embedding': 'embed_tokens',
    'encoder.positional_embedding': 'encoder.embed_positions.weight',
    'decoder.positional_embedding': 'decoder.embed_positions.weight',
    'ln_post': 'layer_norm',
}





def   _UpperCAmelCase       (     UpperCAmelCase      :			Dict      ):
      """simple docstring"""




      __lowerCamelCase     :      List[str]       						=						list(s_dict.keys()      )
      for key in keys:
            __lowerCamelCase     :      Union[str, Any]       						=						key
            for k, v in WHISPER_MAPPING.items():
                  if k in key:
                        __lowerCamelCase     :      Optional[int]       						=						new_key.replace(UpperCAmelCase						,	UpperCAmelCase      )

            print(f"""{key} -> {new_key}"""      )

            __lowerCamelCase     :      Dict       						=						s_dict.pop(UpperCAmelCase      )
      return s_dict





def   _UpperCAmelCase       (     UpperCAmelCase      :			Optional[int]      ):
      """simple docstring"""




      __lowerCamelCase						, __lowerCamelCase     :      Dict       						=						emb.weight.shape
      __lowerCamelCase     :      Tuple       						=						nn.Linear(UpperCAmelCase						,	UpperCAmelCase						,	bias=UpperCAmelCase      )
      __lowerCamelCase     :      Union[str, Any]       						=						emb.weight.data
      return lin_layer





def   _UpperCAmelCase       (     UpperCAmelCase      :			str						,	UpperCAmelCase      :			str      ):
      """simple docstring"""




      os.makedirs(UpperCAmelCase						,	exist_ok=UpperCAmelCase      )
      __lowerCamelCase     :      Any       						=						os.path.basename(UpperCAmelCase      )

      __lowerCamelCase     :      int       						=						url.split("""/"""      )[-2]
      __lowerCamelCase     :      str       						=						os.path.join(UpperCAmelCase						,	UpperCAmelCase      )

      if os.path.exists(UpperCAmelCase      ) and not os.path.isfile(UpperCAmelCase      ):
            raise RuntimeError(f"""{download_target} exists and is not a regular file"""      )

      if os.path.isfile(UpperCAmelCase      ):
            __lowerCamelCase     :      Any       						=						open(UpperCAmelCase						,	"""rb"""      ).read()
            if hashlib.shaaaa(UpperCAmelCase      ).hexdigest() == expected_shaaaa:
                  return model_bytes
            else:
                  warnings.warn(f"""{download_target} exists, but the SHA256 checksum does not match; re-downloading the file"""      )

      with urllib.request.urlopen(UpperCAmelCase      ) as source, open(UpperCAmelCase						,	"""wb"""      ) as output:
            with tqdm(
                total=int(source.info().get("""Content-Length"""      )      )						,	ncols=80						,	unit="""iB"""						,	unit_scale=UpperCAmelCase						,	unit_divisor=1_024      ) as loop:
                  while True:
                        __lowerCamelCase     :      Union[str, Any]       						=						source.read(8_192      )
                        if not buffer:
                              break

                        output.write(UpperCAmelCase      )
                        loop.update(len(UpperCAmelCase      )      )

      __lowerCamelCase     :      List[str]       						=						open(UpperCAmelCase						,	"""rb"""      ).read()
      if hashlib.shaaaa(UpperCAmelCase      ).hexdigest() != expected_shaaaa:
            raise RuntimeError(
                """Model has been downloaded but the SHA256 checksum does not not match. Please retry loading the model."""      )

      return model_bytes





def   _UpperCAmelCase       (     UpperCAmelCase      :			List[Any]						,	UpperCAmelCase      :			Optional[int]      ):
      """simple docstring"""




      if ".pt" not in checkpoint_path:
            __lowerCamelCase     :      Optional[int]       						=						_download(_MODELS[checkpoint_path]      )
      else:
            __lowerCamelCase     :      Union[str, Any]       						=						torch.load(UpperCAmelCase						,	map_location="""cpu"""      )
      __lowerCamelCase     :      Any       						=						original_checkpoint["""dims"""]
      __lowerCamelCase     :      int       						=						original_checkpoint["""model_state_dict"""]
      __lowerCamelCase     :      Tuple       						=						state_dict["""decoder.token_embedding.weight"""]
      remove_ignore_keys_(UpperCAmelCase      )
      rename_keys(UpperCAmelCase      )
      __lowerCamelCase     :      int       						=						True
      __lowerCamelCase     :      Tuple       						=						state_dict["""decoder.layers.0.fc1.weight"""].shape[0]

      __lowerCamelCase     :      Any       						=						WhisperConfig(
          vocab_size=dimensions["""n_vocab"""]						,	encoder_ffn_dim=UpperCAmelCase						,	decoder_ffn_dim=UpperCAmelCase						,	num_mel_bins=dimensions["""n_mels"""]						,	d_model=dimensions["""n_audio_state"""]						,	max_target_positions=dimensions["""n_text_ctx"""]						,	encoder_layers=dimensions["""n_audio_layer"""]						,	encoder_attention_heads=dimensions["""n_audio_head"""]						,	decoder_layers=dimensions["""n_text_layer"""]						,	decoder_attention_heads=dimensions["""n_text_state"""]						,	max_source_positions=dimensions["""n_audio_ctx"""]						,	)

      __lowerCamelCase     :      List[str]       						=						WhisperForConditionalGeneration(UpperCAmelCase      )
      __lowerCamelCase						, __lowerCamelCase     :      str       						=						model.model.load_state_dict(UpperCAmelCase						,	strict=UpperCAmelCase      )
      if len(UpperCAmelCase      ) > 0 and not set(UpperCAmelCase      ) <= {
          "encoder.embed_positions.weights",
          "decoder.embed_positions.weights",
      }:
            raise ValueError(
                """Only `encoder.embed_positions.weights` and `decoder.embed_positions.weights`  are allowed to be missing,"""
                f""" but all the following weights are missing {missing}"""      )

      if tie_embeds:
            __lowerCamelCase     :      Tuple       						=						make_linear_from_emb(model.model.decoder.embed_tokens      )
      else:
            __lowerCamelCase     :      Tuple       						=						proj_out_weights

      model.save_pretrained(UpperCAmelCase      )


if __name__ == "__main__":
    __UpperCamelCase    :						Union[str, Any]     =							argparse.ArgumentParser()
    # # Required parameters
    parser.add_argument('--checkpoint_path', type=str, help='Patht to the downloaded checkpoints')
    parser.add_argument('--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model.')
    __UpperCamelCase    :						Tuple     =							parser.parse_args()

    convert_openai_whisper_to_tfms(args.checkpoint_path, args.pytorch_dump_folder_path)



