'''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()
_A			:      Tuple					      =				logging.get_logger(__name__)


_A			:      List[str]					      =				['''model.decoder.embed_positions.weights''']







def      UpperCamelCase_						(					snake_case_       :       Optional[int]      )					->     Tuple:






 '''simple docstring'''

 if "emb" in name:
  __lowerCAmelCase											=						name.replace("""emb"""						,		"""model.decoder.embed_tokens"""      )
 if "transformer" in name:
  __lowerCAmelCase											=						name.replace("""transformer"""						,		"""model.decoder"""      )
 if "cross_attention" in name:
  __lowerCAmelCase											=						name.replace("""cross_attention"""						,		"""encoder_attn"""      )
 if "linear1" in name:
  __lowerCAmelCase											=						name.replace("""linear1"""						,		"""fc1"""      )
 if "linear2" in name:
  __lowerCAmelCase											=						name.replace("""linear2"""						,		"""fc2"""      )
 if "norm1" in name:
  __lowerCAmelCase											=						name.replace("""norm1"""						,		"""self_attn_layer_norm"""      )
 if "norm_cross" in name:
  __lowerCAmelCase											=						name.replace("""norm_cross"""						,		"""encoder_attn_layer_norm"""      )
 if "norm2" in name:
  __lowerCAmelCase											=						name.replace("""norm2"""						,		"""final_layer_norm"""      )
 if "out_norm" in name:
  __lowerCAmelCase											=						name.replace("""out_norm"""						,		"""model.decoder.layer_norm"""      )
 if "linears" in name:
  __lowerCAmelCase											=						name.replace("""linears"""						,		"""lm_heads"""      )
 if "condition_provider.conditioners.description.output_proj" in name:
  __lowerCAmelCase											=						name.replace("""condition_provider.conditioners.description.output_proj"""						,		"""enc_to_dec_proj"""      )
 return name







def      UpperCamelCase_						(					snake_case_       :       OrderedDict						,		snake_case_       :       int      )					->     Tuple[Dict, Dict]:






 '''simple docstring'''

 __lowerCAmelCase											=						list(state_dict.keys()      )
 __lowerCAmelCase											=						{}
 for key in keys:
  __lowerCAmelCase											=						state_dict.pop(snake_case_      )
  __lowerCAmelCase											=						rename_keys(snake_case_      )
  if "in_proj_weight" in key:
   # split fused qkv proj
   __lowerCAmelCase											=						val[:hidden_size, :]
   __lowerCAmelCase											=						val[hidden_size : 2 * hidden_size, :]
   __lowerCAmelCase											=						val[-hidden_size:, :]
  elif "enc_to_dec_proj" in key:
   __lowerCAmelCase											=						val
  else:
   __lowerCAmelCase											=						val
 return state_dict, enc_dec_proj_state_dict







def      UpperCamelCase_						(					snake_case_       :       str      )					->     MusicgenDecoderConfig:






 '''simple docstring'''

 if checkpoint == "small":
  # default config values
  __lowerCAmelCase											=						10_24
  __lowerCAmelCase											=						24
  __lowerCAmelCase											=						16
 elif checkpoint == "medium":
  __lowerCAmelCase											=						15_36
  __lowerCAmelCase											=						48
  __lowerCAmelCase											=						24
 elif checkpoint == "large":
  __lowerCAmelCase											=						20_48
  __lowerCAmelCase											=						48
  __lowerCAmelCase											=						32
 else:
  raise ValueError(f"""Checkpoint should be one of `['small', 'medium', 'large']`, got {checkpoint}."""      )
 __lowerCAmelCase											=						MusicgenDecoderConfig(
     hidden_size=snake_case_						,		ffn_dim=hidden_size * 4						,		num_hidden_layers=snake_case_						,		num_attention_heads=snake_case_						,		)
 return config







@torch.no_grad()
def      UpperCamelCase_						(					snake_case_       :       Optional[Any]						,		snake_case_       :       List[str]=None						,		snake_case_       :       Any=None						,		snake_case_       :       Dict="cpu"      )					->     Tuple:






 '''simple docstring'''

 __lowerCAmelCase											=						MusicGen.get_pretrained(snake_case_						,		device=snake_case_      )
 __lowerCAmelCase											=						decoder_config_from_checkpoint(snake_case_      )

 __lowerCAmelCase											=						fairseq_model.lm.state_dict()
 __lowerCAmelCase							,       __lowerCAmelCase											=						rename_state_dict(
     snake_case_						,		hidden_size=decoder_config.hidden_size      )

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

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

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

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

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

 # init the composite model
 __lowerCAmelCase											=						MusicgenForConditionalGeneration(text_encoder=snake_case_						,		audio_encoder=snake_case_						,		decoder=snake_case_      )

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

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

 with torch.no_grad():
  __lowerCAmelCase											=						model(input_ids=snake_case_						,		decoder_input_ids=snake_case_      ).logits

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

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

 __lowerCAmelCase											=						MusicgenProcessor(feature_extractor=snake_case_						,		tokenizer=snake_case_      )

 # set the appropriate bos/pad token ids
 __lowerCAmelCase											=						20_48
 __lowerCAmelCase											=						20_48

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

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

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


if __name__ == "__main__":
 _A			:      str					      =				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.'''
 )

 _A			:      Optional[int]					      =				parser.parse_args()
 convert_musicgen_checkpoint(args.checkpoint, args.pytorch_dump_folder, args.push_to_hub)







