Python: Dijkstra算法

Question

在这段代码中，我更多地关注代码可读性，而不是算法复杂性。我并不真正关心main函数中的代码，因为它只是用于测试代码。有些类有一个D前缀来区分它们和常规的Nodes。

我特别想就以下几个问题发表意见：

• 在字典键和name字段中存储节点的名称似乎是错误的。我认为它有一些与并行列表相同的问题。
• DGraph类的唯一功能是作为列表的包装器。另一方面，我喜欢它，因为它更冗长。
• connections应该保留为(node, cost)元组，还是应该创建一个新的DConnection类？
• 我希望避免外部库

class DNode:
    def __init__(self, name):
        self.connections = []
        self.name = name
        self.total_cost = -1
        self.last_connection = None
        self.visited = False

    def add_connection(self, node, cost):
        self.connections.append((node, cost))
        for connection in node.connections:
            if connection[0] is self:
                return
        node.add_connection(self, cost)


class DGraph:
    def __init__(self):
        self.nodes = []

    def new_node(self, name):
        node = DNode(name)
        self.nodes.append(node)
        return node


class Dijkstra:
    def __init__(self, graph, start_node, end_node):
        self.graph = graph
        self.start_node = start_node
        self.end_node = end_node
        self.dlist = DList(graph.nodes, start_node)

    def calculate_shortest_path(self):
        self.recursive_helper(self.start_node)
        node = self.end_node
        node_steps = []
        while node != self.start_node:
            node_steps.append(node)
            node = node.last_connection
        node_steps.append(self.start_node)
        return reversed(node_steps)

    def recursive_helper(self, current_node):
        current_cost = current_node.total_cost
        for connection in current_node.connections:
            neighbor = connection[0]
            if not neighbor.visited:
                total_cost = current_cost + connection[1]
                if neighbor.total_cost == -1 or total_cost < neighbor.total_cost:
                    neighbor.total_cost = total_cost
                    neighbor.last_connection = current_node
        current_node.visited = True
        if self.end_node.visited:
            return
        self.recursive_helper(self.dlist.get_smallest_unvisited_dnode())


class DList:
    def __init__(self, node_list, start_node):
        self.nodes = list(node_list)
        start_node.total_cost = 0

    def get_smallest_unvisited_dnode(self):
        smallest = None
        for node in self.nodes:
            if not node.visited and node.total_cost != -1:
                if smallest is None or node.total_cost < smallest.total_cost:
                    smallest = node
        return smallest


class GraphParser:
    def __init__(self, unparsed_string):
        self.unparsed_string = unparsed_string
        self.nodes = {}
        self.graph = DGraph()

    def create_nodes(self):
        for line in self.unparsed_string.split("\n"):
            if line[0:5] == "node ":
                node_name = line[5:]
                node = self.graph.new_node(node_name)
                self.nodes[node_name] = node

    def create_connections(self):
        for line in self.unparsed_string.split("\n"):
            if line[0:5] == "node ":
                node_name = line[5:]
                node = self.nodes[node_name]
            elif line[0] == "\t":
                connection_data = line[1:].split(",")
                neighbor_name = connection_data[0].strip()
                if len(connection_data) > 0:
                    cost = int(connection_data[1].strip())
                else:
                    cost = 1
                neighbor = self.nodes[neighbor_name]
                node.add_connection(neighbor, cost)

    def get_graph(self):
        self.create_nodes()
        self.create_connections()
        return self.graph


def main():
    filename = "graph.txt"
    f = open(filename, "r")
    content = f.read()
    f.close()

    gp = GraphParser(content)
    graph = gp.get_graph()
    start_node = gp.nodes["S"]
    end_node = gp.nodes["E"]

    d = Dijkstra(graph, start_node, end_node)
    path = d.calculate_shortest_path()
    for i in path:
        print(i.name)

main()

样本graph.txt

node A
    D, 4
    B, 3
    S, 7
node B
    S, 2
    A, 3
    D, 4
    H, 1
node C
    S, 3
    L, 2
node D
    A, 4
    B, 4
    F, 5
node E
    K, 5
    G, 2
node F
    H, 3
    D, 5
node G
    H, 2
    E, 2
node H
    B, 1
    F, 3
    G, 2
node I
    L, 4
    J, 6
    K, 4
node J
    L, 4
    I, 6
    K, 4
node K
    I, 4
    K, 4
node L
    I, 4
    J, 4
node S
    A, 7
    B, 2
    C, 3

