




import gc
import unittest

import numpy as np
import torch

from diffusers import DanceDiffusionPipeline, IPNDMScheduler, UNetaDModel
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS, UNCONDITIONAL_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class       a__	( __lowerCamelCase       , unittest.TestCase				):
     a		:      int		 =     DanceDiffusionPipeline
     a		:      int		 =     UNCONDITIONAL_AUDIO_GENERATION_PARAMS
     a		:      Optional[Any]		 =     PipelineTesterMixin.required_optional_params - {
         """callback""",
         """latents""",
         """callback_steps""",
         """output_type""",
         """num_images_per_prompt""",
     }
     a		:      int		 =     UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
     a		:      str		 =     False
     a		:      str		 =     False







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



          '''simple docstring'''



          torch.manual_seed(0			)
          a     =				UNetaDModel(
              block_out_channels=(32, 32, 64)  ,		extra_in_channels=16  ,		sample_size=512  ,		sample_rate=16000  ,		in_channels=2  ,		out_channels=2  ,		flip_sin_to_cos=__lowercase  ,		use_timestep_embedding=__lowercase  ,		time_embedding_type="fourier"  ,		mid_block_type="UNetMidBlock1D"  ,		down_block_types=("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")  ,		up_block_types=("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")  ,		)
          a     =				IPNDMScheduler()

          a     =				{
              '''unet''': unet,
              '''scheduler''': scheduler,
          }
          return components







     def       lowerCAmelCase_    (     self  ,		A  ,		A=0			)  ->   Dict:



          '''simple docstring'''



          if str(__lowercase			).startswith("mps"			):
               a     =				torch.manual_seed(__lowercase			)
          else:
               a     =				torch.Generator(device=__lowercase			).manual_seed(__lowercase			)
          a     =				{
              '''batch_size''': 1,
              '''generator''': generator,
              '''num_inference_steps''': 4,
          }
          return inputs







     def       lowerCAmelCase_    (     self			)  ->   str:



          '''simple docstring'''



          a     =				'''cpu'''  # ensure determinism for the device-dependent torch.Generator
          a     =				self.get_dummy_components()
          a     =				DanceDiffusionPipeline(**__lowercase			)
          a     =				pipe.to(__lowercase			)
          pipe.set_progress_bar_config(disable=__lowercase			)

          a     =				self.get_dummy_inputs(__lowercase			)
          a     =				pipe(**__lowercase			)
          a     =				output.audios

          a     =				audio[0, -3:, -3:]

          assert audio.shape == (1, 2, components["unet"].sample_size)
          a     =				np.array([-0.7_2_6_5, 1.0_0_0_0, -0.8_3_8_8, 0.1_1_7_5, 0.9_4_9_8, -1.0_0_0_0]			)
          assert np.abs(audio_slice.flatten() - expected_slice			).max() < 1e-2







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



          '''simple docstring'''



          return super().test_save_load_local()







     @skip_mps
     def       lowerCAmelCase_    (     self			)  ->   Dict:



          '''simple docstring'''



          return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3			)







     @skip_mps
     def       lowerCAmelCase_    (     self			)  ->   Any:



          '''simple docstring'''



          return super().test_save_load_optional_components()







     @skip_mps
     def       lowerCAmelCase_    (     self			)  ->   List[Any]:



          '''simple docstring'''



          return super().test_attention_slicing_forward_pass()







     def       lowerCAmelCase_    (     self			)  ->   int:



          '''simple docstring'''



          super().test_inference_batch_single_identical(expected_max_diff=3e-3			)







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







     def       lowerCAmelCase_    (     self			)  ->   Dict:



          '''simple docstring'''



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







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



          '''simple docstring'''



          a     =				torch_device

          a     =				DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"			)
          a     =				pipe.to(__lowercase			)
          pipe.set_progress_bar_config(disable=__lowercase			)

          a     =				torch.manual_seed(0			)
          a     =				pipe(generator=__lowercase  ,		num_inference_steps=100  ,		audio_length_in_s=4.0_9_6			)
          a     =				output.audios

          a     =				audio[0, -3:, -3:]

          assert audio.shape == (1, 2, pipe.unet.sample_size)
          a     =				np.array([-0.0_1_9_2, -0.0_2_3_1, -0.0_3_1_8, -0.0_0_5_9, 0.0_0_0_2, -0.0_0_2_0]			)

          assert np.abs(audio_slice.flatten() - expected_slice			).max() < 1e-2







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



          '''simple docstring'''



          a     =				torch_device

          a     =				DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"  ,		torch_dtype=torch.floataa			)
          a     =				pipe.to(__lowercase			)
          pipe.set_progress_bar_config(disable=__lowercase			)

          a     =				torch.manual_seed(0			)
          a     =				pipe(generator=__lowercase  ,		num_inference_steps=100  ,		audio_length_in_s=4.0_9_6			)
          a     =				output.audios

          a     =				audio[0, -3:, -3:]

          assert audio.shape == (1, 2, pipe.unet.sample_size)
          a     =				np.array([-0.0_3_6_7, -0.0_4_8_8, -0.0_7_7_1, -0.0_5_2_5, -0.0_4_4_4, -0.0_3_4_1]			)

          assert np.abs(audio_slice.flatten() - expected_slice			).max() < 1e-2







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


