bubbles.py 文件源码

def superbubble_nodes(g: AssemblyGraph, source: Node,
                      sink: Node) -> Set[Node]:
    """Find all nodes inside a superbubble."""

    queue = deque([source])
    visited = {source, sink}

    while queue:
        current = queue.popleft()

        for neighbour in g.neighbors_iter(current):
            if neighbour not in visited:
                queue.append(neighbour)
                visited.add(neighbour)

    return visited