


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


      __snake_case 	=       DanceDiffusionPipeline
      __snake_case 	=       UNCONDITIONAL_AUDIO_GENERATION_PARAMS
      __snake_case 	=       PipelineTesterMixin.required_optional_params - {
          """callback""",
          """latents""",
          """callback_steps""",
          """output_type""",
          """num_images_per_prompt""",
      }
      __snake_case 	=       UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
      __snake_case 	=       False
      __snake_case 	=       False




      def 					_snake_case     (    self:	Union[str, Any]    ):
       torch.manual_seed(0    )
       __lowerCamelCase		:							Any		 =						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')						,				)
       __lowerCamelCase		:							Union[str, Any]		 =						IPNDMScheduler()

       __lowerCamelCase		:							int		 =						{
           'unet': unet,
           'scheduler': scheduler,
       }
       return components




      def 					_snake_case     (    self:	Optional[Any]						,				a:	Optional[int]						,				a:	Optional[Any]=0    ):
       if str(a    ).startswith('mps'    ):
        __lowerCamelCase		:							Optional[int]		 =						torch.manual_seed(a    )
       else:
        __lowerCamelCase		:							int		 =						torch.Generator(device=a    ).manual_seed(a    )
       __lowerCamelCase		:							List[str]		 =						{
           'batch_size': 1,
           'generator': generator,
           'num_inference_steps': 4,
       }
       return inputs




      def 					_snake_case     (    self:	List[str]    ):
       __lowerCamelCase		:							Tuple		 =						'cpu'  # ensure determinism for the device-dependent torch.Generator
       __lowerCamelCase		:							Optional[int]		 =						self.get_dummy_components()
       __lowerCamelCase		:							Tuple		 =						DanceDiffusionPipeline(**a    )
       __lowerCamelCase		:							Dict		 =						pipe.to(a    )
       pipe.set_progress_bar_config(disable=a    )

       __lowerCamelCase		:							Optional[int]		 =						self.get_dummy_inputs(a    )
       __lowerCamelCase		:							Optional[int]		 =						pipe(**a    )
       __lowerCamelCase		:							List[Any]		 =						output.audios

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

       assert audio.shape == (1, 2, components["unet"].sample_size)
       __lowerCamelCase		:							List[str]		 =						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 					_snake_case     (    self:	int    ):
       return super().test_save_load_local()




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




      @skip_mps
      def 					_snake_case     (    self:	Optional[int]    ):
       return super().test_save_load_optional_components()




      @skip_mps
      def 					_snake_case     (    self:	Optional[int]    ):
       return super().test_attention_slicing_forward_pass()




      def 					_snake_case     (    self:	Any    ):
       super().test_inference_batch_single_identical(expected_max_diff=3e-3    )



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




      '''simple docstring'''


      def 					_snake_case     (    self:	str    ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()




      def 					_snake_case     (    self:	Optional[Any]    ):
       __lowerCamelCase		:							int		 =						torch_device

       __lowerCamelCase		:							List[str]		 =						DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'    )
       __lowerCamelCase		:							int		 =						pipe.to(a    )
       pipe.set_progress_bar_config(disable=a    )

       __lowerCamelCase		:							Union[str, Any]		 =						torch.manual_seed(0    )
       __lowerCamelCase		:							List[Any]		 =						pipe(generator=a						,				num_inference_steps=100						,				audio_length_in_s=4.0_9_6    )
       __lowerCamelCase		:							str		 =						output.audios

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

       assert audio.shape == (1, 2, pipe.unet.sample_size)
       __lowerCamelCase		:							str		 =						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 					_snake_case     (    self:	Optional[Any]    ):
       __lowerCamelCase		:							List[str]		 =						torch_device

       __lowerCamelCase		:							Tuple		 =						DanceDiffusionPipeline.from_pretrained('harmonai/maestro-150k'						,				torch_dtype=torch.floataa    )
       __lowerCamelCase		:							Optional[Any]		 =						pipe.to(a    )
       pipe.set_progress_bar_config(disable=a    )

       __lowerCamelCase		:							List[Any]		 =						torch.manual_seed(0    )
       __lowerCamelCase		:							Optional[int]		 =						pipe(generator=a						,				num_inference_steps=100						,				audio_length_in_s=4.0_9_6    )
       __lowerCamelCase		:							Optional[Any]		 =						output.audios

       __lowerCamelCase		:							Any		 =						audio[0, -3:, -3:]

       assert audio.shape == (1, 2, pipe.unet.sample_size)
       __lowerCamelCase		:							Any		 =						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





from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_torch_available, is_torch_tpu_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_torch_available():
       import torch

if is_torch_tpu_available(check_device=False):
       import torch_xla.core.xla_model as xm


lowercase_	   =     logging.get_logger(__name__)


@dataclass
class 	A_ (	__UpperCamelCase   ):




      '''simple docstring'''


      __snake_case 	=       [
          """no_inference""",
          """no_cuda""",
          """no_tpu""",
          """no_speed""",
          """no_memory""",
          """no_env_print""",
          """no_multi_process""",
      ]




      def __init__(    self:	Any						,				**a:	Optional[Any]    ):
       for deprecated_arg in self.deprecated_args:
        if deprecated_arg in kwargs:
         __lowerCamelCase		:							str		 =						deprecated_arg[3:]
         setattr(self						,				a						,				not kwargs.pop(a    )    )
         logger.warning(
             F'{deprecated_arg} is depreciated. Please use --no_{positive_arg} or'
             F' {positive_arg}={kwargs[positive_arg]}'    )

       __lowerCamelCase		:							str		 =						kwargs.pop('torchscript'						,				self.torchscript    )
       __lowerCamelCase		:							int		 =						kwargs.pop('torch_xla_tpu_print_metrics'						,				self.torch_xla_tpu_print_metrics    )
       __lowerCamelCase		:							Dict		 =						kwargs.pop('fp16_opt_level'						,				self.fpaa_opt_level    )
       super().__init__(**a    )

      __snake_case 	=       field(default=__UpperCamelCase ,				metadata={"""help""": """Trace the models using torchscript"""}   )
      __snake_case 	=       field(default=__UpperCamelCase ,				metadata={"""help""": """Print Xla/PyTorch tpu metrics"""}   )
      __snake_case 	=       field(
          default="""O1""" ,				metadata={
              """help""": (
                  """For fp16: Apex AMP optimization level selected in ['O0', 'O1', 'O2', and 'O3']. """
                  """See details at https://nvidia.github.io/apex/amp.html"""
              )
          } ,				)




      @cached_property
      def 					_snake_case     (    self:	Dict    ):
       requires_backends(self						,				['torch']    )
       logger.info('PyTorch: setting up devices'    )
       if not self.cuda:
        __lowerCamelCase		:							Dict		 =						torch.device('cpu'    )
        __lowerCamelCase		:							str		 =						0
       elif is_torch_tpu_available():
        __lowerCamelCase		:							Optional[int]		 =						xm.xla_device()
        __lowerCamelCase		:							Dict		 =						0
       else:
        __lowerCamelCase		:							Any		 =						torch.device('cuda' if torch.cuda.is_available() else 'cpu'    )
        __lowerCamelCase		:							Union[str, Any]		 =						torch.cuda.device_count()
       return device, n_gpu




      @property
      def 					_snake_case     (    self:	Optional[Any]    ):
       return is_torch_tpu_available() and self.tpu




      @property
      def 					_snake_case     (    self:	Union[str, Any]    ):
       requires_backends(self						,				['torch']    )
       # TODO(PVP): currently only single GPU is supported
       return torch.cuda.current_device()




      @property
      def 					_snake_case     (    self:	int    ):
       requires_backends(self						,				['torch']    )
       return self._setup_devices[0]




      @property
      def 					_snake_case     (    self:	Union[str, Any]    ):
       requires_backends(self						,				['torch']    )
       return self._setup_devices[1]




      @property
      def 					_snake_case     (    self:	List[Any]    ):
       return self.n_gpu > 0

