





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						):







       """simple docstring"""





       SCREAMING_SNAKE_CASE_   :			Any  			=	DanceDiffusionPipeline
       SCREAMING_SNAKE_CASE_   :			Union[str, Any]  			=	UNCONDITIONAL_AUDIO_GENERATION_PARAMS
       SCREAMING_SNAKE_CASE_   :			List[Any]  			=	PipelineTesterMixin.required_optional_params - {
           '''callback''',
           '''latents''',
           '''callback_steps''',
           '''output_type''',
           '''num_images_per_prompt''',
       }
       SCREAMING_SNAKE_CASE_   :			Optional[int]  			=	UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
       SCREAMING_SNAKE_CASE_   :			int  			=	False
       SCREAMING_SNAKE_CASE_   :			Optional[int]  			=	False







       def  __UpperCAmelCase			(     self						:						Tuple	)    ->       Tuple:
         torch.manual_seed(0	)
         _lowercase						    =			UNetaDModel(
             block_out_channels=(32, 32, 64)    ,extra_in_channels=16    ,sample_size=512    ,sample_rate=1_6000    ,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')    ,)
         _lowercase						    =			IPNDMScheduler()

         _lowercase						    =			{
             'unet': unet,
             'scheduler': scheduler,
         }
         return components







       def  __UpperCAmelCase			(     self						:						str    ,__A						:						Tuple    ,__A						:						int=0	)    ->       Optional[Any]:
         if str(__A	).startswith('mps'	):
           _lowercase						    =			torch.manual_seed(__A	)
         else:
           _lowercase						    =			torch.Generator(device=__A	).manual_seed(__A	)
         _lowercase						    =			{
             'batch_size': 1,
             'generator': generator,
             'num_inference_steps': 4,
         }
         return inputs







       def  __UpperCAmelCase			(     self						:						Optional[int]	)    ->       Union[str, Any]:
         _lowercase						    =			'cpu'  # ensure determinism for the device-dependent torch.Generator
         _lowercase						    =			self.get_dummy_components()
         _lowercase						    =			DanceDiffusionPipeline(**__A	)
         _lowercase						    =			pipe.to(__A	)
         pipe.set_progress_bar_config(disable=__A	)

         _lowercase						    =			self.get_dummy_inputs(__A	)
         _lowercase						    =			pipe(**__A	)
         _lowercase						    =			output.audios

         _lowercase						    =			audio[0, -3:, -3:]

         assert audio.shape == (1, 2, components["unet"].sample_size)
         _lowercase						    =			np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]	)
         assert np.abs(audio_slice.flatten() - expected_slice	).max() < 1e-2







       @skip_mps
       def  __UpperCAmelCase			(     self						:						List[str]	)    ->       Union[str, Any]:
         return super().test_save_load_local()







       @skip_mps
       def  __UpperCAmelCase			(     self						:						Dict	)    ->       Any:
         return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3	)







       @skip_mps
       def  __UpperCAmelCase			(     self						:						Dict	)    ->       List[str]:
         return super().test_save_load_optional_components()







       @skip_mps
       def  __UpperCAmelCase			(     self						:						int	)    ->       Any:
         return super().test_attention_slicing_forward_pass()







       def  __UpperCAmelCase			(     self						:						str	)    ->       List[str]:
         super().test_inference_batch_single_identical(expected_max_diff=3e-3	)





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







       """simple docstring"""





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







       def  __UpperCAmelCase			(     self						:						Any	)    ->       Any:
         _lowercase						    =			torch_device

         _lowercase						    =			DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'	)
         _lowercase						    =			pipe.to(__A	)
         pipe.set_progress_bar_config(disable=__A	)

         _lowercase						    =			torch.manual_seed(0	)
         _lowercase						    =			pipe(generator=__A    ,num_inference_steps=100    ,audio_length_in_s=4.096	)
         _lowercase						    =			output.audios

         _lowercase						    =			audio[0, -3:, -3:]

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         _lowercase						    =			np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]	)

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







       def  __UpperCAmelCase			(     self						:						Dict	)    ->       int:
         _lowercase						    =			torch_device

         _lowercase						    =			DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'    ,torch_dtype=torch.floataa	)
         _lowercase						    =			pipe.to(__A	)
         pipe.set_progress_bar_config(disable=__A	)

         _lowercase						    =			torch.manual_seed(0	)
         _lowercase						    =			pipe(generator=__A    ,num_inference_steps=100    ,audio_length_in_s=4.096	)
         _lowercase						    =			output.audios

         _lowercase						    =			audio[0, -3:, -3:]

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         _lowercase						    =			np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]	)

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

