



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


UpperCamelCase__:   Optional[Any]								=			["model.decoder.embed_positions.weights"]


def  snake_case_     (					_lowerCAmelCase  :						str  )      ->   Tuple:
							if "emb" in name:
														UpperCAmelCase :       int 							=		name.replace('''emb'''   ,      '''model.decoder.embed_tokens'''  )
							if "transformer" in name:
														UpperCAmelCase :       Optional[Any] 							=		name.replace('''transformer'''   ,      '''model.decoder'''  )
							if "cross_attention" in name:
														UpperCAmelCase :       int 							=		name.replace('''cross_attention'''   ,      '''encoder_attn'''  )
							if "linear1" in name:
														UpperCAmelCase :       List[str] 							=		name.replace('''linear1'''   ,      '''fc1'''  )
							if "linear2" in name:
														UpperCAmelCase :       Any 							=		name.replace('''linear2'''   ,      '''fc2'''  )
							if "norm1" in name:
														UpperCAmelCase :       str 							=		name.replace('''norm1'''   ,      '''self_attn_layer_norm'''  )
							if "norm_cross" in name:
														UpperCAmelCase :       str 							=		name.replace('''norm_cross'''   ,      '''encoder_attn_layer_norm'''  )
							if "norm2" in name:
														UpperCAmelCase :       List[str] 							=		name.replace('''norm2'''   ,      '''final_layer_norm'''  )
							if "out_norm" in name:
														UpperCAmelCase :       Dict 							=		name.replace('''out_norm'''   ,      '''model.decoder.layer_norm'''  )
							if "linears" in name:
														UpperCAmelCase :       Optional[int] 							=		name.replace('''linears'''   ,      '''lm_heads'''  )
							if "condition_provider.conditioners.description.output_proj" in name:
														UpperCAmelCase :       List[str] 							=		name.replace('''condition_provider.conditioners.description.output_proj'''   ,      '''enc_to_dec_proj'''  )
							return name


def  snake_case_     (					_lowerCAmelCase  :						OrderedDict   ,      _lowerCAmelCase  :						int  )      ->   Tuple[Dict, Dict]:
							UpperCAmelCase :       int 							=		list(state_dict.keys()  )
							UpperCAmelCase :       Tuple 							=		{}
							for key in keys:
														UpperCAmelCase :       Optional[int] 							=		state_dict.pop(_lowerCAmelCase  )
														UpperCAmelCase :       Union[str, Any] 							=		rename_keys(_lowerCAmelCase  )
														if "in_proj_weight" in key:
																					# split fused qkv proj
																					UpperCAmelCase :       Union[str, Any] 							=		val[:hidden_size, :]
																					UpperCAmelCase :       str 							=		val[hidden_size : 2 * hidden_size, :]
																					UpperCAmelCase :       str 							=		val[-hidden_size:, :]
														elif "enc_to_dec_proj" in key:
																					UpperCAmelCase :       Tuple 							=		val
														else:
																					UpperCAmelCase :       int 							=		val
							return state_dict, enc_dec_proj_state_dict


def  snake_case_     (					_lowerCAmelCase  :						str  )      ->   MusicgenDecoderConfig:
							if checkpoint == "small":
														# default config values
														UpperCAmelCase :       Optional[int] 							=		1024
														UpperCAmelCase :       List[Any] 							=		24
														UpperCAmelCase :       List[Any] 							=		16
							elif checkpoint == "medium":
														UpperCAmelCase :       Tuple 							=		1536
														UpperCAmelCase :       Union[str, Any] 							=		48
														UpperCAmelCase :       Tuple 							=		24
							elif checkpoint == "large":
														UpperCAmelCase :       Tuple 							=		2048
														UpperCAmelCase :       List[Any] 							=		48
														UpperCAmelCase :       List[Any] 							=		32
							else:
														raise ValueError(f"""Checkpoint should be one of `['small', 'medium', 'large']`, got {checkpoint}."""  )
							UpperCAmelCase :       List[Any] 							=		MusicgenDecoderConfig(
							    hidden_size=_lowerCAmelCase   ,      ffn_dim=hidden_size * 4   ,      num_hidden_layers=_lowerCAmelCase   ,      num_attention_heads=_lowerCAmelCase   ,      )
							return config


