




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


__A   :  List[Any]					=      ['model.decoder.embed_positions.weights']

def 						snake_case__       (					_lowerCamelCase				)			->Union[str, Any]:






		"""simple docstring"""


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

def 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase				)			->Tuple[Dict, Dict]:






		"""simple docstring"""


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

def 						snake_case__       (					_lowerCamelCase				)			->MusicgenDecoderConfig:






		"""simple docstring"""


		if checkpoint == "small":
				# default config values
				__lowercase				:		List[Any]												=				10_24
				__lowercase				:		Any												=				24
				__lowercase				:		Optional[int]												=				16
		elif checkpoint == "medium":
				__lowercase				:		Optional[int]												=				15_36
				__lowercase				:		Optional[Any]												=				48
				__lowercase				:		Dict												=				24
		elif checkpoint == "large":
				__lowercase				:		Union[str, Any]												=				20_48
				__lowercase				:		List[str]												=				48
				__lowercase				:		Optional[Any]												=				32
		else:
				raise ValueError(F'Checkpoint should be one of `[\'small\', \'medium\', \'large\']`, got {checkpoint}.'				)
		__lowercase				:		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,  _lowerCamelCase=None,  _lowerCamelCase=None,  _lowerCamelCase="cpu"				)			->Tuple:






		"""simple docstring"""


		__lowercase				:		Dict												=				MusicGen.get_pretrained(_lowerCamelCase,  device=_lowerCamelCase				)
		__lowercase				:		Optional[int]												=				decoder_config_from_checkpoint(_lowerCamelCase				)

		__lowercase				:		Tuple												=				fairseq_model.lm.state_dict()
		__lowercase						,__lowercase				:		int												=				rename_state_dict(
		    _lowerCamelCase,  hidden_size=decoder_config.hidden_size				)

		__lowercase				:		int												=				TaEncoderModel.from_pretrained("t5-base"				)
		__lowercase				:		Optional[int]												=				EncodecModel.from_pretrained("facebook/encodec_32khz"				)
		__lowercase				:		List[Any]												=				MusicgenForCausalLM(_lowerCamelCase				).eval()

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

		with torch.no_grad():
				__lowercase				:		List[Any]												=				model(input_ids=_lowerCamelCase,  decoder_input_ids=_lowerCamelCase				).logits

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

		# now construct the processor
		__lowercase				:		int												=				AutoTokenizer.from_pretrained("t5-base"				)
		__lowercase				:		List[str]												=				AutoFeatureExtractor.from_pretrained("facebook/encodec_32khz",  padding_side="left"				)

		__lowercase				:		Optional[int]												=				MusicgenProcessor(feature_extractor=_lowerCamelCase,  tokenizer=_lowerCamelCase				)

		# set the appropriate bos/pad token ids
		__lowercase				:		int												=				20_48
		__lowercase				:		str												=				20_48

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

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






"""simple docstring"""







from collections import deque

def 						snake_case__       (					_lowerCamelCase				)			->Dict:






		"""simple docstring"""


		__lowercase				:		Any												=				len(_lowerCamelCase				)
		__lowercase				:		str												=				deque()
		__lowercase				:		Tuple												=				[False for _ in range(_lowerCamelCase				)]
		__lowercase				:		Optional[int]												=				[-1 for _ in range(_lowerCamelCase				)]
		__lowercase				:		Dict												=				index_of[:]

		def strong_connect(_lowerCamelCase,  _lowerCamelCase,  _lowerCamelCase				):
				__lowercase				:		List[str]												=				index  # the number when this node is seen
				__lowercase				:		List[str]												=				index  # lowest rank node reachable from here
				index += 1
				stack.append(_lowerCamelCase				)
				__lowercase				:		List[Any]												=				True

				for w in g[v]:
						if index_of[w] == -1:
								__lowercase				:		Optional[int]												=				strong_connect(_lowerCamelCase,  _lowerCamelCase,  _lowerCamelCase				)
								__lowercase				:		List[Any]												=				(
								    lowlink_of[w] if lowlink_of[w] < lowlink_of[v] else lowlink_of[v]
								)
						elif on_stack[w]:
								__lowercase				:		List[Any]												=				(
								    lowlink_of[w] if lowlink_of[w] < lowlink_of[v] else lowlink_of[v]
								)

				if lowlink_of[v] == index_of[v]:
						__lowercase				:		Any												=				[]
						__lowercase				:		List[Any]												=				stack.pop()
						__lowercase				:		Any												=				False
						component.append(_lowerCamelCase				)
						while w != v:
								__lowercase				:		List[str]												=				stack.pop()
								__lowercase				:		Dict												=				False
								component.append(_lowerCamelCase				)
						components.append(_lowerCamelCase				)
				return index

		__lowercase				:		Tuple												=				[]
		for v in range(_lowerCamelCase				):
				if index_of[v] == -1:
						strong_connect(_lowerCamelCase,  0,  _lowerCamelCase				)

		return components

def 						snake_case__       (					_lowerCamelCase,  _lowerCamelCase				)			->Tuple:






		"""simple docstring"""


		__lowercase				:		int												=				[[] for _ in range(_lowerCamelCase				)]
		for u, v in edges:
				g[u].append(_lowerCamelCase				)
		return g


if __name__ == "__main__":
						# Test
						__A   :  int					=      7
						__A   :  int					=      [0, 0, 1, 2, 3, 3, 4, 4, 6]
						__A   :  List[Any]					=      [1, 3, 2, 0, 1, 4, 5, 6, 5]
						__A   :  List[str]					=      [(u, v) for u, v in zip(source, target)]
						__A   :  List[str]					=      create_graph(n_vertices, edges)

						assert [[5], [6], [4], [3, 2, 1, 0]] == tarjan(g)
