
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       _snake_case   (	snake_case_  ,						unittest.TestCase							):
 __lowerCAmelCase  :     Union[str, Any]        =			DanceDiffusionPipeline
 __lowerCAmelCase  :     Optional[Any]        =			UNCONDITIONAL_AUDIO_GENERATION_PARAMS
 __lowerCAmelCase  :     List[Any]        =			PipelineTesterMixin.required_optional_params - {
     """callback""",
     """latents""",
     """callback_steps""",
     """output_type""",
     """num_images_per_prompt""",
 }
 __lowerCAmelCase  :     List[str]        =			UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
 __lowerCAmelCase  :     Any        =			False
 __lowerCAmelCase  :     List[Any]        =			False


 def      lowercase__     (					self):


   '''simple docstring'''




   torch.manual_seed(0)
   lowercase__  :   Any    		=      UNetaDModel(
       block_out_channels=(32, 32, 64)					,  extra_in_channels=16					,  sample_size=5_12					,  sample_rate=1_60_00					,  in_channels=2					,  out_channels=2					,  flip_sin_to_cos=SCREAMING_SNAKE_CASE_					,  use_timestep_embedding=SCREAMING_SNAKE_CASE_					,  time_embedding_type="""fourier"""					,  mid_block_type="""UNetMidBlock1D"""					,  down_block_types=("""DownBlock1DNoSkip""", """DownBlock1D""", """AttnDownBlock1D""")					,  up_block_types=("""AttnUpBlock1D""", """UpBlock1D""", """UpBlock1DNoSkip""")					,  )
   lowercase__  :   int    		=      IPNDMScheduler()

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


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_=0):


   '''simple docstring'''




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


 def      lowercase__     (					self):


   '''simple docstring'''




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

   lowercase__  :   Optional[int]    		=      self.get_dummy_inputs(SCREAMING_SNAKE_CASE_)
   lowercase__  :   Any    		=      pipe(**SCREAMING_SNAKE_CASE_)
   lowercase__  :   Dict    		=      output.audios

   lowercase__  :   int    		=      audio[0, -3:, -3:]

   assert audio.shape == (1, 2, components["unet"].sample_size)
   lowercase__  :   Dict    		=      np.array([-0.7_2_6_5, 1.0_0_0_0, -0.8_3_8_8, 0.1_1_7_5, 0.9_4_9_8, -1.0_0_0_0])
   assert np.abs(audio_slice.flatten() - expected_slice).max() < 1E-2


 @skip_mps
 def      lowercase__     (					self):


   '''simple docstring'''




   return super().test_save_load_local()


 @skip_mps
 def      lowercase__     (					self):


   '''simple docstring'''




   return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3)


 @skip_mps
 def      lowercase__     (					self):


   '''simple docstring'''




   return super().test_save_load_optional_components()


 @skip_mps
 def      lowercase__     (					self):


   '''simple docstring'''




   return super().test_attention_slicing_forward_pass()



 def      lowercase__     (					self):


   '''simple docstring'''




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



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


 def      lowercase__     (					self):


   '''simple docstring'''




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


 def      lowercase__     (					self):


   '''simple docstring'''




   lowercase__  :   Optional[Any]    		=      torch_device

   lowercase__  :   Any    		=      DanceDiffusionPipeline.from_pretrained("""harmonai/maestro-150k""")
   lowercase__  :   Dict    		=      pipe.to(SCREAMING_SNAKE_CASE_)
   pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_)

   lowercase__  :   List[str]    		=      torch.manual_seed(0)
   lowercase__  :   List[str]    		=      pipe(generator=SCREAMING_SNAKE_CASE_					,  num_inference_steps=1_00					,  audio_length_in_s=4.0_9_6)
   lowercase__  :   str    		=      output.audios

   lowercase__  :   List[Any]    		=      audio[0, -3:, -3:]

   assert audio.shape == (1, 2, pipe.unet.sample_size)
   lowercase__  :   int    		=      np.array([-0.0_1_9_2, -0.0_2_3_1, -0.0_3_1_8, -0.0_0_5_9, 0.0_0_0_2, -0.0_0_2_0])

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



 def      lowercase__     (					self):


   '''simple docstring'''




   lowercase__  :   str    		=      torch_device

   lowercase__  :   Optional[Any]    		=      DanceDiffusionPipeline.from_pretrained("""harmonai/maestro-150k"""					,  torch_dtype=torch.floataa)
   lowercase__  :   Optional[int]    		=      pipe.to(SCREAMING_SNAKE_CASE_)
   pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_)

   lowercase__  :   Tuple    		=      torch.manual_seed(0)
   lowercase__  :   Dict    		=      pipe(generator=SCREAMING_SNAKE_CASE_					,  num_inference_steps=1_00					,  audio_length_in_s=4.0_9_6)
   lowercase__  :   int    		=      output.audios

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

   assert audio.shape == (1, 2, pipe.unet.sample_size)
   lowercase__  :   List[str]    		=      np.array([-0.0_3_6_7, -0.0_4_8_8, -0.0_7_7_1, -0.0_5_2_5, -0.0_4_4_4, -0.0_3_4_1])

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








