Dijkstra's algorithm using collections.namedtuple

Question

I decided to try out some named tuples by implementing Dijkstra's algorithm to find the cheapest routes in a file like this (where each line represents node_a is connected with node_b with a cost of n):

1 6 14
1 2 7
1 3 9
2 3 1
2 4 15
3 6 2
3 4 11
4 5 6
5 6 9

However, something that caught my attention is that some of the lines got really long:

import sys
from collections import namedtuple

INFINITY = 999999

class UndirectedGraph:
    def __init__(self, node_list):
        self.Node = namedtuple('Node', ['coming_from', 'cost'])
        self.node_dict = self.get_nodes(node_list)
        self.create_connections(node_list)
        self.size = len(self.node_dict)

    def get_nodes(self, node_list):
        '''
        Gets a list of tuples (node1, node2, weight_of_connection) and
        distribute theses nodes through a dictionaire 
        '''
        node_dict = {}
        for line in node_list:
            a_node, another_node = line[0], line[1]
            node_dict[a_node] = [] 
            node_dict[another_node] = []    
        return node_dict

    def create_connections(self, raw_node_list):
        '''
        Creates the connection between the nodes in the nodes dict
        '''
        try:
            for n in raw_node_list:
                current_node, neighbor, cost = n
                self.node_dict[current_node].append(self.get_new_node(neighbor, cost))
                self.node_dict[neighbor].append(self.get_new_node(current_node, cost))
        except:
            print("General error: {}".format(sys.exc_info()[0]))
            raise

    def dijkstra(self, source):
        '''
        Applies the dijkstra algorithm for finding the shortest path to every 
        other node coming from source. 
        Returns a list containing the label of the early node and the cost of the 
        total path to the given node
        '''
        if source not in self.node_dict:
            raise ValueError('Node informed does not exist')

        # first slot of costs_array is a sentinel
        costs_array = [self.get_new_node(INFINITY, INFINITY)] * (self.size + 1)
        visited_nodes = set()
        current_label = source
        costs_array[source] = self.get_new_node(source, 0)

        # transverse through all nodes
        for i in range(self.size):
            visited_nodes.add(current_label)
            for neighbor in self.node_dict[current_label]:
                if neighbor.coming_from not in visited_nodes and \
                    self.has_lower_cost(current_label, neighbor, costs_array):
                    costs_array[neighbor.coming_from] = self.get_minimum_cost(current_label, \
                        neighbor, costs_array)
            current_label = self.get_cheapest_neighbor(current_label, costs_array, visited_nodes)
        return costs_array[1:]

    def get_cheapest_neighbor(self, current, costs_array, visited_nodes):
        '''
        Returns the index in costs_array for the cheapest available node
        '''
        lowest_value = lowest_index = INFINITY
        for node in self.node_dict[current]:
            if node.coming_from not in visited_nodes and \
                    node.cost < lowest_value:
                lowest_value = node.cost
                lowest_index = node.coming_from
        return lowest_index

    def has_lower_cost(self, current, neighbor, costs_array):
        '''
        Returns True if the new cost calculation is less than previous value
        '''
        if costs_array[neighbor.coming_from].cost is INFINITY:
            return True
        return (costs_array[current].cost + neighbor.cost) < costs_array[neighbor.coming_from].cost

    def get_minimum_cost(self, current, neighbor, costs_array):
        '''
        Returns a new node with an appropriate new cost
        '''
        return self.get_new_node(current, costs_array[current].cost + neighbor.cost)

    def get_new_node(self, node_label, node_cost):
        '''
        Creates and returns a Node named tuple
        '''
        return self.Node(coming_from=node_label, cost=node_cost)


def main():
    # read nodes from file
    with open('dij.txt', 'r') as f:
        nodes_list = [[int(item) for item in line.split()] for line in f]
    my_graph = UndirectedGraph(node_list=nodes_list)
    start_node = 1
    minimum_paths = my_graph.dijkstra(start_node)
    print('Minimum path to every node starting from {} is:'.format(start_node))
    for i, n in enumerate(minimum_paths, start=1):
        print('{}: coming from {} with cost of {}'.format(i, n.coming_from, n.cost))


