





import logging
import os
from dataclasses import dataclass, field
from typing import Dict, Optional

import numpy as np
from utils_multiple_choice import MultipleChoiceDataset, Split, processors

import transformers
from transformers import (
    AutoConfig,
    AutoModelForMultipleChoice,
    AutoTokenizer,
    DataCollatorWithPadding,
    EvalPrediction,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import is_main_process


SCREAMING_SNAKE_CASE_   :      str			=   logging.getLogger(__name__)



def  SCREAMING_SNAKE_CASE				(      snake_case						,   snake_case							)		->    Any:
		return (preds == labels).mean()

@dataclass
class 		snake_case_   :







	'''simple docstring'''
	__UpperCamelCase   =		field(
	    metadata={'''help''': '''Path to pretrained model or model identifier from huggingface.co/models'''}						)
	__UpperCamelCase   =		field(
	    default=UpperCAmelCase_     ,     metadata={'''help''': '''Pretrained config name or path if not the same as model_name'''}						)
	__UpperCamelCase   =		field(
	    default=UpperCAmelCase_     ,     metadata={'''help''': '''Pretrained tokenizer name or path if not the same as model_name'''}						)
	__UpperCamelCase   =		field(
	    default=UpperCAmelCase_     ,     metadata={'''help''': '''Where do you want to store the pretrained models downloaded from huggingface.co'''}     ,     )





@dataclass
class 		snake_case_   :







	'''simple docstring'''
	__UpperCamelCase   =		field(metadata={'''help''': '''The name of the task to train on: ''' + ''', '''.join(processors.keys()						)}						)
	__UpperCamelCase   =		field(metadata={'''help''': '''Should contain the data files for the task.'''}						)
	__UpperCamelCase   =		field(
	    default=1_28     ,     metadata={
	        '''help''': (
	            '''The maximum total input sequence length after tokenization. Sequences longer '''
	            '''than this will be truncated, sequences shorter will be padded.'''
	        )
	    }     ,     )
	__UpperCamelCase   =		field(
	    default=UpperCAmelCase_     ,     metadata={'''help''': '''Overwrite the cached training and evaluation sets'''}						)





def  SCREAMING_SNAKE_CASE				(      )		->    List[str]:
		# See all possible arguments in src/transformers/training_args.py
		# or by passing the --help flag to this script.
		# We now keep distinct sets of args, for a cleaner separation of concerns.

		__lowercase											=     HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)							)
		__lowercase						,      __lowercase						,      __lowercase											=     parser.parse_args_into_dataclasses()

		if (
		    os.path.exists(training_args.output_dir							)
		    and os.listdir(training_args.output_dir							)
		    and training_args.do_train
		    and not training_args.overwrite_output_dir
		):
				raise ValueError(
				    F"Output directory ({training_args.output_dir}) already exists and is not empty. Use"
				    ' --overwrite_output_dir to overcome.'							)

		# Setup logging
		logging.basicConfig(
		    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'						,   datefmt='%m/%d/%Y %H:%M:%S'						,   level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN						,   )
		logger.warning(
		    'Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s'						,   training_args.local_rank						,   training_args.device						,   training_args.n_gpu						,   bool(training_args.local_rank != -1							)						,   training_args.fpaa						,   )
		# Set the verbosity to info of the Transformers logger (on main process only):
		if is_main_process(training_args.local_rank							):
				transformers.utils.logging.set_verbosity_info()
				transformers.utils.logging.enable_default_handler()
				transformers.utils.logging.enable_explicit_format()
		logger.info('Training/evaluation parameters %s'						,   snake_case							)

		# Set seed
		set_seed(training_args.seed							)

		try:
				__lowercase											=     processors[data_args.task_name]()
				__lowercase											=     processor.get_labels()
				__lowercase											=     len(snake_case							)
		except KeyError:
				raise ValueError('Task not found: %s' % (data_args.task_name)							)

		# Load pretrained model and tokenizer
		#
		# Distributed training:
		# The .from_pretrained methods guarantee that only one local process can concurrently
		# download model & vocab.

		__lowercase											=     AutoConfig.from_pretrained(
		    model_args.config_name if model_args.config_name else model_args.model_name_or_path						,   num_labels=snake_case						,   finetuning_task=data_args.task_name						,   cache_dir=model_args.cache_dir						,   )
		__lowercase											=     AutoTokenizer.from_pretrained(
		    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path						,   cache_dir=model_args.cache_dir						,   )
		__lowercase											=     AutoModelForMultipleChoice.from_pretrained(
		    model_args.model_name_or_path						,   from_tf=bool('.ckpt' in model_args.model_name_or_path							)						,   config=snake_case						,   cache_dir=model_args.cache_dir						,   )

		# Get datasets
		__lowercase											=     (
		    MultipleChoiceDataset(
		        data_dir=data_args.data_dir						,   tokenizer=snake_case						,   task=data_args.task_name						,   max_seq_length=data_args.max_seq_length						,   overwrite_cache=data_args.overwrite_cache						,   mode=Split.train						,   )
		    if training_args.do_train
		    else None
		)
		__lowercase											=     (
		    MultipleChoiceDataset(
		        data_dir=data_args.data_dir						,   tokenizer=snake_case						,   task=data_args.task_name						,   max_seq_length=data_args.max_seq_length						,   overwrite_cache=data_args.overwrite_cache						,   mode=Split.dev						,   )
		    if training_args.do_eval
		    else None
		)

		def compute_metrics(snake_case							) -> Dict:
				__lowercase											=     np.argmax(p.predictions						,   axis=1							)
				return {"acc": simple_accuracy(snake_case						,   p.label_ids							)}

		# Data collator
		__lowercase											=     DataCollatorWithPadding(snake_case						,   pad_to_multiple_of=8							) if training_args.fpaa else None

		# Initialize our Trainer
		__lowercase											=     Trainer(
		    model=snake_case						,   args=snake_case						,   train_dataset=snake_case						,   eval_dataset=snake_case						,   compute_metrics=snake_case						,   data_collator=snake_case						,   )

		# Training
		if training_args.do_train:
				trainer.train(
				    model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path							) else None							)
				trainer.save_model()
				# For convenience, we also re-save the tokenizer to the same directory,
				# so that you can share your model easily on huggingface.co/models =)
				if trainer.is_world_master():
						tokenizer.save_pretrained(training_args.output_dir							)

    # Evaluation
		__lowercase											=     {}
		if training_args.do_eval:
				logger.info('*** Evaluate ***'							)

				__lowercase											=     trainer.evaluate()

				__lowercase											=     os.path.join(training_args.output_dir						,   'eval_results.txt'							)
				if trainer.is_world_master():
						with open(snake_case						,   'w'							) as writer:
								logger.info('***** Eval results *****'							)
								for key, value in result.items():
										logger.info('  %s = %s'						,   snake_case						,   snake_case							)
										writer.write('%s = %s\n' % (key, value)							)

								results.update(snake_case							)

		return results



