





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


SCREAMING_SNAKE_CASE_   :      List[str]			=   {
    '''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  SCREAMING_SNAKE_CASE				(      snake_case							)		->    Any:
		__lowercase											=     ['layers', 'blocks']
		for k in ignore_keys:
				state_dict.pop(snake_case						,   snake_case							)


SCREAMING_SNAKE_CASE_   :      Tuple			=   {
    '''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  SCREAMING_SNAKE_CASE				(      snake_case							)		->    Optional[int]:
		__lowercase											=     list(s_dict.keys()							)
		for key in keys:
				__lowercase											=     key
				for k, v in WHISPER_MAPPING.items():
						if k in key:
								__lowercase											=     new_key.replace(snake_case						,   snake_case							)

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

				__lowercase											=     s_dict.pop(snake_case							)
		return s_dict



def  SCREAMING_SNAKE_CASE				(      snake_case							)		->    Optional[Any]:
		__lowercase						,      __lowercase											=     emb.weight.shape
		__lowercase											=     nn.Linear(snake_case						,   snake_case						,   bias=snake_case							)
		__lowercase											=     emb.weight.data
		return lin_layer



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case							)		->    bytes:
		os.makedirs(snake_case						,   exist_ok=snake_case							)
		__lowercase											=     os.path.basename(snake_case							)

		__lowercase											=     url.split('/'							)[-2]
		__lowercase											=     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							):
				__lowercase											=     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:
								__lowercase											=     source.read(8_192							)
								if not buffer:
										break

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

		__lowercase											=     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  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case							)		->    Union[str, Any]:
		if ".pt" not in checkpoint_path:
				__lowercase											=     _download(_MODELS[checkpoint_path]							)
		else:
				__lowercase											=     torch.load(snake_case						,   map_location='cpu'							)
		__lowercase											=     original_checkpoint['dims']
		__lowercase											=     original_checkpoint['model_state_dict']
		__lowercase											=     state_dict['decoder.token_embedding.weight']
		remove_ignore_keys_(snake_case							)
		rename_keys(snake_case							)
		__lowercase											=     True
		__lowercase											=     state_dict['decoder.layers.0.fc1.weight'].shape[0]

		__lowercase											=     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']						,   )

		__lowercase											=     WhisperForConditionalGeneration(snake_case							)
		__lowercase						,      __lowercase											=     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:
				__lowercase											=     make_linear_from_emb(model.model.decoder.embed_tokens							)
		else:
				__lowercase											=     proj_out_weights

		model.save_pretrained(snake_case							)


if __name__ == "__main__":
		SCREAMING_SNAKE_CASE_   :      str			=   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.''')
		SCREAMING_SNAKE_CASE_   :      Dict			=   parser.parse_args()

		convert_openai_whisper_to_tfms(args.checkpoint_path, args.pytorch_dump_folder_path)











import os
from collections import namedtuple

import pytest

from datasets import ClassLabel, Features, Sequence, Value
from datasets.commands.test import TestCommand
from datasets.info import DatasetInfo, DatasetInfosDict


SCREAMING_SNAKE_CASE_   :      Optional[Any]			=   namedtuple(
    '''_TestCommandArgs''',
    [
        '''dataset''',
        '''name''',
        '''cache_dir''',
        '''data_dir''',
        '''all_configs''',
        '''save_infos''',
        '''ignore_verifications''',
        '''force_redownload''',
        '''clear_cache''',
    ],
    defaults=[None, None, None, False, False, False, False, False],
)



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case							)		->    Optional[int]:
		return (abs(source - target							) / target) < 0.01



@pytest.mark.integration
def  SCREAMING_SNAKE_CASE				(      snake_case							)		->    Dict:
		__lowercase											=     _TestCommandArgs(dataset=snake_case						,   all_configs=snake_case						,   save_infos=snake_case							)
		__lowercase											=     TestCommand(*snake_case							)
		test_command.run()
		__lowercase											=     os.path.join(snake_case						,   'README.md'							)
		assert os.path.exists(snake_case							)
		__lowercase											=     DatasetInfosDict.from_directory(snake_case							)
		__lowercase											=     DatasetInfosDict(
		    {
		        'default': DatasetInfo(
		            features=Features(
		                {
		                    'tokens': Sequence(Value('string'							)							),
		                    'ner_tags': Sequence(
		                        ClassLabel(names=['O', 'B-PER', 'I-PER', 'B-ORG', 'I-ORG', 'B-LOC', 'I-LOC']							)							),
		                    'langs': Sequence(Value('string'							)							),
		                    'spans': Sequence(Value('string'							)							),
		                }							)						,   splits=[
		                {
		                    'name': 'train',
		                    'num_bytes': 2_351_563,
		                    'num_examples': 10_000,
		                },
		                {
		                    'name': 'validation',
		                    'num_bytes': 238_418,
		                    'num_examples': 1_000,
		                },
		            ]						,   download_size=3_940_680						,   dataset_size=2_589_981						,   )
		    }							)
		assert dataset_infos.keys() == expected_dataset_infos.keys()
		for key in DatasetInfo._INCLUDED_INFO_IN_YAML:
				__lowercase						,      __lowercase											=     getattr(dataset_infos['default']						,   snake_case							), getattr(expected_dataset_infos['default']						,   snake_case							)
				if key == "num_bytes":
						assert is_apercent_close(snake_case						,   snake_case							)
				elif key == "splits":
						assert list(snake_case							) == list(snake_case							)
						for split in result:
								assert result[split].name == expected[split].name
								assert result[split].num_examples == expected[split].num_examples
								assert is_apercent_close(result[split].num_bytes						,   expected[split].num_bytes							)
				else:
						result == expected




