OpenI
/
graphkit-learn

"""
@author: linlin
@references: Suard F, Rakotomamonjy A, Bensrhair A. Kernel on Bag of Paths For Measuring Similarity of Shapes. InESANN 2007 Apr 25 (pp. 355-360).
"""

import sys
import pathlib
sys.path.insert(0, "../")
import time
import itertools
from tqdm import tqdm

import networkx as nx
import numpy as np

from gklearn.kernels.deltaKernel import deltakernel
from gklearn.utils.graphdataset import get_dataset_attributes


def pathkernel(*args, node_label='atom', edge_label='bond_type'):
    """Calculate mean average path kernels between graphs.

    Parameters
    ----------
    Gn : List of NetworkX graph
        List of graphs between which the kernels are calculated.
    /
    G1, G2 : NetworkX graphs
        2 graphs between which the kernel is calculated.
    node_label : string
        node attribute used as label. The default node label is atom.
    edge_label : string
        edge attribute used as label. The default edge label is bond_type.

    Return
    ------
    Kmatrix/kernel : Numpy matrix/float
        Kernel matrix, each element of which is the path kernel between 2 praphs. / Path kernel between 2 graphs.
    """
    Gn = args[0] if len(args) == 1 else [args[0], args[1]]
    Kmatrix = np.zeros((len(Gn), len(Gn)))
    ds_attrs = get_dataset_attributes(
        Gn,
        attr_names=['node_labeled', 'edge_labeled', 'is_directed'],
        node_label=node_label,
        edge_label=edge_label)
    try:
        some_weight = list(nx.get_edge_attributes(Gn[0],
                                                  edge_label).values())[0]
        weight = edge_label if isinstance(some_weight, float) or isinstance(
            some_weight, int) else None
    except:
        weight = None

    start_time = time.time()

    splist = [
        get_shortest_paths(Gn[i], weight) for i in tqdm(
            range(0, len(Gn)), desc='getting shortest paths', file=sys.stdout)
    ]

    pbar = tqdm(
        total=((len(Gn) + 1) * len(Gn) / 2),
        desc='calculating kernels',
        file=sys.stdout)
    if ds_attrs['node_labeled']:
        if ds_attrs['edge_labeled']:
            for i in range(0, len(Gn)):
                for j in range(i, len(Gn)):
                    Kmatrix[i][j] = _pathkernel_do_l(Gn[i], Gn[j], splist[i],
                                                     splist[j], node_label,
                                                     edge_label)
                    Kmatrix[j][i] = Kmatrix[i][j]
                    pbar.update(1)
        else:
            for i in range(0, len(Gn)):
                for j in range(i, len(Gn)):
                    Kmatrix[i][j] = _pathkernel_do_nl(Gn[i], Gn[j], splist[i],
                                                      splist[j], node_label)
                    Kmatrix[j][i] = Kmatrix[i][j]
                    pbar.update(1)

    else:
        if ds_attrs['edge_labeled']:
            for i in range(0, len(Gn)):
                for j in range(i, len(Gn)):
                    Kmatrix[i][j] = _pathkernel_do_el(Gn[i], Gn[j], splist[i],
                                                      splist[j], edge_label)
                    Kmatrix[j][i] = Kmatrix[i][j]
                    pbar.update(1)
        else:
            for i in range(0, len(Gn)):
                for j in range(i, len(Gn)):
                    Kmatrix[i][j] = _pathkernel_do_unl(Gn[i], Gn[j], splist[i],
                                                       splist[j])
                    Kmatrix[j][i] = Kmatrix[i][j]
                    pbar.update(1)

    run_time = time.time() - start_time
    print(
        "\n --- mean average path kernel matrix of size %d built in %s seconds ---"
        % (len(Gn), run_time))

    return Kmatrix, run_time


