


'''simple docstring'''





import gc
import unittest

import numpy as np
import torch

from diffusers import DDPMScheduler, MidiProcessor, SpectrogramDiffusionPipeline
from diffusers.pipelines.spectrogram_diffusion import SpectrogramContEncoder, SpectrogramNotesEncoder, TaFilmDecoder
from diffusers.utils import require_torch_gpu, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_note_seq, require_onnxruntime

from ..pipeline_params import TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS, TOKENS_TO_AUDIO_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()


__lowerCamelCase       				=	'''./tests/fixtures/elise_format0.mid'''




@unittest.skip("The note-seq package currently throws an error on import"	)
class 							UpperCAmelCase ( _snake_case	,   unittest.TestCase	):
 UpperCAmelCase				=				SpectrogramDiffusionPipeline
 UpperCAmelCase				=				PipelineTesterMixin.required_optional_params - {
     "callback",
     "latents",
     "callback_steps",
     "output_type",
     "num_images_per_prompt",
 }
 UpperCAmelCase				=				False
 UpperCAmelCase				=				False
 UpperCAmelCase				=				TOKENS_TO_AUDIO_GENERATION_PARAMS
 UpperCAmelCase				=				TOKENS_TO_AUDIO_GENERATION_BATCH_PARAMS


 def 					__SCREAMING_SNAKE_CASE				(			self       :  Optional[int]   ):
    torch.manual_seed(0   )
    UpperCAmelCase__  :Optional[Any]            =			SpectrogramNotesEncoder(
        max_length=2_0_4_8 ,					vocab_size=1_5_3_6 ,					d_model=7_6_8 ,					dropout_rate=0.1 ,					num_layers=1 ,					num_heads=1 ,					d_kv=4 ,					d_ff=2_0_4_8 ,					feed_forward_proj='''gated-gelu''' ,					)

    UpperCAmelCase__  :List[str]            =			SpectrogramContEncoder(
        input_dims=1_2_8 ,					targets_context_length=2_5_6 ,					d_model=7_6_8 ,					dropout_rate=0.1 ,					num_layers=1 ,					num_heads=1 ,					d_kv=4 ,					d_ff=2_0_4_8 ,					feed_forward_proj='''gated-gelu''' ,					)

    UpperCAmelCase__  :Dict            =			TaFilmDecoder(
        input_dims=1_2_8 ,					targets_length=2_5_6 ,					max_decoder_noise_time=2_0_0_0_0.0 ,					d_model=7_6_8 ,					num_layers=1 ,					num_heads=1 ,					d_kv=4 ,					d_ff=2_0_4_8 ,					dropout_rate=0.1 ,					)

    UpperCAmelCase__  :List[str]            =			DDPMScheduler()

    UpperCAmelCase__  :List[str]            =			{
        '''notes_encoder''': notes_encoder.eval(),
        '''continuous_encoder''': continuous_encoder.eval(),
        '''decoder''': decoder.eval(),
        '''scheduler''': scheduler,
        '''melgan''': None,
    }
    return components


 def 					__SCREAMING_SNAKE_CASE				(			self       :  int ,					__lowerCamelCase       :  Any ,					__lowerCamelCase       :  str=0   ):
    if str(__lowerCamelCase   ).startswith('''mps'''   ):
       UpperCAmelCase__  :List[str]            =			torch.manual_seed(__lowerCamelCase   )
    else:
       UpperCAmelCase__  :Optional[int]            =			torch.Generator(device=__lowerCamelCase   ).manual_seed(__lowerCamelCase   )
    UpperCAmelCase__  :int            =			{
        '''input_tokens''': [
            [1_1_3_4, 9_0, 1_1_3_5, 1_1_3_3, 1_0_8_0, 1_1_2, 1_1_3_2, 1_0_8_0, 1_1_3_3, 1_0_7_9, 1_3_3, 1_1_3_2, 1_0_7_9, 1_1_3_3, 1] + [0] * 2_0_3_3
        ],
        '''generator''': generator,
        '''num_inference_steps''': 4,
        '''output_type''': '''mel''',
    }
    return inputs


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

    UpperCAmelCase__  :int            =			self.get_dummy_inputs(__lowerCamelCase   )
    UpperCAmelCase__  :List[Any]            =			pipe(**__lowerCamelCase   )
    UpperCAmelCase__  :int            =			output.audios

    UpperCAmelCase__  :Optional[int]            =			mel[0, -3:, -3:]

    assert mel_slice.shape == (3, 3)
    UpperCAmelCase__  :Union[str, Any]            =			np.array(
        [-11.51_29_25, -4.78_82_15, -0.46_17_29_05, -2.05_17_15, -10.53_91_47, -10.97_09_63, -9.09_16_34, 4.0, 4.0]   )
    assert np.abs(mel_slice.flatten() - expected_slice   ).max() < 1e-2


 @skip_mps
 def 					__SCREAMING_SNAKE_CASE				(			self       :  List[str]   ):
    return super().test_save_load_local()


 @skip_mps
 def 					__SCREAMING_SNAKE_CASE				(			self       :  List[Any]   ):
    return super().test_dict_tuple_outputs_equivalent()


 @skip_mps
 def 					__SCREAMING_SNAKE_CASE				(			self       :  List[Any]   ):
    return super().test_save_load_optional_components()


 @skip_mps
 def 					__SCREAMING_SNAKE_CASE				(			self       :  Optional[int]   ):
    return super().test_attention_slicing_forward_pass()


 def 					__SCREAMING_SNAKE_CASE				(			self       :  int   ):
    pass


 def 					__SCREAMING_SNAKE_CASE				(			self       :  List[str]   ):
    pass





 @skip_mps
 def 					__SCREAMING_SNAKE_CASE				(			self       :  Optional[Any]   ):
    return super().test_progress_bar()




