bubbles.py 文件源码

def partition_graph(g: AssemblyGraph) -> Iterator[Tuple[AssemblyGraph, bool]]:
    """This function partitions a directed graph into a set of subgraphs. It
    is partioned in such way that the set of super bubbles of `g` is the same
    as the union of the super bubble sets of all subgraphs returned by this
    function.

    This function yields each partitioned subgraph, together with a flag if
    it is acyclic or not.
    """

    singleton_nodes = []

    for connected_component in networkx.strongly_connected_components(g):
        if len(connected_component) == 1:
            singleton_nodes.extend(connected_component)
            continue

        subgraph = g.subgraph(connected_component)
        for u, v in g.in_edges_iter(connected_component):
            if u not in subgraph:
                subgraph.add_edge('r_', v)

        for u, v in g.out_edges_iter(connected_component):
            if v not in subgraph:
                subgraph.add_edge(u, "re_")

        yield subgraph, False

    # Build subgraph with only singleton strongly connected components
    subgraph = g.subgraph(singleton_nodes)
    for u, v in g.in_edges_iter(singleton_nodes):
        if u not in subgraph:
            subgraph.add_edge('r_', v)

    start_nodes = [n for n in subgraph.nodes_iter()
                   if subgraph.in_degree(n) == 0]
    for n in start_nodes:
        if n == 'r_':
            continue
        subgraph.add_edge('r_', n)

    for u, v in g.out_edges(singleton_nodes):
        if v not in subgraph:
            subgraph.add_edge(u, 're_')

    sink_nodes = [n for n in subgraph.nodes_iter()
                  if subgraph.out_degree(n) == 0]
    for n in sink_nodes:
        if n == 're_':
            continue
        subgraph.add_edge(n, 're_')

    yield subgraph, True