




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				):







     def       lowerCAmelCase_    (     self			)  ->   Optional[int]:



          '''simple docstring'''



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







     @property
     def       lowerCAmelCase_    (     self			)  ->   Dict:



          '''simple docstring'''



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



          '''simple docstring'''



          torch.manual_seed(0			)
          a     =				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       lowerCAmelCase_    (     self			)  ->   str:



          '''simple docstring'''



          torch.manual_seed(0			)
          a     =				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")  ,		)
          a     =				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       lowerCAmelCase_    (     self			)  ->   Optional[Any]:



          '''simple docstring'''



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

          a     =				DDPMScheduler()
          a     =				AudioDiffusionPipeline(vqvae=A  ,		unet=self.dummy_unet  ,		mel=A  ,		scheduler=A			)
          a     =				pipe.to(A			)
          pipe.set_progress_bar_config(disable=A			)

          a     =				torch.Generator(device=A			).manual_seed(42			)
          a     =				pipe(generator=A  ,		steps=4			)
          a     =				output.audios[0]
          a     =				output.images[0]

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

          a     =				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]  ,		)

          a     =				DDIMScheduler()
          a     =				self.dummy_vqvae_and_unet
          a     =				AudioDiffusionPipeline(
              vqvae=self.dummy_vqvae_and_unet[0]  ,		unet=dummy_vqvae_and_unet[1]  ,		mel=A  ,		scheduler=A			)
          a     =				pipe.to(A			)
          pipe.set_progress_bar_config(disable=A			)

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

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

          a     =				self.dummy_unet_condition
          a     =				AudioDiffusionPipeline(
              vqvae=self.dummy_vqvae_and_unet[0]  ,		unet=A  ,		mel=A  ,		scheduler=A			)
          a     =				pipe.to(A			)
          pipe.set_progress_bar_config(disable=A			)

          np.random.seed(0			)
          a     =				torch.rand((1, 1, 10)			)
          a     =				pipe(generator=A  ,		encoding=A			)
          a     =				output.images[0]
          a     =				np.frombuffer(image.tobytes()  ,		dtype="uint8"			)[:10]
          a     =				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				):







     def       lowerCAmelCase_    (     self			)  ->   Optional[Any]:



          '''simple docstring'''



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







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



          '''simple docstring'''



          a     =				torch_device

          a     =				DiffusionPipeline.from_pretrained("teticio/audio-diffusion-ddim-256"			)
          a     =				pipe.to(A			)
          pipe.set_progress_bar_config(disable=A			)

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

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







# Author: OMKAR PATHAK, Nwachukwu Chidiebere

# Use a Python dictionary to construct the graph.
from __future__ import annotations

from pprint import pformat
from typing import Generic, TypeVar

lowercase__				:      str	   = TypeVar("T")




class       a__	( Generic[T]				):







     def __init__(     self  ,		A = True			)  ->   None:



          '''simple docstring'''



          a     =				{}  # dictionary of lists
          a     =				directed







     def       lowerCAmelCase_    (     self  ,		A  ,		A			)  ->   GraphAdjacencyList[T]:



          '''simple docstring'''



          if not self.directed:  # For undirected graphs
               # if both source vertex and destination vertex are both present in the
               # adjacency list, add destination vertex to source vertex list of adjacent
               # vertices and add source vertex to destination vertex list of adjacent
               # vertices.
               if source_vertex in self.adj_list and destination_vertex in self.adj_list:
                    self.adj_list[source_vertex].append(A			)
                    self.adj_list[destination_vertex].append(A			)
               # if only source vertex is present in adjacency list, add destination vertex
               # to source vertex list of adjacent vertices, then create a new vertex with
               # destination vertex as key and assign a list containing the source vertex
               # as it's first adjacent vertex.
               elif source_vertex in self.adj_list:
                    self.adj_list[source_vertex].append(A			)
                    a     =				[source_vertex]
               # if only destination vertex is present in adjacency list, add source vertex
               # to destination vertex list of adjacent vertices, then create a new vertex
               # with source vertex as key and assign a list containing the source vertex
               # as it's first adjacent vertex.
               elif destination_vertex in self.adj_list:
                    self.adj_list[destination_vertex].append(A			)
                    a     =				[destination_vertex]
               # if both source vertex and destination vertex are not present in adjacency
               # list, create a new vertex with source vertex as key and assign a list
               # containing the destination vertex as it's first adjacent vertex also
               # create a new vertex with destination vertex as key and assign a list
               # containing the source vertex as it's first adjacent vertex.
               else:
                    a     =				[destination_vertex]
                    a     =				[source_vertex]
          else:  # For directed graphs
               # if both source vertex and destination vertex are present in adjacency
               # list, add destination vertex to source vertex list of adjacent vertices.
               if source_vertex in self.adj_list and destination_vertex in self.adj_list:
                    self.adj_list[source_vertex].append(A			)
               # if only source vertex is present in adjacency list, add destination
               # vertex to source vertex list of adjacent vertices and create a new vertex
               # with destination vertex as key, which has no adjacent vertex
               elif source_vertex in self.adj_list:
                    self.adj_list[source_vertex].append(A			)
                    a     =				[]
               # if only destination vertex is present in adjacency list, create a new
               # vertex with source vertex as key and assign a list containing destination
               # vertex as first adjacent vertex
               elif destination_vertex in self.adj_list:
                    a     =				[destination_vertex]
               # if both source vertex and destination vertex are not present in adjacency
               # list, create a new vertex with source vertex as key and a list containing
               # destination vertex as it's first adjacent vertex. Then create a new vertex
               # with destination vertex as key, which has no adjacent vertex
               else:
                    a     =				[destination_vertex]
                    a     =				[]

          return self







     def __repr__(     self			)  ->   str:



          '''simple docstring'''



          return pformat(self.adj_list			)