def  SCREAMING_SNAKE_CASE				(      snake_case							)		->    Tuple:
		# For xla_spawn (TPUs)
		main()


if __name__ == "__main__":
		main()











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()
SCREAMING_SNAKE_CASE_   :      Optional[Any]			=   logging.get_logger(__name__)


SCREAMING_SNAKE_CASE_   :      Any			=   ['''model.decoder.embed_positions.weights''']



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



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



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

		__lowercase											=     fairseq_model.lm.state_dict()
		__lowercase						,      __lowercase											=     rename_state_dict(
		    snake_case						,   hidden_size=decoder_config.hidden_size							)

		__lowercase											=     TaEncoderModel.from_pretrained('t5-base'							)
		__lowercase											=     EncodecModel.from_pretrained('facebook/encodec_32khz'							)
		__lowercase											=     MusicgenForCausalLM(snake_case							).eval()

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

		with torch.no_grad():
				__lowercase											=     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
		__lowercase											=     AutoTokenizer.from_pretrained('t5-base'							)
		__lowercase											=     AutoFeatureExtractor.from_pretrained('facebook/encodec_32khz'						,   padding_side='left'							)

		__lowercase											=     MusicgenProcessor(feature_extractor=snake_case						,   tokenizer=snake_case							)

		# set the appropriate bos/pad token ids
		__lowercase											=     2_048
		__lowercase											=     2_048

		# set other default generation config params
		__lowercase											=     int(30 * audio_encoder.config.frame_rate							)
		__lowercase											=     True
		__lowercase											=     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__":
		SCREAMING_SNAKE_CASE_   :      Optional[int]			=   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.'''
		)

		SCREAMING_SNAKE_CASE_   :      Any			=   parser.parse_args()
		convert_musicgen_checkpoint(args.checkpoint, args.pytorch_dump_folder, args.push_to_hub)