@slow
@require_torch_gpu
@require_onnxruntime
@require_note_seq
class 							UpperCAmelCase ( unittest.TestCase	):


 def 					__SCREAMING_SNAKE_CASE				(			self       :  str   ):
    # clean up the VRAM after each test
    super().tearDown()
    gc.collect()
    torch.cuda.empty_cache()


 def 					__SCREAMING_SNAKE_CASE				(			self       :  str   ):
    # TODO - test that pipeline can decode tokens in a callback
    # so that music can be played live
    UpperCAmelCase__  :Optional[Any]            =			torch_device

    UpperCAmelCase__  :List[Any]            =			SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''   )
    UpperCAmelCase__  :str            =			pipe.melgan
    UpperCAmelCase__  :Optional[int]            =			None

    UpperCAmelCase__  :int            =			pipe.to(__lowerCamelCase   )
    pipe.set_progress_bar_config(disable=__lowerCamelCase   )

    def callback(__lowerCamelCase       :  Optional[Any] ,					__lowerCamelCase       :  List[Any]   ):
       # decode mel to audio
       UpperCAmelCase__  :List[Any]            =			melgan(input_features=mel_output.astype(np.floataa   )   )[0]
       assert len(audio[0]   ) == 8_1_9_2_0 * (step + 1)
       # simulate that audio is played
       return audio

    UpperCAmelCase__  :int            =			MidiProcessor()
    UpperCAmelCase__  :int            =			processor(__lowerCamelCase   )

    UpperCAmelCase__  :Dict            =			input_tokens[:3]
    UpperCAmelCase__  :Optional[Any]            =			torch.manual_seed(0   )
    pipe(__lowerCamelCase ,					num_inference_steps=5 ,					generator=__lowerCamelCase ,					callback=__lowerCamelCase ,					output_type='''mel'''   )


 def 					__SCREAMING_SNAKE_CASE				(			self       :  List[Any]   ):
    UpperCAmelCase__  :Tuple            =			torch_device

    UpperCAmelCase__  :Dict            =			SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''   )
    UpperCAmelCase__  :str            =			pipe.to(__lowerCamelCase   )
    pipe.set_progress_bar_config(disable=__lowerCamelCase   )
    UpperCAmelCase__  :Dict            =			MidiProcessor()

    UpperCAmelCase__  :Any            =			processor(__lowerCamelCase   )
    # just run two denoising loops
    UpperCAmelCase__  :List[str]            =			input_tokens[:2]

    UpperCAmelCase__  :str            =			torch.manual_seed(0   )
    UpperCAmelCase__  :List[str]            =			pipe(__lowerCamelCase ,					num_inference_steps=2 ,					generator=__lowerCamelCase   )

    UpperCAmelCase__  :List[Any]            =			output.audios[0]

    assert abs(np.abs(__lowerCamelCase   ).sum() - 3_6_1_2.8_4_1   ) < 1e-1





 def 					__SCREAMING_SNAKE_CASE				(			self       :  int   ):
    UpperCAmelCase__  :List[Any]            =			torch_device

    UpperCAmelCase__  :List[Any]            =			SpectrogramDiffusionPipeline.from_pretrained('''google/music-spectrogram-diffusion'''   )
    UpperCAmelCase__  :Dict            =			pipe.to(__lowerCamelCase   )
    pipe.set_progress_bar_config(disable=__lowerCamelCase   )

    UpperCAmelCase__  :str            =			MidiProcessor()

    UpperCAmelCase__  :List[Any]            =			processor(__lowerCamelCase   )

    # just run 4 denoising loops
    UpperCAmelCase__  :Union[str, Any]            =			input_tokens[:4]

    UpperCAmelCase__  :str            =			torch.manual_seed(0   )
    UpperCAmelCase__  :List[Any]            =			pipe(__lowerCamelCase ,					num_inference_steps=1_0_0 ,					generator=__lowerCamelCase   )

    UpperCAmelCase__  :List[Any]            =			output.audios[0]
    assert abs(np.abs(__lowerCamelCase   ).sum() - 9_3_8_9.1_1_1_1   ) < 5e-2






