


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 							UpperCamelCase__ (						a__				,					unittest.TestCase ):
					__SCREAMING_SNAKE_CASE							:	Tuple            = DanceDiffusionPipeline
					__SCREAMING_SNAKE_CASE							:	Tuple            = UNCONDITIONAL_AUDIO_GENERATION_PARAMS
					__SCREAMING_SNAKE_CASE							:	List[Any]            = PipelineTesterMixin.required_optional_params - {
					    'callback',
					    'latents',
					    'callback_steps',
					    'output_type',
					    'num_images_per_prompt',
					}
					__SCREAMING_SNAKE_CASE							:	int            = UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
					__SCREAMING_SNAKE_CASE							:	Optional[int]            = False
					__SCREAMING_SNAKE_CASE							:	List[Any]            = False

					def 	UpperCAmelCase__							(  self							: Any							):






										'''simple docstring'''






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

										lowercase_					=      {
										    """unet""": unet,
										    """scheduler""": scheduler,
										}
										return components

					def 	UpperCAmelCase__							(  self							: Optional[Any]				,       UpperCamelCase__							: str				,       UpperCamelCase__							: Optional[int]=0							):






										'''simple docstring'''






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

					def 	UpperCAmelCase__							(  self							: Optional[int]							):






										'''simple docstring'''






										lowercase_					=      """cpu"""  # ensure determinism for the device-dependent torch.Generator
										lowercase_					=      self.get_dummy_components()
										lowercase_					=      DanceDiffusionPipeline(**_snake_case							)
										lowercase_					=      pipe.to(_snake_case							)
										pipe.set_progress_bar_config(disable=_snake_case							)

										lowercase_					=      self.get_dummy_inputs(_snake_case							)
										lowercase_					=      pipe(**_snake_case							)
										lowercase_					=      output.audios

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

										assert audio.shape == (1, 2, components["unet"].sample_size)
										lowercase_					=      np.array([-0.7_265, 1.0_000, -0.8_388, 0.1_175, 0.9_498, -1.0_000]							)
										assert np.abs(audio_slice.flatten() - expected_slice							).max() < 1e-2

					@skip_mps
					def 	UpperCAmelCase__							(  self							: Optional[Any]							):






										'''simple docstring'''






										return super().test_save_load_local()

					@skip_mps
					def 	UpperCAmelCase__							(  self							: str							):






										'''simple docstring'''






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

					@skip_mps
					def 	UpperCAmelCase__							(  self							: Tuple							):






										'''simple docstring'''






										return super().test_save_load_optional_components()

					@skip_mps
					def 	UpperCAmelCase__							(  self							: Dict							):






										'''simple docstring'''






										return super().test_attention_slicing_forward_pass()







					def 	UpperCAmelCase__							(  self							: Optional[Any]							):






										'''simple docstring'''






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







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

					def 	UpperCAmelCase__							(  self							: List[str]							):






										'''simple docstring'''






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

					def 	UpperCAmelCase__							(  self							: Union[str, Any]							):






										'''simple docstring'''






										lowercase_					=      torch_device

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

										lowercase_					=      torch.manual_seed(0							)
										lowercase_					=      pipe(generator=_snake_case				,       num_inference_steps=100				,       audio_length_in_s=4.096							)
										lowercase_					=      output.audios

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

										assert audio.shape == (1, 2, pipe.unet.sample_size)
										lowercase_					=      np.array([-0.0_192, -0.0_231, -0.0_318, -0.0_059, 0.0_002, -0.0_020]							)

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







					def 	UpperCAmelCase__							(  self							: List[Any]							):






										'''simple docstring'''






										lowercase_					=      torch_device

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

										lowercase_					=      torch.manual_seed(0							)
										lowercase_					=      pipe(generator=_snake_case				,       num_inference_steps=100				,       audio_length_in_s=4.096							)
										lowercase_					=      output.audios

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

										assert audio.shape == (1, 2, pipe.unet.sample_size)
										lowercase_					=      np.array([-0.0_367, -0.0_488, -0.0_771, -0.0_525, -0.0_444, -0.0_341]							)

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









"""simple docstring"""
import argparse
import logging
import pickle
import random
import time

import numpy as np

from transformers import BertTokenizer, GPTaTokenizer, RobertaTokenizer


