


'''simple docstring'''






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


a_					:    Dict          =    {
    "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 	_A				(lowerCAmelCase__  :Any		)   ->  List[Any]:




     '''simple docstring'''



     _a			      =							['layers', 'blocks']
     for k in ignore_keys:
          state_dict.pop(lowerCAmelCase__				,						lowerCAmelCase__		)


a_					:    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 	_A				(lowerCAmelCase__  :Optional[int]		)   ->  List[Any]:




     '''simple docstring'''



     _a			      =							list(s_dict.keys()		)
     for key in keys:
          _a			      =							key
          for k, v in WHISPER_MAPPING.items():
               if k in key:
                    _a			      =							new_key.replace(lowerCAmelCase__				,						lowerCAmelCase__		)

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

          _a			      =							s_dict.pop(lowerCAmelCase__		)
     return s_dict






def 	_A				(lowerCAmelCase__  :Any		)   ->  int:




     '''simple docstring'''



     _a      ,   _a			      =							emb.weight.shape
     _a			      =							nn.Linear(lowerCAmelCase__				,						lowerCAmelCase__				,						bias=lowerCAmelCase__		)
     _a			      =							emb.weight.data
     return lin_layer






def 	_A				(lowerCAmelCase__  :str				,						lowerCAmelCase__  :str		)   ->  bytes:




     '''simple docstring'''



     os.makedirs(lowerCAmelCase__				,						exist_ok=lowerCAmelCase__		)
     _a			      =							os.path.basename(lowerCAmelCase__		)

     _a			      =							url.split('/'		)[-2]
     _a			      =							os.path.join(lowerCAmelCase__				,						lowerCAmelCase__		)

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

     if os.path.isfile(lowerCAmelCase__		):
          _a			      =							open(lowerCAmelCase__				,						'rb'		).read()
          if hashlib.shaaaa(lowerCAmelCase__		).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(lowerCAmelCase__		) as source, open(lowerCAmelCase__				,						'wb'		) as output:
          with tqdm(
              total=int(source.info().get('Content-Length'		)		)				,						ncols=80				,						unit='iB'				,						unit_scale=lowerCAmelCase__				,						unit_divisor=10_24		) as loop:
               while True:
                    _a			      =							source.read(81_92		)
                    if not buffer:
                         break

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

     _a			      =							open(lowerCAmelCase__				,						'rb'		).read()
     if hashlib.shaaaa(lowerCAmelCase__		).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 	_A				(lowerCAmelCase__  :Optional[int]				,						lowerCAmelCase__  :int		)   ->  Any:




     '''simple docstring'''



     if ".pt" not in checkpoint_path:
          _a			      =							_download(_MODELS[checkpoint_path]		)
     else:
          _a			      =							torch.load(lowerCAmelCase__				,						map_location='cpu'		)
     _a			      =							original_checkpoint['dims']
     _a			      =							original_checkpoint['model_state_dict']
     _a			      =							state_dict['decoder.token_embedding.weight']
     remove_ignore_keys_(lowerCAmelCase__		)
     rename_keys(lowerCAmelCase__		)
     _a			      =							True
     _a			      =							state_dict['decoder.layers.0.fc1.weight'].shape[0]

     _a			      =							WhisperConfig(
         vocab_size=dimensions['n_vocab']				,						encoder_ffn_dim=lowerCAmelCase__				,						decoder_ffn_dim=lowerCAmelCase__				,						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']				,						)

     _a			      =							WhisperForConditionalGeneration(lowerCAmelCase__		)
     _a      ,   _a			      =							model.model.load_state_dict(lowerCAmelCase__				,						strict=lowerCAmelCase__		)
     if len(lowerCAmelCase__		) > 0 and not set(lowerCAmelCase__		) <= {
         "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:
          _a			      =							make_linear_from_emb(model.model.decoder.embed_tokens		)
     else:
          _a			      =							proj_out_weights

     model.save_pretrained(lowerCAmelCase__		)


if __name__ == "__main__":
 a_					:    Tuple          =    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.")
 a_					:    Union[str, Any]          =    parser.parse_args()

 convert_openai_whisper_to_tfms(args.checkpoint_path, args.pytorch_dump_folder_path)





'''simple docstring'''






class  a     :
 def __init__( self      ,				__magic_name__						)					->							Optional[int]:
      _a			      =							n
      _a			      =							[None] * self.n
      _a			      =							0  # index of the first element
      _a			      =							0
      _a			      =							0
 def __len__( self						)					->							int:
      return self.size
 def 							__UpperCAmelCase		( self						)					->							bool:
      return self.size == 0
 def 							__UpperCAmelCase		( self						)					->							Optional[Any]:
      return False if self.is_empty() else self.array[self.front]
 def 							__UpperCAmelCase		( self      ,				__magic_name__						)					->							Optional[Any]:
      if self.size >= self.n:
           raise Exception('QUEUE IS FULL'						)

      _a			      =							data
      _a			      =							(self.rear + 1) % self.n
      self.size += 1
      return self




 def 							__UpperCAmelCase		( self						)					->							Any:
      if self.size == 0:
           raise Exception('UNDERFLOW'						)

      _a			      =							self.array[self.front]
      _a			      =							None
      _a			      =							(self.front + 1) % self.n
      self.size -= 1
      return temp

