


"""simple docstring"""
import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__SCREAMING_SNAKE_CASE							:    List[Any]		=    logging.get_logger(__name__)


__SCREAMING_SNAKE_CASE							:    List[Any]		=    {'vocab_file': 'vocab.json', 'merges_file': 'merges.txt'}

# See all BART models at https://huggingface.co/models?filter=bart
__SCREAMING_SNAKE_CASE							:    int		=    {
    'vocab_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/vocab.json',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/vocab.json',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json',
    },
    'merges_file': {
        'facebook/bart-base': 'https://huggingface.co/facebook/bart-base/resolve/main/merges.txt',
        'facebook/bart-large': 'https://huggingface.co/facebook/bart-large/resolve/main/merges.txt',
        'facebook/bart-large-mnli': 'https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt',
        'facebook/bart-large-cnn': 'https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt',
        'facebook/bart-large-xsum': 'https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt',
        'yjernite/bart_eli5': 'https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt',
    },
}

__SCREAMING_SNAKE_CASE							:    Optional[Any]		=    {
    'facebook/bart-base': 1_024,
    'facebook/bart-large': 1_024,
    'facebook/bart-large-mnli': 1_024,
    'facebook/bart-large-cnn': 1_024,
    'facebook/bart-large-xsum': 1_024,
    'yjernite/bart_eli5': 1_024,
}







@lru_cache()
def 						_a				(      )    ->				Dict:
							snake_case_           =		(
							    list(range(ord("""!"""				)				, ord("""~"""				) + 1				)				) + list(range(ord("""¡"""				)				, ord("""¬"""				) + 1				)				) + list(range(ord("""®"""				)				, ord("""ÿ"""				) + 1				)				)
							)
							snake_case_           =		bs[:]
							snake_case_           =		0
							for b in range(2**8				):
														if b not in bs:
																					bs.append(__lowerCAmelCase				)
																					cs.append(2**8 + n				)
																					n += 1
							snake_case_           =		[chr(__lowerCAmelCase				) for n in cs]
							return dict(zip(__lowerCAmelCase				, __lowerCAmelCase				)				)







def 						_a				(      _SCREAMING_SNAKE_CASE				)    ->				Union[str, Any]:
							snake_case_           =		set()
							snake_case_           =		word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)				)
														snake_case_           =		char
							return pairs





