5
\$\begingroup\$

I have Dijkstra's algorithm:

# ==========================================================================
# We will create a dictionary to represent the graph
# =============================================================================
graph = {
    'a' : {'b':3,'c':4, 'd':7},
    'b' : {'c':1,'f':5},
    'c' : {'f':6,'d':2},
    'd' : {'e':3, 'g':6},
    'e' : {'g':3, 'h':4},
    'f' : {'e':1, 'h':8},
    'g' : {'h':2},
    'h' : {'g':2}
}

def dijkstra(graph, start, goal):
    shortest_distance = {}     # dictionary to record the cost to reach to node. We will constantly update this dictionary as we move along the graph.
    track_predecessor = {}     # dictionary to keep track of path that led to that node.
    unseenNodes = graph.copy() # to iterate through all nodes
    infinity = 99999999999     # infinity can be considered a very large number
    track_path = []            # dictionary to record as we trace back our journey

    # Initially we want to assign 0 as the cost to reach to source node and infinity as cost to all other nodes
    for node in unseenNodes:
        shortest_distance[node] = infinity
    shortest_distance[start] = 0

    # The loop will keep running until we have entirely exhausted the graph, until we have seen all the nodes
    # To iterate through the graph, we need to determine the min_distance_node every time.
    while unseenNodes:
        min_distance_node = None
        for node in unseenNodes:
            if min_distance_node is None:
                min_distance_node = node
            elif shortest_distance[node] < shortest_distance[min_distance_node]:
                min_distance_node = node

        # From the minimum node, what are our possible paths
        path_options = graph[min_distance_node].items()

        # We have to calculate the cost each time for each path we take and only update it if it is lower than the existing cost
        for child_node, weight in path_options:
            if weight + shortest_distance[min_distance_node] < shortest_distance[child_node]:
                shortest_distance[child_node] = weight + shortest_distance[min_distance_node]
                track_predecessor[child_node] = min_distance_node

        # We want to pop out the nodes that we have just visited so that we dont iterate over them again.
        unseenNodes.pop(min_distance_node)

    # Once we have reached the destination node, we want trace back our path and calculate the total accumulated cost.
    currentNode = goal
    while currentNode != start:
        try:
            track_path.insert(0, currentNode)
            currentNode = track_predecessor[currentNode]
        except KeyError:
            print('Path not reachable')
            break
    track_path.insert(0, start)

    #  If the cost is infinity, the node had not been reached.
    if shortest_distance[goal] != infinity:
        print('Shortest distance is ' + str(shortest_distance[goal]))
        print('And the path is ' + str(track_path))

It works fine if I have a small amount of nodes (like in the code), but I have graph with around 480 000 nodes and by my approximation, in such a large graph, it will take 7.5 hours, and that is only 1 way! How could I make it work faster? In OSM, for instance it calculates in seconds!

\$\endgroup\$
4
\$\begingroup\$

PEP-8 Violations

Variables (unseenNodes, currentNode) should be snake_case, not camelCase.

Infinity

99999999999 is no where near infinity. If you want to use infinity, use the real infinity: float("+inf"), or in Python 3.5+, math.inf

Conversion to strings

Python's print() statement automatically converts any non-string object into a string, for printing. You don't have to. This:

    print('Shortest distance is ' + str(shortest_distance[goal]))
    print('And the path is ' + str(track_path))

can be written as:

    print('Shortest distance is', shortest_distance[goal])
    print('And the path is', track_path)

which is slightly shorter to type. As a bonus, it is slightly more efficient, as it does not need to create, and subsequently deallocate, a new string object which is the concatenation of the the strings.

If you are using Python 3.6+, you might want to use f-strings:

    print(f'Shortest distance is {shortest_distance[goal]}')
    print(f'And the path is {track_path}')

which interpolates values directly into the strings.

Return Value

Your function finds the path, prints the result, and does not return anything. This is not useful if you wish to use the discovered path in any other fashion. The function should compute the result and return it. The caller should be responsible for printing the result.

Document the code

True, you have plenty of comments. But the caller can only guess at what the functions arguments are supposed to be, and what the function returns. You should document this with type-hints, and """doc-strings""". Something like:

from typing import Any, Mapping, Tuple, List

Node = Any
Edges = Mapping[Node, float]
Graph = Mapping[Node, Edges]

def dijkstra(graph: Graph, start: Node, goal: Node) -> Tuple[float, List]:
    """
    Find the shortest distance between two nodes in a graph, and
    the path that produces that distance.

    The graph is defined as a mapping from Nodes to a Map of nodes which
    can be directly reached from that node, and the corresponding distance.

    Returns:
        A tuple containing
            - the distance between the start and goal nodes
            - the path as a list of nodes from the start to goal.

    If no path can be found, the distance is returned as infinite, and the
    path is an empty list.
    """

Avoid multiple lookups

In this code:

    min_distance_node = None
    for node in unseenNodes:
        if min_distance_node is None:
            min_distance_node = node
        elif shortest_distance[node] < shortest_distance[min_distance_node]:
            min_distance_node = node