if __name__ == '__main__':
    main()

I'm open to all suggestions on how to make this more pythonic, readable, efficient, or less 'cumbersome'.

The output with this code as the given example is:

Minimum path to every node starting from {} is:
1: coming from 1 with cost of 0
2: coming from 1 with cost of 7
3: coming from 2 with cost of 8
4: coming from 3 with cost of 19
5: coming from 6 with cost of 19
6: coming from 3 with cost of 10

Answer 1

One of the first things that I notice is that you define the Node as an instance variable. Better would be to define it at the top level:

Node = namedtuple('Node', ['coming_from', 'cost'])

Also, in Dijkstra's algorithm, the key point is not the nodes, but the edges. An edge is a tuple (source, destination, weight). But let's first see what we can clean up before tackling that.

Your UndirectedGraph takes a list of things. It claims to be a list of nodes, but it's actually a list of (node1, node2, weight_of_connection) lists: edges. So maybe it would make sense to define a namedtuple Edge:

Edge = namedtuple('Edge', ['source', 'target', 'cost'])

And in main:

edges = []
with open('dij.txt', 'r') as f:
    for line in f:
        source, target, cost = map(int, line.split())
        edges.append(Edge(source, target, cost))
my_graph = UndirectedGraph(node_list=edges)
...

Of course, we should rename node_list to edges, right? Because it's a list of edges. (Don't forget to change node_list to edges in the call as well. Maybe even remove the node_list= completely.

def __init__(self, edges):
    self.node_dict = self.get_nodes(edges)
    self.create_connections(edges)
    self.size = len(self.node_dict)

Now, let's take a look at self.get_nodes. It's supposed to construct a node_dict: a dictionary of nodes to lists (initially empty).

def get_nodes(self, node_list):
    '''
    Gets a list of tuples (node1, node2, weight_of_connection) and
    distribute theses nodes through a dictionaire 
    '''
    node_dict = {}
    for line in node_list:
        a_node, another_node = line[0], line[1]
        node_dict[a_node] = [] 
        node_dict[another_node] = []    
    return node_dict

Because of the Edge thing we used, we can actually write it as follows:

def get_nodes(self, edges):
    '''
    Gets a list of edges and
    distribute theses nodes through a dictionary
    '''
    node_dict = {}
    for edge in edges:
        node_dict[edge.source] = [] 
        node_dict[edge.target] = []    
    return node_dict

But even simpler, we could replace it with a defaultdict:

from collections import defaultdict

...
class UndirectedGraph(object):
    ...
    def get_nodes(self, edges):
        '''
        Gets a list of edges and
        distribute theses nodes through a dictionary
        '''
        return defaultdict(list)

At which point we can just write

def __init__(self, edges):
    self.node_dict = defaultdict(list)
    self.create_connections(edges)
    self.size = len(self.node_dict)

and remove the get_nodes method. Now let's take a look at the create_connections method.

def create_connections(self, raw_node_list):
    '''
    Creates the connection between the nodes in the nodes dict
    '''
    try:
        for n in raw_node_list:
            current_node, neighbor, cost = n
            self.node_dict[current_node].append(self.get_new_node(neighbor, cost))
            self.node_dict[neighbor].append(self.get_new_node(current_node, cost))
    except:
        print("General error: {}".format(sys.exc_info()[0]))
        raise

Why do you even have the try/except? It adds nothing of value. Also, the parameter is now a list of edges, instead of a raw_node_list. Another thing I'd suggest doing is using the attributes of our Edge class we just defined.

def create_connections(self, edges):
    '''
    Creates the connection between the nodes in the nodes dict
    '''
    for edge in edges:
        self.node_dict[edge.source].append(self.get_new_node(edge.target, edge.cost))
        self.node_dict[edge.target].append(self.get_new_node(edge.source, edge.cost))

Now I'd like to take a look at the algorithm itself, and I see several things I'd like to suggest, but don't have time for a full refactoring:

• costs_array. Maybe make it a costs_dict, with a node as a key, and an integer as value. For instance, initialize it as costs_dict = {source: 0}
• You're continuously constructing Node objects, while all you need is knowing the {node_name: path_cost} to continue.