Answer 1

我认为你在使用不需要的类。您的DGraph类是一个类(您已经实现了一半)，而GraphParser类是另一个类。由于关注点的分离，您甚至必须对文件内容进行两次解析。在我看来，如果这是一个简单的函数，需要一个可迭代的行(例如打开的文件)，并从它创建一个dict of Nodes (将名称映射到节点)，则会容易得多：

class Node:
    def __init__(self, name):
        self.connections = set()
        self.name = name
        self.total_cost = -1
        self.last_connection = None
        self.visited = False

    def __str__(self):
        return f"Name: {self.name}, total cost: {self.total_cost}, connections: {[(node.name, cost) for node, cost in self.connections]}"

    __repr__ = __str__


def parse_graph(lines):
    graph = {}
    for line in lines:
        if line.startswith("node "):
            name = line[5:]
            # Get the node if it already exists, otherwise create a new node
            node = graph.setdefault(name, Node(name))
        elif line.startswith(("\t", " "*4)):  # some editors might replace \t with 4 spaces, careful!
            name2, cost = line.strip().split(",")
            # Get the other node if it already exists, otherwise create a new node
            node2 = graph.setdefault(name2, Node(name2))
            # add connection manually, no need to iterate over all existing connections
            # reverse connection added in definition of node2
            node.connections.add((node2, int(cost)))  # int can deal with whitespace
        else:
            raise ValueError(f"Cannot parse line: {line}")
    return graph

这里的关键部分是在第一次看到节点时(甚至作为连接的目标)添加一个节点，并在该节点已经存在时再次返回它。为此，我使用了dict.setdefault，它在功能上相当于：

if name not in graph:
    graph[name] = Node(name)
node = graph[name]

目前还没有对所有作为连接目标引用的节点进行正确的检查，您可能需要添加这样的内容。

它还使用了一个事实，即Node.connections是一个set，因此添加一个已经存在的conenction并不有害，因为元组(Node, cost)是可使用的(使用‘`Node的id )。

对于示例文件，将返回以下图形结构：

# {'A': Name: A, total cost: -1, connections: [('S', 7), ('D', 4), ('B', 3)],
#  'B': Name: B, total cost: -1, connections: [('H', 1), ('D', 4), ('S', 2), ('A', 3)],
#  'C': Name: C, total cost: -1, connections: [('L', 2), ('S', 3)],
#  'D': Name: D, total cost: -1, connections: [('F', 5), ('A', 4), ('B', 4)],
#  'E': Name: E, total cost: -1, connections: [('K', 5), ('G', 2)],
#  'F': Name: F, total cost: -1, connections: [('D', 5), ('H', 3)],
#  'G': Name: G, total cost: -1, connections: [('E', 2), ('H', 2)],
#  'H': Name: H, total cost: -1, connections: [('F', 3), ('B', 1), ('G', 2)],
#  'I': Name: I, total cost: -1, connections: [('L', 4), ('K', 4), ('J', 6)],
#  'J': Name: J, total cost: -1, connections: [('I', 6), ('L', 4), ('K', 4)],
#  'K': Name: K, total cost: -1, connections: [('K', 4), ('J', 4), ('E', 5), ('I', 4)],
#  'L': Name: L, total cost: -1, connections: [('J', 4), ('C', 2), ('I', 4)],
#  'S': Name: S, total cost: -1, connections: [('C', 3), ('A', 7), ('B', 2)]}

您可能需要对Dijkstra类进行稍微的调整，但这不会使其变得更加困难。

对于打开和关闭文件，应该使用with和if __name__ == "__main__":守卫：

def main():
    with open("graph.txt") as file:
        graph = parse_graph(file)

    start_node = graph["S"]
    end_node = graph["E"]

    d = Dijkstra(graph, start_node, end_node)
    path = d.calculate_shortest_path()
    for i in path:
        print(i.name)

if __name__ == "__main__":
    main()

如果您对文件格式有控制，您可以采取这种假设节点存在，如果它提到到一个极端，以减少所需的行数。然后，您的graph.txt可以是(假设所有连接仍然是双向的和成本对称的)：

A, D, 4
A, B, 3
A, S, 7
B, S, 2
B, D, 4
B, H, 1
C, S, 3
C, L, 2
D, F, 5
E, G, 2
E, K, 5
F, H, 3
G, H, 2
I, L, 4
I, K, 4
I, J, 6
J, L, 4
J, K, 4