Added connected components for digraphs

prakhar1989 · prakhar1989 · commit 163a174e5bf0 · 2012-12-05T16:00:52.000+05:30
diff --git a/digraph.py b/digraph.py
@@ -1,4 +1,5 @@
 from graph import graph
+from copy import deepcopy
 class digraph(graph):
     """
     Directed Graph class - made of nodes and edges
@@ -7,7 +8,7 @@ class digraph(graph):
     has_edge, nodes, edges, add_node_attribute, node_attributes
     neighbors, del_node, del_edge, node_order, set_edge_weight,
     get_edge_weight, set_edge_properties, get_edge_properties
-    clear_node_attributes, has_cycle
+    clear_node_attributes, get_transpose
     """
 
     DEFAULT_WEIGHT = 1
@@ -61,3 +62,12 @@ def del_node(self, node):
             if self.has_edge((n, node)):
                 self.del_edge((n, node))
         del(self.node_neighbors[node])
+
+    def get_transpose(self):
+        """ Returns the transpose of the graph
+        with edges reversed and nodes same """
+        digr = deepcopy(self)
+        for (u, v) in self.edges():
+            digr.del_edge((u, v))
+            digr.add_edge((v, u))
+        return digr
diff --git a/digraph_test.py b/digraph_test.py
@@ -94,5 +94,13 @@ def test_clear_node_attributes_method(self):
         self.gr.clear_node_attributes()
         self.assertEqual(self.gr.node_attributes("a"), [])
 
+    def test_get_transpose_method(self):
+        transpose = self.gr.get_transpose()
+        self.assertEqual(len(transpose.nodes()), len(self.gr.nodes()))
+        self.assertEqual(len(transpose.edges()), len(self.gr.edges()))
+        self.assertTrue(self.gr.has_edge(("a", "b")))
+        self.assertTrue(transpose.has_edge(("b", "a")))
+        self.assertFalse(transpose.has_edge(("a", "b")))
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/graph_algorithms.py b/graph_algorithms.py
@@ -93,3 +93,48 @@ def find_sink_node(digr):
     while digr.neighbors(node):
         node = digr.neighbors(node)[0]
     return node
+
+def directed_connected_components(digr):
+    """ Returns a list of strongly connected components
+    in a directed graph using Kosaraju's two pass algorithm """
+    if not digr.DIRECTED:
+        raise Exception("%s is not a directed graph" % digr)
+    finishing_times = DFS_loop(digr.get_transpose())
+    # use finishing_times in descending order
+    nodes_explored, connected_components = [], []
+    for node in finishing_times[::-1]:
+        component = []
+        outer_dfs(digr, node, nodes_explored, component)
+        if component:
+            nodes_explored += component
+            connected_components.append(component)
+    return connected_components
+
+def outer_dfs(digr, node, nodes_explored, path):
+    if node in path or node in nodes_explored: 
+        return False
+    path.append(node)
+    for each in digr.neighbors(node):
+        if each not in path or each not in nodes_explored:
+            outer_dfs(digr, each, nodes_explored, path)
+
+def DFS_loop(digr):
+    """ Core DFS loop used to find strongly connected components
+    in a directed graph """
+    node_explored = [] # list for keeping track of nodes explored
+    finishing_times = [] # list for adding nodes based on their finishing times
+    for node in digr.nodes():
+        if node not in node_explored:
+            leader_node = node
+            inner_DFS(digr, node, node_explored, finishing_times)
+    return finishing_times 
+
+def inner_DFS(digr, node, node_explored, finishing_times):
+    """ Inner DFS used in DFS loop method """
+    node_explored.append(node) # mark explored
+    for each in digr.neighbors(node):
+        if each not in node_explored:
+            inner_DFS(digr, each, node_explored, finishing_times)
+    global finishing_counter
+    # adds nodes based on increasing order of finishing times
+    finishing_times.append(node) 
diff --git a/graph_algorithms_test.py b/graph_algorithms_test.py
@@ -64,5 +64,20 @@ def test_topological_ordering(self):
         self.assertEqual(sum([order[u] < order[v] for (u, v) in 
                          dag.edges()]), len(dag.edges())) # all comparisons are True
 
+    def test_directed_connected_components(self):
+        digr = digraph()
+        digr.add_nodes(["a", "b", "c", "d", "e", "f", "g", "h", "i"])
+        digr.add_edges([("b", "a"), ("a", "c"), ("c", "b"), ("d", "b")])
+        digr.add_edges([("d", "f"), ("f", "e"), ("e", "d"), ("g", "e")])
+        digr.add_edges([("g", "h"), ("h", "i"), ("i", "g")])
+        self.assertEqual(len(directed_connected_components(digr)), 3)
+        digr2 = digraph()
+        digr2.add_nodes(["a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"])
+        digr2.add_edges([("a", "b"), ("b", "c"), ("c", "a"), ("b", "d"), ("d", "e")])
+        digr2.add_edges([("e", "f"), ("f", "g"), ("g", "e"), ("d", "g"), ("i", "f")])
+        digr2.add_edges([("h", "g"), ("c", "h"), ("c", "k"), ("h", "i"), ("i", "j")])
+        digr2.add_edges([("h", "j"), ("j", "k"), ("k", "h")])
+        self.assertEqual(len(directed_connected_components(digr2)), 4)
+
 if __name__ == "__main__":
     unittest.main()
