


"""simple docstring"""






import argparse
import logging

import datasets
from tokenizers import Tokenizer, decoders, normalizers, pre_tokenizers, processors
from tokenizers.models import Unigram
from tokenizers.trainers import UnigramTrainer

from transformers import AlbertTokenizerFast


SCREAMING_SNAKE_CASE__				:  Tuple						= logging.getLogger(__name__)




def        A_					(					)     ->     str:
   a   :	str						=    argparse.ArgumentParser(description='Train a unigram tokenizer on the wikitext dataset.'				)
   parser.add_argument(
       '--dataset_name'       ,   type=_UpperCAmelCase       ,   default='wikitext'       ,   help='Name of the training. Explore datasets at: hf.co/datasets.'       ,   )
   parser.add_argument(
       '--dataset_config'       ,   type=_UpperCAmelCase       ,   default='wikitext-103-raw-v1'       ,   help='Configuration name of the dataset.'				)
   parser.add_argument(
       '--batch_size'       ,   type=_UpperCAmelCase       ,   default=1000       ,   help='Batch size during training.'       ,   )
   parser.add_argument(
       '--vocab_size'       ,   type=_UpperCAmelCase       ,   default=1_0048       ,   help='Size of the desired vocabulary.'       ,   )
   parser.add_argument(
       '--limit'       ,   default=_UpperCAmelCase       ,   type=_UpperCAmelCase       ,   help='Limit the number of shards (used for debugging).'       ,   )
   parser.add_argument(
       '--export_to_hub'       ,   action='store_true'       ,   )

   a   :	Dict						=    parser.parse_args()
   return args




def        A_					(					UpperCAmelCase__				)     ->     str:
   a   :	str						=    datasets.load_dataset(args.dataset_name       ,   args.dataset_config       ,   split='train'				)

   if args.limit is not None:
      a   :	str						=    min(len(_UpperCAmelCase				)       ,   args.limit				)
      a   :	Tuple						=    dataset.select(range(_UpperCAmelCase				)				)
      logger.info(F'Limiting the dataset to {args.limit} entries.'				)

   def batch_iterator():
      for i in range(0       ,   len(_UpperCAmelCase				)       ,   args.batch_size				):
         yield dataset[i : i + args.batch_size]["text"]

    # Prepare the tokenizer.
   a   :	Dict						=    Tokenizer(Unigram()				)
   a   :	List[Any]						=    normalizers.Sequence([normalizers.Replace('``'       ,   '"'				), normalizers.Replace('\'\''       ,   '"'				)]				)
   a   :	str						=    pre_tokenizers.Metaspace()

   # Prepare the trainer.
   a   :	Optional[Any]						=    UnigramTrainer(
       unk_token='<unk>'       ,   special_tokens=['[CLS]', '[SEP]', '<unk>', '<pad>', '[MASK]']       ,   vocab_size=args.vocab_size       ,   )

   logger.info('Training the tokenizer.'				)
   tokenizer.train_from_iterator(batch_iterator()       ,   trainer=_UpperCAmelCase				)
   logger.info('Tokenizer training complete!'				)

   a   :	Optional[Any]						=    tokenizer.token_to_id('[CLS]'				)
   a   :	List[str]						=    tokenizer.token_to_id('[SEP]'				)
   a   :	str						=    processors.TemplateProcessing(
       single='[CLS]:0 $A:0 [SEP]:0'       ,   pair='[CLS]:0 $A:0 [SEP]:0 $B:1 [SEP]:1'       ,   special_tokens=[
           ('[CLS]', cls_token_id),
           ('[SEP]', sep_token_id),
       ]       ,   )
   a   :	List[str]						=    decoders.Metaspace()

   if args.export_to_hub:
      logger.info('Exporting the trained tokenzier to Hub.'				)
      a   :	List[Any]						=    AlbertTokenizerFast(tokenizer_object=_UpperCAmelCase				)
      new_tokenizer.push_to_hub('unigram-tokenizer-dataset'				)


