"""Vertex orbit counts by EVOKE, through the _evoke_cpp extension."""
import networkx as nx
import numpy as np
from _evoke_cpp import evoke_count, evoke_count_csr
from .orbit_converter import OrbitMatrixConverter
[docs]
class EVOKEOrbitCounter:
"""Vertex orbit counts of a graph by EVOKE (graphlets of 4 or 5 vertices).
Parameters
----------
G : networkx.Graph
The graph; self-loops are ignored.
size : {4, 5}, default 4
Graphlet size: 15 orbits for 4, 73 for 5.
"""
def __init__(self, G: nx.Graph, size: int = 4):
if size not in (4, 5):
raise ValueError("EVOKE supports graphlet sizes 4 and 5.")
self.G = G
self.size = size
self.counts = None
self._converter = None
@property
def converter(self):
"""The OrbitMatrixConverter for induced counts, loaded on first use."""
if self._converter is None:
self._converter = OrbitMatrixConverter()
return self._converter
def _nx_to_cpp_adjlist(self):
"""Adjacency lists over positions in sorted node order, without self-loops, and the sorted nodes."""
nodes = sorted(self.G.nodes())
index = {node: i for i, node in enumerate(nodes)}
adj = [[index[nbr] for nbr in self.G.neighbors(node) if nbr != node] for node in nodes]
return adj, nodes
[docs]
def count_orbits(self) -> dict:
"""Count the non-induced orbits: ``{node: list of counts}``, also kept in ``self.counts``."""
adj, nodes = self._nx_to_cpp_adjlist()
rows = evoke_count(adj, size=self.size, parallel=True)
self.counts = {node: row for node, row in zip(nodes, rows)}
return self.counts
[docs]
def get_orbits(self, induced: bool = False) -> np.ndarray:
"""The orbit counts, one row per node in sorted node order.
Parameters
----------
induced : bool, default False
Induced counts instead of non-induced ones (EVOKE's own).
Returns
-------
numpy.ndarray
An ``n x 15`` (size 4) or ``n x 73`` (size 5) int64 array.
"""
if self.counts is None:
self.count_orbits()
num_orbits = 15 if self.size == 4 else 73
orbit_matrix = np.array([self.counts[node] for node in sorted(self.counts)], dtype=np.int64)
orbit_matrix = orbit_matrix.reshape(len(self.counts), num_orbits)
return self.converter.noninduced_to_induced(orbit_matrix) if induced else orbit_matrix
_SKIPPABLE = (((8,), 1), ((14,), 2), ((9, 10, 12, 13), 4))
[docs]
@staticmethod
def count_csr(offsets, nbors, size=4, orbits=None):
"""Non-induced orbit counts of a graph in CSR form.
Parameters
----------
offsets, nbors : array_like of int
The graph over vertices 0..n-1: the neighbours of vertex ``i`` are
``nbors[offsets[i]:offsets[i + 1]]``.
size : {4, 5}, default 4
Graphlet size.
orbits : iterable of int, optional
The orbits needed (None: all). At size 4, routines counting none of them are skipped, and
the orbits only they count are left at 0.
Returns
-------
numpy.ndarray
An ``n x 15`` or ``n x 73`` int64 array.
"""
if size not in (4, 5):
raise ValueError("EVOKE supports graphlet sizes 4 and 5.")
skip = 0
if orbits is not None and size == 4:
need = set(orbits)
skip = sum(bit for group, bit in EVOKEOrbitCounter._SKIPPABLE if need.isdisjoint(group))
return evoke_count_csr(np.ascontiguousarray(offsets, dtype=np.int64),
np.ascontiguousarray(nbors, dtype=np.int64), size, parallel=True, skip=skip)