
"""simple docstring"""




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   _lowercase						(       a__	,   unittest.TestCase				):


     '''simple docstring'''





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



     def  lowerCAmelCase__       (      self  )->					Optional[Any]:
          torch.manual_seed(0  )
          UpperCAmelCase__							: List[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=_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__							: int 							=						IPNDMScheduler()

          UpperCAmelCase__							: Union[str, Any] 							=						{
              'unet': unet,
              'scheduler': scheduler,
          }
          return components



     def  lowerCAmelCase__       (      self ,			__UpperCamelCase ,			__UpperCamelCase=0  )->					Optional[Any]:

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



     def  lowerCAmelCase__       (      self  )->					Any:
          UpperCAmelCase__							: List[str] 							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
          UpperCAmelCase__							: int 							=						self.get_dummy_components()
          UpperCAmelCase__							: Optional[Any] 							=						DanceDiffusionPipeline(**_A  )
          UpperCAmelCase__							: int 							=						pipe.to(_A  )
          pipe.set_progress_bar_config(disable=_A  )

          UpperCAmelCase__							: Union[str, Any] 							=						self.get_dummy_inputs(_A  )
          UpperCAmelCase__							: List[str] 							=						pipe(**_A  )
          UpperCAmelCase__							: List[Any] 							=						output.audios

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

          assert audio.shape == (1, 2, components["unet"].sample_size)
          UpperCAmelCase__							: Optional[Any] 							=						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  lowerCAmelCase__       (      self  )->					str:
          return super().test_save_load_local()



     @skip_mps
     def  lowerCAmelCase__       (      self  )->					str:
          return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3  )



     @skip_mps
     def  lowerCAmelCase__       (      self  )->					Any:
          return super().test_save_load_optional_components()



     @skip_mps
     def  lowerCAmelCase__       (      self  )->					List[Any]:
          return super().test_attention_slicing_forward_pass()



     def  lowerCAmelCase__       (      self  )->					Dict:
          super().test_inference_batch_single_identical(expected_max_diff=3E-3  )







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


     '''simple docstring'''





     def  lowerCAmelCase__       (      self  )->					List[Any]:
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()



     def  lowerCAmelCase__       (      self  )->					Optional[Any]:
          UpperCAmelCase__							: List[Any] 							=						torch_device

          UpperCAmelCase__							: int 							=						DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"  )
          UpperCAmelCase__							: int 							=						pipe.to(_A  )
          pipe.set_progress_bar_config(disable=_A  )

          UpperCAmelCase__							: Optional[int] 							=						torch.manual_seed(0  )
          UpperCAmelCase__							: str 							=						pipe(generator=_A ,			num_inference_steps=1_00 ,			audio_length_in_s=4.096  )
          UpperCAmelCase__							: str 							=						output.audios

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

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

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

          UpperCAmelCase__							: Union[str, Any] 							=						torch.manual_seed(0  )
          UpperCAmelCase__							: Optional[int] 							=						pipe(generator=_A ,			num_inference_steps=1_00 ,			audio_length_in_s=4.096  )
          UpperCAmelCase__							: Union[str, Any] 							=						output.audios

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

          assert audio.shape == (1, 2, pipe.unet.sample_size)
          UpperCAmelCase__							: List[str] 							=						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 timeit

import numpy as np

import datasets
from datasets.arrow_writer import ArrowWriter
from datasets.features.features import _ArrayXD



def 							a__ (				lowerCAmelCase	:       List[str]		):






     '''simple docstring'''
     def wrapper(*lowerCAmelCase	:       Any       ,    **lowerCAmelCase	:       Tuple		):
          UpperCAmelCase__							: Optional[int] 							=						timeit.default_timer()
          UpperCAmelCase__							: int 							=						func(*lowerCAmelCase       ,    **lowerCAmelCase		)
          UpperCAmelCase__							: List[Any] 							=						timeit.default_timer() - starttime
          return delta

     UpperCAmelCase__							: int 							=						func.__name__

     return wrapper



def 							a__ (				lowerCAmelCase	:       dict       ,    lowerCAmelCase	:       Optional[int]=100       ,    lowerCAmelCase	:       List[str]=None		):






     '''simple docstring'''
     UpperCAmelCase__							: str 							=						[]
     UpperCAmelCase__							: Optional[Any] 							=						seq_shapes or {}
     for i in range(lowerCAmelCase		):
          UpperCAmelCase__							: int 							=						{}
          for col_id, (k, v) in enumerate(features.items()		):
               if isinstance(lowerCAmelCase       ,    _ArrayXD		):
                    UpperCAmelCase__							: List[str] 							=						np.random.rand(*v.shape		).astype(v.dtype		)
               elif isinstance(lowerCAmelCase       ,    datasets.Value		):
                    if v.dtype == "string":
                         UpperCAmelCase__							: Dict 							=						"The small grey turtle was surprisingly fast when challenged."
                    else:
                         UpperCAmelCase__							: str 							=						np.random.randint(10       ,    size=1		).astype(v.dtype		).item()
               elif isinstance(lowerCAmelCase       ,    datasets.Sequence		):
                    while isinstance(lowerCAmelCase       ,    datasets.Sequence		):
                         UpperCAmelCase__							: List[str] 							=						v.feature
                    UpperCAmelCase__							: Optional[int] 							=						seq_shapes[k]
                    UpperCAmelCase__							: Optional[int] 							=						np.random.rand(*lowerCAmelCase		).astype(v.dtype		)
               UpperCAmelCase__							: Union[str, Any] 							=						data

          dummy_data.append((i, example)		)

     return dummy_data



def 							a__ (				lowerCAmelCase	:       List[str]       ,    lowerCAmelCase	:       Tuple       ,    lowerCAmelCase	:       List[str]=100       ,    lowerCAmelCase	:       Optional[int]=None		):






     '''simple docstring'''
     UpperCAmelCase__							: int 							=						generate_examples(lowerCAmelCase       ,    num_examples=lowerCAmelCase       ,    seq_shapes=lowerCAmelCase		)

     with ArrowWriter(features=lowerCAmelCase       ,    path=lowerCAmelCase		) as writer:
          for key, record in dummy_data:
               UpperCAmelCase__							: List[Any] 							=						features.encode_example(lowerCAmelCase		)
               writer.write(lowerCAmelCase		)

          UpperCAmelCase__							,	UpperCAmelCase__							: Optional[Any] 							=						writer.finalize()

     if not num_final_examples == num_examples:
          raise ValueError(
              F"Error writing the dataset, wrote {num_final_examples} examples but should have written {num_examples}."		)

     UpperCAmelCase__							: Optional[int] 							=						datasets.Dataset.from_file(filename=lowerCAmelCase       ,    info=datasets.DatasetInfo(features=lowerCAmelCase		)		)

     return dataset



