Coverage for /opt/conda/lib/python3.14/site-packages/medil/_graph.py: 92%
169 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-23 02:29 +0000
1"""Various types and representations of graphs."""
3import numpy as np
4from numpy.random import default_rng
7class UndirectedDependenceGraph(object):
8 r"""Adjacency matrix representation using a 2d numpy array.
10 Upon initialization, this class is fairly standard implementation
11 of an undirected graph. However, upon calling the :meth:`make_aux`
12 method, an auxilliary data structure in the form of several new
13 attributes is created, which are used by
14 :meth:`medil.ecc_algorithms.find_clique_min_cover` according to
15 the algorithm in :cite:`Gramm_2009`.
17 Notes
18 -----
19 The algorithms for finding the minMCM via ECC contain many
20 algebraic operations, so adjacency matrix representation (via
21 NumPy) is most covenient.
23 The diag is used to store information when the graph is reduced,
24 not to indicate self loops, so it is important that diag = 1 at
25 init.
27 """
29 def __init__(self, adj_matrix, verbose=False):
30 # doesn't behave well unless input is nparray
31 self.adj_matrix = adj_matrix
32 self.num_vertices = np.trace(adj_matrix)
33 self.max_num_verts = len(adj_matrix)
34 self.num_edges = np.triu(adj_matrix, 1).sum()
35 self.verbose = verbose
37 def add_edges(self, edges):
38 v_1s = edges[:, 0]
39 v_2s = edges[:, 1]
40 self.adj_matrix[v_1s, v_2s] = 1
41 self.adj_matrix[v_2s, v_1s] = 1
42 self.num_edges = np.triu(self.adj_matrix, 1).sum()
44 def rm_edges(self, edges):
45 v_1s = edges[:, 0]
46 v_2s = edges[:, 1]
47 self.adj_matrix[v_1s, v_2s] = 0
48 self.adj_matrix[v_2s, v_1s] = 0
49 self.num_edges = np.triu(self.adj_matrix, 1).sum()
51 def make_aux(self):
52 # this makes the auxilliary structure described in INITIALIZATION in the paper
54 # find neighbourhood for each vertex
55 # each row corresponds to a unique edge
56 max_num_edges = self.n_choose_2(self.max_num_verts)
57 self.common_neighbors = np.zeros(
58 (max_num_edges, self.max_num_verts), int
59 ) # init
61 # mapping of edges to unique row idx
62 triu_idx = np.triu_indices(self.max_num_verts, 1)
63 nghbrhd_idx = np.zeros((self.max_num_verts, self.max_num_verts), int)
64 nghbrhd_idx[triu_idx] = np.arange(max_num_edges)
65 # nghbrhd_idx += nghbrhd_idx.T
66 self.get_idx = lambda edge: nghbrhd_idx[edge[0], edge[1]]
68 # reverse mapping
69 u, v = np.where(np.triu(np.ones_like(self.adj_matrix), 1))
70 self.get_edge = lambda idx: (u[idx], v[idx])
72 # compute actual neighborhood for each edge = (v_1, v_2)
73 self.nbrs = lambda edge: np.logical_and(
74 self.adj_matrix[edge[0]], self.adj_matrix[edge[1]]
75 )
77 extant_edges = np.transpose(np.triu(self.adj_matrix, 1).nonzero())
78 self.extant_edges_idx = np.fromiter(
79 {self.get_idx(edge) for edge in extant_edges}, dtype=int
80 )
81 extant_nbrs = np.array([self.nbrs(edge) for edge in extant_edges], int)
82 extant_nbrs_idx = np.array([self.get_idx(edge) for edge in extant_edges], int)
84 # from paper: set of N_{u, v} for all edges (u, v)
85 self.common_neighbors[extant_nbrs_idx] = extant_nbrs
87 # number of cliques for each node? assignments? if we set diag=0
88 # num_cliques = common_neighbors.sum(0)
90 # sum() of submatrix of graph containing exactly the rows/columns
91 # corresponding to the nodes in common_neighbors(edge) using
92 # logical indexing:
94 # make mask to identify subgraph (closed common neighborhood of
95 # nodes u, v in edge u,v)
96 mask = lambda edge_idx: np.array(self.common_neighbors[edge_idx], dtype=bool)
98 # make subgraph-adjacency matrix, and then subtract diag and
99 # divide by two to get num edges in subgraph---same as sum() of
100 # triu(subgraph-adjacency matrix) but probably a bit faster
101 nbrhood = lambda edge_idx: self.adj_matrix[mask(edge_idx)][:, mask(edge_idx)]
102 max_num_edges_in_nbrhood = (
103 lambda edge_idx: (nbrhood(edge_idx).sum() - mask(edge_idx).sum()) // 2
104 )
106 # from paper: set of c_{u, v} for all edges (u, v)
107 self.nbrhood_edge_counts = np.array(
108 [
109 max_num_edges_in_nbrhood(edge_idx)
110 for edge_idx in np.arange(max_num_edges)
111 ],
112 int,
113 )
115 # important structs are:
116 # self.common_neighbors
117 # self.nbrhood_edge_counts
118 # # and fun is
119 # self.nbrs
121 @staticmethod
122 def n_choose_2(n):
123 return n * (n - 1) // 2
125 def reducible_copy(self):
126 return ReducibleUndDepGraph(self)
129class ReducibleUndDepGraph(UndirectedDependenceGraph):
130 def __init__(self, udg):
131 self.unreduced = udg.unreduced if hasattr(udg, "unreduced") else udg
132 self.adj_matrix = udg.adj_matrix.copy()
133 self.num_vertices = udg.num_vertices
134 self.num_edges = udg.num_edges
136 self.the_cover = None
137 self.verbose = udg.verbose
139 # from auxilliary structure if needed
140 if not hasattr(udg, "get_idx"):
141 udg.make_aux()
142 self.get_idx = udg.get_idx
144 # need to also update these all when self.cover_edges() is called? already done in rule_1
145 self.common_neighbors = udg.common_neighbors.copy()
146 self.nbrhood_edge_counts = udg.nbrhood_edge_counts.copy()
148 # update when cover_edges() is called, actually maybe just extant_edges?
149 self.extant_edges_idx = udg.extant_edges_idx.copy()
150 self.nbrs = udg.nbrs
151 self.get_edge = udg.get_edge
153 if hasattr(udg, "reduced_away"): # then rule_3 wasn't applied
154 self.reduced_away = udg.reduced_away
156 def reset(self):
157 self.__init__(self.unreduced)
159 def reduzieren(self, k_num_cliques):
160 # reduce by first applying rule 1 and then repeatedly applying
161 # rule 2 or rule 3 (both of which followed by rule 1 again)
162 # until they don't apply
163 if self.verbose:
164 print("\t\treducing:")
165 self.k_num_cliques = k_num_cliques
166 self.reducing = True
167 while self.reducing:
168 self.reducing = False
169 self.rule_1()
170 self.rule_2()
171 if self.k_num_cliques < 0:
172 return
173 if self.reducing:
174 continue
175 self.rule_3()
177 def rule_1(self):
178 # rule_1: Remove isolated vertices and vertices that are only
179 # adjacent to covered edges
181 isolated_verts = np.where(self.adj_matrix.sum(0) + self.adj_matrix.sum(1) == 2)[
182 0
183 ]
184 if len(isolated_verts) > 0: # then Rule 1 is applied
185 if self.verbose:
186 print("\t\t\tapplying Rule 1...")
188 # update auxilliary attributes; LEMMA 2
189 self.adj_matrix[isolated_verts, isolated_verts] = 0
190 self.num_vertices -= len(isolated_verts)
192 # decrease nbrhood edge counts
193 for vert in isolated_verts:
194 open_nbrhood = self.unreduced.adj_matrix[vert].copy()
195 open_nbrhood[vert] = 0
196 idx_nbrhoods_to_update = np.where(self.common_neighbors[:, vert] == 1)[
197 0
198 ]
199 tiled = np.tile(
200 open_nbrhood, (len(idx_nbrhoods_to_update), 1)
201 ) # instead of another loop
202 to_subtract = np.logical_and(
203 tiled, self.common_neighbors[idx_nbrhoods_to_update]
204 ).sum(1)
205 self.nbrhood_edge_counts[idx_nbrhoods_to_update] -= to_subtract
206 # my own addition:
207 # self.nbrhood[:, vert] = 0
209 # remove isolated_verts from common neighborhoods
210 self.common_neighbors[:, isolated_verts] = 0
212 def rule_2(self):
213 # rule_2: If an uncovered edge {u,v} is contained in exactly
214 # one maximal clique C, i.e., the common neighbors of u and v
215 # induce a clique, then add C to the solution, mark its edges
216 # as covered, and decrease k by one
218 score = self.n_choose_2(self.common_neighbors.sum(1)) - self.nbrhood_edge_counts
219 # score of n implies edge is in exactly n+1 maximal cliques,
220 # so we want edges with score 0
222 clique_idxs = np.where(score[self.extant_edges_idx] == 0)[0]
224 if clique_idxs.size > 0:
225 clique_idx = clique_idxs[-1]
226 if self.verbose:
227 print("\t\t\tapplying Rule 2...")
228 clique = self.common_neighbors[self.extant_edges_idx[clique_idx]].copy()
229 self.the_cover = (
230 clique.reshape(1, -1)
231 if self.the_cover is None
232 else np.vstack((self.the_cover, clique))
233 )
234 self.cover_edges()
235 self.k_num_cliques -= 1
236 self.reducing = True
237 # start the loop over so Rule 1 can 'clean up'
238 # self.common_neighbors[clique_idxs[0]] = 0 # zero out row, to update struct? not in paper?
240 def rule_3(self):
241 # rule_3: Consider a vertex v that has at least one
242 # prisoner. If each prisoner is connected to at least one
243 # vertex other than v via an uncovered edge (automatically
244 # given if instance is reduced w.r.t. rules 1 and 2), and the
245 # prisoners dominate the exit,s then delete v. To reconstruct
246 # a solution for the unreduced instance, add v to every clique
247 # containing a prisoner of v.
249 for vert, nbrhood in enumerate(self.adj_matrix):
250 if nbrhood[vert] == 0: # then nbrhood is empty
251 continue
252 exits = np.zeros(self.unreduced.max_num_verts, bool)
253 nbrs = np.flatnonzero(nbrhood)
254 for nbr in nbrs:
255 if (nbrhood.astype(int) - self.adj_matrix[nbr].astype(int) == -1).any():
256 exits[nbr] = True
257 # exits[j] == True iff j is an exit for vert
258 # prisoners[j] == True iff j is a prisoner of vert
259 prisoners = np.logical_and(~exits, self.adj_matrix[vert])
260 prisoners[vert] = False # a vert isn't its own prisoner
262 # check if each exit is adjacent to at least one prisoner
263 prisoners_dominate_exits = self.adj_matrix[prisoners][:, exits].sum(0)
264 if prisoners_dominate_exits.all(): # apply the rule
265 self.reducing = True
266 if self.verbose:
267 print("\t\t\tapplying Rule 3...")
269 # mark edges covered so rule_1 will delete vertex
270 edges = np.array(
271 [
272 [vert, u]
273 for u in np.flatnonzero(self.adj_matrix[vert])
274 if vert != u
275 ]
276 )
277 edges.sort() # so they're in triu, so that self.get_idx works
278 self.cover_edges(edges)
280 # keep track of deleted nodes and their prisoners for reconstructing solution
281 if not hasattr(self, "reduced_away"):
282 self.reduced_away = np.zeros_like(self.adj_matrix, bool)
283 self.reduced_away[vert] = prisoners
284 break
286 def choose_nbrhood(self):
287 # only compute for existing edges
288 common_neighbors = self.common_neighbors[self.extant_edges_idx]
289 nbrhood_edge_counts = self.nbrhood_edge_counts[self.extant_edges_idx]
290 score = self.n_choose_2(common_neighbors.sum(1)) - nbrhood_edge_counts
292 # idx in reduced idx list, not from full edge list
293 chosen_edge_idx = score.argmin()
295 mask = common_neighbors[chosen_edge_idx].astype(bool)
297 chosen_nbrhood = self.adj_matrix.copy()
298 chosen_nbrhood[~mask, :] = 0
299 chosen_nbrhood[:, ~mask] = 0
300 if chosen_nbrhood.sum() <= mask.sum():
301 raise ValueError("This nbrhood has no edges")
302 return chosen_nbrhood
304 def cover_edges(self, edges=None):
305 if edges is None:
306 # always call after updating the cover; only on single, recently added clique
307 if self.the_cover is None:
308 return # self.adj_matrix
310 # change edges to 0 if they're covered
311 clique = self.the_cover[-1]
312 covered = np.where(clique)[0]
314 # trick for getting combinations from idx
315 comb_idx = np.triu_indices(len(covered), 1)
317 # actual pairwise combinations; ie all edges (v_i, v_j) covered by the clique
318 covered_edges = np.empty((len(comb_idx[0]), 2), int)
319 covered_edges[:, 0] = covered[comb_idx[0]]
320 covered_edges[:, 1] = covered[comb_idx[1]]
322 else:
323 covered_edges = edges
325 # cover (remove from reduced_graph) edges
326 self.rm_edges(covered_edges)
327 # update extant_edges_idx
328 rmed_edges_idx = [self.get_idx(edge) for edge in covered_edges]
329 extant_rmed_edges_idx = [
330 edge for edge in rmed_edges_idx if edge in self.extant_edges_idx
331 ]
332 idx_idx = np.array(
333 [np.where(self.extant_edges_idx == idx) for idx in extant_rmed_edges_idx],
334 int,
335 ).flatten()
337 self.extant_edges_idx = np.delete(self.extant_edges_idx, idx_idx)
338 # now here do all the updates to nbrs?----actually probably don't want this? see 2clique house example
340 # update self.common_neighbors
341 # self.common_neighbors[rmed_edges_idx] = 0 # zero out rows covered edges; maybe not necessary, since common_neighbors is (probably?) only called with extant_edges_idx?
343 if self.verbose:
344 print("\t\t\t{} uncovered edges remaining".format(self.num_edges))
346 # if edges is None:
347 # cover_orig = self.the_cover
348 # while self.the_cover.shape[0] > 1:
349 # self.the_cover = self.the_cover[:-1]
350 # self.cover_edges()
351 # self.the_cover = cover_orig
353 def reconstruct_cover(self, the_cover):
354 if not hasattr(self, "reduced_away"): # then rule_3 wasn't applied
355 return the_cover
357 # add the reduced away vert to all covering cliques containing at least one of its prisoners
358 to_expand = np.flatnonzero(self.reduced_away.sum(1))
359 for vert in to_expand:
360 prisoners = self.reduced_away[vert]
361 tiled_prisoners = np.tile(
362 prisoners, (len(the_cover), 1)
363 ) # instead of another loop
364 cliques_to_update_mask = (
365 np.logical_and(tiled_prisoners, the_cover).sum(1).astype(bool)
366 )
367 the_cover[cliques_to_update_mask, vert] = 1
369 return the_cover