I'm also a bit uncertain if your implementation of the algorithm is entirely correct, but I'd need to look that up a bit more.

Answer 2

Definition of Node

Having Node defined as a member of each UndirectedGraph is a bit weird here. It means that the different graphs you'll define will have their own definion of Node. It doesn't seem to be very useful. Unless you really need this, it is probably easier to just define Node before defining UndirectedGraph.

Then, get_new_node becomes :

def get_new_node(self, node_label, node_cost):
    '''
    Creates and returns a Node named tuple
    '''
    return Node(coming_from=node_label, cost=node_cost)

which makes things a bit more complicated than it should be. Who wants to call self.get_new_node when you can call Node ? I'd get rid of the function alltogether.

Infinity is (slightly) bigger than 999999

When implementing the algorithm, you are a bit stuck because you need some value with the meaning "there is no path found so far". Java, C have constant that can be use for such a purpose. In Python, things are a bit more complicated and may depend on the version of Python you use but you'll find suggestions online. Also, you could try to use None for such purposes but you'd need to handle it with special care.

Useless try catch

I do not really understand the point of your try-catch. I am guessing an uncaught exception would print all relevant information (inclusing the traceback) on the standard/error output so there is no point in catching the exceptions just to print things.

Removing duplicated logic

Defining cost = costs_array[neighbor.coming_from].cost in has_lower_cost makes it (from my point of view) easier to understand.

Removing duplicated logic (bis)

get_nodes and create_connection are both iterating over the list and perform similar looking actions but because it is done in 2 steps and because the name used are so different, it is hard to see that we are doing the same thing twice.

Writing :

def __init__(self, node_list):
    self.node_dict = {}
    self.get_nodes(node_list)
    self.create_connections(node_list)
    self.size = len(self.node_dict)

def get_nodes(self, node_list):
    '''
    Gets a list of tuples (node1, node2, weight_of_connection) and
    distribute theses nodes through a dictionaire 
    '''
    for current_node, neighbor, cost in node_list:
        self.node_dict[current_node] = []
        self.node_dict[neighbor] = []

def create_connections(self, node_list):
    '''
    Creates the connection between the nodes in the nodes dict
    '''
    print(node_list)
    for current_node, neighbor, cost in node_list:
        self.node_dict[current_node].append(Node(neighbor, cost))
        self.node_dict[neighbor].append(Node(current_node, cost))

makes things a lot more obvious.

This being said, there is an even better way to do this.

First, you can see that the whole thing could be written in a single function like this (I decided to go back to a function returning a dict to avoid typing self. too many times):

def create_connections(self, node_list):
    '''
    Creates the connection between the nodes in the nodes dict
    '''
    print(node_list)
    node_dict = {}
    for current_node, neighbor, cost in node_list:
        if current_node not in node_dict:
            node_dict[current_node] = []
        node_dict[current_node].append(Node(neighbor, cost))
        if neighbor not in node_dict:
            node_dict[neighbor] = []
        node_dict[neighbor].append(Node(current_node, cost))
    return node_dict

Good thing it that there is a standard function to do "if value in dict then add default value, then do something on the key associated to the value" : setdefault.

The function is now :

def create_connections(self, node_list):
    '''
    Creates the connection between the nodes in the nodes dict
    '''
    print(node_list)
    node_dict = {}
    for current_node, neighbor, cost in node_list:
        node_dict.setdefault(current_node, []).append(Node(neighbor, cost))
        node_dict.setdefault(neighbor, []).append(Node(current_node, cost))
    return node_dict

Reusing builtin function

get_cheapest_neighbor can be rewritten using the min builtin.

def get_cheapest_neighbor(self, current, costs_array, visited_nodes):
    '''
    Returns the index in costs_array for the cheapest available node
    '''
    relevant_nodes = [n for n in self.node_dict[current] if n.coming_from not in visited_nodes]
    if relevant_nodes:
        return min(relevant_nodes, key=operator.itemgetter(1)).coming_from
    return INFINITY

Other quick comments

It might be worth defining a class methode from_file to create the graph from a file.

There is no much point in having self.size.