

'''simple docstring'''






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




  """simple docstring"""




  def 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]			):
   # clean up the VRAM after each test
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     Any			):
   torch.manual_seed(0			)
   __lowercase							  = UNetaDModel(
       sample_size=(3_2, 6_4)					,in_channels=1					,out_channels=1					,layers_per_block=2					,block_out_channels=(1_2_8, 1_2_8)					,down_block_types=('''AttnDownBlock2D''', '''DownBlock2D''')					,up_block_types=('''UpBlock2D''', '''AttnUpBlock2D''')					,)
   return model






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]			):
   torch.manual_seed(0			)
   __lowercase							  = UNetaDConditionModel(
       sample_size=(6_4, 3_2)					,in_channels=1					,out_channels=1					,layers_per_block=2					,block_out_channels=(1_2_8, 1_2_8)					,down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D''')					,up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D''')					,cross_attention_dim=1_0					,)
   return model






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     int			):
   torch.manual_seed(0			)
   __lowercase							  = AutoencoderKL(
       sample_size=(1_2_8, 6_4)					,in_channels=1					,out_channels=1					,latent_channels=1					,layers_per_block=2					,block_out_channels=(1_2_8, 1_2_8)					,down_block_types=('''DownEncoderBlock2D''', '''DownEncoderBlock2D''')					,up_block_types=('''UpDecoderBlock2D''', '''UpDecoderBlock2D''')					,)
   __lowercase							  = UNetaDModel(
       sample_size=(6_4, 3_2)					,in_channels=1					,out_channels=1					,layers_per_block=2					,block_out_channels=(1_2_8, 1_2_8)					,down_block_types=('''AttnDownBlock2D''', '''DownBlock2D''')					,up_block_types=('''UpBlock2D''', '''AttnUpBlock2D''')					,)
   return vqvae, unet






  @slow
  def 					SCREAMING_SNAKE_CASE      (      self   :     List[str]			):
   __lowercase							  = '''cpu'''  # ensure determinism for the device-dependent torch.Generator
   __lowercase							  = Mel(
       x_res=self.dummy_unet.config.sample_size[1]					,y_res=self.dummy_unet.config.sample_size[0]					,)

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

   __lowercase							  = torch.Generator(device=lowercase__			).manual_seed(4_2			)
   __lowercase							  = pipe(generator=lowercase__					,steps=4			)
   __lowercase							  = output.audios[0]
   __lowercase							  = output.images[0]

   __lowercase							  = torch.Generator(device=lowercase__			).manual_seed(4_2			)
   __lowercase							  = pipe(generator=lowercase__					,steps=4					,return_dict=lowercase__			)
   __lowercase							  = 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]
   )
   __lowercase							  = np.frombuffer(image.tobytes()					,dtype='''uint8'''			)[:1_0]
   __lowercase							  = np.frombuffer(image_from_tuple.tobytes()					,dtype='''uint8'''			)[:1_0]
   __lowercase							  = np.array([6_9, 2_5_5, 2_5_5, 2_5_5, 0, 0, 7_7, 1_8_1, 1_2, 1_2_7]			)

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

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

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

   np.random.seed(0			)
   __lowercase							  = np.random.uniform(-1					,1					,((dummy_vqvae_and_unet[0].config.sample_size[1] - 1) * mel.hop_length,)			)
   __lowercase							  = torch.Generator(device=lowercase__			).manual_seed(4_2			)
   __lowercase							  = pipe(raw_audio=lowercase__					,generator=lowercase__					,start_step=5					,steps=1_0			)
   __lowercase							  = 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]
   )
   __lowercase							  = np.frombuffer(image.tobytes()					,dtype='''uint8'''			)[:1_0]
   __lowercase							  = np.array([1_2_0, 1_1_7, 1_1_0, 1_0_9, 1_3_8, 1_6_7, 1_3_8, 1_4_8, 1_3_2, 1_2_1]			)

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

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

   np.random.seed(0			)
   __lowercase							  = torch.rand((1, 1, 1_0)			)
   __lowercase							  = pipe(generator=lowercase__					,encoding=lowercase__			)
   __lowercase							  = output.images[0]
   __lowercase							  = np.frombuffer(image.tobytes()					,dtype='''uint8'''			)[:1_0]
   __lowercase							  = np.array([1_0_7, 1_0_3, 1_2_0, 1_2_7, 1_4_2, 1_2_2, 1_1_3, 1_2_2, 9_7, 1_1_1]			)

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






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




  """simple docstring"""




  def 					SCREAMING_SNAKE_CASE      (      self   :     Dict			):
   # clean up the VRAM after each test
   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()






  def 					SCREAMING_SNAKE_CASE      (      self   :     List[str]			):
   __lowercase							  = torch_device

   __lowercase							  = DiffusionPipeline.from_pretrained('''teticio/audio-diffusion-ddim-256'''			)
   __lowercase							  = pipe.to(lowercase__			)
   pipe.set_progress_bar_config(disable=lowercase__			)

   __lowercase							  = torch.Generator(device=lowercase__			).manual_seed(4_2			)
   __lowercase							  = pipe(generator=lowercase__			)
   __lowercase							  = output.audios[0]
   __lowercase							  = 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]
   __lowercase							  = np.frombuffer(image.tobytes()					,dtype='''uint8'''			)[:1_0]
   __lowercase							  = np.array([1_5_1, 1_6_7, 1_5_4, 1_4_4, 1_2_2, 1_3_4, 1_2_1, 1_0_5, 7_0, 2_6]			)

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