if __name__ == "__main__":
 SCREAMING_SNAKE_CASE__				:  Optional[int]						= parse_args()
 main(args)










import gc
import unittest

import numpy as np
import torch

from diffusers import DDPMScheduler, MidiProcessor, SpectrogramDiffusionPipeline
from diffusers.pipelines.spectrogram_diffusion import SpectrogramContEncoder, SpectrogramNotesEncoder, TaFilmDecoder
from diffusers.utils import require_torch_gpu, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_note_seq, require_onnxruntime

from ..pipeline_params import TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS, TOKENS_TO_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


UpperCAmelCase__  =			"./tests/fixtures/elise_format0.mid"







@unittest.skip('''The note-seq package currently throws an error on import'''		)
class      __lowerCAmelCase       (     A							,		unittest.TestCase		):
			UpperCamelCase     =	SpectrogramDiffusionPipeline
			UpperCamelCase     =	PipelineTesterMixin.required_optional_params - {
			    '''callback''',
			    '''latents''',
			    '''callback_steps''',
			    '''output_type''',
			    '''num_images_per_prompt''',
			}
			UpperCamelCase     =	False
			UpperCamelCase     =	False
			UpperCamelCase     =	TOKENS_TO_AUDIO_GENERATION_PARAMS
			UpperCamelCase     =	TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS



			def        _lowerCamelCase			(		self  :    str)  ->       Any:



						"""simple docstring"""






						torch.manual_seed(0)
						_UpperCAmelCase									=  SpectrogramNotesEncoder(
						    max_length=20_48					,						vocab_size=15_36					,						d_model=7_68					,						dropout_rate=0.1					,						num_layers=1					,						num_heads=1					,						d_kv=4					,						d_ff=20_48					,						feed_forward_proj='gated-gelu'					,						)

						_UpperCAmelCase									=  SpectrogramContEncoder(
						    input_dims=1_28					,						targets_context_length=2_56					,						d_model=7_68					,						dropout_rate=0.1					,						num_layers=1					,						num_heads=1					,						d_kv=4					,						d_ff=20_48					,						feed_forward_proj='gated-gelu'					,						)

						_UpperCAmelCase									=  TaFilmDecoder(
						    input_dims=1_28					,						targets_length=2_56					,						max_decoder_noise_time=2_0_0_0_0.0					,						d_model=7_68					,						num_layers=1					,						num_heads=1					,						d_kv=4					,						d_ff=20_48					,						dropout_rate=0.1					,						)

						_UpperCAmelCase									=  DDPMScheduler()

						_UpperCAmelCase									=  {
						    'notes_encoder': notes_encoder.eval(),
						    'continuous_encoder': continuous_encoder.eval(),
						    'decoder': decoder.eval(),
						    'scheduler': scheduler,
						    'melgan': None,
						}
						return components



			def        _lowerCamelCase			(		self  :    Optional[Any]					,						A  :    str					,						A  :    Union[str, Any]=0)  ->       Optional[Any]:



						"""simple docstring"""






						if str(A).startswith('mps'):
									_UpperCAmelCase									=  torch.manual_seed(A)
						else:
									_UpperCAmelCase									=  torch.Generator(device=A).manual_seed(A)
						_UpperCAmelCase									=  {
						    'input_tokens': [
						        [11_34, 90, 11_35, 11_33, 10_80, 1_12, 11_32, 10_80, 11_33, 10_79, 1_33, 11_32, 10_79, 11_33, 1] + [0] * 20_33
						    ],
						    'generator': generator,
						    'num_inference_steps': 4,
						    'output_type': 'mel',
						}
						return inputs



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



						"""simple docstring"""






						_UpperCAmelCase									=  'cpu'  # ensure determinism for the device-dependent torch.Generator
						_UpperCAmelCase									=  self.get_dummy_components()
						_UpperCAmelCase									=  SpectrogramDiffusionPipeline(**A)
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase									=  self.get_dummy_inputs(A)
						_UpperCAmelCase									=  pipe(**A)
						_UpperCAmelCase									=  output.audios

						_UpperCAmelCase									=  mel[0, -3:, -3:]

						assert mel_slice.shape == (3, 3)
						_UpperCAmelCase									=  np.array(
						    [-1_1.5_1_2_9_2_5, -4.7_8_8_2_1_5, -0.4_6_1_7_2_9_0_5, -2.0_5_1_7_1_5, -1_0.5_3_9_1_4_7, -1_0.9_7_0_9_6_3, -9.0_9_1_6_3_4, 4.0, 4.0])
						assert np.abs(mel_slice.flatten() - expected_slice).max() < 1E-2



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[Any])  ->       Optional[Any]:



						"""simple docstring"""






						return super().test_save_load_local()



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[Any])  ->       List[str]:



						"""simple docstring"""






						return super().test_dict_tuple_outputs_equivalent()



			@skip_mps
			def        _lowerCamelCase			(		self  :    Optional[Any])  ->       Dict:



						"""simple docstring"""






						return super().test_save_load_optional_components()



			@skip_mps
			def        _lowerCamelCase			(		self  :    Union[str, Any])  ->       List[str]:



						"""simple docstring"""






						return super().test_attention_slicing_forward_pass()



			def        _lowerCamelCase			(		self  :    List[Any])  ->       Dict:



						"""simple docstring"""






						pass



			def        _lowerCamelCase			(		self  :    str)  ->       Any:



						"""simple docstring"""






						pass



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



						"""simple docstring"""






						return super().test_progress_bar()