import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment







def __UpperCamelCase( _A       :					Any	,      _A       :					List[str]=()	,      _A       :					List[str]=None	,      _A       :					Dict="no"	,      _A       :					List[str]="29500"     ):
   '''simple docstring'''
   UpperCAmelCase__				:					int 					=	False
   UpperCAmelCase__				:					List[str] 					=	False
   if any(key.startswith('''KAGGLE'''     ) for key in os.environ.keys()     ):
      UpperCAmelCase__				:					List[str] 					=	True
   elif "IPython" in sys.modules:
      UpperCAmelCase__				:					List[Any] 					=	'''google.colab''' in str(sys.modules['''IPython'''].get_ipython()     )

   try:
      UpperCAmelCase__				:					Tuple 					=	PrecisionType(mixed_precision.lower()     )
   except ValueError:
      raise ValueError(
          F'''Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}.'''     )

   if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''	,      _A     ) is not None):
      # TPU launch
      import torch_xla.distributed.xla_multiprocessing as xmp

      if len(AcceleratorState._shared_state     ) > 0:
         raise ValueError(
             '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
             '''your training function. Restart your notebook and make sure no cells initializes an '''
             '''`Accelerator`.'''     )
      if num_processes is None:
         UpperCAmelCase__				:					Tuple 					=	8

      UpperCAmelCase__				:					Optional[Any] 					=	PrepareForLaunch(_A	,      distributed_type='''TPU'''     )
      print(F'''Launching a training on {num_processes} TPU cores.'''     )
      xmp.spawn(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )
   elif in_colab:
      # No need for a distributed launch otherwise as it's either CPU or one GPU.
      if torch.cuda.is_available():
         print('''Launching training on one GPU.'''     )
      else:
         print('''Launching training on one CPU.'''     )
      function(*_A     )
   else:
      if num_processes is None:
         raise ValueError(
             '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.'''     )

      if num_processes > 1:
         # Multi-GPU launch
         from torch.multiprocessing import start_processes
         from torch.multiprocessing.spawn import ProcessRaisedException

         if len(AcceleratorState._shared_state     ) > 0:
            raise ValueError(
                '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
                '''inside your training function. Restart your notebook and make sure no cells initializes an '''
                '''`Accelerator`.'''     )

         if torch.cuda.is_initialized():
            raise ValueError(
                '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
                '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
                '''function.'''     )

         # torch.distributed will expect a few environment variable to be here. We set the ones common to each
         # process here (the other ones will be set be the launcher).
         with patch_environment(
             world_size=_A	,      master_addr='''127.0.01'''	,      master_port=_A	,      mixed_precision=_A     ):
            UpperCAmelCase__				:					str 					=	PrepareForLaunch(_A	,      distributed_type='''MULTI_GPU'''     )
            print(F'''Launching training on {num_processes} GPUs.'''     )
            try:
               start_processes(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )
            except ProcessRaisedException as e:
               if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
                  raise RuntimeError(
                      '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
                      '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
                      '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
                      '''which one is problematic.'''     ) from e

      else:
         # No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
         if is_mps_available():
            UpperCAmelCase__				:					Union[str, Any] 					=	'''1'''
            print('''Launching training on MPS.'''     )
         elif torch.cuda.is_available():
            print('''Launching training on one GPU.'''     )
         else:
            print('''Launching training on CPU.'''     )
         function(*_A     )







def __UpperCamelCase( _A       :					List[str]	,      _A       :					Optional[Any]=()	,      _A       :					str=2     ):
   '''simple docstring'''
   from torch.multiprocessing import start_processes

   with tempfile.NamedTemporaryFile() as tmp_file:
      # torch.distributed will expect a few environment variable to be here. We set the ones common to each
      # process here (the other ones will be set be the launcher).
      with patch_environment(
          world_size=_A	,      master_addr='''127.0.01'''	,      master_port='''29500'''	,      accelerate_mixed_precision='''no'''	,      accelerate_debug_rdv_file=tmp_file.name	,      accelerate_use_cpu='''yes'''	,      ):
         UpperCAmelCase__				:					Optional[int] 					=	PrepareForLaunch(_A	,      debug=_A     )
         start_processes(_A	,      args=_A	,      nprocs=_A	,      start_method='''fork'''     )