'''simple docstring'''






import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


lowerCAmelCase__											=	logging.get_logger(__name__)

lowerCAmelCase__											=	{'''vocab_file''': '''spiece.model'''}

lowerCAmelCase__											=	{
    '''vocab_file''': {
        '''bert_for_seq_generation''': (
            '''https://huggingface.co/google/bert_for_seq_generation_L-24_bbc_encoder/resolve/main/spiece.model'''
        ),
    }
}

lowerCAmelCase__											=	{'''bert_for_seq_generation''': 512}






class 			lowercase_	(lowerCamelCase__      ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE    :  List[Any]							=			VOCAB_FILES_NAMES
  SCREAMING_SNAKE_CASE    :  Dict							=			PRETRAINED_VOCAB_FILES_MAP
  SCREAMING_SNAKE_CASE    :  Optional[int]							=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  SCREAMING_SNAKE_CASE    :  List[int]							=			[]
  SCREAMING_SNAKE_CASE    :  Any							=			['input_ids', 'attention_mask']






  def __init__(      self   :     Optional[Any]					,lowercase__   :     Tuple					,lowercase__   :     Tuple="<s>"					,lowercase__   :     Union[str, Any]="</s>"					,lowercase__   :     str="<unk>"					,lowercase__   :     Tuple="<pad>"					,lowercase__   :     Union[str, Any]="<::::>"					,lowercase__   :     Optional[Dict[str, Any]] = None					,**lowercase__   :     Any					,):
   __lowercase							  = {} if sp_model_kwargs is None else sp_model_kwargs

   # Add extra_ids to the special token list
   super().__init__(
       bos_token=lowercase__					,eos_token=lowercase__					,unk_token=lowercase__					,pad_token=lowercase__					,sep_token=lowercase__					,sp_model_kwargs=self.sp_model_kwargs					,**lowercase__					,)

   __lowercase							  = vocab_file

   __lowercase							  = spm.SentencePieceProcessor(**self.sp_model_kwargs			)
   self.sp_model.Load(lowercase__			)






  @property
  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]			):
   return self.sp_model.get_piece_size()






  def 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]			):
   __lowercase							  = {self.convert_ids_to_tokens(lowercase__			): i for i in range(self.vocab_size			)}
   vocab.update(self.added_tokens_encoder			)
   return vocab






  def __getstate__(      self   :     str			):
   __lowercase							  = self.__dict__.copy()
   __lowercase							  = None
   return state






  def __setstate__(      self   :     Optional[int]					,lowercase__   :     Optional[Any]			):
   __lowercase							  = d

   # for backward compatibility
   if not hasattr(self					,'''sp_model_kwargs'''			):
    __lowercase							  = {}

   __lowercase							  = spm.SentencePieceProcessor(**self.sp_model_kwargs			)
   self.sp_model.Load(self.vocab_file			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     str					,lowercase__   :     str			):
   return self.sp_model.encode(lowercase__					,out_type=lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Tuple					,lowercase__   :     Union[str, Any]			):
   return self.sp_model.piece_to_id(lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Any					,lowercase__   :     Tuple			):
   __lowercase							  = self.sp_model.IdToPiece(lowercase__			)
   return token






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]					,lowercase__   :     int			):
   __lowercase							  = []
   __lowercase							  = ''''''
   for token in tokens:
    # make sure that special tokens are not decoded using sentencepiece model
    if token in self.all_special_tokens:
     out_string += self.sp_model.decode(lowercase__			) + token
     __lowercase							  = []
    else:
     current_sub_tokens.append(lowercase__			)
   out_string += self.sp_model.decode(lowercase__			)
   return out_string.strip()






  def 					SCREAMING_SNAKE_CASE      (      self   :     int					,lowercase__   :     str					,lowercase__   :     Optional[str] = None			):
   if not os.path.isdir(lowercase__			):
    logger.error(F"Vocabulary path ({save_directory}) should be a directory"			)
    return
   __lowercase							  = os.path.join(
       lowercase__					,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']			)

   if os.path.abspath(self.vocab_file			) != os.path.abspath(lowercase__			) and os.path.isfile(self.vocab_file			):
    copyfile(self.vocab_file					,lowercase__			)
   elif not os.path.isfile(self.vocab_file			):
    with open(lowercase__					,'''wb'''			) as fi:
     __lowercase							  = self.sp_model.serialized_model_proto()
     fi.write(lowercase__			)

   return (out_vocab_file,)

