





import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
		from .features import FeatureType


SCREAMING_SNAKE_CASE_	,SCREAMING_SNAKE_CASE_	,SCREAMING_SNAKE_CASE_   :      Any			=   False, False, False

@dataclass
class 		snake_case_   :







	'''simple docstring'''
	__UpperCamelCase   =		None
	__UpperCamelCase   =		True
	__UpperCamelCase   =		True
	__UpperCamelCase   =		None
	# Automatically constructed
	__UpperCamelCase   =		"dict"
	__UpperCamelCase   =		pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}						)
	__UpperCamelCase   =		field(default='''Audio'''     ,     init=UpperCAmelCase_     ,     repr=UpperCAmelCase_						)




	def __call__(						self							:     Optional[Any]   )       -> List[str]:

			'''simple docstring'''




			return self.pa_type




	def    UpperCAmelCase			(						self							:     List[Any]					,      __lowerCamelCase							:     Union[str, bytes, dict]   )       -> dict:

			'''simple docstring'''




			try:
					import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
			except ImportError as err:
					raise ImportError('To support encoding audio data, please install \'soundfile\'.'   ) from err
			if isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					return {"bytes": None, "path": value}
			elif isinstance(__lowerCamelCase					,      __lowerCamelCase   ):
					return {"bytes": value, "path": None}
			elif "array" in value:
					# convert the audio array to wav bytes
					__lowercase											=     BytesIO()
					sf.write(__lowerCamelCase					,      value['array']					,      value['sampling_rate']					,      format='wav'   )
					return {"bytes": buffer.getvalue(), "path": None}
			elif value.get('path'   ) is not None and os.path.isfile(value['path']   ):
					# we set "bytes": None to not duplicate the data if they're already available locally
					if value["path"].endswith('pcm'   ):
							# "PCM" only has raw audio bytes
							if value.get('sampling_rate'   ) is None:
									# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
									raise KeyError('To use PCM files, please specify a \'sampling_rate\' in Audio object'   )
							if value.get('bytes'   ):
									# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
									__lowercase											=     np.frombuffer(value['bytes']					,      dtype=np.intaa   ).astype(np.floataa   ) / 32_767
							else:
									__lowercase											=     np.memmap(value['path']					,      dtype='h'					,      mode='r'   ).astype(np.floataa   ) / 32_767

							__lowercase											=     BytesIO(bytes()   )
							sf.write(__lowerCamelCase					,      __lowerCamelCase					,      value['sampling_rate']					,      format='wav'   )
							return {"bytes": buffer.getvalue(), "path": None}
					else:
							return {"bytes": None, "path": value.get('path'   )}
			elif value.get('bytes'   ) is not None or value.get('path'   ) is not None:
					# store the audio bytes, and path is used to infer the audio format using the file extension
					return {"bytes": value.get('bytes'   ), "path": value.get('path'   )}
			else:
					raise ValueError(
					    F"An audio sample should have one of 'path' or 'bytes' but they are missing or None in {value}."   )




	def    UpperCAmelCase			(						self							:     Any					,      __lowerCamelCase							:     dict					,      __lowerCamelCase							:     Optional[Dict[str, Union[str, bool, None]]] = None   )       -> dict:

			'''simple docstring'''




			if not self.decode:
					raise RuntimeError('Decoding is disabled for this feature. Please use Audio(decode=True) instead.'   )

			__lowercase						,      __lowercase											=     (value['path'], BytesIO(value['bytes']   )) if value['bytes'] is not None else (value['path'], None)
			if path is None and file is None:
					raise ValueError(F"An audio sample should have one of 'path' or 'bytes' but both are None in {value}."   )

			try:
					import librosa
					import soundfile as sf
			except ImportError as err:
					raise ImportError('To support decoding audio files, please install \'librosa\' and \'soundfile\'.'   ) from err

			__lowercase											=     xsplitext(__lowerCamelCase   )[1][1:].lower() if path is not None else None
			if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
					raise RuntimeError(
					    'Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, '
					    'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '   )
			elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
					raise RuntimeError(
					    'Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, '
					    'You can try to update `soundfile` python library: `pip install "soundfile>=0.12.1"`. '   )

			if file is None:
					__lowercase											=     token_per_repo_id or {}
					__lowercase											=     path.split('::'   )[-1]
					try:
							__lowercase											=     string_to_dict(__lowerCamelCase					,      config.HUB_DATASETS_URL   )['repo_id']
							__lowercase											=     token_per_repo_id[repo_id]
					except (ValueError, KeyError):
							__lowercase											=     None

					with xopen(__lowerCamelCase					,      'rb'					,      use_auth_token=__lowerCamelCase   ) as f:
							__lowercase						,      __lowercase											=     sf.read(__lowerCamelCase   )

			else:
					__lowercase						,      __lowercase											=     sf.read(__lowerCamelCase   )

			__lowercase											=     array.T
			if self.mono:
					__lowercase											=     librosa.to_mono(__lowerCamelCase   )
			if self.sampling_rate and self.sampling_rate != sampling_rate:
					__lowercase											=     librosa.resample(__lowerCamelCase					,      orig_sr=__lowerCamelCase					,      target_sr=self.sampling_rate   )
					__lowercase											=     self.sampling_rate

			return {"path": path, "array": array, "sampling_rate": sampling_rate}




	def    UpperCAmelCase			(						self							:     int   )       -> Union["FeatureType", Dict[str, "FeatureType"]]:

			'''simple docstring'''




			from .features import Value

			if self.decode:
					raise ValueError('Cannot flatten a decoded Audio feature.'   )
			return {
			    "bytes": Value('binary'   ),
			    "path": Value('string'   ),
			}




	def    UpperCAmelCase			(						self							:     List[Any]					,      __lowerCamelCase							:     Union[pa.StringArray, pa.StructArray]   )       -> pa.StructArray:

			'''simple docstring'''




			if pa.types.is_string(storage.type   ):
					__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.binary()   )
					__lowercase											=     pa.StructArray.from_arrays([bytes_array, storage]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			elif pa.types.is_binary(storage.type   ):
					__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.string()   )
					__lowercase											=     pa.StructArray.from_arrays([storage, path_array]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			elif pa.types.is_struct(storage.type   ) and storage.type.get_all_field_indices('array'   ):
					__lowercase											=     pa.array([Audio().encode_example(__lowerCamelCase   ) if x is not None else None for x in storage.to_pylist()]   )
			elif pa.types.is_struct(storage.type   ):
					if storage.type.get_field_index('bytes'   ) >= 0:
							__lowercase											=     storage.field('bytes'   )
					else:
							__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.binary()   )
					if storage.type.get_field_index('path'   ) >= 0:
							__lowercase											=     storage.field('path'   )
					else:
							__lowercase											=     pa.array([None] * len(__lowerCamelCase   )					,      type=pa.string()   )
					__lowercase											=     pa.StructArray.from_arrays([bytes_array, path_array]					,      ['bytes', 'path']					,      mask=storage.is_null()   )
			return array_cast(__lowerCamelCase					,      self.pa_type   )






	def    UpperCAmelCase			(						self							:     Union[str, Any]					,      __lowerCamelCase							:     pa.StructArray   )       -> pa.StructArray:

			'''simple docstring'''




			@no_op_if_value_is_null
			def path_to_bytes(__lowerCamelCase							:     Any   ):
					with xopen(__lowerCamelCase					,      'rb'   ) as f:
							__lowercase											=     f.read()
					return bytes_

			__lowercase											=     pa.array(
			    [
			        (path_to_bytes(x['path']   ) if x['bytes'] is None else x['bytes']) if x is not None else None
			        for x in storage.to_pylist()
			    ]					,      type=pa.binary()					,      )
			__lowercase											=     pa.array(
			    [os.path.basename(__lowerCamelCase   ) if path is not None else None for path in storage.field('path'   ).to_pylist()]					,      type=pa.string()					,      )
			__lowercase											=     pa.StructArray.from_arrays([bytes_array, path_array]					,      ['bytes', 'path']					,      mask=bytes_array.is_null()   )
			return array_cast(__lowerCamelCase					,      self.pa_type   )











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)




