





import math
from typing import Any, Callable, List, Optional, Tuple, Union

import numpy as np
import torch

from ...models import TaFilmDecoder
from ...schedulers import DDPMScheduler
from ...utils import is_onnx_available, logging, randn_tensor


if is_onnx_available():
  from ..onnx_utils import OnnxRuntimeModel

from ..pipeline_utils import AudioPipelineOutput, DiffusionPipeline
from .continous_encoder import SpectrogramContEncoder
from .notes_encoder import SpectrogramNotesEncoder


SCREAMING_SNAKE_CASE__       :			Union[str, Any]			      =						logging.get_logger(__name__)  # pylint: disable=invalid-name

SCREAMING_SNAKE_CASE__       :			Tuple			      =						2_5_6







class   __lowerCAmelCase			(					_UpperCamelCase  ):
   _UpperCamelCase :						str     		=				["""melgan"""]





   def __init__( self					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			)				->  None:
      """simple docstring"""





      super().__init__()

      # From MELGAN
      a__	:						Dict      		=  math.log(1E-5      )  # Matches MelGAN training.
      a__	:						Any      		=  4.0  # Largest value for most examples
      a__	:						int      		=  128

      self.register_modules(
          notes_encoder=snake_case					,			continuous_encoder=snake_case					,			decoder=snake_case					,			scheduler=snake_case					,			melgan=snake_case					,			)





   def 	_snake_case ( self					,			snake_case					,			snake_case=(-1.0, 1.0)					,			snake_case=False      )				->  int:
      """simple docstring"""





      a__		,     a__	:						Any      		=  output_range
      if clip:
         a__	:						Tuple      		=  torch.clip(snake_case					,			self.min_value					,			self.max_value      )
      # Scale to [0, 1].
      a__	:						Optional[Any]      		=  (features - self.min_value) / (self.max_value - self.min_value)
      # Scale to [min_out, max_out].
      return zero_one * (max_out - min_out) + min_out





   def 	_snake_case ( self					,			snake_case					,			snake_case=(-1.0, 1.0)					,			snake_case=False      )				->  Optional[int]:
      """simple docstring"""





      a__		,     a__	:						str      		=  input_range
      a__	:						Tuple      		=  torch.clip(snake_case					,			snake_case					,			snake_case      ) if clip else outputs
      # Scale to [0, 1].
      a__	:						str      		=  (outputs - min_out) / (max_out - min_out)
      # Scale to [self.min_value, self.max_value].
      return zero_one * (self.max_value - self.min_value) + self.min_value





   def 	_snake_case ( self					,			snake_case					,			snake_case					,			snake_case      )				->  str:
      """simple docstring"""





      a__	:						str      		=  input_tokens > 0
      a__		,     a__	:						Dict      		=  self.notes_encoder(
          encoder_input_tokens=snake_case					,			encoder_inputs_mask=snake_case      )

      a__		,     a__	:						int      		=  self.continuous_encoder(
          encoder_inputs=snake_case					,			encoder_inputs_mask=snake_case      )

      return [(tokens_encoded, tokens_mask), (continuous_encoded, continuous_mask)]





   def 	_snake_case ( self					,			snake_case					,			snake_case					,			snake_case      )				->  Union[str, Any]:
      """simple docstring"""





      a__	:						Union[str, Any]      		=  noise_time
      if not torch.is_tensor(snake_case      ):
         a__	:						List[Any]      		=  torch.tensor([timesteps]					,			dtype=torch.long					,			device=input_tokens.device      )
      elif torch.is_tensor(snake_case      ) and len(timesteps.shape      ) == 0:
         a__	:						str      		=  timesteps[None].to(input_tokens.device      )

      # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
      a__	:						Tuple      		=  timesteps * torch.ones(input_tokens.shape[0]					,			dtype=timesteps.dtype					,			device=timesteps.device      )

      a__	:						List[Any]      		=  self.decoder(
          encodings_and_masks=snake_case					,			decoder_input_tokens=snake_case					,			decoder_noise_time=snake_case      )
      return logits





   @torch.no_grad()
   def __call__( self					,			snake_case					,			snake_case = None					,			snake_case = 100					,			snake_case = True					,			snake_case = "numpy"					,			snake_case = None					,			snake_case = 1					,			)				->  Union[AudioPipelineOutput, Tuple]:
      """simple docstring"""





      if (callback_steps is None) or (
          callback_steps is not None and (not isinstance(snake_case					,			snake_case      ) or callback_steps <= 0)
      ):
         raise ValueError(
             F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
             F""" {type(snake_case      )}."""      )

      a__	:						Tuple      		=  np.zeros([1, TARGET_FEATURE_LENGTH, self.n_dims]					,			dtype=np.floataa      )
      a__	:						int      		=  np.zeros([1, 0, self.n_dims]					,			np.floataa      )
      a__	:						List[Any]      		=  torch.ones((1, TARGET_FEATURE_LENGTH)					,			dtype=snake_case					,			device=self.device      )

      for i, encoder_input_tokens in enumerate(snake_case      ):
         if i == 0:
            a__	:						Dict      		=  torch.from_numpy(pred_mel[:1].copy()      ).to(
                device=self.device					,			dtype=self.decoder.dtype      )
            # The first chunk has no previous context.
            a__	:						int      		=  torch.zeros((1, TARGET_FEATURE_LENGTH)					,			dtype=snake_case					,			device=self.device      )
         else:
            # The full song pipeline does not feed in a context feature, so the mask
            # will be all 0s after the feature converter. Because we know we're
            # feeding in a full context chunk from the previous prediction, set it
            # to all 1s.
            a__	:						str      		=  ones

         a__	:						Tuple      		=  self.scale_features(
             snake_case					,			output_range=[-1.0, 1.0]					,			clip=snake_case      )

         a__	:						Dict      		=  self.encode(
             input_tokens=torch.IntTensor([encoder_input_tokens]      ).to(device=self.device      )					,			continuous_inputs=snake_case					,			continuous_mask=snake_case					,			)

         # Sample encoder_continuous_inputs shaped gaussian noise to begin loop
         a__	:						Dict      		=  randn_tensor(
             shape=encoder_continuous_inputs.shape					,			generator=snake_case					,			device=self.device					,			dtype=self.decoder.dtype					,			)

         # set step values
         self.scheduler.set_timesteps(snake_case      )

         # Denoising diffusion loop
         for j, t in enumerate(self.progress_bar(self.scheduler.timesteps      )      ):
            a__	:						str      		=  self.decode(
                encodings_and_masks=snake_case					,			input_tokens=snake_case					,			noise_time=t / self.scheduler.config.num_train_timesteps					,			)

            # Compute previous output: x_t -> x_t-1
            a__	:						Any      		=  self.scheduler.step(snake_case					,			snake_case					,			snake_case					,			generator=snake_case      ).prev_sample

         a__	:						Optional[int]      		=  self.scale_to_features(snake_case					,			input_range=[-1.0, 1.0]      )
         a__	:						Optional[int]      		=  mel[:1]
         a__	:						Union[str, Any]      		=  mel.cpu().float().numpy()

         a__	:						Dict      		=  np.concatenate([full_pred_mel, pred_mel[:1]]					,			axis=1      )

         # call the callback, if provided
         if callback is not None and i % callback_steps == 0:
            callback(snake_case					,			snake_case      )

         logger.info("Generated segment"					,			snake_case      )

      if output_type == "numpy" and not is_onnx_available():
         raise ValueError(
             "Cannot return output in 'np' format if ONNX is not available. Make sure to have ONNX installed or set 'output_type' to 'mel'."      )
      elif output_type == "numpy" and self.melgan is None:
         raise ValueError(
             "Cannot return output in 'np' format if melgan component is not defined. Make sure to define `self.melgan` or set 'output_type' to 'mel'."      )

      if output_type == "numpy":
         a__	:						Union[str, Any]      		=  self.melgan(input_features=full_pred_mel.astype(np.floataa      )      )
      else:
         a__	:						Dict      		=  full_pred_mel

      if not return_dict:
         return (output,)

      return AudioPipelineOutput(audios=snake_case      )








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       _A  (      lowerCamelCase   ):
   a__	:						Optional[int]      		=  ["layers", "blocks"]
   for k in ignore_keys:
      state_dict.pop(lowerCamelCase							, lowerCamelCase   )


