From 79cb3a5fcf9e868531fc674a2d5136c9c7c85a2d Mon Sep 17 00:00:00 2001 From: jajupmochi Date: Sun, 20 Dec 2020 14:54:09 +0100 Subject: [PATCH] Add test of equality of graph kernel results for different parameters. --- gklearn/tests/test_graph_kernels.py | 659 ++++++++++++++++++++---------------- 1 file changed, 358 insertions(+), 301 deletions(-) diff --git a/gklearn/tests/test_graph_kernels.py b/gklearn/tests/test_graph_kernels.py index c2ff6ec..40f1d51 100644 --- a/gklearn/tests/test_graph_kernels.py +++ b/gklearn/tests/test_graph_kernels.py @@ -5,6 +5,16 @@ import pytest import multiprocessing import numpy as np +############################################################################## + +def test_list_graph_kernels(): + """ + """ + from gklearn.kernels import GRAPH_KERNELS, list_of_graph_kernels + assert list_of_graph_kernels() == [i for i in GRAPH_KERNELS] + + +############################################################################## def chooseDataset(ds_name): """Choose dataset according to name. @@ -54,201 +64,252 @@ def chooseDataset(ds_name): return dataset -def test_list_graph_kernels(): - """ +def assert_equality(compute_fun, **kwargs): + """Check if outputs are the same using different methods to compute. + + Parameters + ---------- + compute_fun : function + The function to compute the kernel, with the same key word arguments as + kwargs. + **kwargs : dict + The key word arguments over the grid of which the kernel results are + compared. + + Returns + ------- + None. """ - from gklearn.kernels import GRAPH_KERNELS, list_of_graph_kernels - assert list_of_graph_kernels() == [i for i in GRAPH_KERNELS] + from sklearn.model_selection import ParameterGrid + param_grid = ParameterGrid(kwargs) + result_lists = [[], [], []] + for params in list(param_grid): + results = compute_fun(**params) + for rs, lst in zip(results, result_lists): + lst.append(rs) + for lst in result_lists: + for i in range(len(lst[:-1])): + assert np.array_equal(lst[i], lst[i + 1]) @pytest.mark.parametrize('ds_name', ['Alkane', 'AIDS']) @pytest.mark.parametrize('weight,compute_method', [(0.01, 'geo'), (1, 'exp')]) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_CommonWalk(ds_name, parallel, weight, compute_method): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_CommonWalk(ds_name, weight, compute_method): """Test common walk kernel. """ - from gklearn.kernels import CommonWalk - import networkx as nx - - dataset = chooseDataset(ds_name) - dataset.load_graphs([g for g in dataset.graphs if nx.number_of_nodes(g) > 1]) - - try: - graph_kernel = CommonWalk(node_labels=dataset.node_labels, - edge_labels=dataset.edge_labels, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - weight=weight, - compute_method=compute_method) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - except Exception as exception: - assert False, exception + def compute(parallel=None): + from gklearn.kernels import CommonWalk + import networkx as nx + + dataset = chooseDataset(ds_name) + dataset.load_graphs([g for g in dataset.graphs if nx.number_of_nodes(g) > 1]) + + try: + graph_kernel = CommonWalk(node_labels=dataset.node_labels, + edge_labels=dataset.edge_labels, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + weight=weight, + compute_method=compute_method) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None]) @pytest.mark.parametrize('ds_name', ['Alkane', 'AIDS']) @pytest.mark.parametrize('remove_totters', [False]) #[True, False]) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_Marginalized(ds_name, parallel, remove_totters): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_Marginalized(ds_name, remove_totters): """Test marginalized kernel. """ - from gklearn.kernels import Marginalized + def compute(parallel=None): + from gklearn.kernels import Marginalized - dataset = chooseDataset(ds_name) + dataset = chooseDataset(ds_name) - try: - graph_kernel = Marginalized(node_labels=dataset.node_labels, - edge_labels=dataset.edge_labels, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - p_quit=0.5, - n_iteration=2, - remove_totters=remove_totters) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + try: + graph_kernel = Marginalized(node_labels=dataset.node_labels, + edge_labels=dataset.edge_labels, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + p_quit=0.5, + n_iteration=2, + remove_totters=remove_totters) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - except Exception as exception: - assert False, exception + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None]) @pytest.mark.parametrize('ds_name', ['Acyclic']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_SylvesterEquation(ds_name, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_SylvesterEquation(ds_name): """Test sylvester equation kernel. """ - from gklearn.kernels import SylvesterEquation + def compute(parallel=None): + from gklearn.kernels import SylvesterEquation + + dataset = chooseDataset(ds_name) - dataset = chooseDataset(ds_name) + try: + graph_kernel = SylvesterEquation( + ds_infos=dataset.get_dataset_infos(keys=['directed']), + weight=1e-3, + p=None, + q=None, + edge_weight=None) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - try: - graph_kernel = SylvesterEquation( - ds_infos=dataset.get_dataset_infos(keys=['directed']), - weight=1e-3, - p=None, - q=None, - edge_weight=None) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel - except Exception as exception: - assert False, exception + assert_equality(compute, parallel=['imap_unordered', None]) @pytest.mark.parametrize('ds_name', ['Acyclic', 'AIDS']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_ConjugateGradient(ds_name, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_ConjugateGradient(ds_name): """Test conjugate gradient kernel. """ - from gklearn.kernels import ConjugateGradient - from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct - import functools - - dataset = chooseDataset(ds_name) - - mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) - sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} - - try: - graph_kernel = ConjugateGradient( - node_labels=dataset.node_labels, - node_attrs=dataset.node_attrs, - edge_labels=dataset.edge_labels, - edge_attrs=dataset.edge_attrs, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - weight=1e-3, - p=None, - q=None, - edge_weight=None, - node_kernels=sub_kernels, - edge_kernels=sub_kernels) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - except Exception as exception: - assert False, exception + def compute(parallel=None): + from gklearn.kernels import ConjugateGradient + from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct + import functools + + dataset = chooseDataset(ds_name) + + mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) + sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} + + try: + graph_kernel = ConjugateGradient( + node_labels=dataset.node_labels, + node_attrs=dataset.node_attrs, + edge_labels=dataset.edge_labels, + edge_attrs=dataset.edge_attrs, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + weight=1e-3, + p=None, + q=None, + edge_weight=None, + node_kernels=sub_kernels, + edge_kernels=sub_kernels) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None]) @pytest.mark.parametrize('ds_name', ['Acyclic', 'AIDS']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_FixedPoint(ds_name, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_FixedPoint(ds_name): """Test fixed point kernel. """ - from gklearn.kernels import FixedPoint - from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct - import functools - - dataset = chooseDataset(ds_name) - - mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) - sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} - - try: - graph_kernel = FixedPoint( - node_labels=dataset.node_labels, - node_attrs=dataset.node_attrs, - edge_labels=dataset.edge_labels, - edge_attrs=dataset.edge_attrs, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - weight=1e-3, - p=None, - q=None, - edge_weight=None, - node_kernels=sub_kernels, - edge_kernels=sub_kernels) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - except Exception as exception: - assert False, exception + def compute(parallel=None): + from gklearn.kernels import FixedPoint + from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct + import functools + + dataset = chooseDataset(ds_name) + + mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) + sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} + + try: + graph_kernel = FixedPoint( + node_labels=dataset.node_labels, + node_attrs=dataset.node_attrs, + edge_labels=dataset.edge_labels, + edge_attrs=dataset.edge_attrs, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + weight=1e-3, + p=None, + q=None, + edge_weight=None, + node_kernels=sub_kernels, + edge_kernels=sub_kernels) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None]) @pytest.mark.parametrize('ds_name', ['Acyclic']) @pytest.mark.parametrize('sub_kernel', ['exp', 'geo']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_SpectralDecomposition(ds_name, sub_kernel, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_SpectralDecomposition(ds_name, sub_kernel): """Test spectral decomposition kernel. """ - from gklearn.kernels import SpectralDecomposition - - dataset = chooseDataset(ds_name) - - try: - graph_kernel = SpectralDecomposition( - ds_infos=dataset.get_dataset_infos(keys=['directed']), - weight=1e-3, - p=None, - q=None, - edge_weight=None, - sub_kernel=sub_kernel) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - except Exception as exception: - assert False, exception + def compute(parallel=None): + from gklearn.kernels import SpectralDecomposition + + dataset = chooseDataset(ds_name) + + try: + graph_kernel = SpectralDecomposition( + ds_infos=dataset.get_dataset_infos(keys=['directed']), + weight=1e-3, + p=None, + q=None, + edge_weight=None, + sub_kernel=sub_kernel) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None]) # @pytest.mark.parametrize( @@ -301,184 +362,179 @@ def test_SpectralDecomposition(ds_name, sub_kernel, parallel): @pytest.mark.parametrize('ds_name', ['Alkane', 'Acyclic', 'Letter-med', 'AIDS', 'Fingerprint']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_ShortestPath(ds_name, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_ShortestPath(ds_name): """Test shortest path kernel. """ - from gklearn.kernels import ShortestPath - from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct - import functools - - dataset = chooseDataset(ds_name) - - mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) - sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} - try: - graph_kernel = ShortestPath(node_labels=dataset.node_labels, - node_attrs=dataset.node_attrs, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - fcsp=True, - node_kernels=sub_kernels) - gram_matrix1, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - graph_kernel = ShortestPath(node_labels=dataset.node_labels, - node_attrs=dataset.node_attrs, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - fcsp=False, - node_kernels=sub_kernels) - gram_matrix2, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - except Exception as exception: - assert False, exception - - assert np.array_equal(gram_matrix1, gram_matrix2) + def compute(parallel=None, fcsp=None): + from gklearn.kernels import ShortestPath + from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct + import functools + + dataset = chooseDataset(ds_name) + + mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) + sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} + try: + graph_kernel = ShortestPath(node_labels=dataset.node_labels, + node_attrs=dataset.node_attrs, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + fcsp=fcsp, + node_kernels=sub_kernels) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None], fcsp=[True, False]) #@pytest.mark.parametrize('ds_name', ['Alkane', 'Acyclic', 'Letter-med', 'AIDS', 'Fingerprint']) @pytest.mark.parametrize('ds_name', ['Alkane', 'Acyclic', 'Letter-med', 'AIDS', 'Fingerprint', 'Fingerprint_edge', 'Cuneiform']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_StructuralSP(ds_name, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_StructuralSP(ds_name): """Test structural shortest path kernel. """ - from gklearn.kernels import StructuralSP - from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct - import functools - - dataset = chooseDataset(ds_name) - - mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) - sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} - try: - graph_kernel = StructuralSP(node_labels=dataset.node_labels, - edge_labels=dataset.edge_labels, - node_attrs=dataset.node_attrs, - edge_attrs=dataset.edge_attrs, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - fcsp=True, - node_kernels=sub_kernels, - edge_kernels=sub_kernels) - gram_matrix1, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True, normalize=False) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - graph_kernel = StructuralSP(node_labels=dataset.node_labels, - edge_labels=dataset.edge_labels, - node_attrs=dataset.node_attrs, - edge_attrs=dataset.edge_attrs, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - fcsp=False, - node_kernels=sub_kernels, - edge_kernels=sub_kernels) - gram_matrix2, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True, normalize=False) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - - except Exception as exception: - assert False, exception - - assert np.array_equal(gram_matrix1, gram_matrix2) + def compute(parallel=None, fcsp=None): + from gklearn.kernels import StructuralSP + from gklearn.utils.kernels import deltakernel, gaussiankernel, kernelproduct + import functools + + dataset = chooseDataset(ds_name) + + mixkernel = functools.partial(kernelproduct, deltakernel, gaussiankernel) + sub_kernels = {'symb': deltakernel, 'nsymb': gaussiankernel, 'mix': mixkernel} + try: + graph_kernel = StructuralSP(node_labels=dataset.node_labels, + edge_labels=dataset.edge_labels, + node_attrs=dataset.node_attrs, + edge_attrs=dataset.edge_attrs, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + fcsp=fcsp, + node_kernels=sub_kernels, + edge_kernels=sub_kernels) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None], fcsp=[True, False]) @pytest.mark.parametrize('ds_name', ['Alkane', 'AIDS']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) #@pytest.mark.parametrize('k_func', ['MinMax', 'tanimoto', None]) @pytest.mark.parametrize('k_func', ['MinMax', 'tanimoto']) -@pytest.mark.parametrize('compute_method', ['trie', 'naive']) -def test_PathUpToH(ds_name, parallel, k_func, compute_method): +# @pytest.mark.parametrize('compute_method', ['trie', 'naive']) +def test_PathUpToH(ds_name, k_func): """Test path kernel up to length $h$. """ - from gklearn.kernels import PathUpToH - - dataset = chooseDataset(ds_name) - - try: - graph_kernel = PathUpToH(node_labels=dataset.node_labels, - edge_labels=dataset.edge_labels, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - depth=2, k_func=k_func, compute_method=compute_method) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - except Exception as exception: - assert False, exception + def compute(parallel=None, compute_method=None): + from gklearn.kernels import PathUpToH + + dataset = chooseDataset(ds_name) + + try: + graph_kernel = PathUpToH(node_labels=dataset.node_labels, + edge_labels=dataset.edge_labels, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + depth=2, k_func=k_func, compute_method=compute_method) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None], + compute_method=['trie', 'naive']) @pytest.mark.parametrize('ds_name', ['Alkane', 'AIDS']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_Treelet(ds_name, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_Treelet(ds_name): """Test treelet kernel. """ - from gklearn.kernels import Treelet - from gklearn.utils.kernels import polynomialkernel - import functools - - dataset = chooseDataset(ds_name) - - pkernel = functools.partial(polynomialkernel, d=2, c=1e5) - try: - graph_kernel = Treelet(node_labels=dataset.node_labels, - edge_labels=dataset.edge_labels, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - sub_kernel=pkernel) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - except Exception as exception: - assert False, exception + def compute(parallel=None): + from gklearn.kernels import Treelet + from gklearn.utils.kernels import polynomialkernel + import functools + + dataset = chooseDataset(ds_name) + + pkernel = functools.partial(polynomialkernel, d=2, c=1e5) + try: + graph_kernel = Treelet(node_labels=dataset.node_labels, + edge_labels=dataset.edge_labels, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + sub_kernel=pkernel) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None]) @pytest.mark.parametrize('ds_name', ['Acyclic']) #@pytest.mark.parametrize('base_kernel', ['subtree', 'sp', 'edge']) # @pytest.mark.parametrize('base_kernel', ['subtree']) -@pytest.mark.parametrize('parallel', ['imap_unordered', None]) -def test_WLSubtree(ds_name, parallel): +# @pytest.mark.parametrize('parallel', ['imap_unordered', None]) +def test_WLSubtree(ds_name): """Test Weisfeiler-Lehman subtree kernel. """ - from gklearn.kernels import WLSubtree - - dataset = chooseDataset(ds_name) - - try: - graph_kernel = WLSubtree(node_labels=dataset.node_labels, - edge_labels=dataset.edge_labels, - ds_infos=dataset.get_dataset_infos(keys=['directed']), - height=2) - gram_matrix, run_time = graph_kernel.compute(dataset.graphs, - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], - parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) - except Exception as exception: - assert False, exception + def compute(parallel=None): + from gklearn.kernels import WLSubtree + + dataset = chooseDataset(ds_name) + + try: + graph_kernel = WLSubtree(node_labels=dataset.node_labels, + edge_labels=dataset.edge_labels, + ds_infos=dataset.get_dataset_infos(keys=['directed']), + height=2) + gram_matrix, run_time = graph_kernel.compute(dataset.graphs, + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel_list, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1:], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + kernel, run_time = graph_kernel.compute(dataset.graphs[0], dataset.graphs[1], + parallel=parallel, n_jobs=multiprocessing.cpu_count(), verbose=True) + except Exception as exception: + assert False, exception + else: + return gram_matrix, kernel_list, kernel + + assert_equality(compute, parallel=['imap_unordered', None]) if __name__ == "__main__": - test_list_graph_kernels() + test_list_graph_kernels() # test_spkernel('Alkane', 'imap_unordered') -# test_ShortestPath('Alkane', 'imap_unordered') + # test_ShortestPath('Alkane') # test_StructuralSP('Fingerprint_edge', 'imap_unordered') # test_StructuralSP('Alkane', None) # test_StructuralSP('Cuneiform', None) @@ -487,3 +543,4 @@ if __name__ == "__main__": # test_RandomWalk('Acyclic', 'conjugate', None, 'imap_unordered') # test_RandomWalk('Acyclic', 'fp', None, None) # test_RandomWalk('Acyclic', 'spectral', 'exp', 'imap_unordered') +# test_CommonWalk('Alkane', 0.01, 'geo')