you continuously look up shortest_distance[min_distance_node]. In compiled languages, the compiler may be able to perform data-flow analysis, and determine that the value only need to be looked up again if min_distance_node changes. In an interpreted language like Python, where a lookup action can execute user-defined code and change the value, each and every lookup operation must be executed. shortest_distance[min_distance_node] is two variable lookups plus a dictionary indexing operation. Compare with:

    min_distance_node = None
    min_distance = infinity
    for node in unseenNodes:
        distance = shortest_distance[node]
        if min_distance_node is None:
            min_distance_node = node
            min_distance = distance
        elif distance < min_distance:
            min_distance_node = node
            min_distance = distance

This code will run faster, due to less lookups of shortest_distance[min_distance_node] and shortest_distance[node].

But finding the minimum of a list is such a common operation, that Python has a built-in function for doing this: min(iterable, *, key, default). The key argument is used to specify an ordering function ... in this case, a mapping from node to distance. The default can be used to prevent a ValueError if there are no nodes left, which is unnecessary in this case.

    min_distance_node = min(unseenNodes, key=lambda node: shortest_distance[node])

In the same vein:

    for child_node, weight in path_options:
        if weight + shortest_distance[min_distance_node] < shortest_distance[child_node]:
            shortest_distance[child_node] = weight + shortest_distance[min_distance_node]
            track_predecessor[child_node] = min_distance_node

repeatedly looks up shortest_distance[min_distance_node]; again, two variable lookups and a dictionary indexing operation. Again, we can move this out of the loop:

    min_distance = shortest_distance[min_distance_node]
    for child_node, weight in path_options:
        if weight + min_distance < shortest_distance[child_node]:
            shortest_distance[child_node] = weight + min_distance
            track_predecessor[child_node] = min_distance_node

Reducing the Working Set

The code to find the min_distance_node: how many nodes does it check? In your toy graph "a" to "h", on the first iteration, it needs to search 8 nodes. With 480 000 nodes, it would need to search 480 000 nodes! In the second iteration, one node has been removed from unseenNodes, so the it would search one node less. 7 nodes is fine, but 479 999 nodes is a huge number of nodes.

How many nodes does "a" connect to? Only 3. The min_distance_node will become one of those 3 nodes. Searching the remaining nodes (with infinite distances) isn't necessary. If you added to the unseenNodes only the nodes which can be reached at each step of the algorithm, your search space would reduce from several thousand nodes to a couple of hundred.

Moreover, if you maintained these unseenNodes in a sorted order by distance, the min_distance_node would always be the first node in this “priority queue”, and you wouldn’t need to search through the unseenNodes at all.

Maintaining the unseen nodes in a priority queue is easily done through a min-heap structure, which is built into Python (heapq):

from math import inf, isinf
from heapq import heappush, heappop
from typing import Any, Mapping, Tuple, List

Node = Any
Edges = Mapping[Node, float]
Graph = Mapping[Node, Edges]

def dijkstra(graph: Graph, start: Node, goal: Node) -> Tuple[float, List]:
    """
    Find the shortest distance between two nodes in a graph, and
    the path that produces that distance.

    The graph is defined as a mapping from Nodes to a Map of nodes which
    can be directly reached from that node, and the corresponding distance.

    Returns:
        A tuple containing
            - the distance between the start and goal nodes
            - the path as a list of nodes from the start to goal.

    If no path can be found, the distance is returned as infinite, and the
    path is an empty list.
    """

    shortest_distance = {}
    predecessor = {}
    heap = []

    heappush(heap, (0, start, None))

    while heap:

        distance, node, previous = heappop(heap)

        if node in shortest_distance:
            continue

        shortest_distance[node] = distance
        predecessor[node] = previous

        if node == goal:

            path = []
            while node:
                path.append(node)
                node = predecessor[node]

            return distance, path[::-1]

        else:
            for successor, dist in graph[node].items():
                heappush(heap, (distance + dist, successor, node))

    else:

        return inf, []




if __name__ == '__main__':
    graph = {
        'a' : {'b':3, 'c':4, 'd':7},
        'b' : {'c':1, 'f':5},
        'c' : {'f':6, 'd':2},
        'd' : {'e':3, 'g':6},
        'e' : {'g':3, 'h':4},
        'f' : {'e':1, 'h':8},
        'g' : {'h':2},
        'h' : {'g':2}
    }

    distance, path = dijkstra(graph, 'a', 'e')
    if isinf(distance):
        print("No path")
    else:
        print(f"Distance = {distance}, path={path}")

OSM

By 'OSM' do you mean "Open Street Maps"? If so, you are using the wrong algorithm. Map nodes have coordinates, which can be use as "hints", to direct the search in a given direction. See A* Search Algorithm

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Rather than the suggestions here on finding the minimum of a list, OP should be using a priority queue \$\endgroup\$ – BlueRaja - Danny Pflughoeft Jan 21 at 8:56
  • \$\begingroup\$ Thank you a lot for your edits and I will do my routing problem using A*, I appreciate your help \$\endgroup\$ – Alexey Jan 21 at 11:32
  • \$\begingroup\$ @BlueRaja-DannyPflughoeft The suggestions here ends with: “Maintaining the unseen nodes in a priority queue is best done through a min-heap structure, which is built into Python. \$\endgroup\$ – AJNeufeld Jan 21 at 15:18

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.