


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   __lowercase				(  __snake_case ,					unittest.TestCase  ):
      UpperCamelCase								=       DanceDiffusionPipeline
      UpperCamelCase								=       UNCONDITIONAL_AUDIO_GENERATION_PARAMS
      UpperCamelCase								=       PipelineTesterMixin.required_optional_params - {
          '''callback''',
          '''latents''',
          '''callback_steps''',
          '''output_type''',
          '''num_images_per_prompt''',
      }
      UpperCamelCase								=       UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
      UpperCamelCase								=       False
      UpperCamelCase								=       False





      def        _lowercase				(		self						:					List[Any]       )    ->   int:






         """simple docstring"""







         torch.manual_seed(0       )
         UpperCAmelCase							=     UNetaDModel(
             block_out_channels=(3_2, 3_2, 6_4)					,       extra_in_channels=1_6					,       sample_size=5_1_2					,       sample_rate=1_6_0_0_0					,       in_channels=2					,       out_channels=2					,       flip_sin_to_cos=__lowerCamelCase					,       use_timestep_embedding=__lowerCamelCase					,       time_embedding_type="""fourier"""					,       mid_block_type="""UNetMidBlock1D"""					,       down_block_types=("""DownBlock1DNoSkip""", """DownBlock1D""", """AttnDownBlock1D""")					,       up_block_types=("""AttnUpBlock1D""", """UpBlock1D""", """UpBlock1DNoSkip""")					,       )
         UpperCAmelCase							=     IPNDMScheduler()

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





      def        _lowercase				(		self						:					Any					,       __lowerCamelCase						:					List[str]					,       __lowerCamelCase						:					Optional[Any]=0       )    ->   Union[str, Any]:






         """simple docstring"""







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





      def        _lowercase				(		self						:					str       )    ->   List[str]:






         """simple docstring"""







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

         UpperCAmelCase							=     self.get_dummy_inputs(__lowerCamelCase       )
         UpperCAmelCase							=     pipe(**__lowerCamelCase       )
         UpperCAmelCase							=     output.audios

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

         assert audio.shape == (1, 2, components["unet"].sample_size)
         UpperCAmelCase							=     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        _lowercase				(		self						:					Optional[Any]       )    ->   Optional[Any]:






         """simple docstring"""







         return super().test_save_load_local()





      @skip_mps
      def        _lowercase				(		self						:					str       )    ->   Any:






         """simple docstring"""







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





      @skip_mps
      def        _lowercase				(		self						:					List[Any]       )    ->   Optional[int]:






         """simple docstring"""







         return super().test_save_load_optional_components()





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






         """simple docstring"""







         return super().test_attention_slicing_forward_pass()





      def        _lowercase				(		self						:					Union[str, Any]       )    ->   Optional[Any]:






         """simple docstring"""







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






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





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






         """simple docstring"""







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





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






         """simple docstring"""







         UpperCAmelCase							=     torch_device

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

         UpperCAmelCase							=     torch.manual_seed(0       )
         UpperCAmelCase							=     pipe(generator=__lowerCamelCase					,       num_inference_steps=1_0_0					,       audio_length_in_s=4.096       )
         UpperCAmelCase							=     output.audios

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

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         UpperCAmelCase							=     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        _lowercase				(		self						:					int       )    ->   Union[str, Any]:






         """simple docstring"""







         UpperCAmelCase							=     torch_device

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

         UpperCAmelCase							=     torch.manual_seed(0       )
         UpperCAmelCase							=     pipe(generator=__lowerCamelCase					,       num_inference_steps=1_0_0					,       audio_length_in_s=4.096       )
         UpperCAmelCase							=     output.audios

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

         assert audio.shape == (1, 2, pipe.unet.sample_size)
         UpperCAmelCase							=     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







import multiprocessing
import time

from arguments import PretokenizationArguments
from datasets import load_dataset

from transformers import AutoTokenizer, HfArgumentParser
def   _UpperCamelCase				(						lowerCAmelCase_   )						->int:
   UpperCAmelCase							=     {}
   UpperCAmelCase							=     tokenizer(example["""content"""]  ,						truncation=lowerCAmelCase_   )["""input_ids"""]
   UpperCAmelCase							=     len(example["""content"""]   ) / len(output["""input_ids"""]   )
   return output


__a     =			HfArgumentParser(PretokenizationArguments)
__a     =			parser.parse_args()
if args.num_workers is None:
 __a     =			multiprocessing.cpu_count()
__a     =			AutoTokenizer.from_pretrained(args.tokenizer_dir)

__a     =			time.time()
__a     =			load_dataset(args.dataset_name, split="""train""")
print(F"""Dataset loaded in {time.time()-t_start:.2f}s""")

__a     =			time.time()
__a     =			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""")

__a     =			time.time()
ds.push_to_hub(args.tokenized_data_repo)







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