lowercase__				:      int	   = logging.get_logger(__name__)


lowercase__				:      Dict	   = "▁"

lowercase__				:      Union[str, Any]	   = {"vocab_file": "spiece.model"}

lowercase__				:      Union[str, Any]	   = {
    "vocab_file": {
        "google/reformer-crime-and-punishment": (
            "https://huggingface.co/google/reformer-crime-and-punishment/resolve/main/spiece.model"
        )
    }
}

lowercase__				:      Tuple	   = {
    "google/reformer-crime-and-punishment": 524_288,
}




class       a__	( UpperCamelCase__				):
     a		:      List[Any]		 =     VOCAB_FILES_NAMES
     a		:      List[Any]		 =     PRETRAINED_VOCAB_FILES_MAP
     a		:      Any		 =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     a		:      Dict		 =     ["""input_ids""", """attention_mask"""]







     def __init__(     self  ,		A  ,		A="</s>"  ,		A="<unk>"  ,		A=[]  ,		A = None  ,		**A  ,		)  ->   None:



          '''simple docstring'''



          a     =				{} if sp_model_kwargs is None else sp_model_kwargs

          super().__init__(
              eos_token=A  ,		unk_token=A  ,		additional_special_tokens=A  ,		sp_model_kwargs=self.sp_model_kwargs  ,		**A  ,		)

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







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



          '''simple docstring'''



          return self.sp_model.get_piece_size()







     def       lowerCAmelCase_    (     self			)  ->   Dict[str, int]:



          '''simple docstring'''



          a     =				{self.convert_ids_to_tokens(A			): i for i in range(self.vocab_size			)}
          vocab.update(self.added_tokens_encoder			)
          return vocab







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



          '''simple docstring'''



          a     =				self.__dict__.copy()
          a     =				None
          return state







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



          '''simple docstring'''



          a     =				d

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

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







     def       lowerCAmelCase_    (     self  ,		A			)  ->   List[str]:



          '''simple docstring'''



          return self.sp_model.encode(A  ,		out_type=A			)







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



          '''simple docstring'''



          return self.sp_model.piece_to_id(A			)







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



          '''simple docstring'''



          if index < self.sp_model.get_piece_size():
               a     =				self.sp_model.IdToPiece(A			)
          return token







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



          '''simple docstring'''



          a     =				[]
          a     =				""
          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(A			) + token
                    a     =				[]
               else:
                    current_sub_tokens.append(A			)
          out_string += self.sp_model.decode(A			)
          return out_string.strip()







     def       lowerCAmelCase_    (     self  ,		A  ,		A = None			)  ->   Tuple[str]:



          '''simple docstring'''



          if not os.path.isdir(A			):
               logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''			)
               return
          a     =				os.path.join(
              A  ,		(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]			)

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

          return (out_vocab_file,)

