Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.9/dist-packages/networkx/algorithms/isomorphism/temporalisomorphvf2.py: 22%
93 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-10-20 07:00 +0000
1"""
2*****************************
3Time-respecting VF2 Algorithm
4*****************************
6An extension of the VF2 algorithm for time-respecting graph isomorphism
7testing in temporal graphs.
9A temporal graph is one in which edges contain a datetime attribute,
10denoting when interaction occurred between the incident nodes. A
11time-respecting subgraph of a temporal graph is a subgraph such that
12all interactions incident to a node occurred within a time threshold,
13delta, of each other. A directed time-respecting subgraph has the
14added constraint that incoming interactions to a node must precede
15outgoing interactions from the same node - this enforces a sense of
16directed flow.
18Introduction
19------------
21The TimeRespectingGraphMatcher and TimeRespectingDiGraphMatcher
22extend the GraphMatcher and DiGraphMatcher classes, respectively,
23to include temporal constraints on matches. This is achieved through
24a semantic check, via the semantic_feasibility() function.
26As well as including G1 (the graph in which to seek embeddings) and
27G2 (the subgraph structure of interest), the name of the temporal
28attribute on the edges and the time threshold, delta, must be supplied
29as arguments to the matching constructors.
31A delta of zero is the strictest temporal constraint on the match -
32only embeddings in which all interactions occur at the same time will
33be returned. A delta of one day will allow embeddings in which
34adjacent interactions occur up to a day apart.
36Examples
37--------
39Examples will be provided when the datetime type has been incorporated.
42Temporal Subgraph Isomorphism
43-----------------------------
45A brief discussion of the somewhat diverse current literature will be
46included here.
48References
49----------
51[1] Redmond, U. and Cunningham, P. Temporal subgraph isomorphism. In:
52The 2013 IEEE/ACM International Conference on Advances in Social
53Networks Analysis and Mining (ASONAM). Niagara Falls, Canada; 2013:
54pages 1451 - 1452. [65]
56For a discussion of the literature on temporal networks:
58[3] P. Holme and J. Saramaki. Temporal networks. Physics Reports,
59519(3):97–125, 2012.
61Notes
62-----
64Handles directed and undirected graphs and graphs with parallel edges.
66"""
68import networkx as nx
70from .isomorphvf2 import DiGraphMatcher, GraphMatcher
72__all__ = ["TimeRespectingGraphMatcher", "TimeRespectingDiGraphMatcher"]
75class TimeRespectingGraphMatcher(GraphMatcher):
76 def __init__(self, G1, G2, temporal_attribute_name, delta):
77 """Initialize TimeRespectingGraphMatcher.
79 G1 and G2 should be nx.Graph or nx.MultiGraph instances.
81 Examples
82 --------
83 To create a TimeRespectingGraphMatcher which checks for
84 syntactic and semantic feasibility:
86 >>> from networkx.algorithms import isomorphism
87 >>> from datetime import timedelta
88 >>> G1 = nx.Graph(nx.path_graph(4, create_using=nx.Graph()))
90 >>> G2 = nx.Graph(nx.path_graph(4, create_using=nx.Graph()))
92 >>> GM = isomorphism.TimeRespectingGraphMatcher(
93 ... G1, G2, "date", timedelta(days=1)
94 ... )
95 """
96 self.temporal_attribute_name = temporal_attribute_name
97 self.delta = delta
98 super().__init__(G1, G2)
100 def one_hop(self, Gx, Gx_node, neighbors):
101 """
102 Edges one hop out from a node in the mapping should be
103 time-respecting with respect to each other.
104 """
105 dates = []
106 for n in neighbors:
107 if isinstance(Gx, nx.Graph): # Graph G[u][v] returns the data dictionary.
108 dates.append(Gx[Gx_node][n][self.temporal_attribute_name])
109 else: # MultiGraph G[u][v] returns a dictionary of key -> data dictionary.
110 for edge in Gx[Gx_node][
111 n
112 ].values(): # Iterates all edges between node pair.
113 dates.append(edge[self.temporal_attribute_name])
114 if any(x is None for x in dates):
115 raise ValueError("Datetime not supplied for at least one edge.")
116 return not dates or max(dates) - min(dates) <= self.delta
118 def two_hop(self, Gx, core_x, Gx_node, neighbors):
119 """
120 Paths of length 2 from Gx_node should be time-respecting.
121 """
122 return all(
123 self.one_hop(Gx, v, [n for n in Gx[v] if n in core_x] + [Gx_node])
124 for v in neighbors
125 )
127 def semantic_feasibility(self, G1_node, G2_node):
128 """Returns True if adding (G1_node, G2_node) is semantically
129 feasible.
131 Any subclass which redefines semantic_feasibility() must
132 maintain the self.tests if needed, to keep the match() method
133 functional. Implementations should consider multigraphs.
134 """
135 neighbors = [n for n in self.G1[G1_node] if n in self.core_1]
136 if not self.one_hop(self.G1, G1_node, neighbors): # Fail fast on first node.
137 return False
138 if not self.two_hop(self.G1, self.core_1, G1_node, neighbors):
139 return False
140 # Otherwise, this node is semantically feasible!
141 return True
144class TimeRespectingDiGraphMatcher(DiGraphMatcher):
145 def __init__(self, G1, G2, temporal_attribute_name, delta):
146 """Initialize TimeRespectingDiGraphMatcher.
148 G1 and G2 should be nx.DiGraph or nx.MultiDiGraph instances.
150 Examples
151 --------
152 To create a TimeRespectingDiGraphMatcher which checks for
153 syntactic and semantic feasibility:
155 >>> from networkx.algorithms import isomorphism
156 >>> from datetime import timedelta
157 >>> G1 = nx.DiGraph(nx.path_graph(4, create_using=nx.DiGraph()))
159 >>> G2 = nx.DiGraph(nx.path_graph(4, create_using=nx.DiGraph()))
161 >>> GM = isomorphism.TimeRespectingDiGraphMatcher(
162 ... G1, G2, "date", timedelta(days=1)
163 ... )
164 """
165 self.temporal_attribute_name = temporal_attribute_name
166 self.delta = delta
167 super().__init__(G1, G2)
169 def get_pred_dates(self, Gx, Gx_node, core_x, pred):
170 """
171 Get the dates of edges from predecessors.
172 """
173 pred_dates = []
174 if isinstance(Gx, nx.DiGraph): # Graph G[u][v] returns the data dictionary.
175 for n in pred:
176 pred_dates.append(Gx[n][Gx_node][self.temporal_attribute_name])
177 else: # MultiGraph G[u][v] returns a dictionary of key -> data dictionary.
178 for n in pred:
179 for edge in Gx[n][
180 Gx_node
181 ].values(): # Iterates all edge data between node pair.
182 pred_dates.append(edge[self.temporal_attribute_name])
183 return pred_dates
185 def get_succ_dates(self, Gx, Gx_node, core_x, succ):
186 """
187 Get the dates of edges to successors.
188 """
189 succ_dates = []
190 if isinstance(Gx, nx.DiGraph): # Graph G[u][v] returns the data dictionary.
191 for n in succ:
192 succ_dates.append(Gx[Gx_node][n][self.temporal_attribute_name])
193 else: # MultiGraph G[u][v] returns a dictionary of key -> data dictionary.
194 for n in succ:
195 for edge in Gx[Gx_node][
196 n
197 ].values(): # Iterates all edge data between node pair.
198 succ_dates.append(edge[self.temporal_attribute_name])
199 return succ_dates
201 def one_hop(self, Gx, Gx_node, core_x, pred, succ):
202 """
203 The ego node.
204 """
205 pred_dates = self.get_pred_dates(Gx, Gx_node, core_x, pred)
206 succ_dates = self.get_succ_dates(Gx, Gx_node, core_x, succ)
207 return self.test_one(pred_dates, succ_dates) and self.test_two(
208 pred_dates, succ_dates
209 )
211 def two_hop_pred(self, Gx, Gx_node, core_x, pred):
212 """
213 The predecessors of the ego node.
214 """
215 return all(
216 self.one_hop(
217 Gx,
218 p,
219 core_x,
220 self.preds(Gx, core_x, p),
221 self.succs(Gx, core_x, p, Gx_node),
222 )
223 for p in pred
224 )
226 def two_hop_succ(self, Gx, Gx_node, core_x, succ):
227 """
228 The successors of the ego node.
229 """
230 return all(
231 self.one_hop(
232 Gx,
233 s,
234 core_x,
235 self.preds(Gx, core_x, s, Gx_node),
236 self.succs(Gx, core_x, s),
237 )
238 for s in succ
239 )
241 def preds(self, Gx, core_x, v, Gx_node=None):
242 pred = [n for n in Gx.predecessors(v) if n in core_x]
243 if Gx_node:
244 pred.append(Gx_node)
245 return pred
247 def succs(self, Gx, core_x, v, Gx_node=None):
248 succ = [n for n in Gx.successors(v) if n in core_x]
249 if Gx_node:
250 succ.append(Gx_node)
251 return succ
253 def test_one(self, pred_dates, succ_dates):
254 """
255 Edges one hop out from Gx_node in the mapping should be
256 time-respecting with respect to each other, regardless of
257 direction.
258 """
259 time_respecting = True
260 dates = pred_dates + succ_dates
262 if any(x is None for x in dates):
263 raise ValueError("Date or datetime not supplied for at least one edge.")
265 dates.sort() # Small to large.
266 if 0 < len(dates) and not (dates[-1] - dates[0] <= self.delta):
267 time_respecting = False
268 return time_respecting
270 def test_two(self, pred_dates, succ_dates):
271 """
272 Edges from a dual Gx_node in the mapping should be ordered in
273 a time-respecting manner.
274 """
275 time_respecting = True
276 pred_dates.sort()
277 succ_dates.sort()
278 # First out before last in; negative of the necessary condition for time-respect.
279 if (
280 0 < len(succ_dates)
281 and 0 < len(pred_dates)
282 and succ_dates[0] < pred_dates[-1]
283 ):
284 time_respecting = False
285 return time_respecting
287 def semantic_feasibility(self, G1_node, G2_node):
288 """Returns True if adding (G1_node, G2_node) is semantically
289 feasible.
291 Any subclass which redefines semantic_feasibility() must
292 maintain the self.tests if needed, to keep the match() method
293 functional. Implementations should consider multigraphs.
294 """
295 pred, succ = (
296 [n for n in self.G1.predecessors(G1_node) if n in self.core_1],
297 [n for n in self.G1.successors(G1_node) if n in self.core_1],
298 )
299 if not self.one_hop(
300 self.G1, G1_node, self.core_1, pred, succ
301 ): # Fail fast on first node.
302 return False
303 if not self.two_hop_pred(self.G1, G1_node, self.core_1, pred):
304 return False
305 if not self.two_hop_succ(self.G1, G1_node, self.core_1, succ):
306 return False
307 # Otherwise, this node is semantically feasible!
308 return True