

import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()

class 				a  (	UpperCAmelCase							, unittest.TestCase				):
						_lowercase    						=			DanceDiffusionPipeline
						_lowercase    						=			UNCONDITIONAL_AUDIO_GENERATION_PARAMS
						_lowercase    						=			PipelineTesterMixin.required_optional_params - {
						    "callback",
						    "latents",
						    "callback_steps",
						    "output_type",
						    "num_images_per_prompt",
						}
						_lowercase    						=			UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
						_lowercase    						=			False
						_lowercase    						=			False


						def        _UpperCAmelCase				(       self	):







										'''simple docstring'''


										torch.manual_seed(0	)
										_UpperCAmelCase       :		str 		=    UNetaDModel(
										    block_out_channels=(32, 32, 64)						,					extra_in_channels=16						,					sample_size=512						,					sample_rate=16000						,					in_channels=2						,					out_channels=2						,					flip_sin_to_cos=A_						,					use_timestep_embedding=A_						,					time_embedding_type="fourier"						,					mid_block_type="UNetMidBlock1D"						,					down_block_types=("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")						,					up_block_types=("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")						,					)
										_UpperCAmelCase       :		Union[str, Any] 		=    IPNDMScheduler()

										_UpperCAmelCase       :		Union[str, Any] 		=    {
										    "unet": unet,
										    "scheduler": scheduler,
										}
										return components


						def        _UpperCAmelCase				(       self						,					A_						,					A_=0	):







										'''simple docstring'''


										if str(A_	).startswith("mps"	):
														_UpperCAmelCase       :		int 		=    torch.manual_seed(A_	)
										else:
														_UpperCAmelCase       :		Tuple 		=    torch.Generator(device=A_	).manual_seed(A_	)
										_UpperCAmelCase       :		Optional[Any] 		=    {
										    "batch_size": 1,
										    "generator": generator,
										    "num_inference_steps": 4,
										}
										return inputs


						def        _UpperCAmelCase				(       self	):







										'''simple docstring'''


										_UpperCAmelCase       :		Union[str, Any] 		=    "cpu"  # ensure determinism for the device-dependent torch.Generator
										_UpperCAmelCase       :		int 		=    self.get_dummy_components()
										_UpperCAmelCase       :		List[str] 		=    DanceDiffusionPipeline(**A_	)
										_UpperCAmelCase       :		str 		=    pipe.to(A_	)
										pipe.set_progress_bar_config(disable=A_	)

										_UpperCAmelCase       :		Any 		=    self.get_dummy_inputs(A_	)
										_UpperCAmelCase       :		Union[str, Any] 		=    pipe(**A_	)
										_UpperCAmelCase       :		int 		=    output.audios

										_UpperCAmelCase       :		List[str] 		=    audio[0, -3:, -3:]

										assert audio.shape == (1, 2, components["unet"].sample_size)
										_UpperCAmelCase       :		Any 		=    np.array([-0.72_65, 1.00_00, -0.83_88, 0.11_75, 0.94_98, -1.00_00]	)
										assert np.abs(audio_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(expected_max_difference=3e-3	)


						@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'''


										super().test_inference_batch_single_identical(expected_max_diff=3e-3	)







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


						def        _UpperCAmelCase				(       self	):







										'''simple docstring'''


										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()


						def        _UpperCAmelCase				(       self	):







										'''simple docstring'''


										_UpperCAmelCase       :		Union[str, Any] 		=    torch_device

										_UpperCAmelCase       :		Optional[int] 		=    DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"	)
										_UpperCAmelCase       :		Union[str, Any] 		=    pipe.to(A_	)
										pipe.set_progress_bar_config(disable=A_	)

										_UpperCAmelCase       :		Any 		=    torch.manual_seed(0	)
										_UpperCAmelCase       :		Dict 		=    pipe(generator=A_						,					num_inference_steps=100						,					audio_length_in_s=4.0_96	)
										_UpperCAmelCase       :		Tuple 		=    output.audios

										_UpperCAmelCase       :		Dict 		=    audio[0, -3:, -3:]

										assert audio.shape == (1, 2, pipe.unet.sample_size)
										_UpperCAmelCase       :		str 		=    np.array([-0.01_92, -0.02_31, -0.03_18, -0.00_59, 0.00_02, -0.00_20]	)

										assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2





						def        _UpperCAmelCase				(       self	):







										'''simple docstring'''


										_UpperCAmelCase       :		Optional[Any] 		=    torch_device

										_UpperCAmelCase       :		List[str] 		=    DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"						,					torch_dtype=torch.floataa	)
										_UpperCAmelCase       :		Optional[int] 		=    pipe.to(A_	)
										pipe.set_progress_bar_config(disable=A_	)

										_UpperCAmelCase       :		Optional[Any] 		=    torch.manual_seed(0	)
										_UpperCAmelCase       :		Optional[int] 		=    pipe(generator=A_						,					num_inference_steps=100						,					audio_length_in_s=4.0_96	)
										_UpperCAmelCase       :		Union[str, Any] 		=    output.audios

										_UpperCAmelCase       :		Optional[int] 		=    audio[0, -3:, -3:]

										assert audio.shape == (1, 2, pipe.unet.sample_size)
										_UpperCAmelCase       :		List[str] 		=    np.array([-0.03_67, -0.04_88, -0.07_71, -0.05_25, -0.04_44, -0.03_41]	)

										assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2





import argparse
import json
import os
from collections import OrderedDict

import numpy as np
import tensorflow as tf
import torch

def 							__SCREAMING_SNAKE_CASE							(				lowerCAmelCase:  Any )   ->       Optional[int]:
				_UpperCAmelCase       :		Optional[Any] 		=    os.path.join(args.tf_model_dir  ,      "parameters.json" )
				_UpperCAmelCase       :		List[str] 		=    json.loads(open(lowerCAmelCase ).read() )
				if not params:
								raise ValueError(
								    F'It seems that the json file at {parameter_file} is empty. Make sure you have a correct json file.' )
				if not args.output.endswith(".pt" ):
								_UpperCAmelCase       :		Union[str, Any] 		=    args.output + ".pt"
				_UpperCAmelCase       :		Dict 		=    OrderedDict()
				with tf.device("/CPU:0" ):
								_UpperCAmelCase       :		int 		=    tf.train.load_checkpoint(args.tf_model_dir )
								_UpperCAmelCase       :		int 		=    reader.get_variable_to_shape_map()
								for key_name in shapes.keys():
												_UpperCAmelCase       :		Dict 		=    reader.get_tensor(lowerCAmelCase ).astype(np.floataa )
												if key_name.endswith("/adam_m" ) or key_name.endswith("/adam_v" ):
																continue
												if key_name.startswith("pasts/" ):
																if key_name.startswith("pasts/mlp" ):
																				_UpperCAmelCase       :		int 		=    int(key_name[9] )
																elif key_name.startswith("pasts/out" ):
																				_UpperCAmelCase       :		Any 		=    8
																_UpperCAmelCase       :		str 		=    "model.sqout.%d.weight" % (player * 2)  # enter to nn.Sequencial with Tanh, so 2 at a time
																_UpperCAmelCase       :		Optional[Any] 		=    vnp.transpose([1, 0] ).copy()  # Mesh-Tensorflow is a diagonal matrix
																_UpperCAmelCase       :		str 		=    torch.tensor(lowerCAmelCase )
												elif key_name.startswith("model/moe" ):
																_UpperCAmelCase       :		List[str] 		=    int(key_name[9:].split("/" )[0] )
																if key_name.endswith("/switch_gating/kernel" ):
																				_UpperCAmelCase       :		Union[str, Any] 		=    "model.blocks.%d.feed_forward.mlp.router.classifier.weight" % player
																				_UpperCAmelCase       :		List[str] 		=    vnp.transpose([1, 0] ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		Tuple 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/softmlp/kernel" ):
																				_UpperCAmelCase       :		List[str] 		=    "model.blocks.%d.feed_forward.soft_bypass_mlp.weight" % player
																				_UpperCAmelCase       :		List[str] 		=    vnp.transpose([1, 0] ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		Union[str, Any] 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/wo/kernel" ) or key_name.endswith("/wi/kernel" ):
																				_UpperCAmelCase       :		str 		=    key_name[-9:-7]
																				for i in range(16 ):
																								_UpperCAmelCase       :		Optional[int] 		=    "model.blocks.%d.feed_forward.mlp.experts.expert_%d.%s.weight" % (player, i, nlayer)
																								_UpperCAmelCase       :		Optional[int] 		=    (
																								    vnp[i].transpose([1, 0] ).copy()
																								)  # In Mesh-Tensorflow, it is one array, so it is divided
																								_UpperCAmelCase       :		Dict 		=    torch.tensor(lowerCAmelCase )
												elif key_name.startswith("model/mlp" ):
																_UpperCAmelCase       :		Tuple 		=    int(key_name[9:].split("/" )[0] )
																if key_name.endswith("/p1/kernel" ):
																				_UpperCAmelCase       :		Any 		=    "model.blocks.%d.feed_forward.mlp.wi.weight" % player
																				_UpperCAmelCase       :		int 		=    vnp.transpose([1, 0] ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		List[str] 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/p1/bias" ):
																				_UpperCAmelCase       :		Optional[Any] 		=    "model.blocks.%d.feed_forward.mlp.wi.bias" % player
																				_UpperCAmelCase       :		Tuple 		=    vnp.copy()  # same because it is one dimensional
																				_UpperCAmelCase       :		Any 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/p2/kernel" ):
																				_UpperCAmelCase       :		str 		=    "model.blocks.%d.feed_forward.mlp.wo.weight" % player
																				_UpperCAmelCase       :		Union[str, Any] 		=    vnp.transpose([1, 0] ).copy()  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		List[Any] 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/p2/bias" ):
																				_UpperCAmelCase       :		int 		=    "model.blocks.%d.feed_forward.mlp.wo.bias" % player
																				_UpperCAmelCase       :		Optional[int] 		=    vnp.copy()  # same because it is one dimensional
																				_UpperCAmelCase       :		str 		=    torch.tensor(lowerCAmelCase )
												elif key_name.startswith("model/ln" ):
																_UpperCAmelCase       :		Union[str, Any] 		=    int(key_name[8:].split("/" )[0] )
																if key_name.endswith("/b" ):
																				_UpperCAmelCase       :		Union[str, Any] 		=    "model.blocks.%d.feed_forward.norm.bias" % player
																				_UpperCAmelCase       :		int 		=    vnp.copy()  # same because it is one dimensional
																				_UpperCAmelCase       :		Union[str, Any] 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/g" ):
																				_UpperCAmelCase       :		Optional[Any] 		=    "model.blocks.%d.feed_forward.norm.weight" % player
																				_UpperCAmelCase       :		Union[str, Any] 		=    vnp.copy()  # same because it is one dimensional
																				_UpperCAmelCase       :		List[Any] 		=    torch.tensor(lowerCAmelCase )
												elif key_name.startswith("model/att" ):
																_UpperCAmelCase       :		Optional[Any] 		=    int(key_name[9:].split("/" )[0] )
																if key_name.endswith("/qkv/kernel" ):
																				_UpperCAmelCase       :		List[Any] 		=    vnp.copy()  # Compute same dimension as Mesh-tensorflow using einsum
																				_UpperCAmelCase       :		Dict 		=    state[:, 0, :, :]
																				_UpperCAmelCase       :		List[Any] 		=    state[:, 1, :, :]
																				_UpperCAmelCase       :		Tuple 		=    state[:, 2, :, :]
																				_UpperCAmelCase       :		List[str] 		=    (
																				    state_q.reshape([state_q.shape[0], state_q.shape[1] * state_q.shape[2]] )
																				    .transpose([1, 0] )
																				    .copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		int 		=    (
																				    state_k.reshape([state_k.shape[0], state_k.shape[1] * state_k.shape[2]] )
																				    .transpose([1, 0] )
																				    .copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		Tuple 		=    (
																				    state_v.reshape([state_v.shape[0], state_v.shape[1] * state_v.shape[2]] )
																				    .transpose([1, 0] )
																				    .copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		int 		=    "model.blocks.%d.self_attn.self_attn.q_proj.weight" % player
																				_UpperCAmelCase       :		Any 		=    torch.tensor(lowerCAmelCase )
																				_UpperCAmelCase       :		List[Any] 		=    "model.blocks.%d.self_attn.self_attn.k_proj.weight" % player
																				_UpperCAmelCase       :		Any 		=    torch.tensor(lowerCAmelCase )
																				_UpperCAmelCase       :		Union[str, Any] 		=    "model.blocks.%d.self_attn.self_attn.v_proj.weight" % player
																				_UpperCAmelCase       :		str 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/o/kernel" ):
																				_UpperCAmelCase       :		Dict 		=    "model.blocks.%d.self_attn.self_attn.out_proj.weight" % player
																				_UpperCAmelCase       :		Tuple 		=    (
																				    vnp.reshape([vnp.shape[0] * vnp.shape[1], vnp.shape[2]] ).transpose([1, 0] ).copy()
																				)  # Mesh-Tensorflow is a diagonal matrix
																				_UpperCAmelCase       :		List[str] 		=    torch.tensor(lowerCAmelCase )
												elif key_name.startswith("model/an" ):
																_UpperCAmelCase       :		Optional[int] 		=    int(key_name[8:].split("/" )[0] )
																if key_name.endswith("/b" ):
																				_UpperCAmelCase       :		Any 		=    "model.blocks.%d.self_attn.norm.bias" % player
																				_UpperCAmelCase       :		Any 		=    vnp.copy()  # same because it is one dimensional
																				_UpperCAmelCase       :		Dict 		=    torch.tensor(lowerCAmelCase )
																elif key_name.endswith("/g" ):
																				_UpperCAmelCase       :		Dict 		=    "model.blocks.%d.self_attn.norm.weight" % player
																				_UpperCAmelCase       :		Any 		=    vnp.copy()  # same because it is one dimensional
																				_UpperCAmelCase       :		Tuple 		=    torch.tensor(lowerCAmelCase )
												elif (
												    key_name.startswith("model/wte" )
												    or key_name.startswith("model/wpe" )
												    or key_name.startswith("model/ete" )
												):
																_UpperCAmelCase       :		Optional[Any] 		=    {"wte": "embed_tokens", "wpe": "position_embeddings", "ete": "extra_position_embeddings"}[
																    key_name[-3:]
																]
																_UpperCAmelCase       :		Dict 		=    "model.%s.weight" % nlayer
																_UpperCAmelCase       :		Optional[Any] 		=    vnp.copy()  # same in embedded
																_UpperCAmelCase       :		List[Any] 		=    torch.tensor(lowerCAmelCase )
																if key_name.startswith("model/wte" ):
																				_UpperCAmelCase       :		List[str] 		=    "lm_head.weight"
																				_UpperCAmelCase       :		Optional[int] 		=    vnp.copy()  # same in embedded
																				_UpperCAmelCase       :		List[str] 		=    torch.tensor(lowerCAmelCase )
												elif key_name.startswith("model/wob" ):
																_UpperCAmelCase       :		Dict 		=    "final_logits_bias"
																_UpperCAmelCase       :		Optional[Any] 		=    vnp.copy()  # same in embedded
																_UpperCAmelCase       :		str 		=    state.reshape((1, -1) )
																_UpperCAmelCase       :		Dict 		=    torch.tensor(lowerCAmelCase )
												elif key_name == "model/dense/kernel":
																_UpperCAmelCase       :		List[Any] 		=    "model.last_project.weight"
																_UpperCAmelCase       :		int 		=    vnp.transpose([1, 0] ).copy()  # Mesh-Tensorflow is a diagonal matrix
																_UpperCAmelCase       :		str 		=    torch.tensor(lowerCAmelCase )
												elif key_name == "model/dense_1/bias":
																_UpperCAmelCase       :		List[Any] 		=    "model.last_project.bias"
																_UpperCAmelCase       :		Any 		=    vnp.copy()  # same because it is one dimensional
																_UpperCAmelCase       :		Any 		=    torch.tensor(lowerCAmelCase )
				torch.save(lowerCAmelCase  ,      args.output )


if __name__ == "__main__":
					SCREAMING_SNAKE_CASE_        =      argparse.ArgumentParser(
					    description='model converter.', formatter_class=argparse.ArgumentDefaultsHelpFormatter
					)
					parser.add_argument('--tf_model_dir', metavar='PATH', type=str, required=True, help='import model')
					parser.add_argument('--output', metavar='PATH', type=str, required=True, help='output model')
					SCREAMING_SNAKE_CASE_        =      parser.parse_args()
					convert_tf_gptsan_to_pt(args)


