io.py 文件源码

def plot_tasks_dependency_graph(tasks, ax=None):
    """Plot the graph of all inter-dependencies in the provided tasks list."""
    if not NX_AVAILABLE:
        raise ImportError("Install Networkx to plot task dependency graphs.")
    if not MATPLOTLIB_AVAILABLE:
        raise ImportError("Install Matplotlib to plot task dependency graphs.")
    g = nx.DiGraph()
    tasks_dict = {
        task.id: task
        for task in tasks
    }
    for task_id, task in tasks_dict.items():
        for parent_task in task.follows:
            g.add_edge(parent_task.id, task_id)

    nodes_depths = {
        node: 0
        for node in g.nodes()
    }
    for source, lengths in nx.shortest_path_length(g):
        for target, length in lengths.items():
            nodes_depths[target] = max(nodes_depths[target], length)
    levels = [
        sorted([
            node
            for node, depth in nodes_depths.items()
            if depth == this_depth
        ])[::-1]
        for this_depth in range(max(nodes_depths.values()) + 1)
    ]

    def draw_node(x, y, node, ax):
        task = tasks_dict[node]
        text = task.name.replace("_", "\n") + "\nduration: %d" % task.duration
        ax.text(x, y, text, verticalalignment="center",
                horizontalalignment="center",
                bbox={'facecolor': 'white', 'lw': 0})
    return plot_tree_graph(levels, g.edges(), draw_node, width_factor=2, ax=ax)