'''simple docstring'''

from typing import Optional, Tuple, Union

import flax
import flax.linen as nn
import jax
import jax.numpy as jnp
from flax.core.frozen_dict import FrozenDict

from ..configuration_utils import ConfigMixin, flax_register_to_config
from ..utils import BaseOutput
from .embeddings_flax import FlaxTimestepEmbedding, FlaxTimesteps
from .modeling_flax_utils import FlaxModelMixin
from .unet_ad_blocks_flax import (
    FlaxCrossAttnDownBlockaD,
    FlaxCrossAttnUpBlockaD,
    FlaxDownBlockaD,
    FlaxUNetMidBlockaDCrossAttn,
    FlaxUpBlockaD,
)







@flax.struct.dataclass
class 				_lowercase     (				UpperCAmelCase__			):


 '''simple docstring'''






 _SCREAMING_SNAKE_CASE     :       jnp.ndarray







@flax_register_to_config
class 				_lowercase     (				nn.Module  ,							UpperCAmelCase__  ,							UpperCAmelCase__			):


 '''simple docstring'''






 _SCREAMING_SNAKE_CASE     :       int      	=					32
 _SCREAMING_SNAKE_CASE     :       int      	=					4
 _SCREAMING_SNAKE_CASE     :       int      	=					4
 _SCREAMING_SNAKE_CASE     :       Tuple[str]      	=					(
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "CrossAttnDownBlock2D",
     "DownBlock2D",
 )
 _SCREAMING_SNAKE_CASE     :       Tuple[str]      	=					("UpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D", "CrossAttnUpBlock2D")
 _SCREAMING_SNAKE_CASE     :       Union[bool, Tuple[bool]]      	=					False
 _SCREAMING_SNAKE_CASE     :       Tuple[int]      	=					(320, 640, 1280, 1280)
 _SCREAMING_SNAKE_CASE     :       int      	=					2
 _SCREAMING_SNAKE_CASE     :       Union[int, Tuple[int]]      	=					8
 _SCREAMING_SNAKE_CASE     :       Optional[Union[int, Tuple[int]]]      	=					None
 _SCREAMING_SNAKE_CASE     :       int      	=					1280
 _SCREAMING_SNAKE_CASE     :       float      	=					0.0
 _SCREAMING_SNAKE_CASE     :       bool      	=					False
 _SCREAMING_SNAKE_CASE     :       jnp.dtype      	=					jnp.floataa
 _SCREAMING_SNAKE_CASE     :       bool      	=					True
 _SCREAMING_SNAKE_CASE     :       int      	=					0
 _SCREAMING_SNAKE_CASE     :       bool      	=					False
 def a					(      self   :	Dict      ,       SCREAMING_SNAKE_CASE__   :	jax.random.KeyArray							)  ->    FrozenDict:
  # init input tensors
  __lowerCAmelCase											=						(1, self.in_channels, self.sample_size, self.sample_size)
  __lowerCAmelCase											=						jnp.zeros(SCREAMING_SNAKE_CASE__      ,       dtype=jnp.floataa							)
  __lowerCAmelCase											=						jnp.ones((1,)      ,       dtype=jnp.intaa							)
  __lowerCAmelCase											=						jnp.zeros((1, 1, self.cross_attention_dim)      ,       dtype=jnp.floataa							)

  __lowerCAmelCase							,       __lowerCAmelCase											=						jax.random.split(SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase											=						{"""params""": params_rng, """dropout""": dropout_rng}

  return self.init(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__							)["params"]
 def a					(      self   :	int							)  ->    List[str]:
  __lowerCAmelCase											=						self.block_out_channels
  __lowerCAmelCase											=						block_out_channels[0] * 4

  if self.num_attention_heads is not None:
   raise ValueError(
       """At the moment it is not possible to define the number of attention heads via `num_attention_heads` because of a naming issue as described in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131. Passing `num_attention_heads` will only be supported in diffusers v0.19."""							)

  # If `num_attention_heads` is not defined (which is the case for most models)
  # it will default to `attention_head_dim`. This looks weird upon first reading it and it is.
  # The reason for this behavior is to correct for incorrectly named variables that were introduced
  # when this library was created. The incorrect naming was only discovered much later in https://github.com/huggingface/diffusers/issues/2011#issuecomment-1547958131
  # Changing `attention_head_dim` to `num_attention_heads` for 40,000+ configurations is too backwards breaking
  # which is why we correct for the naming here.
  __lowerCAmelCase											=						self.num_attention_heads or self.attention_head_dim

  # input
  __lowerCAmelCase											=						nn.Conv(
      block_out_channels[0]      ,       kernel_size=(3, 3)      ,       strides=(1, 1)      ,       padding=((1, 1), (1, 1))      ,       dtype=self.dtype      ,       )

  # time
  __lowerCAmelCase											=						FlaxTimesteps(
      block_out_channels[0]      ,       flip_sin_to_cos=self.flip_sin_to_cos      ,       freq_shift=self.config.freq_shift							)
  __lowerCAmelCase											=						FlaxTimestepEmbedding(SCREAMING_SNAKE_CASE__      ,       dtype=self.dtype							)

  __lowerCAmelCase											=						self.only_cross_attention
  if isinstance(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__							):
   __lowerCAmelCase											=						(only_cross_attention,) * len(self.down_block_types							)

  if isinstance(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__							):
   __lowerCAmelCase											=						(num_attention_heads,) * len(self.down_block_types							)

  # down
  __lowerCAmelCase											=						[]
  __lowerCAmelCase											=						block_out_channels[0]
  for i, down_block_type in enumerate(self.down_block_types							):
   __lowerCAmelCase											=						output_channel
   __lowerCAmelCase											=						block_out_channels[i]
   __lowerCAmelCase											=						i == len(SCREAMING_SNAKE_CASE__							) - 1

   if down_block_type == "CrossAttnDownBlock2D":
    __lowerCAmelCase											=						FlaxCrossAttnDownBlockaD(
        in_channels=SCREAMING_SNAKE_CASE__      ,       out_channels=SCREAMING_SNAKE_CASE__      ,       dropout=self.dropout      ,       num_layers=self.layers_per_block      ,       num_attention_heads=num_attention_heads[i]      ,       add_downsample=not is_final_block      ,       use_linear_projection=self.use_linear_projection      ,       only_cross_attention=only_cross_attention[i]      ,       use_memory_efficient_attention=self.use_memory_efficient_attention      ,       dtype=self.dtype      ,       )
   else:
    __lowerCAmelCase											=						FlaxDownBlockaD(
        in_channels=SCREAMING_SNAKE_CASE__      ,       out_channels=SCREAMING_SNAKE_CASE__      ,       dropout=self.dropout      ,       num_layers=self.layers_per_block      ,       add_downsample=not is_final_block      ,       dtype=self.dtype      ,       )

   down_blocks.append(SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase											=						down_blocks

  # mid
  __lowerCAmelCase											=						FlaxUNetMidBlockaDCrossAttn(
      in_channels=block_out_channels[-1]      ,       dropout=self.dropout      ,       num_attention_heads=num_attention_heads[-1]      ,       use_linear_projection=self.use_linear_projection      ,       use_memory_efficient_attention=self.use_memory_efficient_attention      ,       dtype=self.dtype      ,       )

  # up
  __lowerCAmelCase											=						[]
  __lowerCAmelCase											=						list(reversed(SCREAMING_SNAKE_CASE__							)							)
  __lowerCAmelCase											=						list(reversed(SCREAMING_SNAKE_CASE__							)							)
  __lowerCAmelCase											=						list(reversed(SCREAMING_SNAKE_CASE__							)							)
  __lowerCAmelCase											=						reversed_block_out_channels[0]
  for i, up_block_type in enumerate(self.up_block_types							):
   __lowerCAmelCase											=						output_channel
   __lowerCAmelCase											=						reversed_block_out_channels[i]
   __lowerCAmelCase											=						reversed_block_out_channels[min(i + 1      ,       len(SCREAMING_SNAKE_CASE__							) - 1							)]

   __lowerCAmelCase											=						i == len(SCREAMING_SNAKE_CASE__							) - 1

   if up_block_type == "CrossAttnUpBlock2D":
    __lowerCAmelCase											=						FlaxCrossAttnUpBlockaD(
        in_channels=SCREAMING_SNAKE_CASE__      ,       out_channels=SCREAMING_SNAKE_CASE__      ,       prev_output_channel=SCREAMING_SNAKE_CASE__      ,       num_layers=self.layers_per_block + 1      ,       num_attention_heads=reversed_num_attention_heads[i]      ,       add_upsample=not is_final_block      ,       dropout=self.dropout      ,       use_linear_projection=self.use_linear_projection      ,       only_cross_attention=only_cross_attention[i]      ,       use_memory_efficient_attention=self.use_memory_efficient_attention      ,       dtype=self.dtype      ,       )
   else:
    __lowerCAmelCase											=						FlaxUpBlockaD(
        in_channels=SCREAMING_SNAKE_CASE__      ,       out_channels=SCREAMING_SNAKE_CASE__      ,       prev_output_channel=SCREAMING_SNAKE_CASE__      ,       num_layers=self.layers_per_block + 1      ,       add_upsample=not is_final_block      ,       dropout=self.dropout      ,       dtype=self.dtype      ,       )

   up_blocks.append(SCREAMING_SNAKE_CASE__							)
   __lowerCAmelCase											=						output_channel
  __lowerCAmelCase											=						up_blocks

  # out
  __lowerCAmelCase											=						nn.GroupNorm(num_groups=32      ,       epsilon=1e-5							)
  __lowerCAmelCase											=						nn.Conv(
      self.out_channels      ,       kernel_size=(3, 3)      ,       strides=(1, 1)      ,       padding=((1, 1), (1, 1))      ,       dtype=self.dtype      ,       )





 def __call__(      self   :	Dict      ,       SCREAMING_SNAKE_CASE__   :	int      ,       SCREAMING_SNAKE_CASE__   :	Tuple      ,       SCREAMING_SNAKE_CASE__   :	Union[str, Any]      ,       SCREAMING_SNAKE_CASE__   :	Optional[Any]=None      ,       SCREAMING_SNAKE_CASE__   :	List[str]=None      ,       SCREAMING_SNAKE_CASE__   :	bool = True      ,       SCREAMING_SNAKE_CASE__   :	bool = False      ,       )  ->    Union[FlaxUNetaDConditionOutput, Tuple]:
  # 1. time
  if not isinstance(SCREAMING_SNAKE_CASE__      ,       jnp.ndarray							):
   __lowerCAmelCase											=						jnp.array([timesteps]      ,       dtype=jnp.intaa							)
  elif isinstance(SCREAMING_SNAKE_CASE__      ,       jnp.ndarray							) and len(timesteps.shape							) == 0:
   __lowerCAmelCase											=						timesteps.astype(dtype=jnp.floataa							)
   __lowerCAmelCase											=						jnp.expand_dims(SCREAMING_SNAKE_CASE__      ,       0							)

  __lowerCAmelCase											=						self.time_proj(SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase											=						self.time_embedding(SCREAMING_SNAKE_CASE__							)

  # 2. pre-process
  __lowerCAmelCase											=						jnp.transpose(SCREAMING_SNAKE_CASE__      ,       (0, 2, 3, 1)							)
  __lowerCAmelCase											=						self.conv_in(SCREAMING_SNAKE_CASE__							)

  # 3. down
  __lowerCAmelCase											=						(sample,)
  for down_block in self.down_blocks:
   if isinstance(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__							):
    __lowerCAmelCase							,       __lowerCAmelCase											=						down_block(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__      ,       deterministic=not train							)
   else:
    __lowerCAmelCase							,       __lowerCAmelCase											=						down_block(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__      ,       deterministic=not train							)
   down_block_res_samples += res_samples

  if down_block_additional_residuals is not None:
   __lowerCAmelCase											=						()

   for down_block_res_sample, down_block_additional_residual in zip(
       SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__							):
    down_block_res_sample += down_block_additional_residual
    new_down_block_res_samples += (down_block_res_sample,)

   __lowerCAmelCase											=						new_down_block_res_samples

  # 4. mid
  __lowerCAmelCase											=						self.mid_block(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__      ,       deterministic=not train							)

  if mid_block_additional_residual is not None:
   sample += mid_block_additional_residual

  # 5. up
  for up_block in self.up_blocks:
   __lowerCAmelCase											=						down_block_res_samples[-(self.layers_per_block + 1) :]
   __lowerCAmelCase											=						down_block_res_samples[: -(self.layers_per_block + 1)]
   if isinstance(SCREAMING_SNAKE_CASE__      ,       SCREAMING_SNAKE_CASE__							):
    __lowerCAmelCase											=						up_block(
        SCREAMING_SNAKE_CASE__      ,       temb=SCREAMING_SNAKE_CASE__      ,       encoder_hidden_states=SCREAMING_SNAKE_CASE__      ,       res_hidden_states_tuple=SCREAMING_SNAKE_CASE__      ,       deterministic=not train      ,       )
   else:
    __lowerCAmelCase											=						up_block(SCREAMING_SNAKE_CASE__      ,       temb=SCREAMING_SNAKE_CASE__      ,       res_hidden_states_tuple=SCREAMING_SNAKE_CASE__      ,       deterministic=not train							)

        # 6. post-process
  __lowerCAmelCase											=						self.conv_norm_out(SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase											=						nn.silu(SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase											=						self.conv_out(SCREAMING_SNAKE_CASE__							)
  __lowerCAmelCase											=						jnp.transpose(SCREAMING_SNAKE_CASE__      ,       (0, 3, 1, 2)							)

  if not return_dict:
   return (sample,)

  return FlaxUNetaDConditionOutput(sample=SCREAMING_SNAKE_CASE__							)






