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



   def 							_SCREAMING_SNAKE_CASE  (self						:   Any					)						->      Dict:


          '''simple docstring'''


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



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


          '''simple docstring'''


          torch.manual_seed(0					)
          snake_case			:  List[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 							_SCREAMING_SNAKE_CASE  (self						:   Optional[int]					)						->      Tuple:


          '''simple docstring'''


          torch.manual_seed(0					)
          snake_case			:  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 							_SCREAMING_SNAKE_CASE  (self						:   Union[str, Any]					)						->      int:


          '''simple docstring'''


          torch.manual_seed(0					)
          snake_case			:  List[Any]     							= 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")       ,		)
          snake_case			:  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 							_SCREAMING_SNAKE_CASE  (self						:   List[str]					)						->      int:


          '''simple docstring'''


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

          snake_case			:  Tuple     							= DDPMScheduler()
          snake_case			:  int     							= AudioDiffusionPipeline(vqvae=snake_case__       ,		unet=self.dummy_unet       ,		mel=snake_case__       ,		scheduler=snake_case__					)
          snake_case			:  str     							= pipe.to(snake_case__					)
          pipe.set_progress_bar_config(disable=snake_case__					)

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

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

          snake_case			:  Optional[int]     							= 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			:  Optional[Any]     							= DDIMScheduler()
          snake_case			:  List[str]     							= self.dummy_vqvae_and_unet
          snake_case			:  List[Any]     							= AudioDiffusionPipeline(
              vqvae=self.dummy_vqvae_and_unet[0]       ,		unet=dummy_vqvae_and_unet[1]       ,		mel=snake_case__       ,		scheduler=snake_case__					)
          snake_case			:  List[str]     							= pipe.to(snake_case__					)
          pipe.set_progress_bar_config(disable=snake_case__					)

          np.random.seed(0					)
          snake_case			:  int     							= np.random.uniform(-1       ,		1       ,		((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)					)
          snake_case			:  Union[str, Any]     							= torch.Generator(device=snake_case__					).manual_seed(42					)
          snake_case			:  Optional[Any]     							= pipe(raw_audio=snake_case__       ,		generator=snake_case__       ,		start_step=5       ,		steps=10					)
          snake_case			:  Optional[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]
          )
          snake_case			:  Optional[int]     							= np.frombuffer(image.tobytes()       ,		dtype="uint8"					)[:10]
          snake_case			:  Dict     							= 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

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

          np.random.seed(0					)
          snake_case			:  Tuple     							= torch.rand((1, 1, 10)					)
          snake_case			:  Union[str, Any]     							= pipe(generator=snake_case__       ,		encoding=snake_case__					)
          snake_case			:  Union[str, Any]     							= output.images[0]
          snake_case			:  List[str]     							= np.frombuffer(image.tobytes()       ,		dtype="uint8"					)[:10]
          snake_case			:  Optional[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     UpperCAmelCase						(		unittest.TestCase					):



   def 							_SCREAMING_SNAKE_CASE  (self						:   Union[str, Any]					)						->      List[Any]:


          '''simple docstring'''


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



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


          '''simple docstring'''


          snake_case			:  str     							= torch_device

          snake_case			:  int     							= DiffusionPipeline.from_pretrained("teticio/audio-diffusion-ddim-256"					)
          snake_case			:  List[Any]     							= pipe.to(snake_case__					)
          pipe.set_progress_bar_config(disable=snake_case__					)

          snake_case			:  int     							= torch.Generator(device=snake_case__					).manual_seed(42					)
          snake_case			:  Dict     							= pipe(generator=snake_case__					)
          snake_case			:  int     							= output.audios[0]
          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]
          snake_case			:  List[str]     							= np.frombuffer(image.tobytes()       ,		dtype="uint8"					)[:10]
          snake_case			:  List[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







from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


__lowerCamelCase				=			{
    """configuration_chinese_clip""": [
        """CHINESE_CLIP_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """ChineseCLIPConfig""",
        """ChineseCLIPOnnxConfig""",
        """ChineseCLIPTextConfig""",
        """ChineseCLIPVisionConfig""",
    ],
    """processing_chinese_clip""": ["""ChineseCLIPProcessor"""],
}

try:
   if not is_vision_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   __lowerCamelCase				=			["""ChineseCLIPFeatureExtractor"""]
   __lowerCamelCase				=			["""ChineseCLIPImageProcessor"""]

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   __lowerCamelCase				=			[
       """CHINESE_CLIP_PRETRAINED_MODEL_ARCHIVE_LIST""",
       """ChineseCLIPModel""",
       """ChineseCLIPPreTrainedModel""",
       """ChineseCLIPTextModel""",
       """ChineseCLIPVisionModel""",
   ]

if TYPE_CHECKING:
   from .configuration_chinese_clip import (
       CHINESE_CLIP_PRETRAINED_CONFIG_ARCHIVE_MAP,
       ChineseCLIPConfig,
       ChineseCLIPOnnxConfig,
       ChineseCLIPTextConfig,
       ChineseCLIPVisionConfig,
   )
   from .processing_chinese_clip import ChineseCLIPProcessor

   try:
      if not is_vision_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .feature_extraction_chinese_clip import ChineseCLIPFeatureExtractor, ChineseCLIPImageProcessor

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_chinese_clip import (
          CHINESE_CLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
          ChineseCLIPModel,
          ChineseCLIPPreTrainedModel,
          ChineseCLIPTextModel,
          ChineseCLIPVisionModel,
      )

else:
   import sys

   __lowerCamelCase				=			_LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)