@torch.no_grad()
def  snake_case_     (					_lowerCAmelCase  :						List[str]   ,      _lowerCAmelCase  :						List[Any]=None   ,      _lowerCAmelCase  :						Optional[Any]=None   ,      _lowerCAmelCase  :						Dict="cpu"  )      ->   str:
							UpperCAmelCase :       List[str] 							=		MusicGen.get_pretrained(_lowerCAmelCase   ,      device=_lowerCAmelCase  )
							UpperCAmelCase :       Dict 							=		decoder_config_from_checkpoint(_lowerCAmelCase  )

							UpperCAmelCase :       Any 							=		fairseq_model.lm.state_dict()
							UpperCAmelCase							,		UpperCAmelCase :       Optional[int] 							=		rename_state_dict(
							    _lowerCAmelCase   ,      hidden_size=decoder_config.hidden_size  )

							UpperCAmelCase :       Tuple 							=		TaEncoderModel.from_pretrained('''t5-base'''  )
							UpperCAmelCase :       Dict 							=		EncodecModel.from_pretrained('''facebook/encodec_32khz'''  )
							UpperCAmelCase :       Optional[int] 							=		MusicgenForCausalLM(_lowerCAmelCase  ).eval()

							# load all decoder weights - expect that we'll be missing embeddings and enc-dec projection
							UpperCAmelCase							,		UpperCAmelCase :       Optional[int] 							=		decoder.load_state_dict(_lowerCAmelCase   ,      strict=_lowerCAmelCase  )

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

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

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

							# init the composite model
							UpperCAmelCase :       Any 							=		MusicgenForConditionalGeneration(text_encoder=_lowerCAmelCase   ,      audio_encoder=_lowerCAmelCase   ,      decoder=_lowerCAmelCase  )

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

							# check we can do a forward pass
							UpperCAmelCase :       Union[str, Any] 							=		torch.arange(0   ,      8   ,      dtype=torch.long  ).reshape(2   ,      -1  )
							UpperCAmelCase :       List[Any] 							=		input_ids.reshape(2 * 4   ,      -1  )

							with torch.no_grad():
														UpperCAmelCase :       List[str] 							=		model(input_ids=_lowerCAmelCase   ,      decoder_input_ids=_lowerCAmelCase  ).logits

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

							# now construct the processor
							UpperCAmelCase :       Any 							=		AutoTokenizer.from_pretrained('''t5-base'''  )
							UpperCAmelCase :       Optional[Any] 							=		AutoFeatureExtractor.from_pretrained('''facebook/encodec_32khz'''   ,      padding_side='''left'''  )

							UpperCAmelCase :       int 							=		MusicgenProcessor(feature_extractor=_lowerCAmelCase   ,      tokenizer=_lowerCAmelCase  )

							# set the appropriate bos/pad token ids
							UpperCAmelCase :       Optional[int] 							=		2048
							UpperCAmelCase :       List[Any] 							=		2048

							# set other default generation config params
							UpperCAmelCase :       Tuple 							=		int(30 * audio_encoder.config.frame_rate  )
							UpperCAmelCase :       Optional[Any] 							=		True
							UpperCAmelCase :       str 							=		3.0

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

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


if __name__ == "__main__":
						UpperCamelCase__:   Dict								=			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."
						)

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










'''simple docstring'''






import warnings
from contextlib import contextmanager

from ....processing_utils import ProcessorMixin

