

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    snake_case_       (  self    ):

      '''simple docstring'''
      # clean up the VRAM after each test
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()





  @property
  def    snake_case_       (  self    ):

      '''simple docstring'''
      torch.manual_seed(0    )
      snake_case							:		Optional[int]        =		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    snake_case_       (  self    ):

      '''simple docstring'''
      torch.manual_seed(0    )
      snake_case							:		Optional[Any]        =		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    snake_case_       (  self    ):

      '''simple docstring'''
      torch.manual_seed(0    )
      snake_case							:		str        =		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""")   ,)
      snake_case							:		List[str]        =		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    snake_case_       (  self    ):

      '''simple docstring'''
      snake_case							:		Tuple        =		"""cpu"""  # ensure determinism for the device-dependent torch.Generator
      snake_case							:		Optional[Any]        =		Mel(
          x_res=self.dummy_unet.config.sample_size[1]   ,y_res=self.dummy_unet.config.sample_size[0]   ,)

      snake_case							:		List[str]        =		DDPMScheduler()
      snake_case							:		Dict        =		AudioDiffusionPipeline(vqvae=_UpperCAmelCase   ,unet=self.dummy_unet   ,mel=_UpperCAmelCase   ,scheduler=_UpperCAmelCase    )
      snake_case							:		int        =		pipe.to(_UpperCAmelCase    )
      pipe.set_progress_bar_config(disable=_UpperCAmelCase    )

      snake_case							:		int        =		torch.Generator(device=_UpperCAmelCase    ).manual_seed(42    )
      snake_case							:		int        =		pipe(generator=_UpperCAmelCase   ,steps=4    )
      snake_case							:		Any        =		output.audios[0]
      snake_case							:		List[str]        =		output.images[0]

      snake_case							:		Dict        =		torch.Generator(device=_UpperCAmelCase    ).manual_seed(42    )
      snake_case							:		Optional[Any]        =		pipe(generator=_UpperCAmelCase   ,steps=4   ,return_dict=_UpperCAmelCase    )
      snake_case							:		Optional[Any]        =		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]
      )
      snake_case							:		int        =		np.frombuffer(image.tobytes()   ,dtype="""uint8"""    )[:10]
      snake_case							:		List[Any]        =		np.frombuffer(image_from_tuple.tobytes()   ,dtype="""uint8"""    )[:10]
      snake_case							:		Tuple        =		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

      snake_case							:		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]   ,)

      snake_case							:		Tuple        =		DDIMScheduler()
      snake_case							:		Dict        =		self.dummy_vqvae_and_unet
      snake_case							:		List[Any]        =		AudioDiffusionPipeline(
          vqvae=self.dummy_vqvae_and_unet[0]   ,unet=dummy_vqvae_and_unet[1]   ,mel=_UpperCAmelCase   ,scheduler=_UpperCAmelCase    )
      snake_case							:		Optional[Any]        =		pipe.to(_UpperCAmelCase    )
      pipe.set_progress_bar_config(disable=_UpperCAmelCase    )

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

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

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

      np.random.seed(0    )
      snake_case							:		List[str]        =		torch.rand((1, 1, 10)    )
      snake_case							:		Dict        =		pipe(generator=_UpperCAmelCase   ,encoding=_UpperCAmelCase    )
      snake_case							:		Optional[Any]        =		output.images[0]
      snake_case							:		str        =		np.frombuffer(image.tobytes()   ,dtype="""uint8"""    )[:10]
      snake_case							:		int        =		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    snake_case_       (  self    ):

      '''simple docstring'''
      # clean up the VRAM after each test
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()





  def    snake_case_       (  self    ):

      '''simple docstring'''
      snake_case							:		str        =		torch_device

      snake_case							:		List[str]        =		DiffusionPipeline.from_pretrained("""teticio/audio-diffusion-ddim-256"""    )
      snake_case							:		Dict        =		pipe.to(_UpperCAmelCase    )
      pipe.set_progress_bar_config(disable=_UpperCAmelCase    )

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

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











from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


snake_case__   :						Optional[int]													=   {
    """configuration_pegasus_x""": ["""PEGASUS_X_PRETRAINED_CONFIG_ARCHIVE_MAP""", """PegasusXConfig"""],
}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 snake_case__   :						Dict													=   [
     """PEGASUS_X_PRETRAINED_MODEL_ARCHIVE_LIST""",
     """PegasusXForConditionalGeneration""",
     """PegasusXModel""",
     """PegasusXPreTrainedModel""",
 ]


if TYPE_CHECKING:
 from .configuration_pegasus_x import PEGASUS_X_PRETRAINED_CONFIG_ARCHIVE_MAP, PegasusXConfig

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_pegasus_x import (
      PEGASUS_X_PRETRAINED_MODEL_ARCHIVE_LIST,
      PegasusXForConditionalGeneration,
      PegasusXModel,
      PegasusXPreTrainedModel,
  )


else:
 import sys

 snake_case__   :						Optional[Any]													=   _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)





