



"""simple docstring"""





import argparse
from pathlib import Path
from typing import Dict, OrderedDict, Tuple

import torch
from audiocraft.models import MusicGen

from transformers import (
    AutoFeatureExtractor,
    AutoTokenizer,
    EncodecModel,
    MusicgenDecoderConfig,
    MusicgenForConditionalGeneration,
    MusicgenProcessor,
    TaEncoderModel,
)
from transformers.models.musicgen.modeling_musicgen import MusicgenForCausalLM
from transformers.utils import logging


logging.set_verbosity_info()
lowerCamelCase      =		logging.get_logger(__name__)


lowerCamelCase      =		["model.decoder.embed_positions.weights"]
def      a__			(   lowerCAmelCase__   ):




      if "emb" in name:
            UpperCAmelCase_    				=							name.replace("emb"			,    "model.decoder.embed_tokens"   )
      if "transformer" in name:
            UpperCAmelCase_    				=							name.replace("transformer"			,    "model.decoder"   )
      if "cross_attention" in name:
            UpperCAmelCase_    				=							name.replace("cross_attention"			,    "encoder_attn"   )
      if "linear1" in name:
            UpperCAmelCase_    				=							name.replace("linear1"			,    "fc1"   )
      if "linear2" in name:
            UpperCAmelCase_    				=							name.replace("linear2"			,    "fc2"   )
      if "norm1" in name:
            UpperCAmelCase_    				=							name.replace("norm1"			,    "self_attn_layer_norm"   )
      if "norm_cross" in name:
            UpperCAmelCase_    				=							name.replace("norm_cross"			,    "encoder_attn_layer_norm"   )
      if "norm2" in name:
            UpperCAmelCase_    				=							name.replace("norm2"			,    "final_layer_norm"   )
      if "out_norm" in name:
            UpperCAmelCase_    				=							name.replace("out_norm"			,    "model.decoder.layer_norm"   )
      if "linears" in name:
            UpperCAmelCase_    				=							name.replace("linears"			,    "lm_heads"   )
      if "condition_provider.conditioners.description.output_proj" in name:
            UpperCAmelCase_    				=							name.replace("condition_provider.conditioners.description.output_proj"			,    "enc_to_dec_proj"   )
      return name
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__   ):
      UpperCAmelCase_    				=							list(state_dict.keys()   )
      UpperCAmelCase_    				=							{}
      for key in keys:
            UpperCAmelCase_    				=							state_dict.pop(_A   )
            UpperCAmelCase_    				=							rename_keys(_A   )
            if "in_proj_weight" in key:
                  # split fused qkv proj
                  UpperCAmelCase_    				=							val[:hidden_size, :]
                  UpperCAmelCase_    				=							val[hidden_size : 2 * hidden_size, :]
                  UpperCAmelCase_    				=							val[-hidden_size:, :]
            elif "enc_to_dec_proj" in key:
                  UpperCAmelCase_    				=							val
            else:
                  UpperCAmelCase_    				=							val
      return state_dict, enc_dec_proj_state_dict
def      a__			(   lowerCAmelCase__   ):




      if checkpoint == "small":
            # default config values
            UpperCAmelCase_    				=							1024
            UpperCAmelCase_    				=							24
            UpperCAmelCase_    				=							16
      elif checkpoint == "medium":
            UpperCAmelCase_    				=							1536
            UpperCAmelCase_    				=							48
            UpperCAmelCase_    				=							24
      elif checkpoint == "large":
            UpperCAmelCase_    				=							2048
            UpperCAmelCase_    				=							48
            UpperCAmelCase_    				=							32
      else:
            raise ValueError(f"""Checkpoint should be one of `['small', 'medium', 'large']`, got {checkpoint}."""   )
      UpperCAmelCase_    				=							MusicgenDecoderConfig(
          hidden_size=_A			,    ffn_dim=hidden_size * 4			,    num_hidden_layers=_A			,    num_attention_heads=_A			,    )
      return config
