





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 						=      logging.get_logger(__name__)


_a 						=      ["model.decoder.embed_positions.weights"]





def lowerCAmelCase__(__snake_case							)				->      Tuple:







	'''simple docstring'''







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





def lowerCAmelCase__(__snake_case       ,__snake_case							)				->      Tuple[Dict, Dict]:







	'''simple docstring'''







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





def lowerCAmelCase__(__snake_case							)				->      MusicgenDecoderConfig:







	'''simple docstring'''







	if checkpoint == "small":
		# default config values
		lowerCamelCase__								=		1024
		lowerCamelCase__								=		24
		lowerCamelCase__								=		16
	elif checkpoint == "medium":
		lowerCamelCase__								=		1536
		lowerCamelCase__								=		48
		lowerCamelCase__								=		24
	elif checkpoint == "large":
		lowerCamelCase__								=		2048
		lowerCamelCase__								=		48
		lowerCamelCase__								=		32
	else:
		raise ValueError(F'Checkpoint should be one of `[\'small\', \'medium\', \'large\']`, got {checkpoint}.'							)
	lowerCamelCase__								=		MusicgenDecoderConfig(
	    hidden_size=_lowerCAmelCase       ,ffn_dim=hidden_size * 4       ,num_hidden_layers=_lowerCAmelCase       ,num_attention_heads=_lowerCAmelCase       ,)
	return config





@torch.no_grad()
def lowerCAmelCase__(__snake_case       ,__snake_case=None       ,__snake_case=None       ,__snake_case="cpu"							)				->      str:







	'''simple docstring'''







	lowerCamelCase__								=		MusicGen.get_pretrained(_lowerCAmelCase       ,device=_lowerCAmelCase							)
	lowerCamelCase__								=		decoder_config_from_checkpoint(_lowerCAmelCase							)

	lowerCamelCase__								=		fairseq_model.lm.state_dict()
	lowerCamelCase__								=		rename_state_dict(
	    _lowerCAmelCase       ,hidden_size=decoder_config.hidden_size							)

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

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

	with torch.no_grad():
		lowerCamelCase__								=		model(input_ids=_lowerCAmelCase       ,decoder_input_ids=_lowerCAmelCase							).logits

	if logits.shape != (8, 1, 2048):
		raise ValueError('''Incorrect shape for logits'''							)

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

	lowerCamelCase__								=		MusicgenProcessor(feature_extractor=_lowerCAmelCase       ,tokenizer=_lowerCAmelCase							)

	# set the appropriate bos/pad token ids
	lowerCamelCase__								=		2048
	lowerCamelCase__								=		2048

	# set other default generation config params
	lowerCamelCase__								=		int(30 * audio_encoder.config.frame_rate							)
	lowerCamelCase__								=		True
	lowerCamelCase__								=		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 						=      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 						=      parser.parse_args()
			convert_musicgen_checkpoint(args.checkpoint, args.pytorch_dump_folder, args.push_to_hub)








'''simple docstring'''






import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, XLMRobertaTokenizer

