


import argparse
import os
from os.path import isfile

import torch

from transformers import GPTaConfig
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:Optional[Any]   ,    lowercase__	:Optional[int]   ,    lowercase__	:int=0					)							->	str:
						# Format the message.
						if name is None:
												__magic_name__					:				Dict				=   None
						else:
												__magic_name__					:				Union[str, Any]				=   '.' * max(0   ,    spaces - 2					) + '# {:' + str(5_0 - spaces					) + 's}'
												__magic_name__					:				List[Any]				=   fmt.format(lowercase__					)

						# Print and recurse (if needed).
						if isinstance(lowercase__   ,    lowercase__					):
												if msg is not None:
																		print(lowercase__					)
												for k in val.keys():
																		recursive_print(lowercase__   ,    val[k]   ,    spaces + 2					)
						elif isinstance(lowercase__   ,    torch.Tensor					):
												print(lowercase__   ,    ':'   ,    val.size()					)
						else:
												print(lowercase__   ,    ':'   ,    lowercase__					)
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:List[Any]   ,    lowercase__	:Optional[int]   ,    lowercase__	:Optional[Any]   ,    lowercase__	:Any					)							->	Optional[int]:
						# Permutes layout of param tensor to [num_splits * num_heads * hidden_size, :]
						# for compatibility with later versions of NVIDIA Megatron-LM.
						# The inverse operation is performed inside Megatron-LM to read checkpoints:
						# https://github.com/NVIDIA/Megatron-LM/blob/v2.4/megatron/checkpointing.py#L209
						# If param is the weight tensor of the self-attention block, the returned tensor
						# will have to be transposed one more time to be read by HuggingFace GPT2.
						__magic_name__					:				List[Any]				=   param.size()
						# other versions store [num_heads * num_splits * hidden_size, :]
						__magic_name__					:				int				=   (num_heads, num_splits, hidden_size) + input_shape[1:]
						__magic_name__					:				Tuple				=   param.view(*lowercase__					)
						__magic_name__					:				int				=   param.transpose(0   ,    1					).contiguous()
						__magic_name__					:				Tuple				=   param.view(*lowercase__					)
						return param
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:Any   ,    lowercase__	:List[str]					)							->	Optional[Any]:
						__magic_name__					:				Tuple				=   config.n_positions
						__magic_name__					:				List[Any]				=   config.n_layer
						__magic_name__					:				str				=   config.vocab_size
						__magic_name__					:				Any				=   config.n_head
						__magic_name__					:				Optional[Any]				=   config.n_embd // config.n_head

						__magic_name__					:				Union[str, Any]				=   sd_megatron['model.language_model.embedding.word_embeddings.weight'][:vocab_size, :]
						__magic_name__					:				List[str]				=   {
						    'transformer.wte.weight': word_embeddings,
						    'transformer.wpe.weight': sd_megatron['model.language_model.embedding.position_embeddings.weight'],
						    'transformer.ln_f.weight': sd_megatron['model.language_model.encoder.final_layernorm.weight'],
						    'transformer.ln_f.bias': sd_megatron['model.language_model.encoder.final_layernorm.bias'],
						}

						__magic_name__					:				Optional[int]				=   'model.language_model.encoder.layers.'
						for i in range(lowercase__					):
												__magic_name__					:				List[str]				=   torch.tril(torch.ones((n_positions, n_positions)   ,    dtype=torch.bool					)					)
												__magic_name__					:				Tuple				=   causal_mask.view(1   ,    1   ,    lowercase__   ,    lowercase__					)
												__magic_name__					:				int				=   causal_mask
												__magic_name__					:				Optional[Any]				=   torch.tensor(-1e4   ,    dtype=torch.bfloataa					)

												__magic_name__					:				List[Any]				=   sd_megatron[f"{pf}{i}.input_layernorm.weight"]
												__magic_name__					:				Optional[int]				=   sd_megatron[f"{pf}{i}.input_layernorm.bias"]

												__magic_name__					:				Union[str, Any]				=   sd_megatron[f"{pf}{i}.self_attention.query_key_value.weight"]
												__magic_name__					:				str				=   fix_query_key_value_ordering(lowercase__   ,    3   ,    lowercase__   ,    lowercase__					)
												__magic_name__					:				Optional[int]				=   vala.transpose(0   ,    1					).contiguous()

												__magic_name__					:				Optional[int]				=   sd_megatron[f"{pf}{i}.self_attention.query_key_value.bias"]
												__magic_name__					:				Dict				=   fix_query_key_value_ordering(lowercase__   ,    3   ,    lowercase__   ,    lowercase__					)
												__magic_name__					:				Dict				=   vala

												__magic_name__					:				str				=   sd_megatron[f"{pf}{i}.self_attention.dense.weight"].transpose(
												    0   ,    1					)
												__magic_name__					:				Any				=   sd_megatron[f"{pf}{i}.self_attention.dense.bias"]
												__magic_name__					:				int				=   sd_megatron[f"{pf}{i}.post_attention_layernorm.weight"]
												__magic_name__					:				Tuple				=   sd_megatron[f"{pf}{i}.post_attention_layernorm.bias"]
												__magic_name__					:				List[str]				=   sd_megatron[f"{pf}{i}.mlp.dense_h_to_4h.weight"].transpose(0   ,    1					)
												__magic_name__					:				Optional[int]				=   sd_megatron[f"{pf}{i}.mlp.dense_h_to_4h.bias"]
												__magic_name__					:				Tuple				=   sd_megatron[f"{pf}{i}.mlp.dense_4h_to_h.weight"].transpose(
												    0   ,    1					)
												__magic_name__					:				int				=   sd_megatron[f"{pf}{i}.mlp.dense_4h_to_h.bias"]

						# For LM head, transformers' wants the matrix to weight embeddings.
						__magic_name__					:				Tuple				=   word_embeddings

						return sd_hf
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:Any   ,    lowercase__	:List[str]					)							->	Any:
						__magic_name__					:				Tuple				=   6_4_0_0_0
						__magic_name__					:				Union[str, Any]				=   config_megatron['encoder_seq_length']
						__magic_name__					:				Any				=   config_megatron['hidden_size']
						__magic_name__					:				Any				=   config_megatron['num_layers']
						__magic_name__					:				Optional[int]				=   config_megatron['num_attention_heads']
						__magic_name__					:				List[str]				=   config_megatron['ffn_hidden_size']
						__magic_name__					:				List[str]				=   'gelu'
						__magic_name__					:				Optional[int]				=   0.1
						__magic_name__					:				Any				=   0.1
						__magic_name__					:				Union[str, Any]				=   0.1
						__magic_name__					:				List[str]				=   config_megatron['layernorm_epsilon']  # 1e-5
						__magic_name__					:				Any				=   config_megatron['init_method_std']  # 0.02
						__magic_name__					:				Tuple				=   config_megatron['apply_query_key_layer_scaling']  # True
						__magic_name__					:				Union[str, Any]				=   True
						__magic_name__					:				List[str]				=   True

						# This identifies the 6.7B (7B) model which uses a different tokenizer
						if config_megatron["hidden_size"] == 4_0_9_6:
												__magic_name__					:				Dict				=   1  # <|endoftext|>
												__magic_name__					:				List[Any]				=   1  # <|endoftext|>
												__magic_name__					:				Dict				=   0  # <unk>
						else:
												__magic_name__					:				Optional[int]				=   2  # <s>
												__magic_name__					:				str				=   3  # <|endoftext|>
												__magic_name__					:				Union[str, Any]				=   0  # <pad>

						return config_hf
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:Optional[Any]					)							->	Any:
						print(lowercase__					)

						__magic_name__					:				List[str]				=   args.checkpoint_path
						__magic_name__					:				Tuple				=   args.save_path
						if isfile(lowercase__					):
												raise FileNotFoundError(f"ERROR! could not find file {checkpoint_path}"					)

						# Load the model.
						__magic_name__					:				str				=   torch.load(lowercase__   ,    map_location='cpu'					)

						# Load the config.
						__magic_name__					:				Any				=   checkpoint['hyper_parameters']['cfg']
						__magic_name__					:				List[str]				=   GPTaConfig()
						__magic_name__					:				Optional[int]				=   copy_config(config_hf=lowercase__   ,    config_megatron=lowercase__					)
						__magic_name__					:				Optional[Any]				=   ['GPT2LMHeadModel']

						__magic_name__					:				List[Any]				=   checkpoint['state_dict']

						# Convert.
						print('Converting'					)
						__magic_name__					:				Union[str, Any]				=   convert_megatron_checkpoint(lowercase__   ,    lowercase__					)

						# Print the structure of converted state dict.
						if args.print_checkpoint_structure:
												recursive_print(lowercase__   ,    lowercase__					)

						__magic_name__					:				List[str]				=   'GPTSw3Tokenizer'

						# Store the config to file.
						print('Saving config'					)
						config_hf.save_pretrained(lowercase__					)

						# Store the state_dict to file.
						__magic_name__					:				List[str]				=   os.path.join(lowercase__   ,    'pytorch_model.bin'					)
						print(f"Saving checkpoint to \"{output_checkpoint_file}\""					)
						torch.save(lowercase__   ,    lowercase__					)