@torch.no_grad()
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__=None			,    lowerCAmelCase__=None			,    lowerCAmelCase__="cpu"   ):
      UpperCAmelCase_    				=							MusicGen.get_pretrained(_A			,    device=_A   )
      UpperCAmelCase_    				=							decoder_config_from_checkpoint(_A   )

      UpperCAmelCase_    				=							fairseq_model.lm.state_dict()
      UpperCAmelCase_    				=							rename_state_dict(
          _A			,    hidden_size=decoder_config.hidden_size   )

      UpperCAmelCase_    				=							TaEncoderModel.from_pretrained("t5-base"   )
      UpperCAmelCase_    				=							EncodecModel.from_pretrained("facebook/encodec_32khz"   )
      UpperCAmelCase_    				=							MusicgenForCausalLM(_A   ).eval()

      # load all decoder weights - expect that we'll be missing embeddings and enc-dec projection
      UpperCAmelCase_    				=							decoder.load_state_dict(_A			,    strict=_A   )

      for key in missing_keys.copy():
            if key.startswith(("text_encoder", "audio_encoder")   ) or key in EXPECTED_MISSING_KEYS:
                  missing_keys.remove(_A   )

      if len(_A   ) > 0:
            raise ValueError(f"""Missing key(s) in state_dict: {missing_keys}"""   )

      if len(_A   ) > 0:
            raise ValueError(f"""Unexpected key(s) in state_dict: {unexpected_keys}"""   )

      # init the composite model
      UpperCAmelCase_    				=							MusicgenForConditionalGeneration(text_encoder=_A			,    audio_encoder=_A			,    decoder=_A   )

      # load the pre-trained enc-dec projection (from the decoder state dict)
      model.enc_to_dec_proj.load_state_dict(_A   )

      # check we can do a forward pass
      UpperCAmelCase_    				=							torch.arange(0			,    8			,    dtype=torch.long   ).reshape(2			,    -1   )
      UpperCAmelCase_    				=							input_ids.reshape(2 * 4			,    -1   )

      with torch.no_grad():
            UpperCAmelCase_    				=							model(input_ids=_A			,    decoder_input_ids=_A   ).logits

      if logits.shape != (8, 1, 2048):
            raise ValueError("Incorrect shape for logits"   )

      # now construct the processor
      UpperCAmelCase_    				=							AutoTokenizer.from_pretrained("t5-base"   )
      UpperCAmelCase_    				=							AutoFeatureExtractor.from_pretrained("facebook/encodec_32khz"			,    padding_side="left"   )

      UpperCAmelCase_    				=							MusicgenProcessor(feature_extractor=_A			,    tokenizer=_A   )

      # set the appropriate bos/pad token ids
      UpperCAmelCase_    				=							2048
      UpperCAmelCase_    				=							2048

      # set other default generation config params
      UpperCAmelCase_    				=							int(30 * audio_encoder.config.frame_rate   )
      UpperCAmelCase_    				=							True
      UpperCAmelCase_    				=							3.0

      if pytorch_dump_folder is not None:
            Path(_A   ).mkdir(exist_ok=_A   )
            logger.info(f"""Saving model {checkpoint} to {pytorch_dump_folder}"""   )
            model.save_pretrained(_A   )
            processor.save_pretrained(_A   )

      if repo_id:
            logger.info(f"""Pushing model {checkpoint} to {repo_id}"""   )
            model.push_to_hub(_A   )
            processor.push_to_hub(_A   )


if __name__ == "__main__":
  lowerCamelCase      =		argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      """--checkpoint""",
      default="""small""",
      type=str,
      help="""Checkpoint size of the MusicGen model you'd like to convert. Can be one of: `['small', 'medium', 'large']`.""",
  )
  parser.add_argument(
      """--pytorch_dump_folder""",
      required=True,
      default=None,
      type=str,
      help="""Path to the output PyTorch model directory.""",
  )
  parser.add_argument(
      """--push_to_hub""", default=None, type=str, help="""Where to upload the converted model on the 🤗 hub."""
  )
  parser.add_argument(
      """--device""", default="""cpu""", type=str, help="""Torch device to run the conversion, either cpu or cuda."""
  )

  lowerCamelCase      =		parser.parse_args()
  convert_musicgen_checkpoint(args.checkpoint, args.pytorch_dump_folder, args.push_to_hub)









"""simple docstring"""





import argparse
import re

import torch
from CLAP import create_model

from transformers import AutoFeatureExtractor, ClapConfig, ClapModel


lowerCamelCase      =		{
    """text_branch""": """text_model""",
    """audio_branch""": """audio_model.audio_encoder""",
    """attn""": """attention.self""",
    """self.proj""": """output.dense""",
    """attention.self_mask""": """attn_mask""",
    """mlp.fc1""": """intermediate.dense""",
    """mlp.fc2""": """output.dense""",
    """norm1""": """layernorm_before""",
    """norm2""": """layernorm_after""",
    """bn0""": """batch_norm""",
}