from diffusers import AltDiffusionPipeline, AutoencoderKL, DDIMScheduler, PNDMScheduler, UNetaDConditionModel
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class 				SCREAMING_SNAKE_CASE(							A__							,		A__							,		A__							,		unittest.TestCase			):







	"""simple docstring"""






	lowerCamelCase__			 =       AltDiffusionPipeline
	lowerCamelCase__			 =       TEXT_TO_IMAGE_PARAMS
	lowerCamelCase__			 =       TEXT_TO_IMAGE_BATCH_PARAMS
	lowerCamelCase__			 =       TEXT_TO_IMAGE_IMAGE_PARAMS
	lowerCamelCase__			 =       TEXT_TO_IMAGE_IMAGE_PARAMS







	def      A	(						self			:     Dict		)			->		int:
								torch.manual_seed(0		)
								UpperCAmelCase :       str 							=		UNetaDConditionModel(
								    block_out_channels=(32, 64)		,			layers_per_block=2		,			sample_size=32		,			in_channels=4		,			out_channels=4		,			down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')		,			up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')		,			cross_attention_dim=32		,			)
								UpperCAmelCase :       Dict 							=		DDIMScheduler(
								    beta_start=0.0_00_85		,			beta_end=0.0_12		,			beta_schedule='''scaled_linear'''		,			clip_sample=__snake_case		,			set_alpha_to_one=__snake_case		,			)
								torch.manual_seed(0		)
								UpperCAmelCase :       Union[str, Any] 							=		AutoencoderKL(
								    block_out_channels=[32, 64]		,			in_channels=3		,			out_channels=3		,			down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']		,			up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']		,			latent_channels=4		,			)

								# TODO: address the non-deterministic text encoder (fails for save-load tests)
								# torch.manual_seed(0)
								# text_encoder_config = RobertaSeriesConfig(
								#     hidden_size=32,
								#     project_dim=32,
								#     intermediate_size=37,
								#     layer_norm_eps=1e-05,
								#     num_attention_heads=4,
								#     num_hidden_layers=5,
								#     vocab_size=5002,
								# )
								# text_encoder = RobertaSeriesModelWithTransformation(text_encoder_config)

								torch.manual_seed(0		)
								UpperCAmelCase :       Dict 							=		CLIPTextConfig(
								    bos_token_id=0		,			eos_token_id=2		,			hidden_size=32		,			projection_dim=32		,			intermediate_size=37		,			layer_norm_eps=1E-05		,			num_attention_heads=4		,			num_hidden_layers=5		,			pad_token_id=1		,			vocab_size=5002		,			)
								UpperCAmelCase :       List[Any] 							=		CLIPTextModel(__snake_case		)

								UpperCAmelCase :       Optional[Any] 							=		XLMRobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-xlm-roberta'''		)
								UpperCAmelCase :       Optional[int] 							=		77

								UpperCAmelCase :       Optional[int] 							=		{
								    '''unet''': unet,
								    '''scheduler''': scheduler,
								    '''vae''': vae,
								    '''text_encoder''': text_encoder,
								    '''tokenizer''': tokenizer,
								    '''safety_checker''': None,
								    '''feature_extractor''': None,
								}
								return components







	def      A	(						self			:     Optional[Any]		,			__snake_case			:     Dict		,			__snake_case			:     List[str]=0		)			->		Union[str, Any]:
								if str(__snake_case		).startswith('''mps'''		):
															UpperCAmelCase :       str 							=		torch.manual_seed(__snake_case		)
								else:
															UpperCAmelCase :       Tuple 							=		torch.Generator(device=__snake_case		).manual_seed(__snake_case		)
								UpperCAmelCase :       Dict 							=		{
								    '''prompt''': '''A painting of a squirrel eating a burger''',
								    '''generator''': generator,
								    '''num_inference_steps''': 2,
								    '''guidance_scale''': 6.0,
								    '''output_type''': '''numpy''',
								}
								return inputs







	def      A	(						self			:     Union[str, Any]		)			->		List[str]:
								super().test_attention_slicing_forward_pass(expected_max_diff=3E-3		)







	def      A	(						self			:     Tuple		)			->		List[str]:
								super().test_inference_batch_single_identical(expected_max_diff=3E-3		)







	def      A	(						self			:     Tuple		)			->		Optional[int]:
								UpperCAmelCase :       List[str] 							=		'''cpu'''  # ensure determinism for the device-dependent torch.Generator

								UpperCAmelCase :       Any 							=		self.get_dummy_components()
								torch.manual_seed(0		)
								UpperCAmelCase :       Optional[int] 							=		RobertaSeriesConfig(
								    hidden_size=32		,			project_dim=32		,			intermediate_size=37		,			layer_norm_eps=1E-05		,			num_attention_heads=4		,			num_hidden_layers=5		,			vocab_size=5002		,			)
								# TODO: remove after fixing the non-deterministic text encoder
								UpperCAmelCase :       List[str] 							=		RobertaSeriesModelWithTransformation(__snake_case		)
								UpperCAmelCase :       str 							=		text_encoder

								UpperCAmelCase :       Optional[int] 							=		AltDiffusionPipeline(**__snake_case		)
								UpperCAmelCase :       str 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       Optional[int] 							=		self.get_dummy_inputs(__snake_case		)
								UpperCAmelCase :       Optional[int] 							=		'''A photo of an astronaut'''
								UpperCAmelCase :       List[Any] 							=		alt_pipe(**__snake_case		)
								UpperCAmelCase :       Optional[Any] 							=		output.images
								UpperCAmelCase :       str 							=		image[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)
								UpperCAmelCase :       List[str] 							=		np.array(
								    [0.5_74_81_62, 0.60_44_71_45, 0.48_82_12_17, 0.50_10_06_36, 0.5_43_11_85, 0.45_76_36_83, 0.49_65_76_96, 0.48_13_27_33, 0.47_57_30_93]		)

								assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2







	def      A	(						self			:     int		)			->		Union[str, Any]:
								UpperCAmelCase :       Optional[int] 							=		'''cpu'''  # ensure determinism for the device-dependent torch.Generator

								UpperCAmelCase :       int 							=		self.get_dummy_components()
								UpperCAmelCase :       int 							=		PNDMScheduler(skip_prk_steps=__snake_case		)
								torch.manual_seed(0		)
								UpperCAmelCase :       int 							=		RobertaSeriesConfig(
								    hidden_size=32		,			project_dim=32		,			intermediate_size=37		,			layer_norm_eps=1E-05		,			num_attention_heads=4		,			num_hidden_layers=5		,			vocab_size=5002		,			)
								# TODO: remove after fixing the non-deterministic text encoder
								UpperCAmelCase :       Union[str, Any] 							=		RobertaSeriesModelWithTransformation(__snake_case		)
								UpperCAmelCase :       Union[str, Any] 							=		text_encoder
								UpperCAmelCase :       Optional[int] 							=		AltDiffusionPipeline(**__snake_case		)
								UpperCAmelCase :       Dict 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       int 							=		self.get_dummy_inputs(__snake_case		)
								UpperCAmelCase :       Optional[int] 							=		alt_pipe(**__snake_case		)
								UpperCAmelCase :       Optional[int] 							=		output.images
								UpperCAmelCase :       int 							=		image[0, -3:, -3:, -1]

								assert image.shape == (1, 64, 64, 3)
								UpperCAmelCase :       Optional[int] 							=		np.array(
								    [0.51_60_50_93, 0.5_70_72_41, 0.47_36_55_07, 0.50_57_88_86, 0.5_63_38_77, 0.4_64_25_03, 0.5_18_20_81, 0.48_76_34_84, 0.49_08_42_37]		)

								assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2

@slow
@require_torch_gpu
class 				SCREAMING_SNAKE_CASE(							unittest.TestCase			):







	"""simple docstring"""






	def      A	(						self			:     str		)			->		Any:
								# clean up the VRAM after each test
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()







	def      A	(						self			:     List[Any]		)			->		Any:
								# make sure here that pndm scheduler skips prk
								UpperCAmelCase :       List[Any] 							=		AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion'''		,			safety_checker=__snake_case		)
								UpperCAmelCase :       Tuple 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       List[Any] 							=		'''A painting of a squirrel eating a burger'''
								UpperCAmelCase :       Any 							=		torch.manual_seed(0		)
								UpperCAmelCase :       Optional[int] 							=		alt_pipe([prompt]		,			generator=__snake_case		,			guidance_scale=6.0		,			num_inference_steps=20		,			output_type='''np'''		)

								UpperCAmelCase :       Dict 							=		output.images

								UpperCAmelCase :       Optional[Any] 							=		image[0, -3:, -3:, -1]

								assert image.shape == (1, 512, 512, 3)
								UpperCAmelCase :       List[str] 							=		np.array([0.10_10, 0.08_00, 0.07_94, 0.08_85, 0.08_43, 0.07_62, 0.07_69, 0.07_29, 0.05_86]		)

								assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2







	def      A	(						self			:     Tuple		)			->		int:
								UpperCAmelCase :       int 							=		DDIMScheduler.from_pretrained('''BAAI/AltDiffusion'''		,			subfolder='''scheduler'''		)

								UpperCAmelCase :       Tuple 							=		AltDiffusionPipeline.from_pretrained('''BAAI/AltDiffusion'''		,			scheduler=__snake_case		,			safety_checker=__snake_case		)
								UpperCAmelCase :       Dict 							=		alt_pipe.to(__snake_case		)
								alt_pipe.set_progress_bar_config(disable=__snake_case		)

								UpperCAmelCase :       Tuple 							=		'''A painting of a squirrel eating a burger'''
								UpperCAmelCase :       Optional[int] 							=		torch.manual_seed(0		)

								UpperCAmelCase :       List[Any] 							=		alt_pipe([prompt]		,			generator=__snake_case		,			num_inference_steps=2		,			output_type='''numpy'''		)
								UpperCAmelCase :       Dict 							=		output.images

								UpperCAmelCase :       Optional[Any] 							=		image[0, -3:, -3:, -1]

								assert image.shape == (1, 512, 512, 3)
								UpperCAmelCase :       Union[str, Any] 							=		np.array([0.40_19, 0.40_52, 0.38_10, 0.41_19, 0.39_16, 0.39_82, 0.46_51, 0.41_95, 0.53_23]		)

								assert np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2





