

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   ):
			'''simple docstring'''







			def 							__UpperCamelCase (			self							: str	)							->       Optional[int]:

				"""simple docstring"""

				super().tearDown()
				gc.collect()
				torch.cuda.empty_cache()
			@property
			def 							__UpperCamelCase (			self							: Optional[int]	)							->       Tuple:

				"""simple docstring"""

				torch.manual_seed(0	)
				SCREAMING_SNAKE_CASE   :      Tuple				     =       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 							__UpperCamelCase (			self							: Union[str, Any]	)							->       str:

				"""simple docstring"""

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

				"""simple docstring"""

				torch.manual_seed(0	)
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       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")      ,      )
				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       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 							__UpperCamelCase (			self							: str	)							->       Optional[int]:

				"""simple docstring"""

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

				SCREAMING_SNAKE_CASE   :      int				     =       DDPMScheduler()
				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       AudioDiffusionPipeline(vqvae=a      ,      unet=self.dummy_unet      ,      mel=a      ,      scheduler=a	)
				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       pipe.to(a	)
				pipe.set_progress_bar_config(disable=a	)

				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       torch.Generator(device=a	).manual_seed(42	)
				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       pipe(generator=a      ,      steps=4	)
				SCREAMING_SNAKE_CASE   :      Tuple				     =       output.audios[0]
				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       output.images[0]

				SCREAMING_SNAKE_CASE   :      Tuple				     =       torch.Generator(device=a	).manual_seed(42	)
				SCREAMING_SNAKE_CASE   :      Any				     =       pipe(generator=a      ,      steps=4      ,      return_dict=a	)
				SCREAMING_SNAKE_CASE   :      Union[str, 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]
				)
				SCREAMING_SNAKE_CASE   :      Tuple				     =       np.frombuffer(image.tobytes()      ,      dtype="uint8"	)[:10]
				SCREAMING_SNAKE_CASE   :      Any				     =       np.frombuffer(image_from_tuple.tobytes()      ,      dtype="uint8"	)[:10]
				SCREAMING_SNAKE_CASE   :      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

				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   :      Union[str, Any]				     =       DDIMScheduler()
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       self.dummy_vqvae_and_unet
				SCREAMING_SNAKE_CASE   :      Dict				     =       AudioDiffusionPipeline(
				    vqvae=self.dummy_vqvae_and_unet[0]      ,      unet=dummy_vqvae_and_unet[1]      ,      mel=a      ,      scheduler=a	)
				SCREAMING_SNAKE_CASE   :      Any				     =       pipe.to(a	)
				pipe.set_progress_bar_config(disable=a	)

				np.random.seed(0	)
				SCREAMING_SNAKE_CASE   :      Tuple				     =       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=a	).manual_seed(42	)
				SCREAMING_SNAKE_CASE   :      Any				     =       pipe(raw_audio=a      ,      generator=a      ,      start_step=5      ,      steps=10	)
				SCREAMING_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]
				)
				SCREAMING_SNAKE_CASE   :      List[str]				     =       np.frombuffer(image.tobytes()      ,      dtype="uint8"	)[:10]
				SCREAMING_SNAKE_CASE   :      Dict				     =       np.array([120, 117, 110, 109, 138, 167, 138, 148, 132, 121]	)

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

				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       self.dummy_unet_condition
				SCREAMING_SNAKE_CASE   :      Dict				     =       AudioDiffusionPipeline(
				    vqvae=self.dummy_vqvae_and_unet[0]      ,      unet=a      ,      mel=a      ,      scheduler=a	)
				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       pipe.to(a	)
				pipe.set_progress_bar_config(disable=a	)

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







			def 							__UpperCamelCase (			self							: Dict	)							->       str:

				"""simple docstring"""

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






			def 							__UpperCamelCase (			self							: Dict	)							->       int:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE   :      Tuple				     =       torch_device

				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       DiffusionPipeline.from_pretrained("teticio/audio-diffusion-ddim-256"	)
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       pipe.to(a	)
				pipe.set_progress_bar_config(disable=a	)

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

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


from dataclasses import dataclass
from typing import Optional

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .modeling_utils import ModelMixin






@dataclass
class _UpperCamelCase							(							__A   ):
			'''simple docstring'''







			lowerCamelCase__           =42