class 				SCREAMING_SNAKE_CASE(							A__			):







	"""simple docstring"""






	lowerCamelCase__			 =       """MCTCTFeatureExtractor"""
	lowerCamelCase__			 =       """AutoTokenizer"""







	def __init__(						self			:     Dict		,			__snake_case			:     Optional[int]		,			__snake_case			:     List[str]		)			->		str:
								super().__init__(__snake_case		,			__snake_case		)
								UpperCAmelCase :       List[Any] 							=		self.feature_extractor
								UpperCAmelCase :       Union[str, Any] 							=		False







	def __call__(						self			:     Any		,			*__snake_case			:     List[str]		,			**__snake_case			:     Any		)			->		List[Any]:
								# For backward compatibility
								if self._in_target_context_manager:
															return self.current_processor(*__snake_case		,			**__snake_case		)

								if "raw_speech" in kwargs:
															warnings.warn('''Using `raw_speech` as a keyword argument is deprecated. Use `audio` instead.'''		)
															UpperCAmelCase :       int 							=		kwargs.pop('''raw_speech'''		)
								else:
															UpperCAmelCase :       Union[str, Any] 							=		kwargs.pop('''audio'''		,			__snake_case		)
								UpperCAmelCase :       Optional[Any] 							=		kwargs.pop('''sampling_rate'''		,			__snake_case		)
								UpperCAmelCase :       Dict 							=		kwargs.pop('''text'''		,			__snake_case		)
								if len(__snake_case		) > 0:
															UpperCAmelCase :       Any 							=		args[0]
															UpperCAmelCase :       Optional[int] 							=		args[1:]

								if audio is None and text is None:
															raise ValueError('''You need to specify either an `audio` or `text` input to process.'''		)

								if audio is not None:
															UpperCAmelCase :       List[str] 							=		self.feature_extractor(__snake_case		,			*__snake_case		,			sampling_rate=__snake_case		,			**__snake_case		)
								if text is not None:
															UpperCAmelCase :       int 							=		self.tokenizer(__snake_case		,			**__snake_case		)

								if text is None:
															return inputs
								elif audio is None:
															return encodings
								else:
															UpperCAmelCase :       str 							=		encodings['''input_ids''']
															return inputs







	def      A	(						self			:     List[Any]		,			*__snake_case			:     List[Any]		,			**__snake_case			:     List[Any]		)			->		str:
								return self.tokenizer.batch_decode(*__snake_case		,			**__snake_case		)







	def      A	(						self			:     List[Any]		,			*__snake_case			:     int		,			**__snake_case			:     Optional[int]		)			->		Any:
								# For backward compatibility
								if self._in_target_context_manager:
															return self.current_processor.pad(*__snake_case		,			**__snake_case		)

								UpperCAmelCase :       List[Any] 							=		kwargs.pop('''input_features'''		,			__snake_case		)
								UpperCAmelCase :       Optional[Any] 							=		kwargs.pop('''labels'''		,			__snake_case		)
								if len(__snake_case		) > 0:
															UpperCAmelCase :       List[str] 							=		args[0]
															UpperCAmelCase :       List[Any] 							=		args[1:]

								if input_features is not None:
															UpperCAmelCase :       Tuple 							=		self.feature_extractor.pad(__snake_case		,			*__snake_case		,			**__snake_case		)
								if labels is not None:
															UpperCAmelCase :       Optional[int] 							=		self.tokenizer.pad(__snake_case		,			**__snake_case		)

								if labels is None:
															return input_features
								elif input_features is None:
															return labels
								else:
															UpperCAmelCase :       List[str] 							=		labels['''input_ids''']
															return input_features







	def      A	(						self			:     Union[str, Any]		,			*__snake_case			:     Optional[Any]		,			**__snake_case			:     Optional[int]		)			->		Optional[Any]:
								return self.tokenizer.decode(*__snake_case		,			**__snake_case		)







	@contextmanager
	def      A	(						self			:     Any		)			->		Optional[int]:
								warnings.warn(
								    '''`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your '''
								    '''labels by using the argument `text` of the regular `__call__` method (either in the same call as '''
								    '''your audio inputs, or in a separate call.'''		)
								UpperCAmelCase :       Dict 							=		True
								UpperCAmelCase :       List[Any] 							=		self.tokenizer
								yield
								UpperCAmelCase :       Tuple 							=		self.feature_extractor
								UpperCAmelCase :       List[Any] 							=		False





