


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):
      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  _snake_case					(	self						:   Any						)      ->int:






         """simple docstring"""






         torch.manual_seed(0						)
         a__ :List[Any]						  =    UNetaDModel(
             block_out_channels=(32, 32, 64)   ,     extra_in_channels=16   ,     sample_size=512   ,     sample_rate=16000   ,     in_channels=2   ,     out_channels=2   ,     flip_sin_to_cos=_A   ,     use_timestep_embedding=_A   ,     time_embedding_type="fourier"   ,     mid_block_type="UNetMidBlock1D"   ,     down_block_types=("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")   ,     up_block_types=("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")   ,     )
         a__ :int						  =    IPNDMScheduler()

         a__ :Union[str, Any]						  =    {
             'unet': unet,
             'scheduler': scheduler,
         }
         return components




      def  _snake_case					(	self						:   int   ,     __A						:   Dict   ,     __A						:   Dict=0						)      ->List[str]:






         """simple docstring"""






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




      def  _snake_case					(	self						:   str						)      ->Dict:






         """simple docstring"""






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

         a__ :Union[str, Any]						  =    self.get_dummy_inputs(_A						)
         a__ :List[str]						  =    pipe(**_A						)
         a__ :List[Any]						  =    output.audios

         a__ :List[str]						  =    audio[0, -3:, -3:]

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






         """simple docstring"""






         return super().test_save_load_local()




      @skip_mps
      def  _snake_case					(	self						:   Union[str, Any]						)      ->Optional[int]:






         """simple docstring"""






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




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






         """simple docstring"""






         return super().test_save_load_optional_components()




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






         """simple docstring"""






         return super().test_attention_slicing_forward_pass()






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






         """simple docstring"""






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






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




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






         """simple docstring"""






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




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






         """simple docstring"""






         a__ :List[Any]						  =    torch_device

         a__ :int						  =    DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"						)
         a__ :int						  =    pipe.to(_A						)
         pipe.set_progress_bar_config(disable=_A						)

         a__ :Optional[int]						  =    torch.manual_seed(0						)
         a__ :str						  =    pipe(generator=_A   ,     num_inference_steps=100   ,     audio_length_in_s=4.096						)
         a__ :str						  =    output.audios

         a__ :List[str]						  =    audio[0, -3:, -3:]

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






         """simple docstring"""






         a__ :str						  =    torch_device

         a__ :Tuple						  =    DanceDiffusionPipeline.from_pretrained("harmonai/maestro-150k"   ,     torch_dtype=torch.floataa						)
         a__ :Optional[int]						  =    pipe.to(_A						)
         pipe.set_progress_bar_config(disable=_A						)

         a__ :Union[str, Any]						  =    torch.manual_seed(0						)
         a__ :Optional[int]						  =    pipe(generator=_A   ,     num_inference_steps=100   ,     audio_length_in_s=4.096						)
         a__ :Union[str, Any]						  =    output.audios

         a__ :int						  =    audio[0, -3:, -3:]

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






def 				lowerCamelCase__	(      a	:   list[list]				)    ->	list[list]:





			"""simple docstring"""
			a__ :Any						  =    current_set.copy()
			for row_index, row in enumerate(a				):
						a__ :List[str]						  =    row[0]
						for column_index, column in enumerate(a				):
									if magnitude == 0:
												a__ :Tuple						  =    column
												continue
									a__ :Tuple						  =    column / magnitude
    # Subtract to cancel term
			a__ :Any						  =    current_set[0]
			a__ :Dict						  =    [first_row]
			a__ :Dict						  =    current_set[1::]
			for row in current_set:
						a__ :int						  =    []
						# If first term is 0, it is already in form we want, so we preserve it
						if row[0] == 0:
									final_set.append(a				)
									continue
						for column_index in range(len(a				)				):
									temp_row.append(first_row[column_index] - row[column_index]				)
						final_set.append(a				)
			# Create next recursion iteration set
			if len(final_set[0]				) != 3:
						a__ :List[str]						  =    final_set[0]
						a__ :Optional[Any]						  =    []
						a__ :List[Any]						  =    []
						for row in final_set[1::]:
									current_first_column.append(row[0]				)
									next_iteration.append(row[1::]				)
						a__ :Dict						  =    simplify(a				)
						for i in range(len(a				)				):
									resultant[i].insert(0	,  current_first_column[i]				)
						resultant.insert(0	,  a				)
						a__ :Optional[int]						  =    resultant
			return final_set


def 				lowerCamelCase__	(      a	:   list[list]				)    ->	list:





			"""simple docstring"""
			if len(a				) == 0:
						raise IndexError("solve_simultaneous() requires n lists of length n+1"				)
			a__ :Any						  =    len(a				) + 1
			if any(len(a				) != _length for item in equations				):
						raise IndexError("solve_simultaneous() requires n lists of length n+1"				)
			for row in equations:
						if any(not isinstance(a	,  (int, float)				) for column in row				):
									raise ValueError("solve_simultaneous() requires lists of integers"				)
			if len(a				) == 1:
						return [equations[0][-1] / equations[0][0]]
			a__ :Union[str, Any]						  =    equations.copy()
			if any(0 in row for row in data_set				):
						a__ :int						  =    data_set.copy()
						a__ :Dict						  =    []
						for row_index, row in enumerate(a				):
									if 0 not in row:
												a__ :Union[str, Any]						  =    data_set.pop(a				)
												break
						if not full_row:
									raise ValueError("solve_simultaneous() requires at least 1 full equation"				)
						data_set.insert(0	,  a				)
			a__ :Optional[int]						  =    data_set.copy()
			a__ :Any						  =    simplify(a				)
			a__ :str						  =    simplified[::-1]
			a__ :list						  =    []
			for row in simplified:
						a__ :Optional[int]						  =    row[-1]
						if not solutions:
									if row[-2] == 0:
												solutions.append(0				)
												continue
									solutions.append(current_solution / row[-2]				)
									continue
						a__ :Dict						  =    row.copy()[: len(a				) - 1 :]
						while temp_row[0] == 0:
									temp_row.pop(0				)
						if len(a				) == 0:
									solutions.append(0				)
									continue
						a__ :List[Any]						  =    temp_row[1::]
						a__ :Union[str, Any]						  =    temp_row[::-1]
						for column_index, column in enumerate(a				):
									current_solution -= column * solutions[column_index]
						solutions.append(a				)
			a__ :List[str]						  =    []
			for item in solutions:
						final.append(float(round(a	,  5				)				)				)
			return final[::-1]


if __name__ == "__main__":
		import doctest

		doctest.testmod()
		snake_case__			 =						[
		    [2, 1, 1, 1, 1, 4],
		    [1, 2, 1, 1, 1, 5],
		    [1, 1, 2, 1, 1, 6],
		    [1, 1, 1, 2, 1, 7],
		    [1, 1, 1, 1, 2, 8],
		]
		print(solve_simultaneous(eq))
		print(solve_simultaneous([[4, 2]]))


