

'''simple docstring'''



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 		lowerCAmelCase__  ( a					,   unittest.TestCase   ):



 """simple docstring"""



 lowerCAmelCase__       = DanceDiffusionPipeline
 lowerCAmelCase__       = UNCONDITIONAL_AUDIO_GENERATION_PARAMS
 lowerCAmelCase__       = PipelineTesterMixin.required_optional_params - {
     "callback",
     "latents",
     "callback_steps",
     "output_type",
     "num_images_per_prompt",
 }
 lowerCAmelCase__       = UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
 lowerCAmelCase__       = False
 lowerCAmelCase__       = False
 def  UpperCAmelCase__						(						self   :			Optional[int]     )    ->					Optional[int]:


      """simple docstring"""




      torch.manual_seed(0     )
      __SCREAMING_SNAKE_CASE      = 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=__SCREAMING_SNAKE_CASE   ,				use_timestep_embedding=__SCREAMING_SNAKE_CASE   ,				time_embedding_type="""fourier"""   ,				mid_block_type="""UNetMidBlock1D"""   ,				down_block_types=("""DownBlock1DNoSkip""", """DownBlock1D""", """AttnDownBlock1D""")   ,				up_block_types=("""AttnUpBlock1D""", """UpBlock1D""", """UpBlock1DNoSkip""")   ,				)
      __SCREAMING_SNAKE_CASE      = IPNDMScheduler()

      __SCREAMING_SNAKE_CASE      = {
          """unet""": unet,
          """scheduler""": scheduler,
      }
      return components
 def  UpperCAmelCase__						(						self   :			Union[str, Any]   ,				__SCREAMING_SNAKE_CASE   :			List[Any]   ,				__SCREAMING_SNAKE_CASE   :			Union[str, Any]=0     )    ->					List[Any]:


      """simple docstring"""




      if str(__SCREAMING_SNAKE_CASE     ).startswith("""mps"""     ):
           __SCREAMING_SNAKE_CASE      = torch.manual_seed(__SCREAMING_SNAKE_CASE     )
      else:
           __SCREAMING_SNAKE_CASE      = torch.Generator(device=__SCREAMING_SNAKE_CASE     ).manual_seed(__SCREAMING_SNAKE_CASE     )
      __SCREAMING_SNAKE_CASE      = {
          """batch_size""": 1,
          """generator""": generator,
          """num_inference_steps""": 4,
      }
      return inputs
 def  UpperCAmelCase__						(						self   :			Tuple     )    ->					Optional[Any]:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE      = self.get_dummy_inputs(__SCREAMING_SNAKE_CASE     )
      __SCREAMING_SNAKE_CASE      = pipe(**__SCREAMING_SNAKE_CASE     )
      __SCREAMING_SNAKE_CASE      = output.audios

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

      assert audio.shape == (1, 2, components["unet"].sample_size)
      __SCREAMING_SNAKE_CASE      = np.array([-0.7265, 1.0000, -0.8388, 0.1175, 0.9498, -1.0000]     )
      assert np.abs(audio_slice.flatten() - expected_slice     ).max() < 1E-2
 @skip_mps
 def  UpperCAmelCase__						(						self   :			Tuple     )    ->					str:


      """simple docstring"""




      return super().test_save_load_local()
 @skip_mps
 def  UpperCAmelCase__						(						self   :			Union[str, Any]     )    ->					Any:


      """simple docstring"""




      return super().test_dict_tuple_outputs_equivalent(expected_max_difference=3E-3     )
 @skip_mps
 def  UpperCAmelCase__						(						self   :			Optional[Any]     )    ->					Any:


      """simple docstring"""




      return super().test_save_load_optional_components()
 @skip_mps
 def  UpperCAmelCase__						(						self   :			Optional[Any]     )    ->					Optional[int]:


      """simple docstring"""




      return super().test_attention_slicing_forward_pass()


 def  UpperCAmelCase__						(						self   :			int     )    ->					str:


      """simple docstring"""




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







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



 """simple docstring"""



 def  UpperCAmelCase__						(						self   :			Optional[Any]     )    ->					str:


      """simple docstring"""




      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()
 def  UpperCAmelCase__						(						self   :			Dict     )    ->					str:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE      = torch_device

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

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

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

      assert audio.shape == (1, 2, pipe.unet.sample_size)
      __SCREAMING_SNAKE_CASE      = np.array([-0.0192, -0.0231, -0.0318, -0.0059, 0.0002, -0.0020]     )

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


 def  UpperCAmelCase__						(						self   :			Optional[Any]     )    ->					Dict:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE      = torch_device

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

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

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

      assert audio.shape == (1, 2, pipe.unet.sample_size)
      __SCREAMING_SNAKE_CASE      = np.array([-0.0367, -0.0488, -0.0771, -0.0525, -0.0444, -0.0341]     )

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









'''simple docstring'''



import multiprocessing
import time

from arguments import PretokenizationArguments
from datasets import load_dataset

from transformers import AutoTokenizer, HfArgumentParser







def   a__			(    a__		):




     """simple docstring"""


     __SCREAMING_SNAKE_CASE      = {}
     __SCREAMING_SNAKE_CASE      = tokenizer(example["""content"""] ,       truncation=a__		)["""input_ids"""]
     __SCREAMING_SNAKE_CASE      = len(example["""content"""]		) / len(output["""input_ids"""]		)
     return output


UpperCAmelCase					:			Tuple       =	HfArgumentParser(PretokenizationArguments)
UpperCAmelCase					:			Optional[Any]       =	parser.parse_args()
if args.num_workers is None:
  UpperCAmelCase					:			List[Any]       =	multiprocessing.cpu_count()
UpperCAmelCase					:			str       =	AutoTokenizer.from_pretrained(args.tokenizer_dir)

UpperCAmelCase					:			Optional[int]       =	time.time()
UpperCAmelCase					:			Any       =	load_dataset(args.dataset_name, split='train')
print(f"""Dataset loaded in {time.time()-t_start:.2f}s""")

UpperCAmelCase					:			List[str]       =	time.time()
UpperCAmelCase					:			List[str]       =	ds.map(
    tokenize,
    num_proc=args.num_workers,
    remove_columns=[
        'repo_name',
        'path',
        'copies',
        'size',
        'content',
        'license',
        'hash',
        'line_mean',
        'line_max',
        'alpha_frac',
        'autogenerated',
    ],
)
print(f"""Dataset tokenized in {time.time()-t_start:.2f}s""")

UpperCAmelCase					:			List[Any]       =	time.time()
ds.push_to_hub(args.tokenized_data_repo)



print(f"""Data pushed to the hub in {time.time()-t_start:.2f}s""")