@slow
@require_torch_gpu
@require_onnxruntime
@require_note_seq
class      __lowerCAmelCase       (     unittest.TestCase		):



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



						"""simple docstring"""






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



			def        _lowerCamelCase			(		self  :    int)  ->       Tuple:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.melgan
						_UpperCAmelCase									=  None

						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						def callback(A  :    Dict					,						A  :    Optional[int]):
									# decode mel to audio
									_UpperCAmelCase									=  melgan(input_features=mel_output.astype(np.floataa))[0]
									assert len(audio[0]) == 8_19_20 * (step + 1)
									# simulate that audio is played
									return audio

						_UpperCAmelCase									=  MidiProcessor()
						_UpperCAmelCase									=  processor(A)

						_UpperCAmelCase									=  input_tokens[:3]
						_UpperCAmelCase									=  torch.manual_seed(0)
						pipe(A					,						num_inference_steps=5					,						generator=A					,						callback=A					,						output_type='mel')



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



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)
						_UpperCAmelCase									=  MidiProcessor()

						_UpperCAmelCase									=  processor(A)
						# just run two denoising loops
						_UpperCAmelCase									=  input_tokens[:2]

						_UpperCAmelCase									=  torch.manual_seed(0)
						_UpperCAmelCase									=  pipe(A					,						num_inference_steps=2					,						generator=A)

						_UpperCAmelCase									=  output.audios[0]

						assert abs(np.abs(A).sum() - 3_6_1_2.8_4_1) < 1E-1



			def        _lowerCamelCase			(		self  :    List[str])  ->       Tuple:



						"""simple docstring"""






						_UpperCAmelCase									=  torch_device

						_UpperCAmelCase									=  SpectrogramDiffusionPipeline.from_pretrained('google/music-spectrogram-diffusion')
						_UpperCAmelCase									=  pipe.to(A)
						pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase									=  MidiProcessor()

						_UpperCAmelCase									=  processor(A)

						# just run 4 denoising loops
						_UpperCAmelCase									=  input_tokens[:4]

						_UpperCAmelCase									=  torch.manual_seed(0)
						_UpperCAmelCase									=  pipe(A					,						num_inference_steps=1_00					,						generator=A)

						_UpperCAmelCase									=  output.audios[0]
						assert abs(np.abs(A).sum() - 9_3_8_9.1_1_1_1) < 5E-2