class _UpperCamelCase							(							__A							,	__A   ):
			'''simple docstring'''







			@register_to_config
			def __init__(			self							: str      ,      a							: int = 16      ,      a							: int = 88      ,      a							: Optional[int] = None      ,      a							: Optional[int] = None      ,      a							: int = 1      ,      a							: float = 0.0      ,      a							: int = 32      ,      a							: Optional[int] = None      ,      a							: bool = False      ,      a							: Optional[int] = None      ,      a							: str = "geglu"      ,      a							: bool = True      ,      a							: bool = True      ,      )							->       str:

				"""simple docstring"""

				super().__init__()
				SCREAMING_SNAKE_CASE   :      str				     =       num_attention_heads
				SCREAMING_SNAKE_CASE   :      Tuple				     =       attention_head_dim
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       num_attention_heads * attention_head_dim

				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       in_channels

				SCREAMING_SNAKE_CASE   :      List[Any]				     =       torch.nn.GroupNorm(num_groups=a      ,      num_channels=a      ,      eps=1e-6      ,      affine=a	)
				SCREAMING_SNAKE_CASE   :      List[Any]				     =       nn.Linear(a      ,      a	)

				# 3. Define transformers blocks
				SCREAMING_SNAKE_CASE   :      List[str]				     =       nn.ModuleList(
				    [
				        BasicTransformerBlock(
				            a      ,      a      ,      a      ,      dropout=a      ,      cross_attention_dim=a      ,      activation_fn=a      ,      attention_bias=a      ,      double_self_attention=a      ,      norm_elementwise_affine=a      ,      )
				        for d in range(a	)
				    ]	)

				SCREAMING_SNAKE_CASE   :      Optional[int]				     =       nn.Linear(a      ,      a	)






			def 							__UpperCamelCase (			self							: Any      ,      a							: int      ,      a							: Optional[int]=None      ,      a							: List[str]=None      ,      a							: List[str]=None      ,      a							: Union[str, Any]=1      ,      a							: Any=None      ,      a							: bool = True      ,      )							->       Tuple:

				"""simple docstring"""

				SCREAMING_SNAKE_CASE     ,SCREAMING_SNAKE_CASE     ,SCREAMING_SNAKE_CASE     ,SCREAMING_SNAKE_CASE   :      Any				     =       hidden_states.shape
				SCREAMING_SNAKE_CASE   :      Any				     =       batch_frames // num_frames

				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       hidden_states

				SCREAMING_SNAKE_CASE   :      Optional[Any]				     =       hidden_states[None, :].reshape(a      ,      a      ,      a      ,      a      ,      a	)
				SCREAMING_SNAKE_CASE   :      Union[str, Any]				     =       hidden_states.permute(0      ,      2      ,      1      ,      3      ,      4	)

				SCREAMING_SNAKE_CASE   :      Tuple				     =       self.norm(a	)
				SCREAMING_SNAKE_CASE   :      List[Any]				     =       hidden_states.permute(0      ,      3      ,      4      ,      2      ,      1	).reshape(batch_size * height * width      ,      a      ,      a	)

				SCREAMING_SNAKE_CASE   :      List[Any]				     =       self.proj_in(a	)

				# 2. Blocks
				for block in self.transformer_blocks:
					SCREAMING_SNAKE_CASE   :      Optional[int]				     =       block(
					    a      ,      encoder_hidden_states=a      ,      timestep=a      ,      cross_attention_kwargs=a      ,      class_labels=a      ,      )

				# 3. Output
				SCREAMING_SNAKE_CASE   :      List[str]				     =       self.proj_out(a	)
				SCREAMING_SNAKE_CASE   :      int				     =       (
				    hidden_states[None, None, :]
				    .reshape(a      ,      a      ,      a      ,      a      ,      a	)
				    .permute(0      ,      3      ,      4      ,      1      ,      2	)
				    .contiguous()
				)
				SCREAMING_SNAKE_CASE   :      Dict				     =       hidden_states.reshape(a      ,      a      ,      a      ,      a	)

				SCREAMING_SNAKE_CASE   :      int				     =       hidden_states + residual

				if not return_dict:
					return (output,)

				return TransformerTemporalModelOutput(sample=a	)