import argparse
import os

import numpy as np
import tensorflow as tf
import torch

from transformers import BertModel

def 		UpperCamelCase	(      lowercase_ , lowercase_ , lowercase_   )		->	Dict:


  '''simple docstring'''






  lowercase__  :   Dict    		=      ("""dense.weight""", """attention.self.query""", """attention.self.key""", """attention.self.value""")

  lowercase__  :   Union[str, Any]    		=      (
      ("""layer.""", """layer_"""),
      ("""word_embeddings.weight""", """word_embeddings"""),
      ("""position_embeddings.weight""", """position_embeddings"""),
      ("""token_type_embeddings.weight""", """token_type_embeddings"""),
      (""".""", """/"""),
      ("""LayerNorm/weight""", """LayerNorm/gamma"""),
      ("""LayerNorm/bias""", """LayerNorm/beta"""),
      ("""weight""", """kernel"""),
  )

  if not os.path.isdir(lowercase_   ):
    os.makedirs(lowercase_   )

  lowercase__  :   Optional[Any]    		=      model.state_dict()

  def to_tf_var_name(lowercase_   ):
    for patt, repl in iter(lowercase_   ):
      lowercase__  :   str    		=      name.replace(lowercase_ , lowercase_   )
    return F'bert/{name}'

  def create_tf_var(lowercase_ , lowercase_ , lowercase_   ):
    lowercase__  :   List[str]    		=      tf.dtypes.as_dtype(tensor.dtype   )
    lowercase__  :   Tuple    		=      tf.get_variable(dtype=lowercase_ , shape=tensor.shape , name=lowercase_ , initializer=tf.zeros_initializer()   )
    session.run(tf.variables_initializer([tf_var]   )   )
    session.run(lowercase_   )
    return tf_var

  tf.reset_default_graph()
  with tf.Session() as session:
    for var_name in state_dict:
      lowercase__  :   Any    		=      to_tf_var_name(lowercase_   )
      lowercase__  :   Dict    		=      state_dict[var_name].numpy()
      if any(x in var_name for x in tensors_to_transpose   ):
        lowercase__  :   Optional[int]    		=      torch_tensor.T
      lowercase__  :   Any    		=      create_tf_var(tensor=lowercase_ , name=lowercase_ , session=lowercase_   )
      tf.keras.backend.set_value(lowercase_ , lowercase_   )
      lowercase__  :   Optional[int]    		=      session.run(lowercase_   )
      print(F'Successfully created {tf_name}: {np.allclose(lowercase_ , lowercase_   )}'   )

    lowercase__  :   List[Any]    		=      tf.train.Saver(tf.trainable_variables()   )
    saver.save(lowercase_ , os.path.join(lowercase_ , model_name.replace("""-""" , """_"""   ) + """.ckpt"""   )   )

def 		UpperCamelCase	(      lowercase_=None   )		->	Union[str, Any]:


  '''simple docstring'''






  lowercase__  :   List[str]    		=      argparse.ArgumentParser()
  parser.add_argument("""--model_name""" , type=lowercase_ , required=lowercase_ , help="""model name e.g. bert-base-uncased"""   )
  parser.add_argument(
      """--cache_dir""" , type=lowercase_ , default=lowercase_ , required=lowercase_ , help="""Directory containing pytorch model"""   )
  parser.add_argument("""--pytorch_model_path""" , type=lowercase_ , required=lowercase_ , help="""/path/to/<pytorch-model-name>.bin"""   )
  parser.add_argument("""--tf_cache_dir""" , type=lowercase_ , required=lowercase_ , help="""Directory in which to save tensorflow model"""   )
  lowercase__  :   List[str]    		=      parser.parse_args(lowercase_   )

  lowercase__  :   Any    		=      BertModel.from_pretrained(
      pretrained_model_name_or_path=args.model_name , state_dict=torch.load(args.pytorch_model_path   ) , cache_dir=args.cache_dir , )

  convert_pytorch_checkpoint_to_tf(model=lowercase_ , ckpt_dir=args.tf_cache_dir , model_name=args.model_name   )


if __name__ == "__main__":
     main()






