






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 				__A     (			unittest.TestCase ):





				"""simple docstring"""







				def       __lowercase							( self    ):




									"""simple docstring"""

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


				@property
				def       __lowercase							( self    ):




									"""simple docstring"""

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


				@property
				def       __lowercase							( self    ):




									"""simple docstring"""

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


				@property
				def       __lowercase							( self    ):




									"""simple docstring"""

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







				@slow
				def       __lowercase							( self    ):




									"""simple docstring"""

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

									__UpperCamelCase      :					Dict       		=DDPMScheduler()
									__UpperCamelCase      :					int       		=AudioDiffusionPipeline(vqvae=lowerCamelCase__							,  unet=self.dummy_unet							,  mel=lowerCamelCase__							,  scheduler=lowerCamelCase__    )
									__UpperCamelCase      :					str       		=pipe.to(lowerCamelCase__    )
									pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									__UpperCamelCase      :					Union[str, Any]       		=torch.Generator(device=lowerCamelCase__    ).manual_seed(42    )
									__UpperCamelCase      :					List[str]       		=pipe(generator=lowerCamelCase__							,  steps=4    )
									__UpperCamelCase      :					Tuple       		=output.audios[0]
									__UpperCamelCase      :					str       		=output.images[0]

									__UpperCamelCase      :					List[str]       		=torch.Generator(device=lowerCamelCase__    ).manual_seed(42    )
									__UpperCamelCase      :					Optional[Any]       		=pipe(generator=lowerCamelCase__							,  steps=4							,  return_dict=lowerCamelCase__    )
									__UpperCamelCase      :					Dict       		=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]
									)
									__UpperCamelCase      :					Union[str, Any]       		=np.frombuffer(image.tobytes()							,  dtype='uint8'    )[:10]
									__UpperCamelCase      :					Optional[int]       		=np.frombuffer(image_from_tuple.tobytes()							,  dtype='uint8'    )[:10]
									__UpperCamelCase      :					List[Any]       		=np.array([69, 255, 255, 255, 0, 0, 77, 181, 12, 127]    )

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

									__UpperCamelCase      :					List[Any]       		=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]							,  )

									__UpperCamelCase      :					str       		=DDIMScheduler()
									__UpperCamelCase      :					int       		=self.dummy_vqvae_and_unet
									__UpperCamelCase      :					Any       		=AudioDiffusionPipeline(
									    vqvae=self.dummy_vqvae_and_unet[0]							,  unet=dummy_vqvae_and_unet[1]							,  mel=lowerCamelCase__							,  scheduler=lowerCamelCase__    )
									__UpperCamelCase      :					Tuple       		=pipe.to(lowerCamelCase__    )
									pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									np.random.seed(0    )
									__UpperCamelCase      :					Optional[int]       		=np.random.uniform(-1							,  1							,  ((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)    )
									__UpperCamelCase      :					Dict       		=torch.Generator(device=lowerCamelCase__    ).manual_seed(42    )
									__UpperCamelCase      :					Dict       		=pipe(raw_audio=lowerCamelCase__							,  generator=lowerCamelCase__							,  start_step=5							,  steps=10    )
									__UpperCamelCase      :					Union[str, Any]       		=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]
									)
									__UpperCamelCase      :					List[Any]       		=np.frombuffer(image.tobytes()							,  dtype='uint8'    )[:10]
									__UpperCamelCase      :					Union[str, Any]       		=np.array([120, 117, 110, 109, 138, 167, 138, 148, 132, 121]    )

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

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

									np.random.seed(0    )
									__UpperCamelCase      :					Optional[Any]       		=torch.rand((1, 1, 10)    )
									__UpperCamelCase      :					int       		=pipe(generator=lowerCamelCase__							,  encoding=lowerCamelCase__    )
									__UpperCamelCase      :					Optional[Any]       		=output.images[0]
									__UpperCamelCase      :					Dict       		=np.frombuffer(image.tobytes()							,  dtype='uint8'    )[:10]
									__UpperCamelCase      :					List[str]       		=np.array([107, 103, 120, 127, 142, 122, 113, 122, 97, 111]    )

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




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





				"""simple docstring"""







				def       __lowercase							( self    ):




									"""simple docstring"""

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







				def       __lowercase							( self    ):




									"""simple docstring"""

									__UpperCamelCase      :					Any       		=torch_device

									__UpperCamelCase      :					Any       		=DiffusionPipeline.from_pretrained('teticio/audio-diffusion-ddim-256'    )
									__UpperCamelCase      :					str       		=pipe.to(lowerCamelCase__    )
									pipe.set_progress_bar_config(disable=lowerCamelCase__    )

									__UpperCamelCase      :					List[str]       		=torch.Generator(device=lowerCamelCase__    ).manual_seed(42    )
									__UpperCamelCase      :					Any       		=pipe(generator=lowerCamelCase__    )
									__UpperCamelCase      :					List[str]       		=output.audios[0]
									__UpperCamelCase      :					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]
									__UpperCamelCase      :					List[str]       		=np.frombuffer(image.tobytes()							,  dtype='uint8'    )[:10]
									__UpperCamelCase      :					Union[str, Any]       		=np.array([151, 167, 154, 144, 122, 134, 121, 105, 70, 26]    )

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







import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_lowerCamelCase	:		Any							       =   logging.get_logger(__name__)


class 				UpperCamelCase_    ( UpperCAmelCase__     ):

    '''simple docstring'''






    UpperCAmelCase__							  =      '''encoder-decoder'''
    UpperCAmelCase__							  =      True


    def __init__(			self      :	List[str]					, **UpperCAmelCase__      :	Union[str, Any])						->List[Any]:





           '''simple docstring'''

           super().__init__(**UpperCAmelCase__)
           assert (
               "encoder" in kwargs and "decoder" in kwargs
           ), "Config has to be initialized with encoder and decoder config"
           A__       =    kwargs.pop('''encoder''')
           A__       =    encoder_config.pop('''model_type''')
           A__       =    kwargs.pop('''decoder''')
           A__       =    decoder_config.pop('''model_type''')

           from ..auto.configuration_auto import AutoConfig

           A__       =    AutoConfig.for_model(UpperCAmelCase__					, **UpperCAmelCase__)
           A__       =    AutoConfig.for_model(UpperCAmelCase__					, **UpperCAmelCase__)
           A__       =    True


    @classmethod
    def        SCREAMING_SNAKE_CASE							(			cls      :	Union[str, Any]					, UpperCAmelCase__      :	PretrainedConfig					, UpperCAmelCase__      :	PretrainedConfig					, **UpperCAmelCase__      :	Union[str, Any])						->PretrainedConfig:





           '''simple docstring'''

           logger.info('''Set `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config''')
           A__       =    True
           A__       =    True

           return cls(encoder=encoder_config.to_dict()					, decoder=decoder_config.to_dict()					, **UpperCAmelCase__)





    def        SCREAMING_SNAKE_CASE							(			self      :	str)						->Optional[Any]:





           '''simple docstring'''

           A__       =    copy.deepcopy(self.__dict__)
           A__       =    self.encoder.to_dict()
           A__       =    self.decoder.to_dict()
           A__       =    self.__class__.model_type
           return output