'''simple docstring'''





__lowerCamelCase       				=	[
    ['''ab-e1''', '''ac-e3''', '''ad-e5''', '''bc-e4''', '''bd-e2''', '''be-e6''', '''bh-e12''', '''cd-e2''', '''ce-e4''',
     '''de-e1''', '''df-e8''', '''dg-e5''', '''dh-e10''', '''ef-e3''', '''eg-e2''', '''fg-e6''', '''gh-e6''', '''hi-e3'''],
    ['''ab-e1''', '''ac-e3''', '''ad-e5''', '''bc-e4''', '''bd-e2''', '''be-e6''', '''cd-e2''', '''de-e1''', '''df-e8''',
     '''ef-e3''', '''eg-e2''', '''fg-e6'''],
    ['''ab-e1''', '''ac-e3''', '''bc-e4''', '''bd-e2''', '''de-e1''', '''df-e8''', '''dg-e5''', '''ef-e3''', '''eg-e2''',
     '''eh-e12''', '''fg-e6''', '''fh-e10''', '''gh-e6'''],
    ['''ab-e1''', '''ac-e3''', '''bc-e4''', '''bd-e2''', '''bh-e12''', '''cd-e2''', '''df-e8''', '''dh-e10'''],
    ['''ab-e1''', '''ac-e3''', '''ad-e5''', '''bc-e4''', '''bd-e2''', '''cd-e2''', '''ce-e4''', '''de-e1''', '''df-e8''',
     '''dg-e5''', '''ef-e3''', '''eg-e2''', '''fg-e6''']
]


def 					a__    (				UpperCamelCase_	:			List[str]	):
   UpperCAmelCase__  :str            =			set()
   for row in edge_array:
      for item in row:
         distinct_edge.add(item[0]	)
   return list(UpperCamelCase_	)


def 					a__    (				UpperCamelCase_	:			Tuple,      UpperCamelCase_	:			List[Any]	):
   UpperCAmelCase__  :int            =			['''0'''] * len(UpperCamelCase_	)
   for i, row in enumerate(UpperCamelCase_	):
      for item in row:
         if distinct_edge in item[0]:
            UpperCAmelCase__  :List[Any]            =			'''1'''
            break
   return "".join(UpperCamelCase_	)


def 					a__    (				UpperCamelCase_	:			Optional[int]	):
   UpperCAmelCase__  :Any            =			get_distinct_edge(UpperCamelCase_	)
   UpperCAmelCase__  :Dict            =			{}

   for item in distinct_edge:
      UpperCAmelCase__  :List[str]            =			get_bitcode(UpperCamelCase_,      UpperCamelCase_	)
      # print('bit',bit)
      # bt=''.join(bit)
      UpperCAmelCase__  :int            =			bit.count('''1'''	)
      UpperCAmelCase__  :Tuple            =			[s, bit]
   # Store [Distinct edge, WT(Bitcode), Bitcode] in descending order
   UpperCAmelCase__  :str            =			[
       [k, v[0], v[1]]
       for k, v in sorted(frequency_table.items(),      key=lambda UpperCamelCase_	: v[1][0],      reverse=UpperCamelCase_	)
   ]
   return sorted_frequency_table


def 					a__    (				UpperCamelCase_	:			Optional[int]	):
   UpperCAmelCase__  :int            =			{}
   for _, item in enumerate(UpperCamelCase_	):
      nodes.setdefault(item[2],      []	).append(item[0]	)
   return nodes


def 					a__    (				UpperCamelCase_	:			Any	):
   UpperCAmelCase__  :Any            =			{}
   for key, value in nodes.items():
      UpperCAmelCase__  :Optional[int]            =			value
   return cluster


def 					a__    (				UpperCamelCase_	:			Tuple	):
   return [i * 100 / len(UpperCamelCase_	) for i in cluster]


