


"""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__    :							List[Any]            =					{
    '''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__	(lowerCAmelCase_       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    ["layers", "blocks"]
					for k in ignore_keys:
										state_dict.pop(lowerCAmelCase_  ,						lowerCAmelCase_       )


a__    :							Any            =					{
    '''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__	(lowerCAmelCase_       ):




					'''simple docstring'''


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

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

										__SCREAMING_SNAKE_CASE						       =    s_dict.pop(lowerCAmelCase_       )
					return s_dict




def        UpperCAmelCase__	(lowerCAmelCase_       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE       ,  __SCREAMING_SNAKE_CASE						       =    emb.weight.shape
					__SCREAMING_SNAKE_CASE						       =    nn.Linear(lowerCAmelCase_  ,						lowerCAmelCase_  ,						bias=lowerCAmelCase_       )
					__SCREAMING_SNAKE_CASE						       =    emb.weight.data
					return lin_layer




def        UpperCAmelCase__	(lowerCAmelCase_  ,						lowerCAmelCase_       ):




					'''simple docstring'''


					os.makedirs(lowerCAmelCase_  ,						exist_ok=lowerCAmelCase_       )
					__SCREAMING_SNAKE_CASE						       =    os.path.basename(lowerCAmelCase_       )

					__SCREAMING_SNAKE_CASE						       =    url.split("/"       )[-2]
					__SCREAMING_SNAKE_CASE						       =    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_       ):
										__SCREAMING_SNAKE_CASE						       =    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=1024       ) as loop:
															while True:
																				__SCREAMING_SNAKE_CASE						       =    source.read(8192       )
																				if not buffer:
																									break

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

					__SCREAMING_SNAKE_CASE						       =    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        UpperCAmelCase__	(lowerCAmelCase_  ,						lowerCAmelCase_       ):




					'''simple docstring'''


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

					__SCREAMING_SNAKE_CASE						       =    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"]  ,						)

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

					model.save_pretrained(lowerCAmelCase_       )


if __name__ == "__main__":
		a__    :							Optional[int]            =					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__    :							int            =					parser.parse_args()

		convert_openai_whisper_to_tfms(args.checkpoint_path, args.pytorch_dump_folder_path)




"""simple docstring"""







import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModelWithProjection, PreTrainedModel

from ...utils import logging


a__    :							Optional[Any]            =					logging.get_logger(__name__)




class 			UpperCamelCase_     (			UpperCamelCase):

	"""simple docstring"""





	snake_case__    :  int    =      CLIPConfig

	snake_case__    :  str    =      ["CLIPEncoderLayer"]




	def __init__(			self      :    Optional[int]      ,      UpperCAmelCase__      :    CLIPConfig       )	->     Dict:
						super().__init__(UpperCAmelCase__       )

						__SCREAMING_SNAKE_CASE						       =    CLIPVisionModelWithProjection(config.vision_config       )

						__SCREAMING_SNAKE_CASE						       =    nn.Linear(config.vision_config.projection_dim      ,      1       )
						__SCREAMING_SNAKE_CASE						       =    nn.Linear(config.vision_config.projection_dim      ,      1       )




	@torch.no_grad()
	def   UpperCAmelCase_				(			self      :    str      ,      UpperCAmelCase__      :    List[str]      ,      UpperCAmelCase__      :    str      ,      UpperCAmelCase__      :    int=0.5      ,      UpperCAmelCase__      :    Optional[int]=0.5       )	->     Union[str, Any]:
						__SCREAMING_SNAKE_CASE						       =    self.vision_model(UpperCAmelCase__       )[0]

						__SCREAMING_SNAKE_CASE						       =    self.p_head(UpperCAmelCase__       )
						__SCREAMING_SNAKE_CASE						       =    nsfw_detected.flatten()
						__SCREAMING_SNAKE_CASE						       =    nsfw_detected > p_threshold
						__SCREAMING_SNAKE_CASE						       =    nsfw_detected.tolist()

						if any(UpperCAmelCase__       ):
											logger.warning(
											    "Potential NSFW content was detected in one or more images. A black image will be returned instead."
											    " Try again with a different prompt and/or seed."       )

						for idx, nsfw_detected_ in enumerate(UpperCAmelCase__       ):
											if nsfw_detected_:
																__SCREAMING_SNAKE_CASE						       =    np.zeros(images[idx].shape       )

						__SCREAMING_SNAKE_CASE						       =    self.w_head(UpperCAmelCase__       )
						__SCREAMING_SNAKE_CASE						       =    watermark_detected.flatten()
						__SCREAMING_SNAKE_CASE						       =    watermark_detected > w_threshold
						__SCREAMING_SNAKE_CASE						       =    watermark_detected.tolist()

						if any(UpperCAmelCase__       ):
											logger.warning(
											    "Potential watermarked content was detected in one or more images. A black image will be returned instead."
											    " Try again with a different prompt and/or seed."       )

						for idx, watermark_detected_ in enumerate(UpperCAmelCase__       ):
											if watermark_detected_:
																__SCREAMING_SNAKE_CASE						       =    np.zeros(images[idx].shape       )

						return images, nsfw_detected, watermark_detected
