



"""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()
__snake_case							:   List[Any]      =    logging.get_logger(__name__)


__snake_case							:   Dict      =    ['model.decoder.embed_positions.weights']


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


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


def 			_lowercase				(  __snake_case	)						->   MusicgenDecoderConfig:
			if checkpoint == "small":
						# default config values
						__lowerCAmelCase      :  List[str]												=       1_024
						__lowerCAmelCase      :  Union[str, Any]												=       24
						__lowerCAmelCase      :  Optional[Any]												=       16
			elif checkpoint == "medium":
						__lowerCAmelCase      :  Union[str, Any]												=       1_536
						__lowerCAmelCase      :  Any												=       48
						__lowerCAmelCase      :  Dict												=       24
			elif checkpoint == "large":
						__lowerCAmelCase      :  Optional[int]												=       2_048
						__lowerCAmelCase      :  Dict												=       48
						__lowerCAmelCase      :  Any												=       32
			else:
						raise ValueError(F"""Checkpoint should be one of `['small', 'medium', 'large']`, got {checkpoint}."""	)
			__lowerCAmelCase      :  Any												=       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 			_lowercase				(  __snake_case					,__snake_case=None					,__snake_case=None					,__snake_case="cpu"	)						->   int:
			__lowerCAmelCase      :  int												=       MusicGen.get_pretrained(__snake_case					,device=__snake_case	)
			__lowerCAmelCase      :  Optional[int]												=       decoder_config_from_checkpoint(__snake_case	)

			__lowerCAmelCase      :  List[str]												=       fairseq_model.lm.state_dict()
			__lowerCAmelCase						,	__lowerCAmelCase      :  List[str]												=       rename_state_dict(
			    __snake_case					,hidden_size=decoder_config.hidden_size	)

			__lowerCAmelCase      :  List[str]												=       TaEncoderModel.from_pretrained("t5-base"	)
			__lowerCAmelCase      :  int												=       EncodecModel.from_pretrained("facebook/encodec_32khz"	)
			__lowerCAmelCase      :  List[str]												=       MusicgenForCausalLM(__snake_case	).eval()

			# load all decoder weights - expect that we'll be missing embeddings and enc-dec projection
			__lowerCAmelCase						,	__lowerCAmelCase      :  Tuple												=       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      :  str												=       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      :  Tuple												=       torch.arange(0					,8					,dtype=torch.long	).reshape(2					,-1	)
			__lowerCAmelCase      :  Any												=       input_ids.reshape(2 * 4					,-1	)

			with torch.no_grad():
						__lowerCAmelCase      :  Union[str, Any]												=       model(input_ids=__snake_case					,decoder_input_ids=__snake_case	).logits

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

			# now construct the processor
			__lowerCAmelCase      :  Union[str, Any]												=       AutoTokenizer.from_pretrained("t5-base"	)
			__lowerCAmelCase      :  Dict												=       AutoFeatureExtractor.from_pretrained("facebook/encodec_32khz"					,padding_side="left"	)

			__lowerCAmelCase      :  Any												=       MusicgenProcessor(feature_extractor=__snake_case					,tokenizer=__snake_case	)

			# set the appropriate bos/pad token ids
			__lowerCAmelCase      :  List[str]												=       2_048
			__lowerCAmelCase      :  int												=       2_048

			# set other default generation config params
			__lowerCAmelCase      :  Optional[Any]												=       int(30 * audio_encoder.config.frame_rate	)
			__lowerCAmelCase      :  Tuple												=       True
			__lowerCAmelCase      :  Optional[int]												=       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__":
	__snake_case							:   Union[str, Any]      =    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.'
	)

	__snake_case							:   List[Any]      =    parser.parse_args()
	convert_musicgen_checkpoint(args.checkpoint, args.pytorch_dump_folder, args.push_to_hub)




"""simple docstring"""






import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
	from sagemaker import Session, TrainingJobAnalytics
	from sagemaker.huggingface import HuggingFace



@pytest.mark.skipif(
    literal_eval(os.getenv('TEST_SAGEMAKER'  ,		'False'							)							) is not True  ,		reason='Skipping test because should only be run when releasing minor transformers version'  ,		)
@pytest.mark.usefixtures('sm_env'							)
@parameterized_class(
    [
        {
            'framework': 'pytorch',
            'script': 'run_glue.py',
            'model_name_or_path': 'distilbert-base-cased',
            'instance_type': 'ml.g4dn.xlarge',
            'results': {'train_runtime': 6_5_0, 'eval_accuracy': 0.6, 'eval_loss': 0.9},
        },
        {
            'framework': 'tensorflow',
            'script': 'run_tf.py',
            'model_name_or_path': 'distilbert-base-cased',
            'instance_type': 'ml.g4dn.xlarge',
            'results': {'train_runtime': 6_0_0, 'eval_accuracy': 0.3, 'eval_loss': 0.9},
        },
    ]							)
class 			A__   (  unittest.TestCase							):




	'''simple docstring'''





	def 							_SCREAMING_SNAKE_CASE   (     self:		int)				->       Tuple:

				"""simple docstring"""

				if self.framework == "pytorch":
							subprocess.run(
							    F"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()       ,			encoding="utf-8"       ,			check=_SCREAMING_SNAKE_CASE       ,			)
				assert hasattr(self       ,			"env")

	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[Any]       ,			_SCREAMING_SNAKE_CASE:		Any=1)				->       Dict:

				"""simple docstring"""

				return HuggingFace(
				    entry_point=self.script       ,			source_dir=self.env.test_path       ,			role=self.env.role       ,			image_uri=self.env.image_uri       ,			base_job_name=F"""{self.env.base_job_name}-single"""       ,			instance_count=_SCREAMING_SNAKE_CASE       ,			instance_type=self.instance_type       ,			debugger_hook_config=_SCREAMING_SNAKE_CASE       ,			hyperparameters={**self.env.hyperparameters, "model_name_or_path": self.model_name_or_path}       ,			metric_definitions=self.env.metric_definitions       ,			py_version="py36"       ,			)

	def 							_SCREAMING_SNAKE_CASE   (     self:		str       ,			_SCREAMING_SNAKE_CASE:		List[Any])				->       Optional[Any]:

				"""simple docstring"""

				TrainingJobAnalytics(_SCREAMING_SNAKE_CASE).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv""")







	def 							_SCREAMING_SNAKE_CASE   (     self:		str)				->       Optional[Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  Tuple												=       self.create_estimator()

				# run training
				estimator.fit()

				# result dataframe
				__lowerCAmelCase      :  Tuple												=       TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()

				# extract kpis
				__lowerCAmelCase      :  Optional[Any]												=       list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"])
				__lowerCAmelCase      :  Union[str, Any]												=       list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"])
				# get train time from SageMaker job, this includes starting, preprocessing, stopping
				__lowerCAmelCase      :  Tuple												=       (
				    Session().describe_training_job(estimator.latest_training_job.name).get("TrainingTimeInSeconds"       ,			99_9999)
				)

				# assert kpis
				assert train_runtime <= self.results["train_runtime"]
				assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy)
				assert all(t <= self.results["eval_loss"] for t in eval_loss)

				# dump tests result into json file to share in PR
				with open(F"""{estimator.latest_training_job.name}.json"""       ,			"w") as outfile:
							json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}       ,			_SCREAMING_SNAKE_CASE)