if __name__ == "__main__":
			snake_case__  				=			argparse.ArgumentParser()
			parser.add_argument(
			    '--checkpoint_path',
			    type=str,
			    required=True,
			    help='e.g. megatron_gpt--val_loss=2.42-step=38000-consumed_samples=54720000',
			)
			parser.add_argument('--save_path', type=str, required=True, help='e.g. /home/user/gpt-sw3/hf')
			parser.add_argument('--print-checkpoint-structure', action='store_true')
			snake_case__  				=			parser.parse_args()
			main(_args)






import gc
import unittest

import numpy as np
import torch

from diffusers import DDPMScheduler, MidiProcessor, SpectrogramDiffusionPipeline
from diffusers.pipelines.spectrogram_diffusion import SpectrogramContEncoder, SpectrogramNotesEncoder, TaFilmDecoder
from diffusers.utils import require_torch_gpu, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_note_seq, require_onnxruntime

from ..pipeline_params import TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS, TOKENS_TO_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


snake_case__  				=			'./tests/fixtures/elise_format0.mid'



@unittest.skip('''The note-seq package currently throws an error on import'''	)
class      snake_case__	(   _A    ,							unittest.TestCase	):
				__lowerCAmelCase		=   SpectrogramDiffusionPipeline
				__lowerCAmelCase		=   PipelineTesterMixin.required_optional_params - {
				    '''callback''',
				    '''latents''',
				    '''callback_steps''',
				    '''output_type''',
				    '''num_images_per_prompt''',
				}
				__lowerCAmelCase		=   False
				__lowerCAmelCase		=   False
				__lowerCAmelCase		=   TOKENS_TO_AUDIO_GENERATION_PARAMS
				__lowerCAmelCase		=   TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										torch.manual_seed(0   )
										__magic_name__					:				Tuple				=   SpectrogramNotesEncoder(
										    max_length=2048  ,   vocab_size=1536  ,   d_model=768  ,   dropout_rate=0.1  ,   num_layers=1  ,   num_heads=1  ,   d_kv=4  ,   d_ff=2048  ,   feed_forward_proj='gated-gelu'  ,   )

										__magic_name__					:				Union[str, Any]				=   SpectrogramContEncoder(
										    input_dims=128  ,   targets_context_length=256  ,   d_model=768  ,   dropout_rate=0.1  ,   num_layers=1  ,   num_heads=1  ,   d_kv=4  ,   d_ff=2048  ,   feed_forward_proj='gated-gelu'  ,   )

										__magic_name__					:				Optional[Any]				=   TaFilmDecoder(
										    input_dims=128  ,   targets_length=256  ,   max_decoder_noise_time=20000.0  ,   d_model=768  ,   num_layers=1  ,   num_heads=1  ,   d_kv=4  ,   d_ff=2048  ,   dropout_rate=0.1  ,   )

										__magic_name__					:				Any				=   DDPMScheduler()

										__magic_name__					:				Tuple				=   {
										    'notes_encoder': notes_encoder.eval(),
										    'continuous_encoder': continuous_encoder.eval(),
										    'decoder': decoder.eval(),
										    'scheduler': scheduler,
										    'melgan': None,
										}
										return components
				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase  ,   _UpperCAmelCase=0   ):




										"""simple docstring"""




										if str(_UpperCAmelCase   ).startswith('mps'   ):
																__magic_name__					:				str				=   torch.manual_seed(_UpperCAmelCase   )
										else:
																__magic_name__					:				Optional[int]				=   torch.Generator(device=_UpperCAmelCase   ).manual_seed(_UpperCAmelCase   )
										__magic_name__					:				List[str]				=   {
										    'input_tokens': [
										        [1134, 90, 1135, 1133, 1080, 112, 1132, 1080, 1133, 1079, 133, 1132, 1079, 1133, 1] + [0] * 2033
										    ],
										    'generator': generator,
										    'num_inference_steps': 4,
										    'output_type': 'mel',
										}
										return inputs
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Dict				=   'cpu'  # ensure determinism for the device-dependent torch.Generator
										__magic_name__					:				Optional[Any]				=   self.get_dummy_components()
										__magic_name__					:				Dict				=   SpectrogramDiffusionPipeline(**_UpperCAmelCase   )
										__magic_name__					:				Union[str, Any]				=   pipe.to(_UpperCAmelCase   )
										pipe.set_progress_bar_config(disable=_UpperCAmelCase   )

										__magic_name__					:				List[Any]				=   self.get_dummy_inputs(_UpperCAmelCase   )
										__magic_name__					:				Dict				=   pipe(**_UpperCAmelCase   )
										__magic_name__					:				Optional[Any]				=   output.audios

										__magic_name__					:				Dict				=   mel[0, -3:, -3:]

										assert mel_slice.shape == (3, 3)
										__magic_name__					:				Union[str, Any]				=   np.array(
										    [-11.512925, -4.788215, -0.46172905, -2.051715, -10.539147, -10.970963, -9.091634, 4.0, 4.0]   )
										assert np.abs(mel_slice.flatten() - expected_slice   ).max() < 1E-2
				@skip_mps
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										return super().test_save_load_local()
				@skip_mps
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										return super().test_dict_tuple_outputs_equivalent()
				@skip_mps
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										return super().test_save_load_optional_components()
				@skip_mps
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										return super().test_attention_slicing_forward_pass()
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										pass
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										pass





				@skip_mps
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										return super().test_progress_bar()