def 					a__    (				):
   print('''\nNodes\n'''	)
   for key, value in nodes.items():
      print(UpperCamelCase_,      UpperCamelCase_	)
   print('''\nSupport\n'''	)
   print(UpperCamelCase_	)
   print('''\n Cluster \n'''	)
   for key, value in sorted(cluster.items(),      reverse=UpperCamelCase_	):
      print(UpperCamelCase_,      UpperCamelCase_	)
   print('''\n Graph\n'''	)
   for key, value in graph.items():
      print(UpperCamelCase_,      UpperCamelCase_	)
   print('''\n Edge List of Frequent subgraphs \n'''	)
   for edge_list in freq_subgraph_edge_list:
      print(UpperCamelCase_	)


def 					a__    (				UpperCamelCase_	:			Optional[int],      UpperCamelCase_	:			Any,      UpperCamelCase_	:			int,      UpperCamelCase_	:			Dict	):
   for i in cluster[ca]:
      UpperCAmelCase__  :int            =			0
      UpperCAmelCase__  :int            =			ca + 1
      while ca < max(cluster.keys()	):
         for j in cluster[ca]:
            if int(UpperCamelCase_,      2	) & int(UpperCamelCase_,      2	) == int(UpperCamelCase_,      2	):
               if tuple(nodes[i]	) in graph:
                  graph[tuple(nodes[i]	)].append(nodes[j]	)
               else:
                  UpperCAmelCase__  :int            =			[nodes[j]]
               count += 1
         if count == 0:
            UpperCAmelCase__  :List[Any]            =			ca + 1
         else:
            break


def 					a__    (				UpperCamelCase_	:			Optional[int],      UpperCamelCase_	:			List[str]	):
   UpperCAmelCase__  :List[Any]            =			cluster[max(cluster.keys()	)]
   UpperCAmelCase__  :Union[str, Any]            =			'''Header'''
   UpperCAmelCase__  :Optional[int]            =			{}
   for i in x:
      if (["Header"],) in graph:
         graph[(["Header"],)].append(x[i]	)
      else:
         UpperCAmelCase__  :Tuple            =			[x[i]]
   for i in x:
      UpperCAmelCase__  :Dict            =			[['''Header''']]
   UpperCAmelCase__  :Any            =			1
   while i < max(UpperCamelCase_	) - 1:
      create_edge(UpperCamelCase_,      UpperCamelCase_,      UpperCamelCase_,      UpperCamelCase_	)
      UpperCAmelCase__  :Union[str, Any]            =			i + 1
   return graph


def 					a__    (				UpperCamelCase_	:			Optional[Any],      UpperCamelCase_	:			int,      UpperCamelCase_	:			Any,      UpperCamelCase_	:			Dict=None	):
   UpperCAmelCase__  :Tuple            =			(path or []) + [start]
   if start == end:
      paths.append(UpperCamelCase_	)
   for node in graph[start]:
      if tuple(UpperCamelCase_	) not in path:
         my_dfs(UpperCamelCase_,      tuple(UpperCamelCase_	),      UpperCamelCase_,      UpperCamelCase_	)


def 					a__    (				UpperCamelCase_	:			Dict,      UpperCamelCase_	:			int,      UpperCamelCase_	:			List[Any]	):
   UpperCAmelCase__  :List[str]            =			int(s / 100 * (len(UpperCamelCase_	) - 1)	)
   for i in cluster[k]:
      my_dfs(UpperCamelCase_,      tuple(cluster[k][i]	),      (['''Header'''],)	)


def 					a__    (				UpperCamelCase_	:			List[Any]	):
   UpperCAmelCase__  :int            =			[]
   for edges in paths:
      UpperCAmelCase__  :List[Any]            =			[]
      for j in range(len(UpperCamelCase_	) - 1	):
         UpperCAmelCase__  :Any            =			list(edges[j]	)
         for e in temp:
            UpperCAmelCase__  :str            =			(e[0], e[1])
            el.append(UpperCamelCase_	)
      freq_sub_el.append(UpperCamelCase_	)
   return freq_sub_el


def 					a__    (				UpperCamelCase_	:			Dict	):
   for i in range(len(UpperCamelCase_	)	):
      for j in range(len(edge_array[i]	)	):
         UpperCAmelCase__  :Dict            =			edge_array[i][j].split('''-'''	)
         UpperCAmelCase__  :Optional[Any]            =			t


if __name__ == "__main__":
    preprocess(edge_array)
    __lowerCamelCase       				=	get_frequency_table(edge_array)
    __lowerCamelCase       				=	get_nodes(frequency_table)
    __lowerCamelCase       				=	get_cluster(nodes)
    __lowerCamelCase       				=	get_support(cluster)
    __lowerCamelCase       				=	construct_graph(cluster, nodes)
    find_freq_subgraph_given_support(60, cluster, graph)
    __lowerCamelCase       				=	[]
    __lowerCamelCase       				=	freq_subgraphs_edge_list(paths)
    print_all()



