如何从 python 中的列表创建 DAG

Question

我正在使用 networkx 将 (u, v, weights) 手动输入到图形中。但是当输入变得更大时，这种手动插入节点和边将成为一项非常烦人的任务并且容易出错。我正在尝试，但还没有弄清楚如何在没有体力劳动的情况下完成这项任务。

示例输入：

my_list = [“s1[0]”、“d1[0, 2]”、“s2[0]”、“d2[1, 3]”、“d3[0, 2]”、“d4[ 1, 4]", "d5[2, 3]", "d6[1, 4]"]

手动插入：

在将节点插入图表之前，我需要对它们进行编号，因此“s”或“d”的第一次出现可以与后来的类似字符区分开来，例如s1,s2,s3,... 和 d1,d2,d3,... 我知道它类似于 SSA 表单（编译器），但我找不到对我的案例有帮助的东西。

手动将 (u, v, weights) 插入 DiGraph()

my_graph.add_weighted_edges_from([("s1", "d1", 0), ("d1", "s2", 0), ("d1", "d3", 2), ("s2", "d3", 0), (
    "d2", "d4", 1), ("d2", "d5", 3), ("d3", "d5", 2), ("d4", "d6", 1), ("d4", "d6", 4)])

问题：

如何自动将该输入列表(my_list) 转换为 DAG(my_graph)，避免手动插入？

完整代码： 这是我目前所写的。

import networkx as nx
from networkx.drawing.nx_agraph import write_dot, graphviz_layout
from matplotlib import pyplot as plt

my_graph = nx.DiGraph()
my_graph.add_weighted_edges_from([("s1", "d1", 0), ("d1", "s2", 0), ("d1", "d3", 2), ("s2", "d3", 0), (
    "d2", "d4", 1), ("d2", "d5", 3), ("d3", "d5", 2), ("d4", "d6", 1), ("d4", "d6", 4)])


write_dot(my_graph, "graph.dot")

plt.title("draw graph")
pos = graphviz_layout(my_graph, prog='dot')



nx.draw(my_graph, pos, with_labels=True, arrows=True)

plt.show()
plt.clf()

说明：

1. 's' 和 'd' 是一些指令，它们分别需要 1 或 2 个寄存器来执行操作。

2. 在上面的例子中，我们有 2 个 's' 操作和 6 个 'd' 操作，并且有五个寄存器 [0,1,2,3,4]。

3. 每个操作都会执行一些计算并将结果存储在相关的寄存器中。

4. 从输入中我们可以看到 d1 使用了寄存器 0 和 2，所以在这两个寄存器都空闲之前它不能操作。因此，d1 依赖于 s1，因为 s1 在 d1 之前并且正在使用寄存器 0。一旦 s1 完成，d1 就可以操作，因为寄存器 2 已经空闲。

5. 例如我们用 1 初始化所有寄存器。s1 将其输入加倍，而 d1 将两个输入相加并将结果存储在它的第二个寄存器中：

所以在 s1[0] reg-0 * 2 -> 1 * 2 => reg-0 = 2 之后

在 d1[0, 2] reg-0 + reg-2 -> 2 + 1 => reg-0 = 2 和 reg-2 = 3 之后

更新 1： 该图将是基于某些资源 [0...4] 的依赖图，每个节点将需要 1(for 's') 或 2(for 'd ') 的这些资源。

更新 2：两个问题引起了混淆，所以我将它们分开。现在我已经更改了我的输入列表，并且只有一个任务可以将该列表转换为 DAG。我还包括了一个解释部分。

PS：如果你还没有 pip install graphviz，你可能需要它。

Answer 1

好的，现在我对映射的工作原理有了更好的了解，它只是归结为在代码中描述过程，保留哪个操作正在使用哪个资源的映射，并在它使用使用的资源时迭代操作通过前面的操作，我们生成了一条边。我认为这与您正在寻找的内容相符：

import ast
class UniqueIdGenerator:
    def __init__(self, initial=1):
        self.auto_indexing = {}
        self.initial = initial
    def get_unique_name(self, name):
        "adds number after given string to ensure uniqueness."
        if name not in self.auto_indexing:
            self.auto_indexing[name] = self.initial
        unique_idx = self.auto_indexing[name]
        self.auto_indexing[name] += 1
        return f"{name}{unique_idx}"

def generate_DAG(source):
    """
    takes iterable of tuples in format (name, list_of_resources) where
    - name doesn't have to be unique
    - list_of_resources is a list of resources in any hashable format (list of numbers or strings is typical)
    
    generates edges in the format (name1, name2, resource),
    - name1 and name2 are unique-ified versions of names in input
    - resource is the value in the list of resources
    each "edge" represents a handoff of resource, so name1 and name2 use the same resource sequentially.
    """
    # format {resource: name} for each resource in use.
    resources = {}
    g = UniqueIdGenerator()
    for (op, deps) in source:
        op = g.get_unique_name(op)
        for resource in deps:
            # for each resource this operation requires, if a previous operation used it
            if resource in resources:
                # yield the new edge
                yield (resources[resource], op, resource)
            # either first or yielded an edge, this op is now using the resource.
            resources[resource] = op

my_list = ["s[0]", "d[0, 2]", "s[0]", "d[1, 3]", "d[0, 2]", "d[1, 4]", "d[2, 3]", "d[1, 4]"]
data = generate_DAG((a[0], ast.literal_eval(a[1:])) for a in my_list)
print(*data, sep="\n")