






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





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      torch.manual_seed(0      )
      a__	:						Tuple      		=  UNetaDModel(
          block_out_channels=(32, 32, 64)					,			extra_in_channels=16					,			sample_size=512					,			sample_rate=16_000					,			in_channels=2					,			out_channels=2					,			flip_sin_to_cos=__lowerCAmelCase					,			use_timestep_embedding=__lowerCAmelCase					,			time_embedding_type="fourier"					,			mid_block_type="UNetMidBlock1D"					,			down_block_types=("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")					,			up_block_types=("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")					,			)
      a__	:						Tuple      		=  IPNDMScheduler()

      a__	:						List[Any]      		=  {
          "unet": unet,
          "scheduler": scheduler,
      }
      return components





   def 	_snake_case ( self					,			snake_case					,			snake_case=0      )				->  Tuple:
      """simple docstring"""





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





   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





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

      a__	:						Dict      		=  self.get_dummy_inputs(__lowerCAmelCase      )
      a__	:						str      		=  pipe(**__lowerCAmelCase      )
      a__	:						Dict      		=  output.audios

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

      assert audio.shape == (1, 2, components["unet"].sample_size)
      a__	:						Tuple      		=  np.array([-0.7_265, 1.0_000, -0.8_388, 0.1_175, 0.9_498, -1.0_000]      )
      assert np.abs(audio_slice.flatten() - expected_slice      ).max() < 1E-2





   @skip_mps
   def 	_snake_case ( self      )				->  List[str]:
      """simple docstring"""





      return super().test_save_load_local()





   @skip_mps
   def 	_snake_case ( self      )				->  List[str]:
      """simple docstring"""





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





   @skip_mps
   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





      return super().test_save_load_optional_components()





   @skip_mps
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      return super().test_attention_slicing_forward_pass()





   def 	_snake_case ( self      )				->  Union[str, Any]:
      """simple docstring"""





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







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





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





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





   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      a__	:						Tuple      		=  torch_device

      a__	:						Union[str, Any]      		=  DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"      )
      a__	:						Dict      		=  pipe.to(__lowerCAmelCase      )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase      )

      a__	:						Union[str, Any]      		=  torch.manual_seed(0      )
      a__	:						Any      		=  pipe(generator=__lowerCAmelCase					,			num_inference_steps=100					,			audio_length_in_s=4.096      )
      a__	:						Union[str, Any]      		=  output.audios

      a__	:						str      		=  audio[0, -3:, -3:]

      assert audio.shape == (1, 2, pipe.unet.sample_size)
      a__	:						Optional[int]      		=  np.array([-0.0_192, -0.0_231, -0.0_318, -0.0_059, 0.0_002, -0.0_020]      )

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





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      a__	:						Dict      		=  torch_device

      a__	:						Union[str, Any]      		=  DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"					,			torch_dtype=torch.floataa      )
      a__	:						int      		=  pipe.to(__lowerCAmelCase      )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase      )

      a__	:						int      		=  torch.manual_seed(0      )
      a__	:						Optional[int]      		=  pipe(generator=__lowerCAmelCase					,			num_inference_steps=100					,			audio_length_in_s=4.096      )
      a__	:						List[str]      		=  output.audios

      a__	:						Optional[int]      		=  audio[0, -3:, -3:]

      assert audio.shape == (1, 2, pipe.unet.sample_size)
      a__	:						Union[str, Any]      		=  np.array([-0.0_367, -0.0_488, -0.0_771, -0.0_525, -0.0_444, -0.0_341]      )

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








import tempfile
import unittest

from make_student import create_student_by_copying_alternating_layers

from transformers import AutoConfig
from transformers.file_utils import cached_property
from transformers.testing_utils import require_torch


SCREAMING_SNAKE_CASE__       :			List[Any]			      =						"""sshleifer/bart-tiny-random"""
SCREAMING_SNAKE_CASE__       :			Tuple			      =						"""patrickvonplaten/t5-tiny-random"""







@require_torch
class   __lowerCAmelCase			(					unittest.TestCase  ):





   @cached_property
   def 	_snake_case ( self      )				->  List[Any]:
      """simple docstring"""





      return AutoConfig.from_pretrained(snake_case      )





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      a__		,     *a__	:						int      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=1      )
      self.assertEqual(student.config.num_hidden_layers					,			1      )





   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__		,     *a__	:						int      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=snake_case      )





   def 	_snake_case ( self      )				->  List[str]:
      """simple docstring"""





      a__		,     *a__	:						Any      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=snake_case      )
      self.assertEqual(student.config.encoder_layers					,			1      )
      self.assertEqual(student.config.decoder_layers					,			self.teacher_config.encoder_layers      )





   def 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      a__		,     *a__	:						List[str]      		=  create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=1					,			d=1      )
      self.assertEqual(student.config.encoder_layers					,			1      )
      self.assertEqual(student.config.decoder_layers					,			1      )





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      with self.assertRaises(snake_case      ):
         create_student_by_copying_alternating_layers(snake_case					,			tempfile.mkdtemp()					,			e=snake_case					,			d=snake_case      )

