





"""simple docstring"""
import gc
import unittest

import numpy as np
import torch

from diffusers import (
    AudioDiffusionPipeline,
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    DiffusionPipeline,
    Mel,
    UNetaDConditionModel,
    UNetaDModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()






class  _SCREAMING_SNAKE_CASE(       unittest.TestCase  ):




       def    _UpperCamelCase					(						self							)      ->				List[Any]:





             """simple docstring"""






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




       @property
       def    _UpperCamelCase					(						self							)      ->				str:





             """simple docstring"""






             torch.manual_seed(0							)
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  UNetaDModel(
                 sample_size=(32, 64)		,in_channels=1		,out_channels=1		,layers_per_block=2		,block_out_channels=(1_28, 1_28)		,down_block_types=('''AttnDownBlock2D''', '''DownBlock2D''')		,up_block_types=('''UpBlock2D''', '''AttnUpBlock2D''')		,)
             return model




       @property
       def    _UpperCamelCase					(						self							)      ->				Union[str, Any]:





             """simple docstring"""






             torch.manual_seed(0							)
             __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  UNetaDConditionModel(
                 sample_size=(64, 32)		,in_channels=1		,out_channels=1		,layers_per_block=2		,block_out_channels=(1_28, 1_28)		,down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D''')		,up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D''')		,cross_attention_dim=10		,)
             return model




       @property
       def    _UpperCamelCase					(						self							)      ->				int:





             """simple docstring"""






             torch.manual_seed(0							)
             __SCREAMING_SNAKE_CASE       :List[str]		    =  AutoencoderKL(
                 sample_size=(1_28, 64)		,in_channels=1		,out_channels=1		,latent_channels=1		,layers_per_block=2		,block_out_channels=(1_28, 1_28)		,down_block_types=('''DownEncoderBlock2D''', '''DownEncoderBlock2D''')		,up_block_types=('''UpDecoderBlock2D''', '''UpDecoderBlock2D''')		,)
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  UNetaDModel(
                 sample_size=(64, 32)		,in_channels=1		,out_channels=1		,layers_per_block=2		,block_out_channels=(1_28, 1_28)		,down_block_types=('''AttnDownBlock2D''', '''DownBlock2D''')		,up_block_types=('''UpBlock2D''', '''AttnUpBlock2D''')		,)
             return vqvae, unet




       @slow
       def    _UpperCamelCase					(						self							)      ->				str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE       :Optional[int]		    =  '''cpu'''  # ensure determinism for the device-dependent torch.Generator
             __SCREAMING_SNAKE_CASE       :List[str]		    =  Mel(
                 x_res=self.dummy_unet.config.sample_size[1]		,y_res=self.dummy_unet.config.sample_size[0]		,)

             __SCREAMING_SNAKE_CASE       :Tuple		    =  DDPMScheduler()
             __SCREAMING_SNAKE_CASE       :Tuple		    =  AudioDiffusionPipeline(vqvae=SCREAMING_SNAKE_CASE__		,unet=self.dummy_unet		,mel=SCREAMING_SNAKE_CASE__		,scheduler=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE       :List[Any]		    =  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE       :Any		    =  torch.Generator(device=SCREAMING_SNAKE_CASE__							).manual_seed(42							)
             __SCREAMING_SNAKE_CASE       :Any		    =  pipe(generator=SCREAMING_SNAKE_CASE__		,steps=4							)
             __SCREAMING_SNAKE_CASE       :Tuple		    =  output.audios[0]
             __SCREAMING_SNAKE_CASE       :int		    =  output.images[0]

             __SCREAMING_SNAKE_CASE       :Dict		    =  torch.Generator(device=SCREAMING_SNAKE_CASE__							).manual_seed(42							)
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  pipe(generator=SCREAMING_SNAKE_CASE__		,steps=4		,return_dict=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE       :Optional[int]		    =  output[0][0]

             assert audio.shape == (1, (self.dummy_unet.config.sample_size[1] - 1) * mel.hop_length)
             assert (
                 image.height == self.dummy_unet.config.sample_size[0]
                 and image.width == self.dummy_unet.config.sample_size[1]
             )
             __SCREAMING_SNAKE_CASE       :Any		    =  np.frombuffer(image.tobytes()		,dtype='''uint8'''							)[:10]
             __SCREAMING_SNAKE_CASE       :Any		    =  np.frombuffer(image_from_tuple.tobytes()		,dtype='''uint8'''							)[:10]
             __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  np.array([69, 2_55, 2_55, 2_55, 0, 0, 77, 1_81, 12, 1_27]							)

             assert np.abs(image_slice.flatten() - expected_slice							).max() == 0
             assert np.abs(image_from_tuple_slice.flatten() - expected_slice							).max() == 0

             __SCREAMING_SNAKE_CASE       :Dict		    =  Mel(
                 x_res=self.dummy_vqvae_and_unet[0].config.sample_size[1]		,y_res=self.dummy_vqvae_and_unet[0].config.sample_size[0]		,)

             __SCREAMING_SNAKE_CASE       :Optional[int]		    =  DDIMScheduler()
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  self.dummy_vqvae_and_unet
             __SCREAMING_SNAKE_CASE       :List[Any]		    =  AudioDiffusionPipeline(
                 vqvae=self.dummy_vqvae_and_unet[0]		,unet=dummy_vqvae_and_unet[1]		,mel=SCREAMING_SNAKE_CASE__		,scheduler=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE       :Any		    =  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             np.random.seed(0							)
             __SCREAMING_SNAKE_CASE       :Any		    =  np.random.uniform(-1		,1		,((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)							)
             __SCREAMING_SNAKE_CASE       :Dict		    =  torch.Generator(device=SCREAMING_SNAKE_CASE__							).manual_seed(42							)
             __SCREAMING_SNAKE_CASE       :List[str]		    =  pipe(raw_audio=SCREAMING_SNAKE_CASE__		,generator=SCREAMING_SNAKE_CASE__		,start_step=5		,steps=10							)
             __SCREAMING_SNAKE_CASE       :Dict		    =  output.images[0]

             assert (
                 image.height == self.dummy_vqvae_and_unet[0].config.sample_size[0]
                 and image.width == self.dummy_vqvae_and_unet[0].config.sample_size[1]
             )
             __SCREAMING_SNAKE_CASE       :int		    =  np.frombuffer(image.tobytes()		,dtype='''uint8'''							)[:10]
             __SCREAMING_SNAKE_CASE       :List[Any]		    =  np.array([1_20, 1_17, 1_10, 1_09, 1_38, 1_67, 1_38, 1_48, 1_32, 1_21]							)

             assert np.abs(image_slice.flatten() - expected_slice							).max() == 0

             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  self.dummy_unet_condition
             __SCREAMING_SNAKE_CASE       :List[str]		    =  AudioDiffusionPipeline(
                 vqvae=self.dummy_vqvae_and_unet[0]		,unet=SCREAMING_SNAKE_CASE__		,mel=SCREAMING_SNAKE_CASE__		,scheduler=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE       :Dict		    =  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             np.random.seed(0							)
             __SCREAMING_SNAKE_CASE       :Optional[Any]		    =  torch.rand((1, 1, 10)							)
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  pipe(generator=SCREAMING_SNAKE_CASE__		,encoding=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE       :List[str]		    =  output.images[0]
             __SCREAMING_SNAKE_CASE       :List[str]		    =  np.frombuffer(image.tobytes()		,dtype='''uint8'''							)[:10]
             __SCREAMING_SNAKE_CASE       :Any		    =  np.array([1_07, 1_03, 1_20, 1_27, 1_42, 1_22, 1_13, 1_22, 97, 1_11]							)

             assert np.abs(image_slice.flatten() - expected_slice							).max() == 0






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




       def    _UpperCamelCase					(						self							)      ->				str:





             """simple docstring"""






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




       def    _UpperCamelCase					(						self							)      ->				str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE       :str		    =  torch_device

             __SCREAMING_SNAKE_CASE       :str		    =  DiffusionPipeline.from_pretrained('''teticio/audio-diffusion-ddim-256'''							)
             __SCREAMING_SNAKE_CASE       :List[str]		    =  pipe.to(SCREAMING_SNAKE_CASE__							)
             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE       :int		    =  torch.Generator(device=SCREAMING_SNAKE_CASE__							).manual_seed(42							)
             __SCREAMING_SNAKE_CASE       :str		    =  pipe(generator=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE       :Dict		    =  output.audios[0]
             __SCREAMING_SNAKE_CASE       :List[str]		    =  output.images[0]

             assert audio.shape == (1, (pipe.unet.config.sample_size[1] - 1) * pipe.mel.hop_length)
             assert image.height == pipe.unet.config.sample_size[0] and image.width == pipe.unet.config.sample_size[1]
             __SCREAMING_SNAKE_CASE       :int		    =  np.frombuffer(image.tobytes()		,dtype='''uint8'''							)[:10]
             __SCREAMING_SNAKE_CASE       :Any		    =  np.array([1_51, 1_67, 1_54, 1_44, 1_22, 1_34, 1_21, 1_05, 70, 26]							)

             assert np.abs(image_slice.flatten() - expected_slice							).max() == 0






"""simple docstring"""
import unittest

from datasets import load_dataset

from transformers.pipelines import pipeline
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_torch, slow






@is_pipeline_test
@require_torch
class  _SCREAMING_SNAKE_CASE(       unittest.TestCase  ):




       @require_torch
       def    _UpperCamelCase					(						self							)      ->				str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE       :Dict		    =  pipeline(
                 task='''zero-shot-audio-classification'''		,model='''hf-internal-testing/tiny-clap-htsat-unfused'''							)
             __SCREAMING_SNAKE_CASE       :Any		    =  load_dataset('''ashraq/esc50'''							)
             __SCREAMING_SNAKE_CASE       :int		    =  dataset['''train''']['''audio'''][-1]['''array''']
             __SCREAMING_SNAKE_CASE       :Dict		    =  audio_classifier(SCREAMING_SNAKE_CASE__		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']							)
             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[{'''score''': 0.5_0_1, '''label''': '''Sound of a dog'''}, {'''score''': 0.4_9_9, '''label''': '''Sound of vaccum cleaner'''}]		,)




       @unittest.skip('''No models are available in TF'''							)
       def    _UpperCamelCase					(						self							)      ->				Optional[int]:





             """simple docstring"""






             pass




       @slow
       @require_torch
       def    _UpperCamelCase					(						self							)      ->				Tuple:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE       :Optional[int]		    =  pipeline(
                 task='''zero-shot-audio-classification'''		,model='''laion/clap-htsat-unfused'''		,)
             # This is an audio of a dog
             __SCREAMING_SNAKE_CASE       :List[Any]		    =  load_dataset('''ashraq/esc50'''							)
             __SCREAMING_SNAKE_CASE       :Tuple		    =  dataset['''train''']['''audio'''][-1]['''array''']
             __SCREAMING_SNAKE_CASE       :str		    =  audio_classifier(SCREAMING_SNAKE_CASE__		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']							)

             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[
                     {'''score''': 0.9_9_9, '''label''': '''Sound of a dog'''},
                     {'''score''': 0.0_0_1, '''label''': '''Sound of vaccum cleaner'''},
                 ]		,)

             __SCREAMING_SNAKE_CASE       :Dict		    =  audio_classifier([audio] * 5		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']							)
             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[
                     [
                         {'''score''': 0.9_9_9, '''label''': '''Sound of a dog'''},
                         {'''score''': 0.0_0_1, '''label''': '''Sound of vaccum cleaner'''},
                     ],
                 ]
                 * 5		,)
             __SCREAMING_SNAKE_CASE       :Union[str, Any]		    =  audio_classifier(
                 [audio] * 5		,candidate_labels=['''Sound of a dog''', '''Sound of vaccum cleaner''']		,batch_size=5							)
             self.assertEqual(
                 nested_simplify(SCREAMING_SNAKE_CASE__							)		,[
                     [
                         {'''score''': 0.9_9_9, '''label''': '''Sound of a dog'''},
                         {'''score''': 0.0_0_1, '''label''': '''Sound of vaccum cleaner'''},
                     ],
                 ]
                 * 5		,)




       @unittest.skip('''No models are available in TF'''							)
       def    _UpperCamelCase					(						self							)      ->				List[str]:





             """simple docstring"""






             pass