




"""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 :	Any       =   ['''model.decoder.embed_positions.weights''']

def     snake_case__							(       _lowerCamelCase					)      ->int:






		"""simple docstring"""


		if "emb" in name:
				__lowercase					:		List[Any]        =					name.replace("emb",	"model.decoder.embed_tokens"					)
		if "transformer" in name:
				__lowercase					:		Union[str, Any]        =					name.replace("transformer",	"model.decoder"					)
		if "cross_attention" in name:
				__lowercase					:		int        =					name.replace("cross_attention",	"encoder_attn"					)
		if "linear1" in name:
				__lowercase					:		List[Any]        =					name.replace("linear1",	"fc1"					)
		if "linear2" in name:
				__lowercase					:		List[Any]        =					name.replace("linear2",	"fc2"					)
		if "norm1" in name:
				__lowercase					:		Any        =					name.replace("norm1",	"self_attn_layer_norm"					)
		if "norm_cross" in name:
				__lowercase					:		Dict        =					name.replace("norm_cross",	"encoder_attn_layer_norm"					)
		if "norm2" in name:
				__lowercase					:		Dict        =					name.replace("norm2",	"final_layer_norm"					)
		if "out_norm" in name:
				__lowercase					:		Dict        =					name.replace("out_norm",	"model.decoder.layer_norm"					)
		if "linears" in name:
				__lowercase					:		Any        =					name.replace("linears",	"lm_heads"					)
		if "condition_provider.conditioners.description.output_proj" in name:
				__lowercase					:		str        =					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					:		Tuple        =					list(state_dict.keys()					)
		__lowercase					:		Any        =					{}
		for key in keys:
				__lowercase					:		List[str]        =					state_dict.pop(_lowerCamelCase					)
				__lowercase					:		Tuple        =					rename_keys(_lowerCamelCase					)
				if "in_proj_weight" in key:
						# split fused qkv proj
						__lowercase					:		Optional[Any]        =					val[:hidden_size, :]
						__lowercase					:		List[str]        =					val[hidden_size : 2 * hidden_size, :]
						__lowercase					:		Union[str, Any]        =					val[-hidden_size:, :]
				elif "enc_to_dec_proj" in key:
						__lowercase					:		Union[str, Any]        =					val
				else:
						__lowercase					:		Dict        =					val
		return state_dict, enc_dec_proj_state_dict

def     snake_case__							(       _lowerCamelCase					)      ->MusicgenDecoderConfig:






		"""simple docstring"""


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






		"""simple docstring"""


		__lowercase					:		Tuple        =					MusicGen.get_pretrained(_lowerCamelCase,	device=_lowerCamelCase					)
		__lowercase					:		Union[str, Any]        =					decoder_config_from_checkpoint(_lowerCamelCase					)

		__lowercase					:		int        =					fairseq_model.lm.state_dict()
		__lowercase    ,__lowercase					:		Union[str, Any]        =					rename_state_dict(
		    _lowerCamelCase,	hidden_size=decoder_config.hidden_size					)

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

		# load all decoder weights - expect that we'll be missing embeddings and enc-dec projection
		__lowercase    ,__lowercase					:		Optional[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					:		Tuple        =					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					:		Tuple        =					input_ids.reshape(2 * 4,	-1					)

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

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

		# set the appropriate bos/pad token ids
		__lowercase					:		Tuple        =					20_48
		__lowercase					:		Any        =					20_48

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

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







from typing import Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
		from PIL import Image

		from ..image_utils import load_image

if is_torch_available():
		from ..models.auto.modeling_auto import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING

SCREAMING_SNAKE_CASE_   :      Union[str, Any]			=   logging.get_logger(__name__)

@add_end_docstrings(UpperCAmelCase_						)
class 		snake_case_   (							UpperCAmelCase_						):







	'''simple docstring'''




	def __init__(						self							:     Optional[int]					,      *__lowerCamelCase							:     Optional[int]					,      **__lowerCamelCase							:     List[Any]   )       -> Optional[Any]:

			'''simple docstring'''




			super().__init__(*__lowerCamelCase					,      **__lowerCamelCase   )
			self.check_model_type(__lowerCamelCase   )




	def    UpperCAmelCase			(						self							:     Optional[int]					,      __lowerCamelCase							:     int=None					,      __lowerCamelCase							:     List[Any]=None					,      __lowerCamelCase							:     Any=None					,      **__lowerCamelCase							:     Optional[int]   )       -> Any:

			'''simple docstring'''




			__lowercase						,      __lowercase											=     {}, {}
			if padding is not None:
					__lowercase											=     padding
			if truncation is not None:
					__lowercase											=     truncation
			if top_k is not None:
					__lowercase											=     top_k
			return preprocess_params, {}, postprocess_params




	def __call__(						self							:     str					,      __lowerCamelCase							:     Union["Image.Image", str]					,      __lowerCamelCase							:     str = None					,      **__lowerCamelCase							:     Dict   )       -> Tuple:

			'''simple docstring'''




			if isinstance(__lowerCamelCase					,      (Image.Image, str)   ) and isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					__lowercase											=     {'image': image, 'question': question}
			else:
					__lowercase											=     image
			__lowercase											=     super().__call__(__lowerCamelCase					,      **__lowerCamelCase   )
			return results




	def    UpperCAmelCase			(						self							:     List[Any]					,      __lowerCamelCase							:     List[str]					,      __lowerCamelCase							:     Optional[Any]=False					,      __lowerCamelCase							:     Optional[Any]=False   )       -> Optional[int]:

			'''simple docstring'''




			__lowercase											=     load_image(inputs['image']   )
			__lowercase											=     self.tokenizer(
			    inputs['question']					,      return_tensors=self.framework					,      padding=__lowerCamelCase					,      truncation=__lowerCamelCase   )
			__lowercase											=     self.image_processor(images=__lowerCamelCase					,      return_tensors=self.framework   )
			model_inputs.update(__lowerCamelCase   )
			return model_inputs




	def    UpperCAmelCase			(						self							:     str					,      __lowerCamelCase							:     Tuple   )       -> Any:

			'''simple docstring'''




			__lowercase											=     self.model(**__lowerCamelCase   )
			return model_outputs






	def    UpperCAmelCase			(						self							:     int					,      __lowerCamelCase							:     Any					,      __lowerCamelCase							:     int=5   )       -> Tuple:

			'''simple docstring'''




			if top_k > self.model.config.num_labels:
					__lowercase											=     self.model.config.num_labels

			if self.framework == "pt":
					__lowercase											=     model_outputs.logits.sigmoid()[0]
					__lowercase						,      __lowercase											=     probs.topk(__lowerCamelCase   )
			else:
					raise ValueError(F"Unsupported framework: {self.framework}"   )

			__lowercase											=     scores.tolist()
			__lowercase											=     ids.tolist()
			return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(__lowerCamelCase					,      __lowerCamelCase   )]




