'''simple docstring'''






from collections.abc import Generator
from math import sin





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			bytes				):
     if len(__snake_case				) != 3_2:
          raise ValueError('Input must be of length 32'				)

     _A    		=  B''
     for i in [3, 2, 1, 0]:
          little_endian += string_aa[8 * i : 8 * i + 8]
     return little_endian





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			int				):
     if i < 0:
          raise ValueError('Input must be non-negative'				)

     _A    		=  format(__snake_case   ,		'08x'				)[-8:]
     _A    		=  B''
     for i in [3, 2, 1, 0]:
          little_endian_hex += hex_rep[2 * i : 2 * i + 2].encode('utf-8'				)
     return little_endian_hex





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			bytes				):
     _A    		=  B''
     for char in message:
          bit_string += format(__snake_case   ,		'08b'				).encode('utf-8'				)
     _A    		=  format(len(__snake_case				)   ,		'064b'				).encode('utf-8'				)

     # Pad bit_string to a multiple of 512 chars
     bit_string += b"1"
     while len(__snake_case				) % 5_1_2 != 4_4_8:
          bit_string += b"0"
     bit_string += to_little_endian(start_len[3_2:]				) + to_little_endian(start_len[:3_2]				)

     return bit_string





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			bytes				):
     if len(__snake_case				) % 5_1_2 != 0:
          raise ValueError('Input must have length that\'s a multiple of 512'				)

     for pos in range(0   ,		len(__snake_case				)   ,		5_1_2				):
          _A    		=  bit_string[pos : pos + 5_1_2]
          _A    		=  []
          for i in range(0   ,		5_1_2   ,		3_2				):
               block_words.append(int(to_little_endian(block[i : i + 3_2]				)   ,		2				)				)
          yield block_words





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			int				):
     if i < 0:
          raise ValueError('Input must be non-negative'				)

     _A    		=  format(__snake_case   ,		'032b'				)
     _A    		=  ''
     for c in i_str:
          new_str += "1" if c == "0" else "0"
     return int(__snake_case   ,		2				)





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			int   ,		__snake_case							:			int				):
     return (a + b) % 2**3_2





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			int   ,		__snake_case							:			int				):
     if i < 0:
          raise ValueError('Input must be non-negative'				)
     if shift < 0:
          raise ValueError('Shift must be non-negative'				)
     return ((i << shift) ^ (i >> (3_2 - shift))) % 2**3_2





def  _SCREAMING_SNAKE_CASE		(      __snake_case							:			bytes				):
     _A    		=  preprocess(__snake_case				)

     _A    		=  [int(2**3_2 * abs(sin(i + 1				)				)				) for i in range(6_4				)]

     # Starting states
     _A    		=  0X67_452_301
     _A    		=  0XEF_CDA_B89
     _A    		=  0X98_BAD_CFE
     _A    		=  0X10_325_476

     _A    		=  [
         7,
         1_2,
         1_7,
         2_2,
         7,
         1_2,
         1_7,
         2_2,
         7,
         1_2,
         1_7,
         2_2,
         7,
         1_2,
         1_7,
         2_2,
         5,
         9,
         1_4,
         2_0,
         5,
         9,
         1_4,
         2_0,
         5,
         9,
         1_4,
         2_0,
         5,
         9,
         1_4,
         2_0,
         4,
         1_1,
         1_6,
         2_3,
         4,
         1_1,
         1_6,
         2_3,
         4,
         1_1,
         1_6,
         2_3,
         4,
         1_1,
         1_6,
         2_3,
         6,
         1_0,
         1_5,
         2_1,
         6,
         1_0,
         1_5,
         2_1,
         6,
         1_0,
         1_5,
         2_1,
         6,
         1_0,
         1_5,
         2_1,
     ]

     # Process bit string in chunks, each with 16 32-char words
     for block_words in get_block_words(__snake_case				):
          _A    		=  aa
          _A    		=  ba
          _A    		=  ca
          _A    		=  da

          # Hash current chunk
          for i in range(6_4				):
               if i <= 1_5:
                    # f = (b & c) | (not_32(b) & d)     # Alternate definition for f
                    _A    		=  d ^ (b & (c ^ d))
                    _A    		=  i
               elif i <= 3_1:
                    # f = (d & b) | (not_32(d) & c)     # Alternate definition for f
                    _A    		=  c ^ (d & (b ^ c))
                    _A    		=  (5 * i + 1) % 1_6
               elif i <= 4_7:
                    _A    		=  b ^ c ^ d
                    _A    		=  (3 * i + 5) % 1_6
               else:
                    _A    		=  c ^ (b | not_aa(__snake_case				))
                    _A    		=  (7 * i) % 1_6
               _A    		=  (f + a + added_consts[i] + block_words[g]) % 2**3_2
               _A    		=  d
               _A    		=  c
               _A    		=  b
               _A    		=  sum_aa(__snake_case   ,		left_rotate_aa(__snake_case   ,		shift_amounts[i]				)				)

          # Add hashed chunk to running total
          _A    		=  sum_aa(__snake_case   ,		__snake_case				)
          _A    		=  sum_aa(__snake_case   ,		__snake_case				)
          _A    		=  sum_aa(__snake_case   ,		__snake_case				)
          _A    		=  sum_aa(__snake_case   ,		__snake_case				)

     _A    		=  reformat_hex(__snake_case				) + reformat_hex(__snake_case				) + reformat_hex(__snake_case				) + reformat_hex(__snake_case				)
     return digest