def _pathkernel_do_l(G1, G2, sp1, sp2, node_label, edge_label):
    """Calculate mean average path kernel between 2 fully-labeled graphs.

    Parameters
    ----------
    G1, G2 : NetworkX graphs
        2 graphs between which the kernel is calculated.
    sp1, sp2 : list of list
        List of shortest paths of 2 graphs, where each path is represented by a list of nodes.
    node_label : string
        node attribute used as label. The default node label is atom.
    edge_label : string
        edge attribute used as label. The default edge label is bond_type.

    Return
    ------
    kernel : float
        Path Kernel between 2 graphs.
    """
    # calculate kernel
    kernel = 0
    #     if len(sp1) == 0 or len(sp2) == 0:
    #         return 0 # @todo: should it be zero?
    for path1 in sp1:
        for path2 in sp2:
            if len(path1) == len(path2):
                kernel_path = (G1.node[path1[0]][node_label] == G2.node[path2[
                    0]][node_label])
                if kernel_path:
                    for i in range(1, len(path1)):
                        # kernel = 1 if all corresponding nodes and edges in the 2 paths have same labels, otherwise 0
                        if G1[path1[i - 1]][path1[i]][edge_label] != G2[path2[i - 1]][path2[i]][edge_label] or G1.node[path1[i]][node_label] != G2.node[path2[i]][node_label]:
                            kernel_path = 0
                            break
                    kernel += kernel_path  # add up kernels of all paths

    kernel = kernel / (len(sp1) * len(sp2))  # calculate mean average

    return kernel


def _pathkernel_do_nl(G1, G2, sp1, sp2, node_label):
    """Calculate mean average path kernel between 2 node-labeled graphs.
    """
    # calculate kernel
    kernel = 0
    #     if len(sp1) == 0 or len(sp2) == 0:
    #         return 0 # @todo: should it be zero?
    for path1 in sp1:
        for path2 in sp2:
            if len(path1) == len(path2):
                kernel_path = 1
                for i in range(0, len(path1)):
                    # kernel = 1 if all corresponding nodes in the 2 paths have same labels, otherwise 0
                    if G1.node[path1[i]][node_label] != G2.node[path2[i]][node_label]:
                        kernel_path = 0
                        break
                kernel += kernel_path

    kernel = kernel / (len(sp1) * len(sp2))  # calculate mean average

    return kernel


def _pathkernel_do_el(G1, G2, sp1, sp2, edge_label):
    """Calculate mean average path kernel between 2 edge-labeled graphs.
    """
    # calculate kernel
    kernel = 0
    for path1 in sp1:
        for path2 in sp2:
            if len(path1) == len(path2):
                if len(path1) == 0:
                    kernel += 1
                else:
                    kernel_path = 1
                    for i in range(0, len(path1) - 1):
                        # kernel = 1 if all corresponding edges in the 2 paths have same labels, otherwise 0
                        if G1[path1[i]][path1[i + 1]][edge_label] != G2[path2[
                                i]][path2[i + 1]][edge_label]:
                            kernel_path = 0
                            break
                    kernel += kernel_path

    kernel = kernel / (len(sp1) * len(sp2))  # calculate mean average

    return kernel


def _pathkernel_do_unl(G1, G2, sp1, sp2):
    """Calculate mean average path kernel between 2 unlabeled graphs.
    """
    # calculate kernel
    kernel = 0
    for path1 in sp1:
        for path2 in sp2:
            if len(path1) == len(path2):
                kernel += 1

    kernel = kernel / (len(sp1) * len(sp2))  # calculate mean average

    return kernel


def get_shortest_paths(G, weight):
    """Get all shortest paths of a graph.

    Parameters
    ----------
    G : NetworkX graphs
        The graphs whose paths are calculated.
    weight : string/None
        edge attribute used as weight to calculate the shortest path.

    Return
    ------
    sp : list of list
        List of shortest paths of the graph, where each path is represented by a list of nodes.
    """
    sp = []
    for n1, n2 in itertools.combinations(G.nodes(), 2):
        try:
            sp.append(nx.shortest_path(G, n1, n2, weight=weight))
        except nx.NetworkXNoPath:  # nodes not connected
            sp.append([])
    # add single nodes as length 0 paths.
    sp += [[n] for n in G.nodes()]
    return sp