Repository navigation
Expand file tree
/
Copy pathgraph_analyzer.py
More file actions
271 lines (204 loc) · 9.06 KB
/
Copy pathgraph_analyzer.py
File metadata and controls
271 lines (204 loc) · 9.06 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
import sys
from collections import OrderedDict
from inspect import cleandoc
from itertools import chain, repeat
from typing import List, Tuple
import mgp
try:
import networkx as nx
except ImportError as import_error:
sys.stderr.write(
(
"\n"
"NOTE: Please install networkx to be able to use graph_analyzer "
"module. Using Python:\n" + sys.version + "\n"
)
)
raise import_error
# Imported last because it also depends on networkx.
from mgp_networkx import MemgraphMultiDiGraph # noqa E402
_MAX_LIST_SIZE = 10
@mgp.read_proc
def help() -> mgp.Record(name=str, value=str):
"""Shows manual page for graph_analyzer."""
records = []
def make_records(name, doc):
return (mgp.Record(name=n, value=v) for n, v in zip(chain([name], repeat("")), cleandoc(doc).splitlines()))
for func in (help, analyze, analyze_subgraph):
records.extend(make_records("Procedure '{}'".format(func.__name__), func.__doc__))
for m, v in _get_analysis_mapping().items():
records.extend(make_records("Analysis '{}'".format(m), v.__doc__))
return records
@mgp.read_proc
def analyze(context: mgp.ProcCtx, analyses: mgp.Nullable[List[str]] = None) -> mgp.Record(name=str, value=str):
"""
Shows graph information.
In case of multiple results, only the first 10 will be shown.
The optional parameter is a list of graph analyses to run.
If NULL, all available analyses are run.
Example call (give all information):
CALL graph_analyzer.analyze() YIELD *;
Example call (with parameter):
CALL graph_analyzer.analyze(['nodes', 'edges']) YIELD *;
"""
g = MemgraphMultiDiGraph(ctx=context)
recs = _analyze_graph(context, g, analyses)
return [mgp.Record(name=name, value=value) for name, value in recs]
@mgp.read_proc
def analyze_subgraph(
context: mgp.ProcCtx,
vertices: mgp.List[mgp.Vertex],
edges: mgp.List[mgp.Edge],
analyses: mgp.Nullable[List[str]] = None,
) -> mgp.Record(name=str, value=str):
"""
Shows subgraph information.
In case of multiple results, only the first 10 will be shown.
The optional parameter is a list of graph analyses to run.
If NULL, all available analyses are run.
Example call (give all information):
MATCH (n)-[e]->(m) WITH
collect(n) AS nodes,
collect(e) AS edges
CALL graph_analyzer.analyze_subgraph(nodes, edges) YIELD *
RETURN name, value;
Example call (with parameter):
MATCH (n)-[e]->(m) WITH
collect(n) AS nodes,
collect(e) AS edges
CALL graph_analyzer.analyze_subgraph(nodes, edges, ['nodes', 'edges'])
YIELD *
RETURN name, value;
"""
vertices, edges = map(set, [vertices, edges])
g = nx.subgraph_view(
MemgraphMultiDiGraph(ctx=context),
lambda n: n in vertices,
lambda n1, n2, e: e in edges,
)
recs = _analyze_graph(context, g, analyses)
return [mgp.Record(name=name, value=value) for name, value in recs]
def _get_analysis_mapping():
return OrderedDict(
[
("nodes", _number_of_nodes),
("edges", _number_of_edges),
("bridges", _bridges),
("articulation_points", _articulation_points),
("avg_degree", _avg_degree),
("sorted_nodes_degree", _sorted_nodes_degree),
("self_loops", _self_loops),
("is_bipartite", _is_bipartite),
("is_planar", _is_planar),
("is_biconnected: ", _is_biconnected),
("is_weakly_connected", _is_weakly_connected),
("number_of_weakly_components", _weakly_components),
("is_strongly_connected", _is_strongly_connected),
("strongly_components", _strongly_components),
("is_dag", _is_dag),
("is_eulerian", _is_eulerian),
("is_forest", _is_forest),
("is_tree", _is_tree),
]
)
def _get_analysis_func(name: str):
_name_to_proc = _get_analysis_mapping()
return _name_to_proc.get(name.lower())
def _get_analysis_funcs():
return _get_analysis_mapping().values()
def _analyze_graph(context: mgp.ProcCtx, g: nx.MultiDiGraph, analyses: List[str]) -> List[Tuple[str, str]]:
functions = _get_analysis_funcs() if analyses is None else [_get_analysis_func(name) for name in analyses]
records = []
for index, f in enumerate(functions):
context.check_must_abort()
if f is None:
raise KeyError("Graph analysis is not supported: " + analyses[index])
name, value = f(g)
if isinstance(value, (list, set, tuple)):
value = list(value)[:_MAX_LIST_SIZE]
records.append((name, str(value)))
return records
def _number_of_nodes(g: nx.MultiDiGraph) -> Tuple[str, int]:
"""Returns number of nodes."""
return "Number of nodes", nx.number_of_nodes(g)
def _number_of_edges(g: nx.MultiDiGraph) -> Tuple[str, int]:
"""Returns number of edges."""
return "Number of edges", nx.number_of_edges(g)
def _avg_degree(g: nx.MultiDiGraph) -> Tuple[str, float]:
"""Returns average degree."""
_, number_of_nodes = _number_of_nodes(g)
_, number_of_edges = _number_of_edges(g)
avg_degree = 0 if number_of_nodes == 0 else number_of_edges / number_of_nodes
return "Average degree", avg_degree
def _sorted_nodes_degree(g: nx.MultiDiGraph) -> Tuple[str, List[int]]:
"""Returns list of sorted nodes degree. [(node_id, degree), ...]"""
nodes_degree = [(n, g.degree(n)) for n in g.nodes()]
nodes_degree.sort(key=lambda x: x[1], reverse=True)
return "Sorted nodes degree", nodes_degree
def _self_loops(g: nx.MultiDiGraph) -> Tuple[str, int]:
"""Returns number of self loops."""
return "Self loops", sum((1 if e[0] == e[1] else 0 for e in g.edges()))
def _is_bipartite(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Checks if graph is bipartite."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.algorithms.bipartite.basic.is_bipartite(g)
return "Is bipartite", ret
def _is_planar(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Checks if graph is planar."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.algorithms.planarity.check_planarity(g)[0]
return "Is planar", ret
def _is_biconnected(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Check if graph is biconnected."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.is_biconnected(nx.MultiDiGraph.to_undirected(g))
return "Is biconnected", ret
def _is_weakly_connected(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Check if graph is weakly connected."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.is_weakly_connected(g)
return "Is weakly connected", ret
def _is_strongly_connected(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Checks if graph is strongly connected."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.is_strongly_connected(g)
return "Is strongly connected", ret
def _is_dag(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Check if graph is directed acyclic graph (DAG)"""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.algorithms.dag.is_directed_acyclic_graph(g)
return "Is DAG", ret
def _is_eulerian(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Checks if graph is Eulerian."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.algorithms.euler.is_eulerian(g)
return "Is eulerian", ret
def _is_forest(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Checks if graph is forest, all components must be trees."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.algorithms.tree.recognition.is_forest(g)
return "Is forest", ret
def _is_tree(g: nx.MultiDiGraph) -> Tuple[str, bool]:
"""Checks if graph is tree."""
_, number_of_nodes = _number_of_nodes(g)
ret = False if number_of_nodes == 0 else nx.algorithms.tree.recognition.is_tree(g)
return "Is tree", ret
def _bridges(g: nx.MultiDiGraph) -> Tuple[str, int]:
"""Returns number of bridges, multiple edges between same nodes are
mapped to one edge."""
return "Number of bridges", sum(1 for _ in nx.bridges(nx.Graph(g)))
def _articulation_points(g: nx.MultiDiGraph):
"""Returns number of articulation points."""
undirected = nx.MultiDiGraph.to_undirected(g)
return (
"Number of articulation points",
sum(1 for _ in nx.articulation_points(undirected)),
)
def _weakly_components(g: nx.MultiDiGraph):
"""Returns number of weakly components."""
comps = nx.algorithms.components.number_weakly_connected_components(g)
return "Number of weakly connected components", comps
def _strongly_components(g: nx.MultiDiGraph):
"""Returns number of strongly connected components."""
comps = nx.algorithms.components.number_strongly_connected_components(g)
return "Number of strongly connected components", comps