if __name__ == "__main__":
   import doctest

   doctest.testmod()





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





     """simple docstring"""







     SCREAMING_SNAKE_CASE_		 =	DanceDiffusionPipeline
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE_		 =	PipelineTesterMixin.required_optional_params - {
         '''callback''',
         '''latents''',
         '''callback_steps''',
         '''output_type''',
         '''num_images_per_prompt''',
     }
     SCREAMING_SNAKE_CASE_		 =	UNCONDITIONAL_AUDIO_GENERATION_BATCH_PARAMS
     SCREAMING_SNAKE_CASE_		 =	False
     SCREAMING_SNAKE_CASE_		 =	False






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Any:




           """simple docstring"""







           torch.manual_seed(0	)
           UpperCamelCase__     : List[str]											=					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=__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''')    ,    )
           UpperCamelCase__     : Union[str, Any]											=					IPNDMScheduler()

           UpperCamelCase__     : List[str]											=					{
               '''unet''': unet,
               '''scheduler''': scheduler,
           }
           return components






     def  __SCREAMING_SNAKE_CASE      (				self    ,    __SCREAMING_SNAKE_CASE    ,    __SCREAMING_SNAKE_CASE=0	)	-> Any:




           """simple docstring"""







           if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 UpperCamelCase__     : Optional[Any]											=					torch.manual_seed(__SCREAMING_SNAKE_CASE	)
           else:
                 UpperCamelCase__     : Optional[Any]											=					torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : int											=					{
               '''batch_size''': 1,
               '''generator''': generator,
               '''num_inference_steps''': 4,
           }
           return inputs






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




           """simple docstring"""







           UpperCamelCase__     : str											=					'''cpu'''  # ensure determinism for the device-dependent torch.Generator
           UpperCamelCase__     : Dict											=					self.get_dummy_components()
           UpperCamelCase__     : str											=					DanceDiffusionPipeline(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[Any]											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					self.get_dummy_inputs(__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : Dict											=					pipe(**__SCREAMING_SNAKE_CASE	)
           UpperCamelCase__     : List[str]											=					output.audios

           UpperCamelCase__     : Dict											=					audio[0, -3:, -3:]

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




           """simple docstring"""







           return super().test_save_load_local()






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







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






     @skip_mps
     def  __SCREAMING_SNAKE_CASE      (				self	)	-> str:




           """simple docstring"""







           return super().test_save_load_optional_components()






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




           """simple docstring"""







           return super().test_attention_slicing_forward_pass()






     def  __SCREAMING_SNAKE_CASE      (				self	)	-> Tuple:




           """simple docstring"""







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




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





     """simple docstring"""







     def  __SCREAMING_SNAKE_CASE      (				self	)	-> int:




           """simple docstring"""







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






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




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Any											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''	)
           UpperCamelCase__     : int											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Optional[Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : str											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

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




           """simple docstring"""







           UpperCamelCase__     : Any											=					torch_device

           UpperCamelCase__     : Union[str, Any]											=					DanceDiffusionPipeline.from_pretrained('''harmonai/maestro-150k'''    ,    torch_dtype=torch.floataa	)
           UpperCamelCase__     : Tuple											=					pipe.to(__SCREAMING_SNAKE_CASE	)
           pipe.set_progress_bar_config(disable=__SCREAMING_SNAKE_CASE	)

           UpperCamelCase__     : int											=					torch.manual_seed(0	)
           UpperCamelCase__     : Union[str, Any]											=					pipe(generator=__SCREAMING_SNAKE_CASE    ,    num_inference_steps=1_0_0    ,    audio_length_in_s=4.096	)
           UpperCamelCase__     : List[Any]											=					output.audios

           UpperCamelCase__     : List[Any]											=					audio[0, -3:, -3:]

           assert audio.shape == (1, 2, pipe.unet.sample_size)
           UpperCamelCase__     : Optional[Any]											=					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