SCREAMING_SNAKE_CASE__       :			List[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   ):
   a__	:						Tuple      		=  list(s_dict.keys()   )
   for key in keys:
      a__	:						Optional[Any]      		=  key
      for k, v in WHISPER_MAPPING.items():
         if k in key:
            a__	:						Optional[int]      		=  new_key.replace(lowerCamelCase							, lowerCamelCase   )

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

      a__	:						Dict      		=  s_dict.pop(lowerCamelCase   )
   return s_dict






def       _A  (      lowerCamelCase   ):
   a__		,     a__	:						Any      		=  emb.weight.shape
   a__	:						Optional[Any]      		=  nn.Linear(lowerCamelCase							, lowerCamelCase							, bias=lowerCamelCase   )
   a__	:						Optional[Any]      		=  emb.weight.data
   return lin_layer






def       _A  (      lowerCamelCase							, lowerCamelCase   ):
   os.makedirs(lowerCamelCase							, exist_ok=lowerCamelCase   )
   a__	:						Optional[Any]      		=  os.path.basename(lowerCamelCase   )

   a__	:						List[Any]      		=  url.split("/"   )[-2]
   a__	:						Tuple      		=  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__	:						Any      		=  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:
            a__	:						Optional[Any]      		=  source.read(8192   )
            if not buffer:
               break

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

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

   a__	:						Tuple      		=  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__	:						Optional[Any]      		=  WhisperForConditionalGeneration(lowerCamelCase   )
   a__		,     a__	:						Tuple      		=  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__	:						Any      		=  make_linear_from_emb(model.model.decoder.embed_tokens   )
   else:
      a__	:						str      		=  proj_out_weights

   model.save_pretrained(lowerCamelCase   )


if __name__ == "__main__":
  SCREAMING_SNAKE_CASE__       :			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.""")
  SCREAMING_SNAKE_CASE__       :			List[str]			      =						parser.parse_args()

  convert_openai_whisper_to_tfms(args.checkpoint_path, args.pytorch_dump_folder_path)

