Add file

pub_crawl_script.py

@@ -0,0 +1,172 @@
+import osmnx as ox
+import pandas as pd
+import networkx as nx
+import matplotlib.pyplot as plt
+from ortools.constraint_solver import pywrapcp
+from ortools.constraint_solver import routing_enums_pb2
+
+class pub_crawl:
+    def __init__(self, df, G):
+        '''Initialise a pub_crawl instance
+        
+        df: pd.Dataframe containing pubs 'name' and coordinates ('latitude', 'longitude')
+        G: nx.MultiDiGraph of the map area
+        '''
+        self.df = df
+        self.G = G
+
+        self.pub_names = df['name'].to_list()
+        self.initial_route = self.pub_names
+
+        self.optimal_route = None
+        self.optimal_distance = None
+
+        self.pub_nodes = self.create_pub_nodes()
+    
+    def create_pub_nodes(self):
+        # Dictionary of pub names and coordinates
+        pubs_dict = self.df.drop('address', axis=1).set_index('name').T.to_dict('list')
+
+        # Get graph nodes for each pub
+        pub_nodes = {}
+        for k, v in pubs_dict.items():
+            pub_nodes[k] = ox.nearest_nodes(self.G, X = v[1], Y = v[0])
+        
+        return pub_nodes
+    
+    def get_route_length(self, p0, p1):
+        # Find length in meters of shortest path from pub p0 to pub p1
+        route = nx.shortest_path(self.G, self.pub_nodes[p0], self.pub_nodes[p1], weight='length')
+        route_lengths = ox.utils_graph.get_route_edge_attributes(self.G, route, attribute = 'length')
+        route_length_total = sum(route_lengths)
+        return route_length_total
+
+    def create_distance_matrix(self, pubs_considered):
+        distance_matrix = []
+        for i in range(len(pubs_considered)):
+            row = []
+            for j in range(len(pubs_considered)):
+                distance = self.get_route_length(pubs_considered[i], pubs_considered[j])
+                row.append(round(distance*1000)) # avoids rounding error in Google's OR-Tools package
+            distance_matrix.append(row)
+        
+        return distance_matrix
+    
+    def plot_map(self):
+        node_colours = ['#FF0000' if i in list(self.pub_nodes.values()) else '#999999' for i in self.G.nodes]
+        fig, ax = ox.plot_graph(self.G, bgcolor='#FFFFFF', node_color=node_colours, show=False, close=False)
+        for _, node in ox.graph_to_gdfs(self.G, nodes=True, edges=False).fillna("").iterrows():
+            for k, v in self.pub_nodes.items():
+                if node.name == v:
+                    c = node["geometry"].centroid
+                    ax.annotate(k, xy=(c.x, c.y), xycoords='data', xytext=(3, -2), textcoords='offset points', size=8)
+        plt.show()
+    
+    def get_route_nodes(self, route):
+        # Find intermediary route nodes for plotting
+        route_nodes = []
+        for i in range(len(route)-1):
+            path = nx.shortest_path(self.G, self.pub_nodes[route[i]], self.pub_nodes[route[i+1]], weight='length')
+            route_nodes.append(path)
+        return route_nodes
+    
+    def plot_route(self, route):
+        route_nodes = self.get_route_nodes(route)
+        fig, ax = ox.plot_graph_routes(self.G, route_nodes, bgcolor='#FFFFFF', show=False, close=False, 
+        figsize=(12, 12))
+        for _, node in ox.graph_to_gdfs(self.G, nodes=True, edges=False).fillna("").iterrows():
+            for i, k in enumerate(route):
+                if node.name == self.pub_nodes[k]:
+                    c = node["geometry"].centroid
+                    ax.annotate(f'{i}: {k}', xy=(c.x, c.y), xycoords='data', xytext=(3, -2), textcoords='offset points', size=8)
+        
+        return fig
+    
+    def create_data(self, start, pubs_considered):
+        data = {}
+        start_index = pubs_considered.index(start)
+        data['distance_matrix'] = self.create_distance_matrix(pubs_considered)
+        data['num_vehicles'] = 1
+        data['depot'] = start_index
+        return data
+    
+    def format_solution(self, manager, routing, solution, pubs_considered):
+        """Formats solution for osmnx plotting"""
+        index = routing.Start(0)
+        route = [index]
+
+        leg_distances = []
+        route_distance = 0
+        while not routing.IsEnd(index):
+            previous_index = index
+            index = solution.Value(routing.NextVar(index))
+            if route[0] == manager.IndexToNode(index):
+                break
+            route.append(manager.IndexToNode(index))
+            leg_distance = routing.GetArcCostForVehicle(previous_index, index, 0)
+            leg_distances.append(leg_distance)
+            route_distance += leg_distance
+        
+        self.optimal_route = [pubs_considered[i] for i in route]
+        self.optimal_distance = round(route_distance/1000)
+    
+    def optimise(self, start_point, pubs_considered):
+
+        def distance_callback(from_index, to_index):
+            """Returns the distance between the two nodes."""
+            # Convert from routing variable Index to distance matrix NodeIndex
+            from_node = manager.IndexToNode(from_index)
+            to_node = manager.IndexToNode(to_index)
+            return data['distance_matrix'][from_node][to_node]
+        
+        data = self.create_data(start_point, pubs_considered)
+
+        # Create the routing index manager
+        manager = pywrapcp.RoutingIndexManager(len(data['distance_matrix']),
+                                                data['num_vehicles'], data['depot'])
+
+        # Create Routing Model
+        routing = pywrapcp.RoutingModel(manager)
+
+        transit_callback_index = routing.RegisterTransitCallback(distance_callback)
+
+        # Define cost of each arc
+        routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)
+
+        # Settings for simualted annealing
+        search_parameters = pywrapcp.DefaultRoutingSearchParameters()
+        search_parameters.local_search_metaheuristic = (
+            routing_enums_pb2.LocalSearchMetaheuristic.SIMULATED_ANNEALING)
+        search_parameters.time_limit.seconds = 5
+        search_parameters.log_search = True
+
+        # Solve the problem
+        solution = routing.SolveWithParameters(search_parameters)
+
+        # Format solution
+        if solution:
+            self.format_solution(manager, routing, solution, pubs_considered)
+
+if __name__=='main':
+    df = pd.read_csv('galway_pubs.csv')
+    G =  ox.io.load_graphml('galway.graphml')
+
+    crawler = pub_crawl(df, G)
+
+    print('Locations of pubs in Galway to consider:')
+    crawler.plot_map()
+
+    initial_route = crawler.initial_route
+
+    print('Initial route before optimisation (default pub ordering):')
+    crawler.plot_route(initial_route)
+
+    start_pub = 'The Sliding Rock'
+    crawler.optimise(start_pub, ['The Sliding Rock', 'The Salt House', 'Caribou'])
+
+    optimal_route = crawler.optimal_route
+    print('Optimised route:')
+    print(f'Route distance: {crawler.optimal_distance} meters')
+    crawler.plot_route(optimal_route)
+
+