class      __A     (snake_case__):





							'''simple docstring'''

							__lowercase:		Optional[int]							      =    VOCAB_FILES_NAMES
							__lowercase:		List[str]							      =    PRETRAINED_VOCAB_FILES_MAP
							__lowercase:		Optional[int]							      =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__lowercase:		Dict							      =    ["""input_ids""", """attention_mask"""]



							def __init__(				self							:      str		,       UpperCAmelCase_							:      Union[str, Any]		,       UpperCAmelCase_							:      Union[str, Any]		,       UpperCAmelCase_							:      List[Any]="replace"		,       UpperCAmelCase_							:      int="<s>"		,       UpperCAmelCase_							:      Tuple="</s>"		,       UpperCAmelCase_							:      Any="</s>"		,       UpperCAmelCase_							:      List[Any]="<s>"		,       UpperCAmelCase_							:      int="<unk>"		,       UpperCAmelCase_							:      List[str]="<pad>"		,       UpperCAmelCase_							:      Union[str, Any]="<mask>"		,       UpperCAmelCase_							:      str=False		,       **UpperCAmelCase_							:      Tuple		,       )					->Optional[int]:



														"""simple docstring"""



														snake_case_           =		AddedToken(_lowerCAmelCase		,       lstrip=_lowerCAmelCase		,       rstrip=_lowerCAmelCase       ) if isinstance(_lowerCAmelCase		,       _lowerCAmelCase       ) else bos_token
														snake_case_           =		AddedToken(_lowerCAmelCase		,       lstrip=_lowerCAmelCase		,       rstrip=_lowerCAmelCase       ) if isinstance(_lowerCAmelCase		,       _lowerCAmelCase       ) else eos_token
														snake_case_           =		AddedToken(_lowerCAmelCase		,       lstrip=_lowerCAmelCase		,       rstrip=_lowerCAmelCase       ) if isinstance(_lowerCAmelCase		,       _lowerCAmelCase       ) else sep_token
														snake_case_           =		AddedToken(_lowerCAmelCase		,       lstrip=_lowerCAmelCase		,       rstrip=_lowerCAmelCase       ) if isinstance(_lowerCAmelCase		,       _lowerCAmelCase       ) else cls_token
														snake_case_           =		AddedToken(_lowerCAmelCase		,       lstrip=_lowerCAmelCase		,       rstrip=_lowerCAmelCase       ) if isinstance(_lowerCAmelCase		,       _lowerCAmelCase       ) else unk_token
														snake_case_           =		AddedToken(_lowerCAmelCase		,       lstrip=_lowerCAmelCase		,       rstrip=_lowerCAmelCase       ) if isinstance(_lowerCAmelCase		,       _lowerCAmelCase       ) else pad_token

														# Mask token behave like a normal word, i.e. include the space before it
														snake_case_           =		AddedToken(_lowerCAmelCase		,       lstrip=_lowerCAmelCase		,       rstrip=_lowerCAmelCase       ) if isinstance(_lowerCAmelCase		,       _lowerCAmelCase       ) else mask_token

														super().__init__(
														    errors=_lowerCAmelCase		,       bos_token=_lowerCAmelCase		,       eos_token=_lowerCAmelCase		,       unk_token=_lowerCAmelCase		,       sep_token=_lowerCAmelCase		,       cls_token=_lowerCAmelCase		,       pad_token=_lowerCAmelCase		,       mask_token=_lowerCAmelCase		,       add_prefix_space=_lowerCAmelCase		,       **_lowerCAmelCase		,       )

														with open(_lowerCAmelCase		,       encoding="""utf-8"""       ) as vocab_handle:
																					snake_case_           =		json.load(_lowerCAmelCase       )
														snake_case_           =		{v: k for k, v in self.encoder.items()}
														snake_case_           =		errors  # how to handle errors in decoding
														snake_case_           =		bytes_to_unicode()
														snake_case_           =		{v: k for k, v in self.byte_encoder.items()}
														with open(_lowerCAmelCase		,       encoding="""utf-8"""       ) as merges_handle:
																					snake_case_           =		merges_handle.read().split("""\n"""       )[1:-1]
														snake_case_           =		[tuple(merge.split()       ) for merge in bpe_merges]
														snake_case_           =		dict(zip(_lowerCAmelCase		,       range(len(_lowerCAmelCase       )       )       )       )
														snake_case_           =		{}
														snake_case_           =		add_prefix_space

														# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
														snake_case_           =		re.compile(R"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""       )



							@property
							def 		lowerCAmelCase       (				self							:      Tuple       )					->Tuple:



														"""simple docstring"""



														return len(self.encoder       )



							def 		lowerCAmelCase       (				self							:      Optional[Any]       )					->Union[str, Any]:



														"""simple docstring"""



														return dict(self.encoder		,       **self.added_tokens_encoder       )



							def 		lowerCAmelCase       (				self							:      Tuple		,       UpperCAmelCase_							:      Union[str, Any]       )					->int:



														"""simple docstring"""



														if token in self.cache:
																					return self.cache[token]
														snake_case_           =		tuple(_lowerCAmelCase       )
														snake_case_           =		get_pairs(_lowerCAmelCase       )

														if not pairs:
																					return token

														while True:
																					snake_case_           =		min(_lowerCAmelCase		,       key=lambda UpperCAmelCase_       : self.bpe_ranks.get(_lowerCAmelCase		,       float("""inf"""       )       )       )
																					if bigram not in self.bpe_ranks:
																												break
																					snake_case_       ,						snake_case_           =		bigram
																					snake_case_           =		[]
																					snake_case_           =		0
																					while i < len(_lowerCAmelCase       ):
																												try:
																																			snake_case_           =		word.index(_lowerCAmelCase		,       _lowerCAmelCase       )
																												except ValueError:
																																			new_word.extend(word[i:]       )
																																			break
																												else:
																																			new_word.extend(word[i:j]       )
																																			snake_case_           =		j

																												if word[i] == first and i < len(_lowerCAmelCase       ) - 1 and word[i + 1] == second:
																																			new_word.append(first + second       )
																																			i += 2
																												else:
																																			new_word.append(word[i]       )
																																			i += 1
																					snake_case_           =		tuple(_lowerCAmelCase       )
																					snake_case_           =		new_word
																					if len(_lowerCAmelCase       ) == 1:
																												break
																					else:
																												snake_case_           =		get_pairs(_lowerCAmelCase       )
														snake_case_           =		""" """.join(_lowerCAmelCase       )
														snake_case_           =		word
														return word



							def 		lowerCAmelCase       (				self							:      Optional[Any]		,       UpperCAmelCase_							:      int       )					->Optional[Any]:



														"""simple docstring"""



														snake_case_           =		[]
														for token in re.findall(self.pat		,       _lowerCAmelCase       ):
																					snake_case_           =		"""""".join(
																					    self.byte_encoder[b] for b in token.encode("""utf-8"""       )       )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
																					bpe_tokens.extend(bpe_token for bpe_token in self.bpe(_lowerCAmelCase       ).split(""" """       )       )
														return bpe_tokens



							def 		lowerCAmelCase       (				self							:      Optional[Any]		,       UpperCAmelCase_							:      Tuple       )					->List[Any]:



														"""simple docstring"""



														return self.encoder.get(_lowerCAmelCase		,       self.encoder.get(self.unk_token       )       )



							def 		lowerCAmelCase       (				self							:      int		,       UpperCAmelCase_							:      int       )					->List[str]:



														"""simple docstring"""



														return self.decoder.get(_lowerCAmelCase       )



							def 		lowerCAmelCase       (				self							:      Union[str, Any]		,       UpperCAmelCase_							:      List[str]       )					->Tuple:



														"""simple docstring"""



														snake_case_           =		"""""".join(_lowerCAmelCase       )
														snake_case_           =		bytearray([self.byte_decoder[c] for c in text]       ).decode("""utf-8"""		,       errors=self.errors       )
														return text



							def 		lowerCAmelCase       (				self							:      Union[str, Any]		,       UpperCAmelCase_							:      Union[str, Any]		,       UpperCAmelCase_							:      Tuple = None       )					->Dict:



														"""simple docstring"""



														if not os.path.isdir(_lowerCAmelCase       ):
																					logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
																					return
														snake_case_           =		os.path.join(
														    _lowerCAmelCase		,       (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]       )
														snake_case_           =		os.path.join(
														    _lowerCAmelCase		,       (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]       )

														with open(_lowerCAmelCase		,       """w"""		,       encoding="""utf-8"""       ) as f:
																					f.write(json.dumps(self.encoder		,       indent=2		,       sort_keys=_lowerCAmelCase		,       ensure_ascii=_lowerCAmelCase       ) + """\n"""       )

														snake_case_           =		0
														with open(_lowerCAmelCase		,       """w"""		,       encoding="""utf-8"""       ) as writer:
																					writer.write("""#version: 0.2\n"""       )
																					for bpe_tokens, token_index in sorted(self.bpe_ranks.items()		,       key=lambda UpperCAmelCase_       : kv[1]       ):
																												if index != token_index:
																																			logger.warning(
																																			    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																																			    """ Please check that the tokenizer is not corrupted!"""       )
																																			snake_case_           =		token_index
																												writer.write(""" """.join(_lowerCAmelCase       ) + """\n"""       )
																												index += 1

														return vocab_file, merge_file



							def 		lowerCAmelCase       (				self							:      Tuple		,       UpperCAmelCase_							:      Dict		,       UpperCAmelCase_							:      Optional[Any] = None       )					->Optional[Any]:



														"""simple docstring"""



														if token_ids_a is None:
																					return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
														snake_case_           =		[self.cls_token_id]
														snake_case_           =		[self.sep_token_id]
														return cls + token_ids_a + sep + sep + token_ids_a + sep



							def 		lowerCAmelCase       (				self							:      Tuple		,       UpperCAmelCase_							:      List[str]		,       UpperCAmelCase_							:      List[str] = None		,       UpperCAmelCase_							:      List[str] = False       )					->List[Any]:



														"""simple docstring"""



														if already_has_special_tokens:
																					return super().get_special_tokens_mask(
																					    token_ids_a=_lowerCAmelCase		,       token_ids_a=_lowerCAmelCase		,       already_has_special_tokens=_lowerCAmelCase       )

														if token_ids_a is None:
																					return [1] + ([0] * len(_lowerCAmelCase       )) + [1]
														return [1] + ([0] * len(_lowerCAmelCase       )) + [1, 1] + ([0] * len(_lowerCAmelCase       )) + [1]



							def 		lowerCAmelCase       (				self							:      Optional[Any]		,       UpperCAmelCase_							:      Any		,       UpperCAmelCase_							:      Optional[Any] = None       )					->Tuple:



														"""simple docstring"""



														snake_case_           =		[self.sep_token_id]
														snake_case_           =		[self.cls_token_id]

														if token_ids_a is None:
																					return len(cls + token_ids_a + sep       ) * [0]
														return len(cls + token_ids_a + sep + sep + token_ids_a + sep       ) * [0]



							def 		lowerCAmelCase       (				self							:      int		,       UpperCAmelCase_							:      Dict		,       UpperCAmelCase_							:      List[Any]=False		,       **UpperCAmelCase_							:      Optional[int]       )					->str:



														"""simple docstring"""



														snake_case_           =		kwargs.pop("""add_prefix_space"""		,       self.add_prefix_space       )
														if (is_split_into_words or add_prefix_space) and (len(_lowerCAmelCase       ) > 0 and not text[0].isspace()):
																					snake_case_           =		""" """ + text
														return (text, kwargs)







"""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()
__SCREAMING_SNAKE_CASE							:    Union[str, Any]		=    logging.get_logger(__name__)


__SCREAMING_SNAKE_CASE							:    int		=    ['model.decoder.embed_positions.weights']







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







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







def 						_a				(      _SCREAMING_SNAKE_CASE				)    ->				MusicgenDecoderConfig:
							if checkpoint == "small":
														# default config values
														snake_case_           =		1_024
														snake_case_           =		24
														snake_case_           =		16
							elif checkpoint == "medium":
														snake_case_           =		1_536
														snake_case_           =		48
														snake_case_           =		24
							elif checkpoint == "large":
														snake_case_           =		2_048
														snake_case_           =		48
														snake_case_           =		32
							else:
														raise ValueError(f"""Checkpoint should be one of `['small', 'medium', 'large']`, got {checkpoint}."""				)
							snake_case_           =		MusicgenDecoderConfig(
							    hidden_size=_SCREAMING_SNAKE_CASE				, ffn_dim=hidden_size * 4				, num_hidden_layers=_SCREAMING_SNAKE_CASE				, num_attention_heads=_SCREAMING_SNAKE_CASE				, )
							return config







@torch.no_grad()
def 						_a				(      _SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE=None				, _SCREAMING_SNAKE_CASE=None				, _SCREAMING_SNAKE_CASE="cpu"				)    ->				Tuple:
							snake_case_           =		MusicGen.get_pretrained(_SCREAMING_SNAKE_CASE				, device=_SCREAMING_SNAKE_CASE				)
							snake_case_           =		decoder_config_from_checkpoint(_SCREAMING_SNAKE_CASE				)

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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


if __name__ == "__main__":
				__SCREAMING_SNAKE_CASE							:    Tuple		=    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							:    int		=    parser.parse_args()
				convert_musicgen_checkpoint(args.checkpoint, args.pytorch_dump_folder, args.push_to_hub)



