

"""simple docstring"""






import timeit

import numpy as np

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

def 		lowerCamelCase__		(					_lowerCamelCase   ):

    '''simple docstring'''



    def wrapper(*_lowerCamelCase ,   **_lowerCamelCase   ):
        _lowerCAmelCase       :   List[Any]  =    timeit.default_timer()
        _lowerCAmelCase       :   Tuple  =    func(*__UpperCAmelCase ,   **__UpperCAmelCase   )
        _lowerCAmelCase       :   Dict  =    timeit.default_timer() - starttime
        return delta

    _lowerCAmelCase       :   List[str]  =    func.__name__

    return wrapper

def 		lowerCamelCase__		(					_lowerCamelCase ,   _lowerCamelCase=100 ,   _lowerCamelCase=None   ):

    '''simple docstring'''



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

        dummy_data.append((i, example)   )

    return dummy_data

def 		lowerCamelCase__		(					_lowerCamelCase ,   _lowerCamelCase ,   _lowerCamelCase=100 ,   _lowerCamelCase=None   ):

    '''simple docstring'''



    _lowerCAmelCase       :   Optional[int]  =    generate_examples(__UpperCAmelCase ,   num_examples=__UpperCAmelCase ,   seq_shapes=__UpperCAmelCase   )

    with ArrowWriter(features=__UpperCAmelCase ,   path=__UpperCAmelCase   ) as writer:
        for key, record in dummy_data:
            _lowerCAmelCase       :   str  =    features.encode_example(__UpperCAmelCase   )
            writer.write(__UpperCAmelCase   )

        _lowerCAmelCase,					_lowerCAmelCase       :   int  =    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}."""   )

    _lowerCAmelCase       :   Dict  =    datasets.Dataset.from_file(filename=__UpperCAmelCase ,   info=datasets.DatasetInfo(features=__UpperCAmelCase   )   )

    return dataset




"""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 	__UpperCamelCase    (    a__							,							unittest.TestCase						):
  _UpperCAmelCase          =		DanceDiffusionPipeline
  _UpperCAmelCase          =		UNCONDITIONAL_AUDIO_GENERATION_PARAMS
  _UpperCAmelCase          =		PipelineTesterMixin.required_optional_params - {
      "callback",
      "latents",
      "callback_steps",
      "output_type",
      "num_images_per_prompt",
  }
  _UpperCAmelCase          =		UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
  _UpperCAmelCase          =		False
  _UpperCAmelCase          =		False

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      torch.manual_seed(0      )
      _lowerCAmelCase       :   List[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       :   int  =    IPNDMScheduler()

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

  def __lowerCamelCase		(		self    ,_A    ,_A=0      ):
      '''simple docstring'''

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

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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

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

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

      assert audio.shape == (1, 2, components["unet"].sample_size)
      _lowerCAmelCase       :   Optional[Any]  =    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 __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return super().test_save_load_local()

  @skip_mps
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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

  @skip_mps
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return super().test_save_load_optional_components()

  @skip_mps
  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      return super().test_attention_slicing_forward_pass()







  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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







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

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

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

  def __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   List[Any]  =    torch_device

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

      _lowerCAmelCase       :   Optional[int]  =    torch.manual_seed(0      )
      _lowerCAmelCase       :   str  =    pipe(generator=_A    ,num_inference_steps=100    ,audio_length_in_s=4.0_9_6      )
      _lowerCAmelCase       :   str  =    output.audios

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

      assert audio.shape == (1, 2, pipe.unet.sample_size)
      _lowerCAmelCase       :   Union[str, Any]  =    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 __lowerCamelCase		(		self      ):
      '''simple docstring'''

      _lowerCAmelCase       :   str  =    torch_device

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

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

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

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





