



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


_snake_case		=  {
    "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  ):





							'''simple docstring'''




							_lowerCAmelCase     :							Union[str, Any]    =    ["layers", "blocks"]
							for k in ignore_keys:
														state_dict.pop(snake_case_				, snake_case_  )


_snake_case		=  {
    "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  ):





							'''simple docstring'''




							_lowerCAmelCase     :							Any    =    list(s_dict.keys()  )
							for key in keys:
														_lowerCAmelCase     :							Any    =    key
														for k, v in WHISPER_MAPPING.items():
																					if k in key:
																												_lowerCAmelCase     :							Any    =    new_key.replace(snake_case_				, snake_case_  )

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

														_lowerCAmelCase     :							Optional[Any]    =    s_dict.pop(snake_case_  )
							return s_dict






def 				A				(					_lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase   ,    _lowerCAmelCase     :							str    =    emb.weight.shape
							_lowerCAmelCase     :							Any    =    nn.Linear(snake_case_				, snake_case_				, bias=snake_case_  )
							_lowerCAmelCase     :							Union[str, Any]    =    emb.weight.data
							return lin_layer






def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							os.makedirs(snake_case_				, exist_ok=snake_case_  )
							_lowerCAmelCase     :							Dict    =    os.path.basename(snake_case_  )

							_lowerCAmelCase     :							str    =    url.split("/"  )[-2]
							_lowerCAmelCase     :							str    =    os.path.join(snake_case_				, snake_case_  )

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

							if os.path.isfile(snake_case_  ):
														_lowerCAmelCase     :							Tuple    =    open(snake_case_				, "rb"  ).read()
														if hashlib.shaaaa(snake_case_  ).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(snake_case_  ) as source, open(snake_case_				, "wb"  ) as output:
														with tqdm(
														    total=int(source.info().get("Content-Length"  )  )				, ncols=80				, unit="iB"				, unit_scale=snake_case_				, unit_divisor=1_024  ) as loop:
																					while True:
																												_lowerCAmelCase     :							Optional[Any]    =    source.read(8_192  )
																												if not buffer:
																																			break

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

							_lowerCAmelCase     :							Any    =    open(snake_case_				, "rb"  ).read()
							if hashlib.shaaaa(snake_case_  ).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				, _lowerCamelCase  ):





							'''simple docstring'''




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

							_lowerCAmelCase     :							Optional[Any]    =    WhisperConfig(
							    vocab_size=dimensions["n_vocab"]				, encoder_ffn_dim=snake_case_				, decoder_ffn_dim=snake_case_				, 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[Any]    =    WhisperForConditionalGeneration(snake_case_  )
							_lowerCAmelCase   ,    _lowerCAmelCase     :							Union[str, Any]    =    model.model.load_state_dict(snake_case_				, strict=snake_case_  )
							if len(snake_case_  ) > 0 and not set(snake_case_  ) <= {
							    "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     :							Optional[Any]    =    make_linear_from_emb(model.model.decoder.embed_tokens  )
							else:
														_lowerCAmelCase     :							Dict    =    proj_out_weights

							model.save_pretrained(snake_case_  )


if __name__ == "__main__":
						_snake_case		=  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.")
						_snake_case		=  parser.parse_args()

						convert_openai_whisper_to_tfms(args.checkpoint_path, args.pytorch_dump_folder_path)







import copy
import inspect
import unittest

from transformers import AutoBackbone
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import require_timm, require_torch, torch_device
from transformers.utils.import_utils import is_torch_available

from ...test_backbone_common import BackboneTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor


if is_torch_available():
    import torch

    from transformers import TimmBackbone, TimmBackboneConfig

from ...test_pipeline_mixin import PipelineTesterMixin

class 		SCREAMING_SNAKE_CASE__				:


      def __init__(self       :      str				,			a__       :      Dict				,			a__       :      Tuple=None				,			a__       :      List[Any]=None				,			a__       :      Dict=None				,			a__       :      Union[str, Any]="resnet50"				,			a__       :      Dict=3				,			a__       :      str=32				,			a__       :      int=3				,			a__       :      Dict=True				,			a__       :      Any=True				,			):





         """simple docstring"""



         __snake_case							       =   parent
         __snake_case							       =   out_indices if out_indices is not None else [4]
         __snake_case							       =   stage_names
         __snake_case							       =   out_features
         __snake_case							       =   backbone
         __snake_case							       =   batch_size
         __snake_case							       =   image_size
         __snake_case							       =   num_channels
         __snake_case							       =   use_pretrained_backbone
         __snake_case							       =   is_training


      def    a (self       :      Union[str, Any]      ):





         """simple docstring"""



         __snake_case							       =   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]      )
         __snake_case							       =   self.get_config()

         return config, pixel_values


      def    a (self       :      Any      ):





         """simple docstring"""



         return TimmBackboneConfig(
             image_size=self.image_size				,			num_channels=self.num_channels				,			out_features=self.out_features				,			out_indices=self.out_indices				,			stage_names=self.stage_names				,			use_pretrained_backbone=self.use_pretrained_backbone				,			backbone=self.backbone				,			)


      def    a (self       :      List[Any]				,			a__       :      int				,			a__       :      int      ):





         """simple docstring"""



         __snake_case							       =   TimmBackbone(config=a__      )
         model.to(a__      )
         model.eval()
         with torch.no_grad():
            __snake_case							       =   model(a__      )
         self.parent.assertEqual(
             result.feature_map[-1].shape				,			(self.batch_size, model.channels[-1], 14, 14)				,			)


      def    a (self       :      str      ):





         """simple docstring"""



         __snake_case							       =   self.prepare_config_and_inputs()
         __snake_case       ,	__snake_case							       =   config_and_inputs
         __snake_case							       =   {'''pixel_values''': pixel_values}
         return config, inputs_dict






@require_torch
@require_timm
class 		SCREAMING_SNAKE_CASE__				(       _UpperCAmelCase						,      _UpperCAmelCase						,      _UpperCAmelCase						,      unittest.TestCase  ):
      A_						:				Union[str, Any]							=    (TimmBackbone,) if is_torch_available() else ()
      A_						:				Optional[Any]							=    {'feature-extraction': TimmBackbone} if is_torch_available() else {}
      A_						:				List[Any]							=    False
      A_						:				Dict							=    False
      A_						:				Any							=    False
      A_						:				List[Any]							=    False


      def    a (self       :      Tuple      ):





         """simple docstring"""



         __snake_case							       =   TimmBackboneModelTester(self      )
         __snake_case							       =   ConfigTester(self				,			config_class=a__				,			has_text_modality=a__      )


      def    a (self       :      Any      ):





         """simple docstring"""



         self.config_tester.create_and_test_config_to_json_string()
         self.config_tester.create_and_test_config_to_json_file()
         self.config_tester.create_and_test_config_from_and_save_pretrained()
         self.config_tester.create_and_test_config_with_num_labels()
         self.config_tester.check_config_can_be_init_without_params()
         self.config_tester.check_config_arguments_init()


      def    a (self       :      int      ):





         """simple docstring"""



         __snake_case							       =   '''resnet18'''
         __snake_case							       =   '''microsoft/resnet-18'''

         __snake_case							       =   AutoBackbone.from_pretrained(a__				,			use_timm_backbone=a__      )
         __snake_case							       =   AutoBackbone.from_pretrained(a__      )

         self.assertEqual(len(timm_model.out_features      )				,			len(transformers_model.out_features      )      )
         self.assertEqual(len(timm_model.stage_names      )				,			len(transformers_model.stage_names      )      )
         self.assertEqual(timm_model.channels				,			transformers_model.channels      )
         # Out indices are set to the last layer by default. For timm models, we don't know
         # the number of layers in advance, so we set it to (-1,), whereas for transformers
         # models, we set it to [len(stage_names) - 1] (kept for backward compatibility).
         self.assertEqual(timm_model.out_indices				,			(-1,)      )
         self.assertEqual(transformers_model.out_indices				,			[len(timm_model.stage_names      ) - 1]      )

         __snake_case							       =   AutoBackbone.from_pretrained(a__				,			use_timm_backbone=a__				,			out_indices=[1, 2, 3]      )
         __snake_case							       =   AutoBackbone.from_pretrained(a__				,			out_indices=[1, 2, 3]      )

         self.assertEqual(timm_model.out_indices				,			transformers_model.out_indices      )
         self.assertEqual(len(timm_model.out_features      )				,			len(transformers_model.out_features      )      )
         self.assertEqual(timm_model.channels				,			transformers_model.channels      )


      @unittest.skip('''TimmBackbone doesn\'t support feed forward chunking'''      )
      def    a (self       :      str      ):





         """simple docstring"""



         pass


      @unittest.skip('''TimmBackbone doesn\'t have num_hidden_layers attribute'''      )
      def    a (self       :      int      ):





         """simple docstring"""



         pass


      @unittest.skip('''TimmBackbone initialization is managed on the timm side'''      )
      def    a (self       :      Union[str, Any]      ):





         """simple docstring"""



         pass


      @unittest.skip('''TimmBackbone models doesn\'t have inputs_embeds'''      )
      def    a (self       :      Optional[int]      ):





         """simple docstring"""



         pass


      @unittest.skip('''TimmBackbone models doesn\'t have inputs_embeds'''      )
      def    a (self       :      int      ):





         """simple docstring"""



         pass


      @unittest.skip('''TimmBackbone model cannot be created without specifying a backbone checkpoint'''      )
      def    a (self       :      Tuple      ):





         """simple docstring"""



         pass


      @unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone'''      )
      def    a (self       :      int      ):





         """simple docstring"""



         pass


      @unittest.skip('''model weights aren\'t tied in TimmBackbone.'''      )
      def    a (self       :      Optional[Any]      ):





         """simple docstring"""



         pass


      @unittest.skip('''model weights aren\'t tied in TimmBackbone.'''      )
      def    a (self       :      Tuple      ):





         """simple docstring"""



         pass


      @unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone'''      )
      def    a (self       :      Dict      ):





         """simple docstring"""



         pass


      @unittest.skip('''Only checkpoints on timm can be loaded into TimmBackbone'''      )
      def    a (self       :      List[Any]      ):





         """simple docstring"""



         pass


      @unittest.skip('''TimmBackbone doesn\'t have hidden size info in its configuration.'''      )
      def    a (self       :      Optional[Any]      ):





         """simple docstring"""



         pass


      @unittest.skip('''TimmBackbone doesn\'t support output_attentions.'''      )
      def    a (self       :      List[Any]      ):





         """simple docstring"""



         pass


      @unittest.skip('''Safetensors is not supported by timm.'''      )
      def    a (self       :      Tuple      ):





         """simple docstring"""



         pass


      @unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.'''      )
      def    a (self       :      Tuple      ):





         """simple docstring"""



         pass


      def    a (self       :      Tuple      ):





         """simple docstring"""



         __snake_case       ,	__snake_case							       =   self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            __snake_case							       =   model_class(a__      )
            __snake_case							       =   inspect.signature(model.forward      )
            # signature.parameters is an OrderedDict => so arg_names order is deterministic
            __snake_case							       =   [*signature.parameters.keys()]

            __snake_case							       =   ['''pixel_values''']
            self.assertListEqual(arg_names[:1]				,			a__      )


      def    a (self       :      Dict      ):





         """simple docstring"""



         __snake_case       ,	__snake_case							       =   self.model_tester.prepare_config_and_inputs_for_common()
         __snake_case							       =   True
         __snake_case							       =   self.has_attentions

         # no need to test all models as different heads yield the same functionality
         __snake_case							       =   self.all_model_classes[0]
         __snake_case							       =   model_class(a__      )
         model.to(a__      )

         __snake_case							       =   self._prepare_for_class(a__				,			a__      )
         __snake_case							       =   model(**a__      )
         __snake_case							       =   outputs[0][-1]

         # Encoder-/Decoder-only models
         __snake_case							       =   outputs.hidden_states[0]
         hidden_states.retain_grad()

         if self.has_attentions:
            __snake_case							       =   outputs.attentions[0]
            attentions.retain_grad()

         output.flatten()[0].backward(retain_graph=a__      )

         self.assertIsNotNone(hidden_states.grad      )

         if self.has_attentions:
            self.assertIsNotNone(attentions.grad      )


      def    a (self       :      Optional[int]      ):





         """simple docstring"""



         __snake_case       ,	__snake_case							       =   self.model_tester.prepare_config_and_inputs_for_common()

         for model_class in self.all_model_classes:
            __snake_case							       =   model_class(a__      )
            model.to(a__      )
            model.eval()
            __snake_case							       =   model(**a__      )

            self.assertEqual(len(result.feature_maps      )				,			len(config.out_indices      )      )
            self.assertEqual(len(model.channels      )				,			len(config.out_indices      )      )

            # Check output of last stage is taken if out_features=None, out_indices=None
            __snake_case							       =   copy.deepcopy(a__      )
            __snake_case							       =   None
            __snake_case							       =   model_class(a__      )
            model.to(a__      )
            model.eval()
            __snake_case							       =   model(**a__      )

            self.assertEqual(len(result.feature_maps      )				,			1      )
            self.assertEqual(len(model.channels      )				,			1      )

            # Check backbone can be initialized with fresh weights
            __snake_case							       =   copy.deepcopy(a__      )
            __snake_case							       =   False
            __snake_case							       =   model_class(a__      )
            model.to(a__      )
            model.eval()
            __snake_case							       =   model(**a__      )