logging.basicConfig(
    format='''%(asctime)s - %(levelname)s - %(name)s - %(message)s''', datefmt='''%m/%d/%Y %H:%M:%S''', level=logging.INFO
)
__UpperCamelCase     :						str						       =  logging.getLogger(__name__)

def 		_SCREAMING_SNAKE_CASE			():
 lowerCAmelCase    				= argparse.ArgumentParser(
     description='Preprocess the data to avoid re-doing it several times by (tokenization + token_to_ids).'			)
 parser.add_argument('--file_path'  ,     type=_UpperCAmelCase  ,     default='data/dump.txt'  ,     help='The path to the data.'			)
 parser.add_argument('--tokenizer_type'  ,     type=_UpperCAmelCase  ,     default='bert'  ,     choices=['bert', 'roberta', 'gpt2']			)
 parser.add_argument('--tokenizer_name'  ,     type=_UpperCAmelCase  ,     default='bert-base-uncased'  ,     help='The tokenizer to use.'			)
 parser.add_argument('--dump_file'  ,     type=_UpperCAmelCase  ,     default='data/dump'  ,     help='The dump file prefix.'			)
 lowerCAmelCase    				= parser.parse_args()

 logger.info(F'Loading Tokenizer ({args.tokenizer_name})'			)
 if args.tokenizer_type == "bert":
  lowerCAmelCase    				= BertTokenizer.from_pretrained(args.tokenizer_name			)
  lowerCAmelCase    				= tokenizer.special_tokens_map['cls_token']  # `[CLS]`
  lowerCAmelCase    				= tokenizer.special_tokens_map['sep_token']  # `[SEP]`
 elif args.tokenizer_type == "roberta":
  lowerCAmelCase    				= RobertaTokenizer.from_pretrained(args.tokenizer_name			)
  lowerCAmelCase    				= tokenizer.special_tokens_map['cls_token']  # `<s>`
  lowerCAmelCase    				= tokenizer.special_tokens_map['sep_token']  # `</s>`
 elif args.tokenizer_type == "gpt2":
  lowerCAmelCase    				= GPTaTokenizer.from_pretrained(args.tokenizer_name			)
  lowerCAmelCase    				= tokenizer.special_tokens_map['bos_token']  # `<|endoftext|>`
  lowerCAmelCase    				= tokenizer.special_tokens_map['eos_token']  # `<|endoftext|>`

 logger.info(F'Loading text from {args.file_path}'			)
 with open(args.file_path  ,     'r'  ,     encoding='utf8'			) as fp:
  lowerCAmelCase    				= fp.readlines()

 logger.info('Start encoding'			)
 logger.info(F'{len(_UpperCAmelCase			)} examples to process.'			)

 lowerCAmelCase    				= []
 lowerCAmelCase    				= 0
 lowerCAmelCase    				= 1_0000
 lowerCAmelCase    				= time.time()
 for text in data:
  lowerCAmelCase    				= F'{bos} {text.strip()} {sep}'
  lowerCAmelCase    				= tokenizer.encode(_UpperCAmelCase  ,     add_special_tokens=_UpperCAmelCase			)
  rslt.append(_UpperCAmelCase			)

  iter += 1
  if iter % interval == 0:
   lowerCAmelCase    				= time.time()
   logger.info(F'{iter} examples processed. - {(end-start):.2f}s/{interval}expl'			)
   lowerCAmelCase    				= time.time()
 logger.info('Finished binarization'			)
 logger.info(F'{len(_UpperCAmelCase			)} examples processed.'			)

 lowerCAmelCase    				= F'{args.dump_file}.{args.tokenizer_name}.pickle'
 lowerCAmelCase    				= tokenizer.vocab_size
 if vocab_size < (1 << 16):
  lowerCAmelCase    				= [np.uintaa(_UpperCAmelCase			) for d in rslt]
 else:
  lowerCAmelCase    				= [np.intaa(_UpperCAmelCase			) for d in rslt]
 random.shuffle(rslt_			)
 logger.info(F'Dump to {dp_file}'			)
 with open(_UpperCAmelCase  ,     'wb'			) as handle:
  pickle.dump(rslt_  ,     _UpperCAmelCase  ,     protocol=pickle.HIGHEST_PROTOCOL			)


if __name__ == "__main__":
       main()





