"""OrbitSI's filter (FilterEngine): LDF, NLF, the orbit filter and optional arc consistency."""
from collections import Counter
import networkx as nx
import numpy as np
from ..graphindex import CandidateSubgraph, GraphIndex, gather
[docs]
class FilterEngine:
"""OrbitSI's filter: the candidate data vertices of each pattern vertex.
LDF (label and degree), then NLF (neighbour-label frequencies, on the full data graph), then one
orbit pass: the subgraph of the data graph induced by every remaining candidate is orbit-counted,
and a candidate v of pattern vertex u survives if each of v's non-induced orbit counts there is at
least u's in the pattern. Orbits that every pattern vertex has 0 of are neither compared nor,
where the counter can skip them, counted.
The data graph is processed as integer arrays (:class:`orbitsi.graphindex.GraphIndex`, built once
per graph and cached). The candidates of all pattern vertices are held together in two flat
arrays, so LDF, NLF and the orbit comparison each take a fixed number of array operations,
whatever the pattern's size. The candidate subgraph is built in CSR form and handed to the orbit
counter directly when it provides ``count_csr``.
Parameters
----------
data_graph, pattern_graph : networkx.Graph
The data graph and the pattern, with node labels in the ``'label'`` attribute.
orbit_counter_class : type
:class:`~orbitsi.orbit.EVOKEOrbitCounter` or :class:`~orbitsi.orbit.ORCAOrbitCounter`.
graphlet_size : {4, 5}, default 4
Graphlet size of the orbit pass.
arc_consistency : bool, default False
Run arc consistency after NLF (so the orbit-counted subgraph is induced by its survivors) and
again after the orbit pass: v stays a candidate of u only if, for every pattern neighbour u'
of u, some data neighbour of v is a candidate of u', repeated until nothing changes. Not in
OrbitSI as published.
contraction : bool, default True
Before the orbit pass, repeat NLF with neighbours counted inside the candidate subgraph, while
that shrinks it by at least 5% a round, so the orbits are counted on a smaller subgraph. Not
in OrbitSI as published.
edge_thinning : bool, default True
Keep in the orbit-counted subgraph only the edges some pattern edge can map onto. Not in
OrbitSI as published.
"""
def __init__(self, data_graph, pattern_graph, orbit_counter_class, graphlet_size=4, arc_consistency=False,
contraction=True, edge_thinning=True):
self.data_graph = data_graph
self.edge_thinning = edge_thinning
self.arc_consistency = arc_consistency
self.contraction = contraction
self.pattern_graph = pattern_graph
self.graphlet_size = graphlet_size
self.orbit_counter_class = orbit_counter_class
self.pattern_labels = nx.get_node_attributes(pattern_graph, "label")
self.index = GraphIndex.of(data_graph)
self._pu = list(pattern_graph.nodes)
self.pattern_orbits = self._compute_orbits(pattern_graph, orbit_counter_class)
def _compute_orbits(self, graph, counter_class):
"""Non-induced orbit vector per node."""
count_csr = getattr(counter_class, "count_csr", None)
if count_csr is not None:
pix = GraphIndex.of(graph)
orbit_matrix = count_csr(pix.offsets, pix.nbors, self.graphlet_size)
return dict(zip(pix.nodes, orbit_matrix.tolist()))
counter = counter_class(graph, size=self.graphlet_size)
orbit_matrix = counter.get_orbits(induced=False)
return {node: orbit_matrix[i].tolist() for i, node in enumerate(sorted(graph.nodes()))}
def _split(self, flat):
"""{pattern vertex: its candidates (array of data vertex indices)}."""
cu, cv = flat
b = np.searchsorted(cu, np.arange(len(self._pu) + 1)).tolist()
return {u: cv[b[i]:b[i + 1]] for i, u in enumerate(self._pu)}
def _join(self, cands):
"""The flat arrays of {pattern vertex: candidate array}."""
parts = [cands[u] for u in self._pu]
lens = np.fromiter(map(len, parts), dtype=np.int64, count=len(parts))
cv = np.concatenate(parts).astype(np.int64, copy=False) if parts else np.zeros(0, dtype=np.int64)
return np.repeat(np.arange(len(parts), dtype=np.int64), lens), cv
def _check(self, cu, cv):
"""(cu, cv), or None if some pattern vertex has no candidate left."""
k = len(self._pu)
if len(cu) == 0 and k:
return None
return (cu, cv) if np.bincount(cu, minlength=k).all() else None
def _ldf(self):
"""LDF: per pattern vertex, the data vertices with its label and at least its degree; None if some set is empty."""
ix, P = self.index, self.pattern_graph
parts, memo = [], {}
for u in self._pu:
code = ix.label_code.get(self.pattern_labels.get(u))
if code is None:
return None
key = (code, P.degree[u])
rows = memo.get(key)
if rows is None:
cnt = int(np.searchsorted(ix.neg_deg[code], -key[1], side="right"))
if cnt == 0:
return None
rows = memo[key] = np.sort(ix.by_label_deg[code][:cnt])
parts.append(rows)
if not parts:
return np.zeros(0, dtype=np.int64), np.zeros(0, dtype=np.int64)
lens = np.fromiter(map(len, parts), dtype=np.int64, count=len(parts))
return np.repeat(np.arange(len(parts), dtype=np.int64), lens), np.concatenate(parts)
def _ldf_nlf(self):
"""LDF then NLF (as _ldf and _nlf), by data label: the vertices of a label, by decreasing degree,
get one column of neighbour counts per label some pattern vertex of that label requires; a
pattern vertex's LDF candidates are a prefix (degree >= its own), and its NLF checks are
comparisons of that prefix of its required columns."""
ix, P = self.index, self.pattern_graph
req = self._requirements()
if req is None:
return None
codes, counts = req
if counts.size and counts.max() > ix.NLF_CAP:
raise ValueError("pattern vertex degree above GraphIndex.NLF_CAP")
groups = {}
for i, u in enumerate(self._pu):
code = ix.label_code.get(self.pattern_labels.get(u))
if code is None:
return None
groups.setdefault(code, []).append(i)
parts = [None] * len(self._pu)
for g, members in groups.items():
byd, negd = ix.by_label_deg[g], ix.neg_deg[g]
cnt = {i: int(np.searchsorted(negd, -P.degree[self._pu[i]], side="right")) for i in members}
if min(cnt.values()) == 0:
return None
top = byd[:max(cnt.values())]
cols = {}
for i in members:
ok = None
for j in range(codes.shape[1]):
c = int(counts[i, j])
if c == 0:
break
code = int(codes[i, j])
col = cols.get(code)
if col is None:
col = cols[code] = ix.nlf_count(top, np.full(len(top), code, dtype=np.int64))
t = col[:cnt[i]] >= c
ok = t if ok is None else ok & t
rows = byd[:cnt[i]] if ok is None else byd[:cnt[i]][ok]
if len(rows) == 0:
return None
parts[i] = np.sort(rows)
lens = np.fromiter(map(len, parts), dtype=np.int64, count=len(parts))
cv = np.concatenate(parts) if parts else np.zeros(0, dtype=np.int64)
return np.repeat(np.arange(len(parts), dtype=np.int64), lens), cv
def _requirements(self):
"""Per pattern vertex (row), its neighbour-label requirements, most demanding first: label code
and count, padded with count 0; or None if some required label is not in the data graph."""
if not hasattr(self, "_req"):
self._req = self._build_requirements()
return self._req
def _build_requirements(self):
"""Computes _requirements: per pattern vertex, (label code, count) pairs, most demanding first."""
ix, P, plab = self.index, self.pattern_graph, self.pattern_labels
reqs = []
for u in self._pu:
r = []
for lab, c in Counter(plab[w] for w in P.neighbors(u)).items():
code = ix.label_code.get(lab)
if code is None:
return None
r.append((c, code))
r.sort(reverse=True)
reqs.append(r)
R = max(map(len, reqs), default=0)
codes = np.zeros((len(reqs), R), dtype=np.int64)
counts = np.zeros((len(reqs), R), dtype=np.int64)
for i, r in enumerate(reqs):
for j, (c, code) in enumerate(r):
counts[i, j], codes[i, j] = c, code
return codes, counts
def _nlf(self, flat, lookup=None):
"""v stays a candidate of u if, for every label, v has at least as many neighbours with it as u
has: lookup(vertices, label codes) gives the counts (default GraphIndex.nlf_count, the whole
data graph). One requirement per round, each round over every pattern vertex's survivors."""
req = self._requirements()
if req is None:
return None
codes, counts = req
if counts.size and counts.max() > self.index.NLF_CAP:
raise ValueError("pattern vertex degree above GraphIndex.NLF_CAP")
lookup = lookup or self.index.nlf_count
cu, cv = flat
for j in range(codes.shape[1]):
need = counts[cu, j]
sel = np.flatnonzero(need > 0)
if len(sel) == 0:
continue
bad = sel[lookup(cv[sel], codes[cu[sel], j]) < need[sel]]
if len(bad):
keep = np.ones(len(cu), dtype=bool)
keep[bad] = False
cu, cv = cu[keep], cv[keep]
return self._check(cu, cv)
def _contract(self, flat, min_shrink=0.05):
"""NLF again with neighbours counted inside the candidate subgraph (the subgraph induced by all
candidates), which only shrinks as candidates go: an embedding maps u's neighbours to distinct
candidates adjacent to u's image. Repeated while the subgraph loses at least min_shrink of its
vertices per round (any number of rounds is exact)."""
ix = self.index
req = self._requirements()
if req is None:
return None
codes = np.unique(req[0][req[1] > 0])
if len(codes) == 0:
return flat
K = len(codes)
col = np.full(len(ix.label_code), -1, dtype=np.int64)
col[codes] = np.arange(K)
local = np.full(ix.n, -1, dtype=np.int64)
member = np.zeros(ix.n, dtype=bool)
member[flat[1]] = True
S = np.flatnonzero(member)
owner, nb = gather(ix.offsets, ix.nbors, S)
keep = member[nb]
owner, nb = owner[keep], nb[keep]
c = col[ix.labels[nb]]
keep = c >= 0
src, dst, dcol = S[owner[keep]], nb[keep], c[keep]
del owner, nb, c, keep
while True:
local[S] = np.arange(len(S))
cnt = np.bincount(local[src] * K + dcol, minlength=len(S) * K)
before = len(flat[0])
flat = self._nlf(flat, lambda v, code: cnt[local[v] * K + col[code]])
local[S] = -1
if flat is None or len(flat[0]) == before:
return flat
member[:] = False
member[flat[1]] = True
S_next = np.flatnonzero(member)
if len(S_next) > (1 - min_shrink) * len(S):
return flat
S = S_next
keep = member[src] & member[dst]
src, dst, dcol = src[keep], dst[keep], dcol[keep]
def _arc(self, cands):
"""Arc consistency to a fixpoint (see the class docstring), by support counting (AC-4): for
every pattern edge (u, w) and candidate v of u, sup[u, w][v] counts v's data neighbours among
w's candidates. Supports are counted once; when candidates of w are removed, only the supports
of their own data neighbours drop, and a candidate whose support reaches 0 is removed in turn.
The work is one scan of the candidates' neighbourhoods plus the edges of removed vertices.
Returns None if a candidate set empties."""
ix, P = self.index, self.pattern_graph
nbrs = {u: [w for w in P.neighbors(u) if w != u] for u in P.nodes}
rows0 = cands
alive = {u: np.ones(len(r), dtype=bool) for u, r in rows0.items()}
member = {}
for u, r in rows0.items():
m = np.zeros(ix.n, dtype=bool)
m[r] = True
member[u] = m
sup = {}
for u in P.nodes:
if not nbrs[u]:
continue
owner, nb = gather(ix.offsets, ix.nbors, rows0[u])
for w in nbrs[u]:
sup[u, w] = np.bincount(owner[member[w][nb]], minlength=len(rows0[u]))
pending = {}
for u in P.nodes:
if not nbrs[u]:
continue
dead = np.zeros(len(rows0[u]), dtype=bool)
for w in nbrs[u]:
dead |= sup[u, w] == 0
if dead.any():
alive[u] &= ~dead
if not alive[u].any():
return None
member[u][rows0[u][dead]] = False
pending[u] = [rows0[u][dead]]
while pending:
u, removed = pending.popitem()
_, nb = gather(ix.offsets, ix.nbors, np.concatenate(removed))
for x in nbrs[u]:
sel = nb[member[x][nb]]
if len(sel) == 0:
continue
s = sup[x, u]
s -= np.bincount(np.searchsorted(rows0[x], sel), minlength=len(s))
newly = alive[x] & (s <= 0)
if newly.any():
alive[x] &= ~newly
if not alive[x].any():
return None
gone = rows0[x][newly]
member[x][gone] = False
pending.setdefault(x, []).append(gone)
return {u: rows0[u][alive[u]] for u in rows0}
def _orbit(self, flat):
"""The orbit pass: count orbits on the candidate subgraph (edge-thinned if set) and keep the candidates whose counts reach their pattern vertex's in every orbit some pattern vertex has. Returns (flat, subgraph); flat is None if some set empties."""
ix = self.index
cu, cv = flat
member = np.zeros(ix.n, dtype=bool)
member[cv] = True
sub = CandidateSubgraph(ix, np.flatnonzero(member))
count_csr = getattr(self.orbit_counter_class, "count_csr", None)
offsets, nbors, local = sub.csr()
if self.edge_thinning:
offsets, nbors = self._usable_edges(offsets, nbors, cu, local[cv], sub)
pat = np.array([self.pattern_orbits[u] for u in self._pu], dtype=np.int64).reshape(len(self._pu), -1)
need = np.flatnonzero(pat.max(axis=0, initial=0) > 0) if len(self._pu) else np.zeros(0, dtype=np.int64)
if count_csr is not None:
orbits = count_csr(offsets, nbors, self.graphlet_size, orbits=need.tolist())
else:
G = nx.Graph()
G.add_nodes_from(sub.nodes)
ids = np.array(sub.nodes, dtype=object)
G.add_edges_from(zip(ids[np.repeat(np.arange(len(ids)), np.diff(offsets))].tolist(), ids[nbors].tolist()))
order = np.argsort(np.argsort(ids))
orbits = self.orbit_counter_class(G, size=self.graphlet_size).get_orbits(induced=False)[order]
lv = local[cv]
if len(need) == 0 or len(cu) == 0:
return self._check(cu, cv), sub
vecs, pid = np.unique(pat[:, need], axis=0, return_inverse=True)
pid = pid.reshape(-1)
if len(sub.rows) * len(vecs) <= 4 * len(cu):
dominates = (orbits[:, need][:, None, :] >= vecs[None, :, :]).all(axis=2)
keep = dominates[lv, pid[cu]]
cu, cv = cu[keep], cv[keep]
else:
for j in need.tolist():
keep = orbits[lv, j] >= pat[cu, j]
if not keep.all():
cu, cv, lv = cu[keep], cv[keep], lv[keep]
return self._check(cu, cv), sub
EDGE_MASK_MAX = 64
def _usable_edges(self, offsets, nbors, cu, lv, sub):
"""The candidate subgraph's CSR (local vertices) without the edges no pattern edge can map onto:
(a, b) stays only if some pattern edge (u, w) has a a candidate of u and b one of w (with at
most EDGE_MASK_MAX pattern vertices; else if some pattern edge joins their labels). Every
embedding uses only kept edges, so a true candidate's orbit counts there still reach its
pattern vertex's, and the counts -- and their cost -- only drop."""
P, k, n = self.pattern_graph, len(self._pu), len(offsets) - 1
if len(nbors) == 0 or k == 0:
return offsets, nbors
pos = {u: i for i, u in enumerate(self._pu)}
edges = [(pos[u], pos[w]) for u, w in P.edges if u != w]
own = np.repeat(np.arange(n, dtype=np.int64), np.diff(offsets))
if k <= self.EDGE_MASK_MAX:
adj = np.zeros(k, dtype=np.uint64)
for i, j in edges:
adj[i] |= np.uint64(1) << np.uint64(j)
adj[j] |= np.uint64(1) << np.uint64(i)
mine = np.zeros(n, dtype=np.uint64)
reach = np.zeros(n, dtype=np.uint64)
np.bitwise_or.at(mine, lv, np.left_shift(np.uint64(1), cu.astype(np.uint64)))
np.bitwise_or.at(reach, lv, adj[cu])
keep = (reach[own] & mine[nbors]) != 0
else:
ix = self.index
codes = [ix.label_code.get(self.pattern_labels.get(u)) for u in self._pu]
distinct = sorted(set(codes))
col = np.full(len(ix.label_code), -1, dtype=np.int64)
col[distinct] = np.arange(len(distinct))
ok = np.zeros((len(distinct), len(distinct)), dtype=bool)
for i, j in edges:
a, b = col[codes[i]], col[codes[j]]
ok[a, b] = ok[b, a] = True
lab = col[ix.labels[sub.rows]]
keep = ok[lab[own], lab[nbors]]
if keep.all():
return offsets, nbors
new = np.zeros(n + 1, dtype=np.int64)
np.cumsum(np.bincount(own[keep], minlength=n), out=new[1:])
return new, nbors[keep]
def _ids(self, flat):
"""{pattern vertex: candidate node IDs} from the flat arrays."""
return {u: self.index.to_ids(rows) for u, rows in self._split(flat).items()}
def _rows(self, candidate_sets):
"""Flat arrays of {pattern vertex: candidate node IDs}, or None if some set is empty."""
to = self.index.to_index
cands = {u: np.array(sorted(to(v) for v in candidate_sets.get(u, ())), dtype=np.int64) for u in self._pu}
return self._check(*self._join(cands))
[docs]
def run(self):
"""Run the whole filter.
Returns
-------
pattern_orbits : dict
``{pattern node: orbit vector}``.
candidate_sets : dict
``{pattern node: candidate data node IDs}``, in data-graph node order.
subgraph : orbitsi.graphindex.CandidateSubgraph or None
The candidate-induced subgraph the orbits were counted on.
If some pattern vertex has no candidate left, returns ``({}, {}, None)``.
"""
flat = self._ldf_nlf()
if flat is not None and self.arc_consistency:
flat = self._arc_flat(flat)
if flat is not None and self.contraction:
flat = self._contract(flat)
if flat is None:
return {}, {}, None
flat, subgraph = self._orbit(flat)
if flat is not None and self.arc_consistency:
flat = self._arc_flat(flat)
if flat is None:
return {}, {}, None
return self.pattern_orbits, self._ids(flat), subgraph
def _arc_flat(self, flat):
"""Arc consistency on the flat arrays; None if some set empties."""
cands = self._arc(self._split(flat))
return None if cands is None else self._join(cands)
[docs]
def ldf_filter(self):
"""LDF alone: ``{pattern node: candidates}``, or ``{}`` if some pattern vertex has none."""
flat = self._ldf()
return {} if flat is None else self._ids(flat)
[docs]
def nlf_filter(self, candidate_sets):
"""NLF on the full data graph, applied to ``candidate_sets``; ``{}`` if some set empties."""
flat = self._rows(candidate_sets)
flat = None if flat is None else self._nlf(flat)
return {} if flat is None else self._ids(flat)
[docs]
def arc_consistency_filter(self, candidate_sets):
"""Arc consistency to a fixpoint, applied to ``candidate_sets``; ``{}`` if some set empties."""
flat = self._rows(candidate_sets)
flat = None if flat is None else self._arc_flat(flat)
return {} if flat is None else self._ids(flat)
[docs]
def orbit_filter(self, candidate_sets):
"""The orbit pass on the subgraph induced by ``candidate_sets``.
Returns
-------
candidate_sets : dict
The survivors, or ``{}`` if some set empties.
subgraph : orbitsi.graphindex.CandidateSubgraph or None
The subgraph the orbits were counted on.
"""
flat = self._rows(candidate_sets)
if flat is None:
return {}, None
flat, sub = self._orbit(flat)
return ({} if flat is None else self._ids(flat)), sub
[docs]
def orbit_filter_full(self, candidate_sets):
"""The orbit pass with orbits counted on the whole data graph instead (not used by :meth:`run`)."""
ix = self.index
flat = self._rows(candidate_sets)
if flat is None:
return {}
sub = CandidateSubgraph(ix, np.arange(ix.n, dtype=np.int64))
offsets, nbors, local = sub.csr()
count_csr = getattr(self.orbit_counter_class, "count_csr", None)
if count_csr is not None:
orbits = count_csr(offsets, nbors, self.graphlet_size)
else:
order = np.argsort(np.argsort(np.array(sub.nodes, dtype=object)))
orbits = self.orbit_counter_class(self.data_graph, size=self.graphlet_size).get_orbits()[order]
cu, cv = flat
pat = np.array([self.pattern_orbits[u] for u in self._pu], dtype=np.int64).reshape(len(self._pu), -1)
keep = (orbits[cv] >= pat[cu]).all(axis=1)
flat = self._check(cu[keep], cv[keep])
return {} if flat is None else self._ids(flat)
[docs]
def printCandidateSets(self, candidate_sets):
"""Print each pattern vertex's candidates."""
if not candidate_sets:
print("No match possible after filtering.")
else:
print("\n=== Candidate Sets ===")
for u, cands in candidate_sets.items():
print(f"Pattern Node {u} → Candidates: {cands}")