lowerCamelCase      =		AutoFeatureExtractor.from_pretrained("""laion/clap-htsat-unfused""", truncation="""rand_trunc""")
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__=False   ):
      UpperCAmelCase_				,					UpperCAmelCase_    				=							create_model(
          "HTSAT-tiny"			,    "roberta"			,    lowerCAmelCase__			,    precision="fp32"			,    device="cuda:0" if torch.cuda.is_available() else "cpu"			,    enable_fusion=lowerCAmelCase__			,    fusion_type="aff_2d" if enable_fusion else None			,    )
      return model, model_cfg
def      a__			(   lowerCAmelCase__   ):
      UpperCAmelCase_    				=							{}

      UpperCAmelCase_    				=							r".*sequential.(\d+).*"
      UpperCAmelCase_    				=							r".*_projection.(\d+).*"

      for key, value in state_dict.items():
            # check if any key needs to be modified
            for key_to_modify, new_key in KEYS_TO_MODIFY_MAPPING.items():
                  if key_to_modify in key:
                        UpperCAmelCase_    				=							key.replace(lowerCAmelCase__			,    lowerCAmelCase__   )

            if re.match(lowerCAmelCase__			,    lowerCAmelCase__   ):
                  # replace sequential layers with list
                  UpperCAmelCase_    				=							re.match(lowerCAmelCase__			,    lowerCAmelCase__   ).group(1   )

                  UpperCAmelCase_    				=							key.replace(f"""sequential.{sequential_layer}."""			,    f"""layers.{int(lowerCAmelCase__   )//3}.linear."""   )
            elif re.match(lowerCAmelCase__			,    lowerCAmelCase__   ):
                  UpperCAmelCase_    				=							int(re.match(lowerCAmelCase__			,    lowerCAmelCase__   ).group(1   )   )

                  # Because in CLAP they use `nn.Sequential`...
                  UpperCAmelCase_    				=							1 if projecton_layer == 0 else 2

                  UpperCAmelCase_    				=							key.replace(f"""_projection.{projecton_layer}."""			,    f"""_projection.linear{transformers_projection_layer}."""   )

            if "audio" and "qkv" in key:
                  # split qkv into query key and value
                  UpperCAmelCase_    				=							value
                  UpperCAmelCase_    				=							mixed_qkv.size(0   ) // 3

                  UpperCAmelCase_    				=							mixed_qkv[:qkv_dim]
                  UpperCAmelCase_    				=							mixed_qkv[qkv_dim : qkv_dim * 2]
                  UpperCAmelCase_    				=							mixed_qkv[qkv_dim * 2 :]

                  UpperCAmelCase_    				=							query_layer
                  UpperCAmelCase_    				=							key_layer
                  UpperCAmelCase_    				=							value_layer
            else:
                  UpperCAmelCase_    				=							value

      return model_state_dict
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__=False   ):
      UpperCAmelCase_				,					UpperCAmelCase_    				=							init_clap(lowerCAmelCase__			,    enable_fusion=lowerCAmelCase__   )

      clap_model.eval()
      UpperCAmelCase_    				=							clap_model.state_dict()
      UpperCAmelCase_    				=							rename_state_dict(lowerCAmelCase__   )

      UpperCAmelCase_    				=							ClapConfig()
      UpperCAmelCase_    				=							enable_fusion
      UpperCAmelCase_    				=							ClapModel(lowerCAmelCase__   )

      # ignore the spectrogram embedding layer
      model.load_state_dict(lowerCAmelCase__			,    strict=lowerCAmelCase__   )

      model.save_pretrained(lowerCAmelCase__   )
      transformers_config.save_pretrained(lowerCAmelCase__   )


if __name__ == "__main__":
  lowerCamelCase      =		argparse.ArgumentParser()
  parser.add_argument("""--pytorch_dump_folder_path""", default=None, type=str, help="""Path to the output PyTorch model.""")
  parser.add_argument("""--checkpoint_path""", default=None, type=str, help="""Path to fairseq checkpoint""")
  parser.add_argument("""--config_path""", default=None, type=str, help="""Path to hf config.json of model to convert""")
  parser.add_argument("""--enable_fusion""", action="""store_true""", help="""Whether to enable fusion or not""")
  lowerCamelCase      =		parser.parse_args()

  convert_clap_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.enable_fusion)