@slow
@require_torch_gpu
@require_onnxruntime
@require_note_seq
class      snake_case__	(   unittest.TestCase	):
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Tuple				=   torch_device

										__magic_name__					:				Optional[int]				=   SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'   )
										__magic_name__					:				Dict				=   pipe.melgan
										__magic_name__					:				int				=   None

										__magic_name__					:				List[str]				=   pipe.to(_UpperCAmelCase   )
										pipe.set_progress_bar_config(disable=_UpperCAmelCase   )

										def callback(_UpperCAmelCase  ,   _UpperCAmelCase   ):
																# decode mel to audio
																__magic_name__					:				Any				=   melgan(input_features=mel_output.astype(np.floataa   )   )[0]
																assert len(audio[0]   ) == 8_1920 * (step + 1)
																# simulate that audio is played
																return audio

										__magic_name__					:				Optional[Any]				=   MidiProcessor()
										__magic_name__					:				str				=   processor(_UpperCAmelCase   )

										__magic_name__					:				Optional[int]				=   input_tokens[:3]
										__magic_name__					:				Optional[Any]				=   torch.manual_seed(0   )
										pipe(_UpperCAmelCase  ,   num_inference_steps=5  ,   generator=_UpperCAmelCase  ,   callback=_UpperCAmelCase  ,   output_type='mel'   )
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Tuple				=   torch_device

										__magic_name__					:				str				=   SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'   )
										__magic_name__					:				List[str]				=   pipe.to(_UpperCAmelCase   )
										pipe.set_progress_bar_config(disable=_UpperCAmelCase   )
										__magic_name__					:				Union[str, Any]				=   MidiProcessor()

										__magic_name__					:				List[Any]				=   processor(_UpperCAmelCase   )
										# just run two denoising loops
										__magic_name__					:				List[Any]				=   input_tokens[:2]

										__magic_name__					:				int				=   torch.manual_seed(0   )
										__magic_name__					:				int				=   pipe(_UpperCAmelCase  ,   num_inference_steps=2  ,   generator=_UpperCAmelCase   )

										__magic_name__					:				List[str]				=   output.audios[0]

										assert abs(np.abs(_UpperCAmelCase   ).sum() - 3612.841   ) < 1E-1





				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				Any				=   torch_device

										__magic_name__					:				List[Any]				=   SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion'   )
										__magic_name__					:				Any				=   pipe.to(_UpperCAmelCase   )
										pipe.set_progress_bar_config(disable=_UpperCAmelCase   )

										__magic_name__					:				List[str]				=   MidiProcessor()

										__magic_name__					:				Union[str, Any]				=   processor(_UpperCAmelCase   )

										# just run 4 denoising loops
										__magic_name__					:				List[Any]				=   input_tokens[:4]

										__magic_name__					:				Dict				=   torch.manual_seed(0   )
										__magic_name__					:				Union[str, Any]				=   pipe(_UpperCAmelCase  ,   num_inference_steps=100  ,   generator=_UpperCAmelCase   )

										__magic_name__					:				Dict				=   output.audios[0]
										assert abs(np.abs(_UpperCAmelCase   ).sum() - 9389.1111   ) < 5E-2


