|
|
""" |
|
|
Unit tests for dedensification and graph summarization |
|
|
""" |
|
|
import pytest |
|
|
|
|
|
import networkx as nx |
|
|
|
|
|
|
|
|
class TestDirectedDedensification: |
|
|
def build_original_graph(self): |
|
|
original_matrix = [ |
|
|
("1", "BC"), |
|
|
("2", "ABC"), |
|
|
("3", ["A", "B", "6"]), |
|
|
("4", "ABC"), |
|
|
("5", "AB"), |
|
|
("6", ["5"]), |
|
|
("A", ["6"]), |
|
|
] |
|
|
graph = nx.DiGraph() |
|
|
for source, targets in original_matrix: |
|
|
for target in targets: |
|
|
graph.add_edge(source, target) |
|
|
return graph |
|
|
|
|
|
def build_compressed_graph(self): |
|
|
compressed_matrix = [ |
|
|
("1", "BC"), |
|
|
("2", ["ABC"]), |
|
|
("3", ["A", "B", "6"]), |
|
|
("4", ["ABC"]), |
|
|
("5", "AB"), |
|
|
("6", ["5"]), |
|
|
("A", ["6"]), |
|
|
("ABC", "ABC"), |
|
|
] |
|
|
compressed_graph = nx.DiGraph() |
|
|
for source, targets in compressed_matrix: |
|
|
for target in targets: |
|
|
compressed_graph.add_edge(source, target) |
|
|
return compressed_graph |
|
|
|
|
|
def test_empty(self): |
|
|
""" |
|
|
Verify that an empty directed graph results in no compressor nodes |
|
|
""" |
|
|
G = nx.DiGraph() |
|
|
compressed_graph, c_nodes = nx.dedensify(G, threshold=2) |
|
|
assert c_nodes == set() |
|
|
|
|
|
@staticmethod |
|
|
def densify(G, compressor_nodes, copy=True): |
|
|
""" |
|
|
Reconstructs the original graph from a dedensified, directed graph |
|
|
|
|
|
Parameters |
|
|
---------- |
|
|
G: dedensified graph |
|
|
A networkx graph |
|
|
compressor_nodes: iterable |
|
|
Iterable of compressor nodes in the dedensified graph |
|
|
inplace: bool, optional (default: False) |
|
|
Indicates if densification should be done inplace |
|
|
|
|
|
Returns |
|
|
------- |
|
|
G: graph |
|
|
A densified networkx graph |
|
|
""" |
|
|
if copy: |
|
|
G = G.copy() |
|
|
for compressor_node in compressor_nodes: |
|
|
all_neighbors = set(nx.all_neighbors(G, compressor_node)) |
|
|
out_neighbors = set(G.neighbors(compressor_node)) |
|
|
for out_neighbor in out_neighbors: |
|
|
G.remove_edge(compressor_node, out_neighbor) |
|
|
in_neighbors = all_neighbors - out_neighbors |
|
|
for in_neighbor in in_neighbors: |
|
|
G.remove_edge(in_neighbor, compressor_node) |
|
|
for out_neighbor in out_neighbors: |
|
|
G.add_edge(in_neighbor, out_neighbor) |
|
|
G.remove_node(compressor_node) |
|
|
return G |
|
|
|
|
|
def setup_method(self): |
|
|
self.c_nodes = ("ABC",) |
|
|
|
|
|
def test_dedensify_edges(self): |
|
|
""" |
|
|
Verifies that dedensify produced the correct edges to/from compressor |
|
|
nodes in a directed graph |
|
|
""" |
|
|
G = self.build_original_graph() |
|
|
compressed_G = self.build_compressed_graph() |
|
|
compressed_graph, c_nodes = nx.dedensify(G, threshold=2) |
|
|
for s, t in compressed_graph.edges(): |
|
|
o_s = "".join(sorted(s)) |
|
|
o_t = "".join(sorted(t)) |
|
|
compressed_graph_exists = compressed_graph.has_edge(s, t) |
|
|
verified_compressed_exists = compressed_G.has_edge(o_s, o_t) |
|
|
assert compressed_graph_exists == verified_compressed_exists |
|
|
assert len(c_nodes) == len(self.c_nodes) |
|
|
|
|
|
def test_dedensify_edge_count(self): |
|
|
""" |
|
|
Verifies that dedensify produced the correct number of compressor nodes |
|
|
in a directed graph |
|
|
""" |
|
|
G = self.build_original_graph() |
|
|
original_edge_count = len(G.edges()) |
|
|
c_G, c_nodes = nx.dedensify(G, threshold=2) |
|
|
compressed_edge_count = len(c_G.edges()) |
|
|
assert compressed_edge_count <= original_edge_count |
|
|
compressed_G = self.build_compressed_graph() |
|
|
assert compressed_edge_count == len(compressed_G.edges()) |
|
|
|
|
|
def test_densify_edges(self): |
|
|
""" |
|
|
Verifies that densification produces the correct edges from the |
|
|
original directed graph |
|
|
""" |
|
|
compressed_G = self.build_compressed_graph() |
|
|
original_graph = self.densify(compressed_G, self.c_nodes, copy=True) |
|
|
G = self.build_original_graph() |
|
|
for s, t in G.edges(): |
|
|
assert G.has_edge(s, t) == original_graph.has_edge(s, t) |
|
|
|
|
|
def test_densify_edge_count(self): |
|
|
""" |
|
|
Verifies that densification produces the correct number of edges in the |
|
|
original directed graph |
|
|
""" |
|
|
compressed_G = self.build_compressed_graph() |
|
|
compressed_edge_count = len(compressed_G.edges()) |
|
|
original_graph = self.densify(compressed_G, self.c_nodes) |
|
|
original_edge_count = len(original_graph.edges()) |
|
|
assert compressed_edge_count <= original_edge_count |
|
|
G = self.build_original_graph() |
|
|
assert original_edge_count == len(G.edges()) |
|
|
|
|
|
|
|
|
class TestUnDirectedDedensification: |
|
|
def build_original_graph(self): |
|
|
""" |
|
|
Builds graph shown in the original research paper |
|
|
""" |
|
|
original_matrix = [ |
|
|
("1", "CB"), |
|
|
("2", "ABC"), |
|
|
("3", ["A", "B", "6"]), |
|
|
("4", "ABC"), |
|
|
("5", "AB"), |
|
|
("6", ["5"]), |
|
|
("A", ["6"]), |
|
|
] |
|
|
graph = nx.Graph() |
|
|
for source, targets in original_matrix: |
|
|
for target in targets: |
|
|
graph.add_edge(source, target) |
|
|
return graph |
|
|
|
|
|
def test_empty(self): |
|
|
""" |
|
|
Verify that an empty undirected graph results in no compressor nodes |
|
|
""" |
|
|
G = nx.Graph() |
|
|
compressed_G, c_nodes = nx.dedensify(G, threshold=2) |
|
|
assert c_nodes == set() |
|
|
|
|
|
def setup_method(self): |
|
|
self.c_nodes = ("6AB", "ABC") |
|
|
|
|
|
def build_compressed_graph(self): |
|
|
compressed_matrix = [ |
|
|
("1", ["B", "C"]), |
|
|
("2", ["ABC"]), |
|
|
("3", ["6AB"]), |
|
|
("4", ["ABC"]), |
|
|
("5", ["6AB"]), |
|
|
("6", ["6AB", "A"]), |
|
|
("A", ["6AB", "ABC"]), |
|
|
("B", ["ABC", "6AB"]), |
|
|
("C", ["ABC"]), |
|
|
] |
|
|
compressed_graph = nx.Graph() |
|
|
for source, targets in compressed_matrix: |
|
|
for target in targets: |
|
|
compressed_graph.add_edge(source, target) |
|
|
return compressed_graph |
|
|
|
|
|
def test_dedensify_edges(self): |
|
|
""" |
|
|
Verifies that dedensify produced correct compressor nodes and the |
|
|
correct edges to/from the compressor nodes in an undirected graph |
|
|
""" |
|
|
G = self.build_original_graph() |
|
|
c_G, c_nodes = nx.dedensify(G, threshold=2) |
|
|
v_compressed_G = self.build_compressed_graph() |
|
|
for s, t in c_G.edges(): |
|
|
o_s = "".join(sorted(s)) |
|
|
o_t = "".join(sorted(t)) |
|
|
has_compressed_edge = c_G.has_edge(s, t) |
|
|
verified_has_compressed_edge = v_compressed_G.has_edge(o_s, o_t) |
|
|
assert has_compressed_edge == verified_has_compressed_edge |
|
|
assert len(c_nodes) == len(self.c_nodes) |
|
|
|
|
|
def test_dedensify_edge_count(self): |
|
|
""" |
|
|
Verifies that dedensify produced the correct number of edges in an |
|
|
undirected graph |
|
|
""" |
|
|
G = self.build_original_graph() |
|
|
c_G, c_nodes = nx.dedensify(G, threshold=2, copy=True) |
|
|
compressed_edge_count = len(c_G.edges()) |
|
|
verified_original_edge_count = len(G.edges()) |
|
|
assert compressed_edge_count <= verified_original_edge_count |
|
|
verified_compressed_G = self.build_compressed_graph() |
|
|
verified_compressed_edge_count = len(verified_compressed_G.edges()) |
|
|
assert compressed_edge_count == verified_compressed_edge_count |
|
|
|
|
|
|
|
|
@pytest.mark.parametrize( |
|
|
"graph_type", [nx.Graph, nx.DiGraph, nx.MultiGraph, nx.MultiDiGraph] |
|
|
) |
|
|
def test_summarization_empty(graph_type): |
|
|
G = graph_type() |
|
|
summary_graph = nx.snap_aggregation(G, node_attributes=("color",)) |
|
|
assert nx.is_isomorphic(summary_graph, G) |
|
|
|
|
|
|
|
|
class AbstractSNAP: |
|
|
node_attributes = ("color",) |
|
|
|
|
|
def build_original_graph(self): |
|
|
pass |
|
|
|
|
|
def build_summary_graph(self): |
|
|
pass |
|
|
|
|
|
def test_summary_graph(self): |
|
|
original_graph = self.build_original_graph() |
|
|
summary_graph = self.build_summary_graph() |
|
|
|
|
|
relationship_attributes = ("type",) |
|
|
generated_summary_graph = nx.snap_aggregation( |
|
|
original_graph, self.node_attributes, relationship_attributes |
|
|
) |
|
|
relabeled_summary_graph = self.deterministic_labels(generated_summary_graph) |
|
|
assert nx.is_isomorphic(summary_graph, relabeled_summary_graph) |
|
|
|
|
|
def deterministic_labels(self, G): |
|
|
node_labels = list(G.nodes) |
|
|
node_labels = sorted(node_labels, key=lambda n: sorted(G.nodes[n]["group"])[0]) |
|
|
node_labels.sort() |
|
|
|
|
|
label_mapping = {} |
|
|
for index, node in enumerate(node_labels): |
|
|
label = "Supernode-%s" % index |
|
|
label_mapping[node] = label |
|
|
|
|
|
return nx.relabel_nodes(G, label_mapping) |
|
|
|
|
|
|
|
|
class TestSNAPNoEdgeTypes(AbstractSNAP): |
|
|
relationship_attributes = () |
|
|
|
|
|
def test_summary_graph(self): |
|
|
original_graph = self.build_original_graph() |
|
|
summary_graph = self.build_summary_graph() |
|
|
|
|
|
relationship_attributes = ("type",) |
|
|
generated_summary_graph = nx.snap_aggregation( |
|
|
original_graph, self.node_attributes |
|
|
) |
|
|
relabeled_summary_graph = self.deterministic_labels(generated_summary_graph) |
|
|
assert nx.is_isomorphic(summary_graph, relabeled_summary_graph) |
|
|
|
|
|
def build_original_graph(self): |
|
|
nodes = { |
|
|
"A": {"color": "Red"}, |
|
|
"B": {"color": "Red"}, |
|
|
"C": {"color": "Red"}, |
|
|
"D": {"color": "Red"}, |
|
|
"E": {"color": "Blue"}, |
|
|
"F": {"color": "Blue"}, |
|
|
"G": {"color": "Blue"}, |
|
|
"H": {"color": "Blue"}, |
|
|
"I": {"color": "Yellow"}, |
|
|
"J": {"color": "Yellow"}, |
|
|
"K": {"color": "Yellow"}, |
|
|
"L": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("A", "B"), |
|
|
("A", "C"), |
|
|
("A", "E"), |
|
|
("A", "I"), |
|
|
("B", "D"), |
|
|
("B", "J"), |
|
|
("B", "F"), |
|
|
("C", "G"), |
|
|
("D", "H"), |
|
|
("I", "J"), |
|
|
("J", "K"), |
|
|
("I", "L"), |
|
|
] |
|
|
G = nx.Graph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target in edges: |
|
|
G.add_edge(source, target) |
|
|
|
|
|
return G |
|
|
|
|
|
def build_summary_graph(self): |
|
|
nodes = { |
|
|
"Supernode-0": {"color": "Red"}, |
|
|
"Supernode-1": {"color": "Red"}, |
|
|
"Supernode-2": {"color": "Blue"}, |
|
|
"Supernode-3": {"color": "Blue"}, |
|
|
"Supernode-4": {"color": "Yellow"}, |
|
|
"Supernode-5": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("Supernode-0", "Supernode-0"), |
|
|
("Supernode-0", "Supernode-1"), |
|
|
("Supernode-0", "Supernode-2"), |
|
|
("Supernode-0", "Supernode-4"), |
|
|
("Supernode-1", "Supernode-3"), |
|
|
("Supernode-4", "Supernode-4"), |
|
|
("Supernode-4", "Supernode-5"), |
|
|
] |
|
|
G = nx.Graph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target in edges: |
|
|
G.add_edge(source, target) |
|
|
|
|
|
supernodes = { |
|
|
"Supernode-0": {"A", "B"}, |
|
|
"Supernode-1": {"C", "D"}, |
|
|
"Supernode-2": {"E", "F"}, |
|
|
"Supernode-3": {"G", "H"}, |
|
|
"Supernode-4": {"I", "J"}, |
|
|
"Supernode-5": {"K", "L"}, |
|
|
} |
|
|
nx.set_node_attributes(G, supernodes, "group") |
|
|
return G |
|
|
|
|
|
|
|
|
class TestSNAPUndirected(AbstractSNAP): |
|
|
def build_original_graph(self): |
|
|
nodes = { |
|
|
"A": {"color": "Red"}, |
|
|
"B": {"color": "Red"}, |
|
|
"C": {"color": "Red"}, |
|
|
"D": {"color": "Red"}, |
|
|
"E": {"color": "Blue"}, |
|
|
"F": {"color": "Blue"}, |
|
|
"G": {"color": "Blue"}, |
|
|
"H": {"color": "Blue"}, |
|
|
"I": {"color": "Yellow"}, |
|
|
"J": {"color": "Yellow"}, |
|
|
"K": {"color": "Yellow"}, |
|
|
"L": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("A", "B", "Strong"), |
|
|
("A", "C", "Weak"), |
|
|
("A", "E", "Strong"), |
|
|
("A", "I", "Weak"), |
|
|
("B", "D", "Weak"), |
|
|
("B", "J", "Weak"), |
|
|
("B", "F", "Strong"), |
|
|
("C", "G", "Weak"), |
|
|
("D", "H", "Weak"), |
|
|
("I", "J", "Strong"), |
|
|
("J", "K", "Strong"), |
|
|
("I", "L", "Strong"), |
|
|
] |
|
|
G = nx.Graph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, type in edges: |
|
|
G.add_edge(source, target, type=type) |
|
|
|
|
|
return G |
|
|
|
|
|
def build_summary_graph(self): |
|
|
nodes = { |
|
|
"Supernode-0": {"color": "Red"}, |
|
|
"Supernode-1": {"color": "Red"}, |
|
|
"Supernode-2": {"color": "Blue"}, |
|
|
"Supernode-3": {"color": "Blue"}, |
|
|
"Supernode-4": {"color": "Yellow"}, |
|
|
"Supernode-5": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("Supernode-0", "Supernode-0", "Strong"), |
|
|
("Supernode-0", "Supernode-1", "Weak"), |
|
|
("Supernode-0", "Supernode-2", "Strong"), |
|
|
("Supernode-0", "Supernode-4", "Weak"), |
|
|
("Supernode-1", "Supernode-3", "Weak"), |
|
|
("Supernode-4", "Supernode-4", "Strong"), |
|
|
("Supernode-4", "Supernode-5", "Strong"), |
|
|
] |
|
|
G = nx.Graph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, type in edges: |
|
|
G.add_edge(source, target, types=[{"type": type}]) |
|
|
|
|
|
supernodes = { |
|
|
"Supernode-0": {"A", "B"}, |
|
|
"Supernode-1": {"C", "D"}, |
|
|
"Supernode-2": {"E", "F"}, |
|
|
"Supernode-3": {"G", "H"}, |
|
|
"Supernode-4": {"I", "J"}, |
|
|
"Supernode-5": {"K", "L"}, |
|
|
} |
|
|
nx.set_node_attributes(G, supernodes, "group") |
|
|
return G |
|
|
|
|
|
|
|
|
class TestSNAPDirected(AbstractSNAP): |
|
|
def build_original_graph(self): |
|
|
nodes = { |
|
|
"A": {"color": "Red"}, |
|
|
"B": {"color": "Red"}, |
|
|
"C": {"color": "Green"}, |
|
|
"D": {"color": "Green"}, |
|
|
"E": {"color": "Blue"}, |
|
|
"F": {"color": "Blue"}, |
|
|
"G": {"color": "Yellow"}, |
|
|
"H": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("A", "C", "Strong"), |
|
|
("A", "E", "Strong"), |
|
|
("A", "F", "Weak"), |
|
|
("B", "D", "Strong"), |
|
|
("B", "E", "Weak"), |
|
|
("B", "F", "Strong"), |
|
|
("C", "G", "Strong"), |
|
|
("C", "F", "Strong"), |
|
|
("D", "E", "Strong"), |
|
|
("D", "H", "Strong"), |
|
|
("G", "E", "Strong"), |
|
|
("H", "F", "Strong"), |
|
|
] |
|
|
G = nx.DiGraph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, type in edges: |
|
|
G.add_edge(source, target, type=type) |
|
|
|
|
|
return G |
|
|
|
|
|
def build_summary_graph(self): |
|
|
nodes = { |
|
|
"Supernode-0": {"color": "Red"}, |
|
|
"Supernode-1": {"color": "Green"}, |
|
|
"Supernode-2": {"color": "Blue"}, |
|
|
"Supernode-3": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("Supernode-0", "Supernode-1", [{"type": "Strong"}]), |
|
|
("Supernode-0", "Supernode-2", [{"type": "Weak"}, {"type": "Strong"}]), |
|
|
("Supernode-1", "Supernode-2", [{"type": "Strong"}]), |
|
|
("Supernode-1", "Supernode-3", [{"type": "Strong"}]), |
|
|
("Supernode-3", "Supernode-2", [{"type": "Strong"}]), |
|
|
] |
|
|
G = nx.DiGraph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, types in edges: |
|
|
G.add_edge(source, target, types=types) |
|
|
|
|
|
supernodes = { |
|
|
"Supernode-0": {"A", "B"}, |
|
|
"Supernode-1": {"C", "D"}, |
|
|
"Supernode-2": {"E", "F"}, |
|
|
"Supernode-3": {"G", "H"}, |
|
|
"Supernode-4": {"I", "J"}, |
|
|
"Supernode-5": {"K", "L"}, |
|
|
} |
|
|
nx.set_node_attributes(G, supernodes, "group") |
|
|
return G |
|
|
|
|
|
|
|
|
class TestSNAPUndirectedMulti(AbstractSNAP): |
|
|
def build_original_graph(self): |
|
|
nodes = { |
|
|
"A": {"color": "Red"}, |
|
|
"B": {"color": "Red"}, |
|
|
"C": {"color": "Red"}, |
|
|
"D": {"color": "Blue"}, |
|
|
"E": {"color": "Blue"}, |
|
|
"F": {"color": "Blue"}, |
|
|
"G": {"color": "Yellow"}, |
|
|
"H": {"color": "Yellow"}, |
|
|
"I": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("A", "D", ["Weak", "Strong"]), |
|
|
("B", "E", ["Weak", "Strong"]), |
|
|
("D", "I", ["Strong"]), |
|
|
("E", "H", ["Strong"]), |
|
|
("F", "G", ["Weak"]), |
|
|
("I", "G", ["Weak", "Strong"]), |
|
|
("I", "H", ["Weak", "Strong"]), |
|
|
("G", "H", ["Weak", "Strong"]), |
|
|
] |
|
|
G = nx.MultiGraph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, types in edges: |
|
|
for type in types: |
|
|
G.add_edge(source, target, type=type) |
|
|
|
|
|
return G |
|
|
|
|
|
def build_summary_graph(self): |
|
|
nodes = { |
|
|
"Supernode-0": {"color": "Red"}, |
|
|
"Supernode-1": {"color": "Blue"}, |
|
|
"Supernode-2": {"color": "Yellow"}, |
|
|
"Supernode-3": {"color": "Blue"}, |
|
|
"Supernode-4": {"color": "Yellow"}, |
|
|
"Supernode-5": {"color": "Red"}, |
|
|
} |
|
|
edges = [ |
|
|
("Supernode-1", "Supernode-2", [{"type": "Weak"}]), |
|
|
("Supernode-2", "Supernode-4", [{"type": "Weak"}, {"type": "Strong"}]), |
|
|
("Supernode-3", "Supernode-4", [{"type": "Strong"}]), |
|
|
("Supernode-3", "Supernode-5", [{"type": "Weak"}, {"type": "Strong"}]), |
|
|
("Supernode-4", "Supernode-4", [{"type": "Weak"}, {"type": "Strong"}]), |
|
|
] |
|
|
G = nx.MultiGraph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, types in edges: |
|
|
for type in types: |
|
|
G.add_edge(source, target, type=type) |
|
|
|
|
|
supernodes = { |
|
|
"Supernode-0": {"A", "B"}, |
|
|
"Supernode-1": {"C", "D"}, |
|
|
"Supernode-2": {"E", "F"}, |
|
|
"Supernode-3": {"G", "H"}, |
|
|
"Supernode-4": {"I", "J"}, |
|
|
"Supernode-5": {"K", "L"}, |
|
|
} |
|
|
nx.set_node_attributes(G, supernodes, "group") |
|
|
return G |
|
|
|
|
|
|
|
|
class TestSNAPDirectedMulti(AbstractSNAP): |
|
|
def build_original_graph(self): |
|
|
nodes = { |
|
|
"A": {"color": "Red"}, |
|
|
"B": {"color": "Red"}, |
|
|
"C": {"color": "Green"}, |
|
|
"D": {"color": "Green"}, |
|
|
"E": {"color": "Blue"}, |
|
|
"F": {"color": "Blue"}, |
|
|
"G": {"color": "Yellow"}, |
|
|
"H": {"color": "Yellow"}, |
|
|
} |
|
|
edges = [ |
|
|
("A", "C", ["Weak", "Strong"]), |
|
|
("A", "E", ["Strong"]), |
|
|
("A", "F", ["Weak"]), |
|
|
("B", "D", ["Weak", "Strong"]), |
|
|
("B", "E", ["Weak"]), |
|
|
("B", "F", ["Strong"]), |
|
|
("C", "G", ["Weak", "Strong"]), |
|
|
("C", "F", ["Strong"]), |
|
|
("D", "E", ["Strong"]), |
|
|
("D", "H", ["Weak", "Strong"]), |
|
|
("G", "E", ["Strong"]), |
|
|
("H", "F", ["Strong"]), |
|
|
] |
|
|
G = nx.MultiDiGraph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, types in edges: |
|
|
for type in types: |
|
|
G.add_edge(source, target, type=type) |
|
|
|
|
|
return G |
|
|
|
|
|
def build_summary_graph(self): |
|
|
nodes = { |
|
|
"Supernode-0": {"color": "Red"}, |
|
|
"Supernode-1": {"color": "Blue"}, |
|
|
"Supernode-2": {"color": "Yellow"}, |
|
|
"Supernode-3": {"color": "Blue"}, |
|
|
} |
|
|
edges = [ |
|
|
("Supernode-0", "Supernode-1", ["Weak", "Strong"]), |
|
|
("Supernode-0", "Supernode-2", ["Weak", "Strong"]), |
|
|
("Supernode-1", "Supernode-2", ["Strong"]), |
|
|
("Supernode-1", "Supernode-3", ["Weak", "Strong"]), |
|
|
("Supernode-3", "Supernode-2", ["Strong"]), |
|
|
] |
|
|
G = nx.MultiDiGraph() |
|
|
for node in nodes: |
|
|
attributes = nodes[node] |
|
|
G.add_node(node, **attributes) |
|
|
|
|
|
for source, target, types in edges: |
|
|
for type in types: |
|
|
G.add_edge(source, target, type=type) |
|
|
|
|
|
supernodes = { |
|
|
"Supernode-0": {"A", "B"}, |
|
|
"Supernode-1": {"C", "D"}, |
|
|
"Supernode-2": {"E", "F"}, |
|
|
"Supernode-3": {"G", "H"}, |
|
|
} |
|
|
nx.set_node_attributes(G, supernodes, "group") |
|
|
return G |
|
|
|
