python 实现有向图和相关的DFS，循环检测算法

Posted 2021-05-10

class DiGraph(object):
    def __init__(self):
        super(DiGraph, self).__init__()

        self.g = dict()

    def add_node_and_neighbors(self, node, neighbors):
        assert node
        assert isinstance(node, basestring)

        self.g[node] = set(neighbors)

    def add_edge(self, from_table_name, to_table_name):

        assert isinstance(from_table_name, basestring)
        assert isinstance(to_table_name, basestring)

        if from_table_name in self.g:
            self.g[from_table_name].add(to_table_name)
        else:
            self.g[from_table_name] = set(to_table_name)

    def num_nodes(self):
        """
        Number of vertices (nodes)
        :return:
        :rtype:
        """
        return len(self.g)

    def num_edges(self):
        """

        :return: number of edges in this graph
        :rtype: int
        """
        count = 0
        for node in self.g:
            count += len(self.g[node])

        return count

    def degree(self, node):
        """
        :param node:
        :type node: str
        :return: number of adjacent (neighboring) vertices (nodes)
        :rtype: int
        """
        assert isinstance(node, basestring)

        if node not in self.g:
            raise Exception("Node %s not present in graph!" % node)

        return len(self.g[node])

    def has_self_loops(self):
        """
        A "self-loop" is a node that points to itself.
        Note that this isn't the same as detecting a 'cycle' in a graph.

        :return:
        :rtype: bool
        """
        for node in self.g:
            for neighbor in self.g[node]:
                if neighbor == node:
                    print("Node %s has a self-loop!" % node)
                    return node, True

        return None, False

    def neighbors(self, node):
        """

        :param node:
        :type node: basestring
        :return: set of neighboring nodes
        :rtype: frozenset
        """
        assert isinstance(node, basestring)

        if node not in self.g:
            raise Exception("Node %s not present in graph!" % node)

        return frozenset(self.g[node])

    def __str__(self):
        s = StringIO()
        pprint(self.g, s)
        return s.getvalue()

    def dependencies(self, start_node, resolved_deps):
        for neighbor in self.neighbors(start_node):
            if neighbor not in resolved_deps:
                self.dependencies(neighbor, resolved_deps)

        resolved_deps.add(start_node)

    def reverse(self):
        r = DiGraph()

        for n in self.g:
            for adj in self.neighbors(n):
                r.add_edge(adj, n)

        return r


class DirectedCycle(object):
    def __init__(self, di_graph):
        super(DirectedCycle, self).__init__()
        assert isinstance(di_graph, DiGraph)

        self.di_graph = di_graph

        # this is the set of nodes that have been visited.
        # Note that a node being in this set is not an indication of a cycle: a cycle is present only if
        # it is in the on_stack list
        self.marked = set()

        # This keeps track of the nodes that are in the path of the node where the dfs originally began.
        # i.e, when doing the recursive dfs, there is a cycle, if we encounter a node that was traversed as part of this dfs
        self.on_stack = list()

        # This is a list of nodes that form a cycle; this is the list used to construct the 'path' of the cycle.
        self.cycle = list()

        # This is an associative array mapping the child node to its parent node. In this class, this array is used to construct the path of the cycle.
        self.edge_to = dict()

        for i, node in enumerate(self.di_graph.g):
            if node not in self.marked:
                # print("Detecting cycles beginning with %s, iteration # %s " % (node, i))
                self.dfs(node)

    def dfs(self, node):

        # Mark that this node has been visited
        self.marked.add(node)

        # Mark that this node is part of a dfs traversal; not the pop at the end of the for loop, marking that the dfs traversal is done..
        self.on_stack.append(node)

        for neighbor_node in self.di_graph.neighbors(node):
            if len(self.cycle):
                # this is to break infinite looping?
                return
            elif neighbor_node not in self.marked:
                # we have reached an adjacent ("neighbor") node that we haven't visited; so mark the edgeTo[child-node] = parent-node and do a dfs on this node
                self.edge_to[neighbor_node] = node
                self.dfs(neighbor_node)
            elif neighbor_node in self.on_stack:
                # so : we have reached an adjacent node has been part of this dfs traversal call stack....so, there is a link from the current node - the one
                # pointed to by the variable 'node' to this 'neighbor_node'. This edge (node -> neighbor_node) forms the last link of the 'cycle'.
                # So, we traverse the edge_to associative array, till we reach the node pointed to by the current 'neighbor_node' variable (since that's the last link to the cycle)
                # and we add the last two nodes to the cycle list : the node pointed to by the variable 'node' and the node pointed to by the variable 'neighbor_node'.

                child_node = self.edge_to[node]
                while child_node != neighbor_node:
                    self.cycle.append(child_node)
                else:
                    self.cycle.append(node)
                    self.cycle.append(neighbor_node)

        # we are done with the for-loop : we did the entire dfs recursive search across the graph, and now, we take the node that started this dfs sub-search off the stack
        assert self.on_stack.pop() == node


    @property
    def path_str(self):
        return '->'.join(self.cycle)

    def has_cycle(self):
        return len(self.cycle) is not 0


class DepthFirstSearch(object):
    def __init__(self, graph, start_node):
        super(DepthFirstSearch, self).__init__()
        assert isinstance(graph, DiGraph)
        self.graph = graph
        self.count = 0
        self.marked = set()
        self.edge_to = dict()

        self.dfs(start_node)

    def dfs(self, node):
        assert isinstance(node, basestring)
        self.marked.add(node)

        for neighboring_node in self.graph.neighbors(node):
            if neighboring_node not in self.marked:
                # edge_to is an associative array, mapping child to parent.
                self.edge_to[neighboring_node] = node
                self.dfs(neighboring_node)

    def path_length(self):
        len(self.marked)

    def path(self):
        pass

    def has_path_to(self, node):
        """
        Checks if there is a path from start_node to the node passed as the parameter...

        :param node:
        :type node:
        :return:
        :rtype:
        """
        assert node
        assert isinstance(node, basestring)

        # self.marked is the set of nodes visited while doing a depth first search from 'from_node'.
        # if the passed 'node' is in this set
        return node in self.marked